13、重叠IO模型

1、课程目标

1. 理解重叠I/O的异步概念
2. 掌握OVERLAPPED结构和完成例程
3. 实现基于重叠I/O的高性能服务器
4. 了解事件通知和完成例程两种模式

2、名词解释

名词	全称	解释
Overlapped I/O	重叠I/O	异步I/O操作，调用立即返回
OVERLAPPED	-	重叠操作结构体
WSAOVERLAPPED	-	Winsock重叠操作结构
WSASend	-	异步发送函数
WSARecv	-	异步接收函数
WSABUF	-	数据缓冲区描述符
Completion Routine	完成例程	I/O完成时的回调函数
Alertable Wait	可警醒等待	允许APC执行的等待状态
APC	Asynchronous Procedure Call	异步过程调用

3、使用工具

• Visual Studio 2022
• xperf/WPA（性能分析）
• Process Monitor（I/O监控）
• WinDbg（调试异步操作）

4、技术原理

4.1、重叠I/O工作原理

应用程序                          内核
    |                              |
    | WSARecv(socket, &wsabuf, ..., &overlapped, callback)
    |----------------------------->|
    |                              | 记录I/O请求
    |<-- 立即返回 WSA_IO_PENDING --|
    |                              |
    | 继续执行其他任务              | 异步执行I/O
    |                              |
    | SleepEx/WaitForSingleObjectEx (可警醒等待)
    |----------------------------->|
    |                              |
    |      【进入可警醒状态】        | I/O完成
    |                              | 触发完成通知
    |                              |
    |<-- 事件信号 或 APC回调 -------|
    |                              |
    | 处理完成的I/O                 |

4.2、OVERLAPPED结构

typedef struct _OVERLAPPED {
    ULONG_PTR Internal;         // 系统使用
    ULONG_PTR InternalHigh;     // 系统使用
    union {
        struct {
            DWORD Offset;       // 文件偏移低32位
            DWORD OffsetHigh;   // 文件偏移高32位
        };
        PVOID Pointer;
    };
    HANDLE hEvent;              // 完成事件（事件通知模式）
} OVERLAPPED;

// WSABUF结构
typedef struct _WSABUF {
    ULONG   len;    // 缓冲区长度
    CHAR*   buf;    // 缓冲区指针
} WSABUF;

4.3、两种完成通知模式

// 模式1：事件通知
// 创建事件对象
WSAEVENT event = WSACreateEvent();

// 设置OVERLAPPED事件
WSAOVERLAPPED overlapped = {0};
overlapped.hEvent = event;

// 发起异步操作
WSARecv(socket, &wsabuf, 1, NULL, &flags, &overlapped, NULL);

// 等待事件
WSAWaitForMultipleEvents(1, &event, TRUE, INFINITE, FALSE);

// 获取结果
WSAGetOverlappedResult(socket, &overlapped, &bytesTransferred, 
                       FALSE, &flags);


// 模式2：完成例程（回调函数）
void CALLBACK CompletionRoutine(
    DWORD dwError,
    DWORD cbTransferred,
    LPWSAOVERLAPPED lpOverlapped,
    DWORD dwFlags
);

// 发起异步操作（指定回调）
WSARecv(socket, &wsabuf, 1, NULL, &flags, &overlapped, 
        CompletionRoutine);

// 进入可警醒等待状态
SleepEx(INFINITE, TRUE);  // 或 WaitForSingleObjectEx

5、代码实现

5.1、基于事件通知的重叠I/O服务器

#include <winsock2.h>
#include <windows.h>
#include <stdio.h>

#pragma comment(lib, "ws2_32.lib")

#define MAX_BUFFER 4096
#define MAX_CLIENTS 64

// 每个连接的I/O上下文
typedef struct _IO_CONTEXT {
    WSAOVERLAPPED   overlapped;
    SOCKET          socket;
    WSABUF          wsabuf;
    char            buffer[MAX_BUFFER];
    DWORD           bytesTransferred;
    DWORD           flags;
    int             opType;     // 0=recv, 1=send
    char            ip[32];
    int             port;
} IO_CONTEXT;

