c - 如何在c/c++中异步运行线程的特定功能？

Question

性能调优:向多个管道写入数据

现在我在单线程中完成:

for(unsigned int i = 0; i < myvector.size();)
{
    tmp_pipe = myvector[i];
    fSuccess = WriteFile( tmp_pipe, &Time, sizeof(double), &dwWritten, NULL );
    if(!fSuccess)
    {
        myvector.erase(myvector.begin()+i);
        printf("Client pipe closed\r\n");
        continue;
    }
    fSuccess = WriteFile( tmp_pipe, &BufferLen, sizeof(long), &dwWritten, NULL );
    if(!fSuccess)
    {
        myvector.erase(myvector.begin()+i);
        printf("Client pipe closed\r\n");
        continue;
    }
    fSuccess = WriteFile( tmp_pipe, pBuffer, BufferLen, &dwWritten, NULL );
    if(!fSuccess)
    {
        myvector.erase(myvector.begin()+i);
        printf("Client pipe closed\r\n");
        continue;
    }
    i++;
}

结果是第一个pipe获取数据最快，最后一个pipe最慢。

我正在考虑在单独的线程中执行此操作，以便对每个 pipe 进行同等处理。

但是如何在c/c++中异步运行线程的特定函数(主线程应该立即返回)？

Answer 1

您可以使用 CreateThread函数创建一个新线程并将管道句柄作为参数传递给线程函数:

DWORD PipeThread(LPVOID param) {
  HANDLE hPipe = (HANDLE)param;
  // Do the WriteFile operations here
  return 0;
}

for(unsigned int i = 0; i < myvector.size(); i++)
  CreateThread(NULL, 0, PipeThread, myvector[i], 0, NULL);

请注意， vector 类不是线程安全的，因此如果您不同步对它们的访问，您将面临 myvector.erase 的问题，例如。使用临界区。

---

更新:既然你提到了高频，你可以使用I/O completion ports而不是每个管道的单独线程。然后，您可以将重叠 I/O 与 WriteFile 一起使用以异步执行写入，并且您可以只有一个额外的线程来监听写入的完成:

// Initial setup: add pipe handles to a completion port
HANDLE hPort = CreateCompletionPort(myvector[0], NULL, 0, 1);
for (unsigned int i = 1; i < myvector.size(); i++)
  CreateCompletionPort(myvector[i], hPort, 0, 0);

// Start thread
CreateThread(NULL, 0, PipeThread, hPort, 0, NULL);

// Do this as many times as you want
for(unsigned int i = 0; i < myvector.size(); i++) {
  OVERLAPPED *ov = new OVERLAPPED;
  ZeroMemory(ov, sizeof ov);
  WriteFile(myvector[i], buffer, size, NULL, ov);
  // If pipe handle was closed, WriteFile will fail immediately
  // Otherwise I/O is performed asynchronously
}

// Close the completion port at the end
// This should automatically free the thread
CloseHandle(hPort);

---

DWRD PipeThread(LPVOID param) {
  HANDLE hPort = (HANDLE)param;
  DWORD nBytes;
  ULONG_PTR key;
  LPOVERLAPPED ov;

  // Continuously loop for I/O completion
  while (GetQueuedCompletionStatus(hPort, &nBytes, &key, &ov, INFINITE)) {
    if (ov != NULL) {
      delete ov;
      // Do anything else you may want to do here
    }
  }

  return 0;
}