Синхронизация выходных потоков из разных потоков
Я пишу приложение, которое записывает в файл XML некоторые данные из разных потоков. Я пытаюсь синхронизировать его, используя основной объект Event, но в файле я получаю неверные данные. Я получаю следующий результат
<file path="somePath" />
<file path="somePath" <file path="somePath" /> />....
но я ожидаю получить
<file path="somePath" />
<file path="somePath" />
<file path="somePath" />
Смотрите ниже мой псевдокод. Что в этом плохого?
unsigned int WINAPI MyThread(void *p)
{
std::wofstream outstr;
outstr.open("indexingtest.xml", std::ios::app);
do
{
if(somePredicat1)
{
WaitForSingleObject(hEvent, INFINITE);
outstr <<"<file path=\""<< sFileName << "\"\n";
outstr <<"\tsize=\""<< fileSize << "\" />\n";
ReleaseMutex(hMutex);
}
if(somePredicat3)
{
MyThread(sFileName);
}
}while(somePredicat2);
outstr.close();
FindClose( hSearch );
return 0;
}
int _tmain(int argc, TCHAR *argv[])
{
hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
//hMutex = CreateMutex(NULL, FALSE, 0);
unsigned int ThreadID;
HANDLE hThread1 = (HANDLE)_beginthreadex(NULL, 0, MyThread, L"D:\\*", 0, &ThreadID);
HANDLE hThread2 = (HANDLE)_beginthreadex(NULL, 0, MyThread, L"C:\\*", 0, &ThreadID);
SetEvent(hEvent);
std::wcout << "\a" << std::endl;
WaitForSingleObject( hThread1, INFINITE );
return 0;
}
Более конкретный код
HANDLE hMutex = CreateMutex(NULL,FALSE, 0);
wchar_t** GetAllFilesImpl( wchar_t const* folder, wchar_t** res, size_t* pAllocated, size_t* pUsed )
{
HANDLE hSearch;
WIN32_FIND_DATAW fileinfo;
size_t allocatedMemory = 0;hSearch = FindFirstFileW( folder, &fileinfo );
if( hSearch != INVALID_HANDLE_VALUE ) {
do {
wchar_t* sFileName, ** tmp, sTmp[ 1024 ];
long fileSize = 0;
long creationDate;
/* ignore ., .. */
if( !wcscmp(fileinfo.cFileName, L".") ||
!wcscmp(fileinfo.cFileName, L"..") )
continue;
sFileName = PathCreator( folder, fileinfo.cFileName );
fileSize = fileinfo.nFileSizeLow;
creationDate = fileinfo.ftCreationTime.dwHighDateTime;if(fileSize)
{
WaitForSingleObject(hMutex, INFINITE);
std::wofstream outstr;
outstr.open("indexingtest.xml", std::ios::app);
outstr.seekp(std::ios_base::end);
outstr <<"<file path=\""<< sFileName << "\"\n";
outstr <<"\tsize=\""<< fileSize << "\" />\n";
outstr.seekp(std::ios_base::end);
outstr.close();
wprintf( L"%s\n", sFileName);
ReleaseMutex(hMutex);
}
tmp = AddToArray( res, pAllocated, pUsed, sFileName );
if( !tmp ) return FreeAllFilesMemory(res), NULL;
res = tmp;
if( fileinfo.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY ) {
wcscpy_s( sTmp, sFileName );
wcscat_s( sTmp, L"\\*" );
tmp = GetAllFilesImpl( sTmp, res, pAllocated, pUsed );
if( !tmp ) return NULL;
res = tmp;
}
} while( FindNextFileW(hSearch, &fileinfo) );
FindClose( hSearch );
}
return res;
}
unsigned int WINAPI GetAllFiles( void* folder )
{
size_t nAllocated = 0, nUsed = 0;
wchar_t** res = GetAllFilesImpl( (wchar_t *)folder, NULL, &nAllocated, &nUsed );
if( res ) {
/* to indicate end of result add a NULL string */
wchar_t** tmp = AddToArray( res, &nAllocated, &nUsed, NULL );
if( !tmp ) return FreeAllFilesMemory(res), -1;
res = tmp;
}
std::wcout << "\a" << std::endl;
return 0;
}
int _tmain(int argc, TCHAR *argv[])
{
Sleep(1000);
unsigned int ThreadID;
HANDLE hThreads[3];
hThreads[0] = (HANDLE)_beginthreadex(NULL, 0, GetAllFiles, L"D:\\*", 0, &ThreadID);
hThreads[1] = (HANDLE)_beginthreadex(NULL, 0, GetAllFiles, L"C:\\Users\\Andrew\\Desktop\\*", 0, &ThreadID);
hThreads[2] = (HANDLE)_beginthreadex(NULL, 0, GetAllFiles, L"E:\\*", 0, &ThreadID);
unsigned int dw = WaitForMultipleObjects(3, hThreads, TRUE, INFINITE);
CloseHandle(hFile);
printf("finished\n");
return 0;
}
Решение
большой проблема в том, что каждый поток открывает файл отдельно. Вместо этого откройте файл до вы создаете потоки, а затем используете мьютекс для синхронизации записей в файл.
В псевдокоде:
std::wofstream output_file;
void my_thread()
{
do
{
if (some_condition)
{
lock_mutex();
do_output();
unlock_mutex();
}
} while (condition);
}
int main()
{
output_file.open(...);
create_thread();
create_thread();
output_file.close();
}
Другие решения
-
Вы должны дождаться всех потоков, прежде чем завершать код
HANDLE aThread[2]; ... aThread[0] = (HANDLE)_beginthreadex(... aThread[1] = (HANDLE)_beginthreadex(... WaitForMultipleObjects(THREADCOUNT, aThread, TRUE, INFINITE); -
Вы ждете события, прежде чем сделать вывод. Когда закончите с выходом
вы отпускаете мьютекс. Это не имеет смысла. Вы должны ждать мьютекса и
событие. Оба потока получают свое ожидание, выпущенное, когда событие установлено. Таким образом
мьютекс ничего не делает
Когда вы помещаете дескриптор события и дескриптор мьютекса в один массив, вы можете использовать
WaitForMultipleObjects для этого тоже:HANDLE hVarious[2]; hVarious[0] = CreateEvent(NULL, TRUE, FALSE, NULL); // Note: this is a manual reset event. // Thus is stays set until explicitly reset hVarious[1] = CreateMutex(NULL, FALSE, 0); // and now start the threads: aThread[0] = (HANDLE)_beginthreadex(... aThread[1] = (HANDLE)_beginthreadex(... // and set the event: SetEvent(hEvent);WaitForMultipleObjects(2, aThread, TRUE, INFINITE);Поток должен выглядеть так:
unsigned int WINAPI MyThread(void *p) { do { if(somePredicat1) { // wait for the mutex AND the event WaitForMultipleObjects(2, hVarious, TRUE, INFINITE); // do the file stuff in the mutex protected part std::wofstream outstr; outstr.open("indexingtest.xml", std::ios::app); outstr <<"<file path=\""<< sFileName << "\"\n"; outstr <<"\tsize=\""<< fileSize << "\" />\n"; outstr.close(); FindClose( hSearch ); ReleaseMutex(hVarious[1]); } }while(somePredicat2); return 0; }
Помните: Мьютекс создан для защиты ресурсов в параллельных приложениях.
Я понятия не имею о somePredicat1 а также somePredicat1, Эти параметры также могут создавать проблемы при использовании в разных потоках. Однако, неправильный вывод, который вы наблюдали, вызван неправильным использованием мьютекса.
Изменить после комментария:
if(somePredicat3)
{
MyThread(sFileName);
}
а. Поток вызывается сам по себе как функция без закрытия файла.
б. Вы должны предоставить более подробную информацию о том, что somePredicat3, somePredicat2, and для.
somePredicat1
с. Вы должны защитить выходной файл с какой-то исключительностью, потому что он используется
более чем на один поток. Вы также можете использовать Объект критического сечения сделать это.