// 服务器上下文
typedef struct _OVERLAP_SERVER {
    SOCKET          listenSocket;
    IO_CONTEXT      clients[MAX_CLIENTS];
    WSAEVENT        events[MAX_CLIENTS + 1];
    int             clientCount;
    BOOL            running;
} OVERLAP_SERVER;

// 初始化I/O上下文
void InitIOContext(IO_CONTEXT* ctx, SOCKET socket, 
                   const char* ip, int port) {
    ZeroMemory(ctx, sizeof(IO_CONTEXT));
    ctx->socket = socket;
    ctx->wsabuf.buf = ctx->buffer;
    ctx->wsabuf.len = MAX_BUFFER;
    strncpy(ctx->ip, ip, sizeof(ctx->ip));
    ctx->port = port;
    
    // 创建事件
    ctx->overlapped.hEvent = WSACreateEvent();
}

// 发起异步接收
BOOL PostRecv(IO_CONTEXT* ctx) {
    ctx->opType = 0;
    ctx->flags = 0;
    ctx->wsabuf.len = MAX_BUFFER;
    ZeroMemory(&ctx->overlapped, sizeof(WSAOVERLAPPED));
    ctx->overlapped.hEvent = WSACreateEvent();
    
    int result = WSARecv(
        ctx->socket,
        &ctx->wsabuf,
        1,
        &ctx->bytesTransferred,
        &ctx->flags,
        &ctx->overlapped,
        NULL        // 不使用完成例程
    );
    
    if (result == SOCKET_ERROR) {
        int err = WSAGetLastError();
        if (err != WSA_IO_PENDING) {
            return FALSE;
        }
    }
    
    return TRUE;
}

// 发起异步发送
BOOL PostSend(IO_CONTEXT* ctx, const char* data, int len) {
    ctx->opType = 1;
    memcpy(ctx->buffer, data, len);
    ctx->wsabuf.len = len;
    ZeroMemory(&ctx->overlapped, sizeof(WSAOVERLAPPED));
    ctx->overlapped.hEvent = WSACreateEvent();
    
    int result = WSASend(
        ctx->socket,
        &ctx->wsabuf,
        1,
        &ctx->bytesTransferred,
        0,
        &ctx->overlapped,
        NULL
    );
    
    if (result == SOCKET_ERROR) {
        int err = WSAGetLastError();
        if (err != WSA_IO_PENDING) {
            return FALSE;
        }
    }
    
    return TRUE;
}

