来自不同线程的同步输出流

Question

我正在写一个应用程序，它向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; 
} 

Answer 1

你拥有的大问题是，每个线程单独打开该文件。请在之前打开文件，然后使用互斥锁将写入同步到文件。

在伪代码：

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(); 
} 

Answer 2

1. 你应该等待所有线程结束代码

   HANDLE aThread[2]; 
   
   ... 
   
   aThread[0] = (HANDLE)_beginthreadex(... 
   aThread[1] = (HANDLE)_beginthreadex(... 
   
   WaitForMultipleObjects(THREADCOUNT, aThread, TRUE, INFINITE); 
   

2. 之前，你在等待的事件你做你的输出之前。当完成输出 时，您释放该互斥锁。这根本不符合逻辑。你应该等待这个互斥体和这个事件。事件设置时，两个线程都获得了等待释放。因此， 互斥体不会执行任何操作。 当你把事件处理和互斥处理成一个阵列，可以使用 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); 
    } 

一个。该线程本身作为一个函数被调用，而不关闭文件。

b。您应该提供有关somePredicat3, somePredicat2, and somePredicat1的更多详细信息。

c。你必须用某种排他性保护输出文件，因为它被多个线程使用了 。您也可以使用Critical Section Object这样做。