1

我想讨论 Windows 中潜在的异步（重叠）I/O 实现，因为有很多方法可以实现它。Windows 中的重叠 I/O 提供了异步处理数据的能力，即操作的执行是非阻塞的。

编辑：这个问题的目的一方面是讨论改进我自己的实现，另一方面是讨论替代实现。什么异步 I/O 实现在并行重 I/O 上最有意义，什么在小型单线程应用程序中最有意义。

我将引用MSDN：

当一个函数被同步执行时，它在操作完成之前不会返回。这意味着调用线程的执行可以无限期地阻塞，同时等待耗时的操作完成。调用重叠操作的函数可以立即返回，即使操作尚未完成。这使得一个耗时的 I/O 操作可以在后台执行，而调用线程可以自由地执行其他任务。例如，单个线程可以在不同的句柄上同时执行 I/O 操作，甚至可以在同一个句柄上同时执行读写操作。

我假设读者熟悉重叠 I/O 的基本概念。

异步 I/O 的另一种解决方案是完成端口，但这不是本次讨论的主题。有关其他 I/O 概念的更多信息，请参见 MSDN“关于文件管理 > 输入和输出 (I/O) > I/O 概念”

我想在这里展示我的（C/C++）实现并分享它以供讨论。

这是我的扩展 OVERLAPPED 结构，称为IoOperation：

struct IoOperation : OVERLAPPED {
    HANDLE Handle;
    unsigned int Operation;
    char* Buffer;
    unsigned int BufferSize;
}

每次调用ReadFile或WriteFile调用异步操作时都会创建此结构。该Handle字段应使用相应的设备/文件句柄进行初始化。Operation是一个用户定义的字段，它告诉我们调用了什么操作。该字段Buffer是指向先前分配的具有给定大小的内存块的指针BufferSize。当然，这个结构体可以随意扩展。它可以包含操作结果，实际传输的大小等。

我们需要的第一件事是每次完成重叠 I/O 时都会发出一个（自动重置）事件句柄。

HANDLE hEvent = CreateEvent(0, FALSE, FALSE, 0);

首先，我决定对所有异步操作只使用一个事件。然后我决定用RegisterWaitForSingleObject的线程池线程注册这个事件。

HANDLE hWait = 0;
....
RegisterWaitForSingleObject(
    &hWait,
    hEvent,
    WaitOrTimerCallback,
    this,
    INFINITE,
    WT_EXECUTEINPERSISTENTTHREAD | WT_EXECUTELONGFUNCTION
);

因此，每次发出此事件信号时，WaitOrTimerCallback都会调用我的回调。

异步操作的初始化如下：

IoOperation* Io = new IoOperation(hFile, hEvent, IoOperation::Write, Data, DataSize);
if (IoQueue->Enqueue(Io)) {
    WriteFile(hFile, Io->Buffer, Io->BufferSize, 0, Io);
}

GetOverlappedResult每个操作都排队并在我的WaitOrTimerCallback回调中成功调用后被删除。代替new一直在这里调用，我们可以使用内存池来避免内存碎片并加快分配速度。

VOID CALLBACK WaitOrTimerCallback(PVOID Parameter, BOOLEAN TimerOrWaitFired) {
    list<IoOperation*>::iterator it = IoQueue.begin();
    while (it != IoQueue.end()) {
        bool IsComplete = true;
        DWORD Transfered = 0;

        IoOperation* Io = *it;
        if (GetOverlappedResult(Io->Handle, Io, &Transfered, FALSE)) {
            if (Io->Operation == IoOperation::Read) {
                // Handle Read, virtual OnRead(), SetEvent, etc.
            } else if (Io->Operation == IoOperation::Write) {
                // Handle Read, virtual OnWrite(), SetEvent, etc.
            } else {
                // ...
            }
        } else {
            if (GetLastError() == ERROR_IO_INCOMPLETE) {
                IsComplete = false;
            } else {
                // Handle Error
            }
        }
        if (IsComplete) {
            delete Io;
            it = IoQueue.erase(it);
        } else {
            it++;
        }
    }
}

当然，为了多线程安全，我们在访问 I/O 队列时需要一个锁保护（临界区）。

这种实现方式有优点也有缺点。

优点：

  • 在持久化线程池线程中执行，无需手动创建线程
  • 只需要一个事件
  • 每个操作都在一个 I/O 队列中排队（可以稍后调用 CancelIoEx）

缺点：

  • I/O 队列需要额外的内存/cpu 时间
  • 对所有排队的 I/O 甚至未完成的 I/O 调用 GetOverlappedResult
4

0 回答 0