// 服务器主循环（事件通知模式）
void ServerRunEvent(OVERLAP_SERVER* server) {
    while (server->running) {
        // 构建事件数组
        WSAEVENT events[MAX_CLIENTS + 1];
        events[0] = WSACreateEvent();  // 监听用
        
        int count = 1;
        for (int i = 0; i < server->clientCount; i++) {
            events[count++] = server->clients[i].overlapped.hEvent;
        }
        
        // 等待事件
        DWORD result = WSAWaitForMultipleEvents(
            count, events, FALSE, 1000, FALSE
        );
        
        if (result == WSA_WAIT_TIMEOUT) {
            continue;
        }
        
        if (result == WSA_WAIT_FAILED) {
            break;
        }
        
        int index = result - WSA_WAIT_EVENT_0;
        
        if (index == 0) {
            // 处理新连接
            struct sockaddr_in clientAddr;
            int addrLen = sizeof(clientAddr);
            
            SOCKET clientSocket = accept(server->listenSocket,
                                        (struct sockaddr*)&clientAddr,
                                        &addrLen);
            
            if (clientSocket != INVALID_SOCKET && 
                server->clientCount < MAX_CLIENTS) {
                char* ip = inet_ntoa(clientAddr.sin_addr);
                int port = ntohs(clientAddr.sin_port);
                
                IO_CONTEXT* ctx = &server->clients[server->clientCount++];
                InitIOContext(ctx, clientSocket, ip, port);
                PostRecv(ctx);
                
                printf("[+] 新连接: %s:%d\n", ip, port);
            }
        } else {
            // 处理客户端I/O完成
            int clientIndex = index - 1;
            IO_CONTEXT* ctx = &server->clients[clientIndex];
            
            DWORD bytesTransferred, flags;
            BOOL success = WSAGetOverlappedResult(
                ctx->socket,
                &ctx->overlapped,
                &bytesTransferred,
                FALSE,
                &flags
            );
            
            WSAResetEvent(ctx->overlapped.hEvent);
            
            if (!success || bytesTransferred == 0) {
                // 连接断开
                printf("[-] 断开: %s:%d\n", ctx->ip, ctx->port);
                closesocket(ctx->socket);
                WSACloseEvent(ctx->overlapped.hEvent);
                
                // 移除客户端
                for (int i = clientIndex; i < server->clientCount - 1; i++) {
                    server->clients[i] = server->clients[i + 1];
                }
                server->clientCount--;
                continue;
            }
            
            if (ctx->opType == 0) {
                // 接收完成
                ctx->buffer[bytesTransferred] = '\0';
                printf("[%s:%d] %s\n", ctx->ip, ctx->port, ctx->buffer);
                
                // 回显
                char response[MAX_BUFFER];
                sprintf(response, "Echo: %s", ctx->buffer);
                PostSend(ctx, response, strlen(response));
            } else {
                // 发送完成，继续接收
                PostRecv(ctx);
            }
        }
    }
}

5.2、基于完成例程的重叠I/O服务器

// 扩展I/O上下文（包含自引用）
typedef struct _IO_CONTEXT_EX {
    WSAOVERLAPPED   overlapped;
    SOCKET          socket;
    WSABUF          wsabuf;
    char            buffer[MAX_BUFFER];
    int             opType;
    char            ip[32];
    int             port;
    struct _OVERLAP_SERVER_EX* server;  // 反向引用
} IO_CONTEXT_EX;

typedef struct _OVERLAP_SERVER_EX {
    SOCKET          listenSocket;
    IO_CONTEXT_EX*  clients[MAX_CLIENTS];
    int             clientCount;
    BOOL            running;
    CRITICAL_SECTION cs;
} OVERLAP_SERVER_EX;

// 完成例程
void CALLBACK RecvCompletion(
    DWORD dwError,
    DWORD cbTransferred,
    LPWSAOVERLAPPED lpOverlapped,
    DWORD dwFlags
) {
    // 获取上下文（OVERLAPPED是结构体第一个成员）
    IO_CONTEXT_EX* ctx = (IO_CONTEXT_EX*)lpOverlapped;
    
    if (dwError != 0 || cbTransferred == 0) {
        printf("[-] 断开: %s:%d\n", ctx->ip, ctx->port);
        closesocket(ctx->socket);
        
        // 从服务器移除（需要加锁）
        OVERLAP_SERVER_EX* server = ctx->server;
        EnterCriticalSection(&server->cs);
        for (int i = 0; i < server->clientCount; i++) {
            if (server->clients[i] == ctx) {
                server->clients[i] = server->clients[--server->clientCount];
                break;
            }
        }
        LeaveCriticalSection(&server->cs);
        
        HeapFree(GetProcessHeap(), 0, ctx);
        return;
    }
    
    ctx->buffer[cbTransferred] = '\0';
    printf("[%s:%d] %s\n", ctx->ip, ctx->port, ctx->buffer);
    
    // 回显
    char response[MAX_BUFFER];
    int len = sprintf(response, "Echo: %s", ctx->buffer);
    
    ctx->opType = 1;
    memcpy(ctx->buffer, response, len);
    ctx->wsabuf.len = len;
    
    ZeroMemory(&ctx->overlapped, sizeof(WSAOVERLAPPED));
    
    WSASend(ctx->socket, &ctx->wsabuf, 1, NULL, 0,
            &ctx->overlapped, SendCompletion);
}

