“完成端口”模型(IOCP)

转 http://hi.baidu.com/firebird/blog/item/f592b3193a02814542a9adeb.html
转：http://neora.javaeye.com/blog/402336

//////////////////////////////////////////////////////////////////////////
/// “完成端口”模型(IOCP)(基于重叠端口(异步)的LF线程池框架)
/// I/O完成端口是应用程序使用线程池来进行处理异步的I/O请求的一种机制。
/// 使用I/O完成端口比在I/O请求时创建线程更快更有效
/// 应用程序发出一些异步的I/O请求，当这些请求完成时，设备驱动将这些工作项目排序到完成端口
/// 这样在完成端口上等待的线程池，便可以处理这些完成的I/O
/// 首先，说说主线程：
/// 1.创建完成端口对象
/// 2.创建工作者线程（这里工作者线程的数量是按照CPU的个数来决定的，这样可以达到最佳性能）
/// 3.创建监听套接字，绑定，监听，然后程序进入循环
/// 4.在循环中，我做了以下几件事情：
/// (1).接受一个客户端连接
/// (2).将该客户端套接字与完成端口绑定到一起(还是调用CreateIoCompletionPort，但这次的
///    作用不同)，注意，按道理来讲，此时传递给CreateIoCompletionPort的第三个参数应该是
///   一个完成键，一般来讲，程序都是传递一个单句柄数据结构的地址，该单句柄数据包含了和
///   该客户端连接有关的信息，由于我们只关心套接字句柄，所以直接将套接字句柄作为完成键传递；
/// (3).触发一个WSARecv异步调用，这次又用到了“尾随数据”，使接收数据所用的缓冲区紧跟
///    在WSAOVERLAPPED对象之后，此外，还有操作类型等重要信息。
/// 在工作者线程的循环中，我们
///   1.调用GetQueuedCompletionStatus取得本次I/O的相关信息（例如套接字句柄、传送的字节数、单I/O数据结构的地址等等）
///   2.通过单I/O数据结构找到接收数据缓冲区，然后将数据原封不动的发送到客户端
///   3.再次触发一个WSARecv异步操作
//////////////////////////////////////////////////////////////////////////

#include <WINSOCK2.H>
#include <stdio.h>
#define PORT 5150
#define MSGSIZE 1024
#pragma comment(lib, "ws2_32.lib")
typedef enum
{
 RECV_POSTED
}OPERATION_TYPE;
typedef struct
{
 WSAOVERLAPPED overlap;
 WSABUF Buffer;
 char szMessage[MSGSIZE];
 DWORD NumberOfBytesRecvd;
 DWORD Flags;
 OPERATION_TYPE OperationType;
}PER_IO_OPERATION_DATA, *LPPER_IO_OPERATION_DATA;
DWORD WINAPI WorkerThread(LPVOID);
int main()
{
 WSADATA wsaData;
 SOCKET sListen, sClient;
 SOCKADDR_IN local, client;
 DWORD i, dwThreadId;
 int iaddrSize = sizeof(SOCKADDR_IN);
 HANDLE CompletionPort = INVALID_HANDLE_VALUE;
 SYSTEM_INFO systeminfo;
 LPPER_IO_OPERATION_DATA lpPerIOData = NULL;
 // Initialize Windows Socket library

 WSAStartup(0x0202, &wsaData);
 // Create completion port//创建一个完成端口对象

 // Winsock将使用这个对象为任意数量的套接字句柄管理I/O请求

 CompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
 //成功创建完完成端口对象后，便可以使用这个对象关联套接字的句柄了

 // Create worker thread

 GetSystemInfo(&systeminfo);
 for (i = 0; i < systeminfo.dwNumberOfProcessors; i++)
 {
  CreateThread(NULL, 0, WorkerThread, CompletionPort, 0, &dwThreadId);
 }
 // Create listening socket

 sListen = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
 // Bind

 local.sin_addr.S_un.S_addr = htonl(INADDR_ANY);
 local.sin_family = AF_INET;
 local.sin_port = htons(PORT);
 bind(sListen, (struct sockaddr *)&local, sizeof(SOCKADDR_IN));
 // Listen

 listen(sListen, 3);
 while (TRUE)
 {
  // Accept a connection

  sClient = accept(sListen, (struct sockaddr *)&client, &iaddrSize);
  printf("Accepted client:%s:%d\n", inet_ntoa(client.sin_addr), ntohs(client.sin_port));
  // Associate the newly arrived client socket with completion port

  // 将新到来的client、Socket对象关联到前面创建的完成端口的对象

  CreateIoCompletionPort((HANDLE)sClient, CompletionPort, (DWORD)sClient, 0);
  // Launch an asynchronous operation for new arrived connection

  lpPerIOData = (LPPER_IO_OPERATION_DATA)HeapAlloc(
   GetProcessHeap(),
   HEAP_ZERO_MEMORY,
   sizeof(PER_IO_OPERATION_DATA));
  lpPerIOData->Buffer.len = MSGSIZE;
  lpPerIOData->Buffer.buf = lpPerIOData->szMessage;
  lpPerIOData->OperationType = RECV_POSTED;
  /// 向完成端口关联套接字句柄之后，就可以通过在套接字投递重叠发送和接受请求处理I/O了

  WSARecv(sClient,
   &lpPerIOData->Buffer,
   1,
   &lpPerIOData->NumberOfBytesRecvd,
   &lpPerIOData->Flags,
   &lpPerIOData->overlap,
   NULL);
 }
 PostQueuedCompletionStatus(CompletionPort, 0xFFFFFFFF, 0, NULL);
 CloseHandle(CompletionPort);
 closesocket(sListen);
 WSACleanup();
 return 0;
}
DWORD WINAPI WorkerThread(LPVOID CompletionPortID)
{
 HANDLE CompletionPort=(HANDLE)CompletionPortID;
 DWORD dwBytesTransferred;
 SOCKET sClient;
 LPPER_IO_OPERATION_DATA lpPerIOData = NULL;
 while (TRUE)
 {
  /// 在I/O操作完成时，I/O系统会向完成端口对象发送一个完成通知包

  /// I/O完成端口以先进先出的方式为这些封包排队。

  /// 应用程序使用GetQueuedCompletionStatus()函数来去的队列中的封包

  GetQueuedCompletionStatus(
   CompletionPort,
   &dwBytesTransferred,
   (PULONG_PTR)&sClient,
   (LPOVERLAPPED *)&lpPerIOData,
   INFINITE);
  if (dwBytesTransferred == 0xFFFFFFFF)
  {
   return 0;
  }
  if (lpPerIOData->OperationType == RECV_POSTED)
  {
   if (dwBytesTransferred == 0)
   {
    // Connection was closed by client

    closesocket(sClient);
    HeapFree(GetProcessHeap(), 0, lpPerIOData);
   }
   else
   {
    lpPerIOData->szMessage[dwBytesTransferred] = '\0';
    send(sClient, lpPerIOData->szMessage, dwBytesTransferred, 0);
    // Launch another asynchronous operation for sClient

    memset(lpPerIOData, 0, sizeof(PER_IO_OPERATION_DATA));
    lpPerIOData->Buffer.len = MSGSIZE;
    lpPerIOData->Buffer.buf = lpPerIOData->szMessage;
    lpPerIOData->OperationType = RECV_POSTED;
    WSARecv(sClient,
     &lpPerIOData->Buffer,
     1,
     &lpPerIOData->NumberOfBytesRecvd,
     &lpPerIOData->Flags,
     &lpPerIOData->overlap,
     NULL);
   }
  }
 }
 return 0;
}