void CALLBACK SendCompletion(
    DWORD dwError,
    DWORD cbTransferred,
    LPWSAOVERLAPPED lpOverlapped,
    DWORD dwFlags
) {
    IO_CONTEXT_EX* ctx = (IO_CONTEXT_EX*)lpOverlapped;
    
    if (dwError != 0) {
        // 发送失败，关闭连接
        closesocket(ctx->socket);
        HeapFree(GetProcessHeap(), 0, ctx);
        return;
    }
    
    // 发送完成，继续接收
    ctx->opType = 0;
    ctx->wsabuf.len = MAX_BUFFER;
    DWORD flags = 0;
    
    ZeroMemory(&ctx->overlapped, sizeof(WSAOVERLAPPED));
    
    WSARecv(ctx->socket, &ctx->wsabuf, 1, NULL, &flags,
            &ctx->overlapped, RecvCompletion);
}

// 接受连接线程
DWORD WINAPI AcceptThread(LPVOID param) {
    OVERLAP_SERVER_EX* server = (OVERLAP_SERVER_EX*)param;
    
    while (server->running) {
        struct sockaddr_in clientAddr;
        int addrLen = sizeof(clientAddr);
        
        SOCKET clientSocket = accept(server->listenSocket,
                                    (struct sockaddr*)&clientAddr,
                                    &addrLen);
        
        if (clientSocket == INVALID_SOCKET) {
            continue;
        }
        
        // 创建I/O上下文
        IO_CONTEXT_EX* ctx = (IO_CONTEXT_EX*)HeapAlloc(
            GetProcessHeap(), HEAP_ZERO_MEMORY, sizeof(IO_CONTEXT_EX)
        );
        
        ctx->socket = clientSocket;
        ctx->wsabuf.buf = ctx->buffer;
        ctx->wsabuf.len = MAX_BUFFER;
        strcpy(ctx->ip, inet_ntoa(clientAddr.sin_addr));
        ctx->port = ntohs(clientAddr.sin_port);
        ctx->server = server;
        
        // 添加到服务器
        EnterCriticalSection(&server->cs);
        if (server->clientCount < MAX_CLIENTS) {
            server->clients[server->clientCount++] = ctx;
        }
        LeaveCriticalSection(&server->cs);
        
        printf("[+] 新连接: %s:%d\n", ctx->ip, ctx->port);
        
        // 发起第一个异步接收
        DWORD flags = 0;
        WSARecv(clientSocket, &ctx->wsabuf, 1, NULL, &flags,
                &ctx->overlapped, RecvCompletion);
    }
    
    return 0;
}

// 主线程（可警醒等待）
void ServerRunCompletion(OVERLAP_SERVER_EX* server) {
    // 启动接受线程
    HANDLE acceptThread = CreateThread(
        NULL, 0, AcceptThread, server, 0, NULL
    );
    
    // 主线程保持可警醒状态
    while (server->running) {
        // SleepEx使线程进入可警醒状态
        // 当有I/O完成时，会调用完成例程
        SleepEx(100, TRUE);  // TRUE表示可警醒
    }
    
    WaitForSingleObject(acceptThread, INFINITE);
    CloseHandle(acceptThread);
}

// 初始化服务器
BOOL InitServerEx(OVERLAP_SERVER_EX* server, int port) {
    WSADATA wsa;
    WSAStartup(MAKEWORD(2, 2), &wsa);
    
    InitializeCriticalSection(&server->cs);
    
    server->listenSocket = socket(AF_INET, SOCK_STREAM, 0);
    
    struct sockaddr_in addr = {0};
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = INADDR_ANY;
    addr.sin_port = htons(port);
    
    bind(server->listenSocket, (struct sockaddr*)&addr, sizeof(addr));
    listen(server->listenSocket, SOMAXCONN);
    
    server->clientCount = 0;
    server->running = TRUE;
    
    printf("[+] 完成例程服务器启动，端口: %d\n", port);
    
    return TRUE;
}

5.3、C2 Agent重叠I/O通信

// Agent异步通信上下文
typedef struct _AGENT_IO_CONTEXT {
    WSAOVERLAPPED   overlapped;
    SOCKET          socket;
    WSABUF          wsabuf;
    char            buffer[8192];
    int             opType;
    BOOL            connected;
} AGENT_IO_CONTEXT;

AGENT_IO_CONTEXT g_AgentIO = {0};

// Agent接收完成例程
void CALLBACK AgentRecvCompletion(
    DWORD dwError,
    DWORD cbTransferred,
    LPWSAOVERLAPPED lpOverlapped,
    DWORD dwFlags
) {
    AGENT_IO_CONTEXT* ctx = (AGENT_IO_CONTEXT*)lpOverlapped;
    
    if (dwError != 0 || cbTransferred == 0) {
        printf("[-] C2连接断开\n");
        ctx->connected = FALSE;
        return;
    }
    
    // 处理接收到的命令
    ProcessC2Command(ctx->buffer, cbTransferred);
    
    // 继续接收
    PostAgentRecv(ctx);
}

// Agent发送完成例程
void CALLBACK AgentSendCompletion(
    DWORD dwError,
    DWORD cbTransferred,
    LPWSAOVERLAPPED lpOverlapped,
    DWORD dwFlags
) {
    // 发送完成，可以继续发送队列中的数据
    printf("[*] 发送完成: %d bytes\n", cbTransferred);
}

// 发起接收
void PostAgentRecv(AGENT_IO_CONTEXT* ctx) {
    ctx->opType = 0;
    ctx->wsabuf.buf = ctx->buffer;
    ctx->wsabuf.len = sizeof(ctx->buffer);
    DWORD flags = 0;
    
    ZeroMemory(&ctx->overlapped, sizeof(WSAOVERLAPPED));
    
    WSARecv(ctx->socket, &ctx->wsabuf, 1, NULL, &flags,
            &ctx->overlapped, AgentRecvCompletion);
}

// 异步发送数据
void AgentAsyncSend(const void* data, int len) {
    AGENT_IO_CONTEXT* ctx = &g_AgentIO;
    
    memcpy(ctx->buffer, data, len);
    ctx->wsabuf.buf = ctx->buffer;
    ctx->wsabuf.len = len;
    ctx->opType = 1;
    
    ZeroMemory(&ctx->overlapped, sizeof(WSAOVERLAPPED));
    
    WSASend(ctx->socket, &ctx->wsabuf, 1, NULL, 0,
            &ctx->overlapped, AgentSendCompletion);
}

// Agent主循环
void AgentMainLoop(const char* c2Host, int c2Port) {
    WSADATA wsa;
    WSAStartup(MAKEWORD(2, 2), &wsa);
    
    while (TRUE) {
        // 连接C2
        g_AgentIO.socket = socket(AF_INET, SOCK_STREAM, 0);
        
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_addr.s_addr = inet_addr(c2Host);
        addr.sin_port = htons(c2Port);
        
        if (connect(g_AgentIO.socket, 
                   (struct sockaddr*)&addr, sizeof(addr)) == 0) {
            g_AgentIO.connected = TRUE;
            printf("[+] C2连接成功\n");
            
            // 发起第一个接收
            PostAgentRecv(&g_AgentIO);
            
            // 保持可警醒状态
            while (g_AgentIO.connected) {
                SleepEx(1000, TRUE);
            }
        }
        
        closesocket(g_AgentIO.socket);
        
        // 重连延迟
        Sleep(30000);
    }
}

6、课后作业

1. 实现文件异步传输

   • 使用重叠I/O读写文件
   • 同时处理网络和文件I/O
   • 实现大文件分片传输

2. 实现多缓冲区接收

   • 使用WSARecv的多缓冲区功能
   • 实现零拷贝接收
   • 优化内存使用

3. 性能对比测试

   • 对比同步/异步I/O性能
   • 测试不同负载下的吞吐量
   • 分析CPU和内存使用