关于完成端口(IOCP)的文章

关于完成端口(IOCP)的文章汇总 - [C/C++]
版权声明：转载时请以超链接形式标明文章原始出处和作者信息及本声明
http://joshuafan.blogbus.com/logs/32007489.html



首先讨论一下I/O Completion Ports试图解决什么样的问题。

写一个IO Intensive服务器程序，对每一个客户请求生成一个新的child process/worker thread来处理，每个process/thread使用同步IO，这是最经典古老的解法了。在这之上的改进是prefork 多个process 或者使用线程池。（使用process或thread，原理都差不多，thread的context switch花销要比process switch要小。为了论述简单，下面只讨论线程。）

这种结构的并发性并不高，哪怕你用C++, C甚至汇编来写，效率都不会很高，究其原因，在于两点：

一．同步IO，每个线程大多数时间在等IO request的结束。IO相对于CPU，那是极极慢的。我翻了翻手里的Computer Architecture, A Quantitative Approach第二版，1996年出的，里面对CPU Register, CPU Cache, RAM, Disk，列的access time如下：


Java代码
Registers:  2-5 nano seconds  
CPU Cache: 3-10 nano seconds  
RAM: 80-400 nano seconds  
Disk: 5 000 000 nano seconds (5 milli seconds) 
Registers:  2-5 nano seconds
CPU Cache: 3-10 nano seconds
RAM: 80-400 nano seconds
Disk: 5 000 000 nano seconds (5 milli seconds)


如今CPU又按照摩尔定律发展了十年后，这个硬盘还是机械式的磁头移来移去读写，尽管如今disk controller都有cache，也在发展，但和CPU相比，差距越来越大。（谁有最新数据可以贴上来。）

二．生成数量大大超过CPU总数的线程。这样做有两个弊端，第一是每个线程要占用内存，Windows底下每个thread自己stack的省缺大小为1M，32位程序下一个用户程序最大能利用的内存也就3G，生成3000个线程，内存就没了。当然有人说64位下面，可以随便浪费，那么，第二个弊端，就无法避免了 ─ 生成大量的线程，CPU必然会花费大量的cpu cycles在线程之间进行切换。如今市场上价格适中的服务器也就2 cpu x 4 core = 8 核而已。生成那么多的线程，CPU在切换线程上花的功夫可能比干正经事还要多。

明白了原因，就可以寻找改进方法。首先，使用异步IO。现在所有主流OS，都提供异步IO（non-blocking IO），连Java这种跨平台的编程环境都在版本1.4里开始支持异步IO了。但是，光有异步IO，这是不够的。论坛里有人发贴子问过，“我的线程发个IO Request，异步IO，直接返回了，然后我的线程干什么？” 异步IO是操作系统提供的机制，我们还需要设计我们程序的结构，使异步IO和线程结合起来，可以充分利用异步IO带来的好处，同时必须控制同时运行线程的数量，减少thread context switch的开销。

IO Completion Port, 是微软针对上述思想，在Windows内核级别，提供的解决方案。

从抽象高度去理解IO Completion Port，可以把它想成一个magic port，一边有一个队列是IO驱动程序处理好的IO数据，另一边是一个小小的线程池，这个port把io数据交给线程池里的线程来处理。同时，别的线程启动了IO异步请求后通知这个port一声，“嘿，注意了，一会儿这个IO handle 会有个数据包传过来要处理。” 这个port回答，“好，我注意一下这个handle。”。



下面我们具体看一下Io Completion Port这个内核对象以及使用。

要创建IoCompletionPort，呼叫Win32函数CreateIoCompletionPort。这个函数一身两用，创建IoCompletionPort也是它，往建好的IoCompletionPort里面加device handle也是它。


Java代码
HANDLE CreateIoCompletionPort(  
  HANDLE hfile,  
  HANDLE hExistingCompPort,  
  ULONG_PTR CompKey,  
  DWORD dwNumberOfConcurrentThreads);  
 
// 创建IoCompletionPort  
 
HANDLE hCp;  
hCp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 4); 
HANDLE CreateIoCompletionPort(
  HANDLE hfile,
  HANDLE hExistingCompPort,
  ULONG_PTR CompKey,
  DWORD dwNumberOfConcurrentThreads);

// 创建IoCompletionPort
HANDLE hCp;
hCp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 4);创建IoCompletionPort头三个参数都是NULL之类的，只有第四个参数，用来配置这个生成的IoCompletionPort所允许同时运行的最大线程数目。

创建好的IoCompletionPort kernel object，拥有两个队列。一个是Device List，包含所有通过这个IoCompletionPort管理的异步IO请求 的Device Handle。另外一个是I/O Completion Queue (FIFO)，Device Handle对应的IO驱动程序处理好的IO数据，放在这个队列里。


为了能在Device List这个队列里面加个entry，用户程序将再一次使用CreateIoCompletionPort 这个函数。


Java代码
CreateIoCompletionPort(myHandle, hCp, myKey, 0); 
CreateIoCompletionPort(myHandle, hCp, myKey, 0);

第一个参数是个IO Handle（Windows不限制handle类型，File, Directory, Serial Port, Parallel port, Mailslot server, Mailslot client, pipe, socket等等都可以），第二个参数是以前创建的IoCompletionPort handle. 这个IO Handle将放到IoCompletionPort handle的Device List那个队列里去。第三个参数是个long整数，是用来identify程序Request Context的。因为现在程序不是一个线程来处理客户Request了，而是不同的线程来处理。在一个线程里按顺序一二三四五来实现程序逻辑的方式是不行了，因此作为程序员你要把逻辑的Context记下来，让不同的线程得到这个Context，根据当前的状态，来执行相关的代码。这个completion key，是找到相映的context的key, index, hash code，pointer, whatever.

第二个队列，IO Completion Queue，是由OS往里面插入entry的。OS在处理好了IO异步请求之后，察看一下这个Device handle是否是放在某个Completion Port里面，如果是，OS就在Completion Port的Completion Queue里面加个Entry。这个Entry包括下列数据。


Java代码
1．Number of bytes transferred  
2．Completion key  
3．Pointer to I/O request’s OVERLAPPED structure  
4．Error code 
1．Number of bytes transferred
2．Completion key
3．Pointer to I/O request’s OVERLAPPED structure
4．Error code


下面来看看IO Completion Port是怎么管理线程的。



前面说Completion Port有个线程池，这种说法并不是很贴切。Completion Port本身并不创建线程，而只是掌管三个thread队列：


Java代码
1．Inactive threads waits IO Completion Port  
2．Active running threads  
3．Threads paused by other reasons, like waiting for something else (i.e. calls WaitForSingleObject, or even stupid but valid, calls Sleep). 
1．Inactive threads waits IO Completion Port
2．Active running threads
3．Threads paused by other reasons, like waiting for something else (i.e. calls WaitForSingleObject, or even stupid but valid, calls Sleep).
线程由程序创建，然后加入第一个队列(waits on IO Completion Port)。为了加入这个队列线程要呼叫一个函数，GetQueuedCompletionStatus。


Java代码
BOOL GetQueuedCompletionStatus(  
  HANDLE       hCompPort,  
  PDWORD       pdwNumBytes,  
  PULONG_PTR   CompKey,  
  OVERLAPPED** ppOverlapped,  
  DWORD        dwMilliseconds); 
BOOL GetQueuedCompletionStatus(
  HANDLE       hCompPort,
  PDWORD       pdwNumBytes,
  PULONG_PTR   CompKey,
  OVERLAPPED** ppOverlapped,
  DWORD        dwMilliseconds);
第一个参数是handle to Completion Port，线程通知OS本线程要加入这个Completion Port的第一个队列。这个函数会block当前线程，使其处于inactive状态。

现在再去看看图二的I/O Completion Queue，OS在一份IO异步请求处理好了后，会在这里插入个entry，Completion Port在收到entry后，看看线程池里面有没有空闲没事做的线程，如果有，不要忘记我们创建这个Completion Port时候规定了个最大同时运行线程数量，如果当前运行线程数量小于这个最大值，那么就把这个线程放到第二个（active running）的队列上去，让这个线程运行起来。前面不是说线程在GetQueuedCompletionStatus上面block了么，现在这个函数返回了，继续运行程序的代码。通过这个最大同时运行线程数量，保证了不会有太多的线程在运行，Viola! 本文开头分析的几个问题全解决了。即是异步IO，又把异步IO和线程池结合了起来，还控制了当前运行线程数量。It’s BEAUTIFUL!

这个线程处理完程序逻辑后，呼叫一下GetQueuedCompletionStatus，又回到了第一个队列。有意思的是这个队列的逻辑是Last In First Out。如果又有IO数据等待线程处理，这个线程可以继续执行，不用进行Context Switch，典型的能者多劳啊，越能干的人干的越多。

这个线程在处理程序逻辑的过程中，可能会因为别的原因而变成inactive，比如在等别的资源（WaitForSingleObject），或者变态一点，自己来了个Sleep，这时线程就给放到第三个队列去了。

这里有个有趣的现象，假如开始我们在第一个队列里面放三个线程，而最大同时运行线程数量设为2，在两个线程跑起来之后，第三个就不跑了，如果这时运行中的某个线程因为等别的资源而变为inactive，那么第三个线程也开始跑起来，同时运行线程数量还是2，这时那个等别的资源的线程等到资源了，又开始跑了起来，这时同时运行线程数量就是3，比设定的2要大。

推荐的最大同时运行线程数量一般为CPU的总数，但是如果运行的线程还要等别的资源，建议把这个数目稍微设大一点，这样并发率会更高. 关于这一点微软的描述如下:

An I/O completion port is associated with the process that created it and is not shareable between processes. However, a single handle is shareable between threads in the same process. For another article about I/O completion ports, see "Inside I/O Completion Ports" in the Microsoft TechNet Library at http://go.microsoft.com/fwlink/?LinkId=117124

Threads and Concurrency
The most important property of an I/O completion port to consider carefully is the concurrency value. The concurrency value of a completion port is specified when it is created with CreateIoCompletionPort via the NumberOfConcurrentThreads parameter. This value limits the number of runnable threads associated with the completion port. When the total number of runnable threads associated with the completion port reaches the concurrency value, the system blocks the execution of any subsequent threads associated with that completion port until the number of runnable threads drops below the concurrency value.

The most efficient scenario occurs when there are completion packets waiting in the queue, but no waits can be satisfied because the port has reached its concurrency limit. Consider what happens with a concurrency value of one and multiple threads waiting in the GetQueuedCompletionStatus function call. In this case, if the queue always has completion packets waiting, when the running thread calls GetQueuedCompletionStatus, it will not block execution because, as mentioned earlier, the thread queue is LIFO. Instead, this thread will immediately pick up the next queued completion packet. No thread context switches will occur, because the running thread is continually picking up completion packets and the other threads are unable to run.

Note 

In the previous example, the extra threads appear to be useless and never run, but that assumes that the running thread never gets put in a wait state by some other mechanism, terminates, or otherwise closes its associated I/O completion port. Consider all such thread execution ramifications when designing the application.

The best overall maximum value to pick for the concurrency value is the number of CPUs on the computer. If your transaction required a lengthy computation, a larger concurrency value will allow more threads to run. Each completion packet may take longer to finish, but more completion packets will be processed at the same time. You can experiment with the concurrency value in conjunction with profiling tools to achieve the best effect for your application.

The system also allows a thread waiting in GetQueuedCompletionStatus to process a completion packet if another running thread associated with the same I/O completion port enters a wait state for other reasons, for example the SuspendThread function. When the thread in the wait state begins running again, there may be a brief period when the number of active threads exceeds the concurrency value. However, the system quickly reduces this number by not allowing any new active threads until the number of active threads falls below the concurrency value. This is one reason to have your application create more threads in its thread pool than the concurrency value. Thread pool management is beyond the scope of this topic, but a good rule of thumb is to have a minimum of twice as many threads in the thread pool as there are processors on the system. For additional information about thread pooling, see Thread Pools.

关于完成端口的另一个应用:

Threads can use the PostQueuedCompletionStatus function to place completion packets in an I/O completion port's queue. By doing so, the completion port can be used to receive communications from other threads of the process, in addition to receiving I/O completion packets from the I/O system. The PostQueuedCompletionStatus function allows an application to queue its own special-purpose completion packets to the I/O completion port without starting an asynchronous I/O operation. This is useful for notifying worker threads of external events, for example.

完成端口的问题

异步过程调用(apcs)问题：
    只有发overlapped请求的线程才可以提供callback函数（需要一个特定的线程为一个特定的I/O请求服务）。
完成端口(I/O completion)的优点：
    不会限制handle个数，可处理成千上万个连接。I/O completion port允许一个线程将一个请求暂时保存下来，由另一个线程为它做实际服务。
并发模型与线程池：
    在典型的并发模型中，服务器为每一个客户端创建一个线程，如果很多客户同时请求，则这些线程都是运行的，那么CPU就要一个个切换，CPU花费了更多的时间在线程切换，线程确没得到很多CPU时间。到底应该创建多少个线程比较合适呢，微软件帮助文档上讲应该是2*CPU个。但理想条件下最好线程不要切换，而又能象线程池一样，重复利用。I/O完成端口就是使用了线程池。
理解与使用：
第一步：
在我们使用完成端口之前，要调用CreateIoCompletionPort函数先创建完成端口对象。
定义如下：
HANDLE CreateIoCompletionPort(
                                 HANDLE FileHandle,
                                HANDLE ExistingCompletionPort,
                               DWORD CompletionKey,
                               DWORD NumberOfConcurrentThreads
);
FileHandle：
文件或设备的handle, 如果值为INVALID_HANDLE_VALUE则产生一个没有和任何文件handle有关系的port.( 可以用来和完成端口联系的各种句柄，文件，套接字)
ExistingCompletionPort:
NULL时生成一个新port, 否则handle会加到此port上。
CompletionKey：
用户自定义数值，被交给服务的线程。GetQueuedCompletionStatus函数时我们可以完全得到我们在此联系函数中的完成键（申请的内存块）。在GetQueuedCompletionStatus
中可以完封不动的得到这个内存块，并且使用它。
NumberOfConcurrentThreads：
参数NumberOfConcurrentThreads用来指定在一个完成端口上可以并发的线程数量。理想的情况是，一个处理器上只运行一个线程，这样可以避免线程上下文切换的开销。如果这个参数的值为0，那就是告诉系统线程数与处理器数相同。我们可以用下面的代码来创建I/O完成端口。
隐藏在之创建完成端口的秘密：
1． 创建一个完成端口
CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0, dwNumberOfConcurrentThreads);
2． 设备列表，完成端口把它同一个或多个设备相关联。
CreateIoCompletionPort(hDevice, hCompPort, dwCompKey, 0) ;
第二步：
根据处理器个数，创建cpu*2个工作线程：
CreateThread(NULL, 0, ServerWorkerThread, CompletionPort,0, &ThreadID))
与此同时，服务器调用WSASocket，bind, listen, WSAAccept,之后，调用
CreateIoCompletionPort((HANDLE) Accept, CompletionPort... )把一个套接字句柄和一个完成端口绑定到一起。完成端口又同一个或多个设备相关联着，所以以套接字为基础，投递发送和请求，对I/O处理。接着，可以依赖完成端口，接收有关I/O操作完成情况的通知。再看程序里：
WSARecv(Accept, &(PerIoData->DataBuf), 1, &RecvBytes, &Flags,
&(PerIoData->Overlapped), NULL)开始调用，这里象前面讲过的一样，既然是异步I/O，所以WSASend和WSARecv的调用会立即返回。
系统处理：
当一个设备的异步I/O请求完成之后，系统会检查该设备是否关联了一个完成端口，如果是，系统就向该完成端口的I/O完成队列中加入完成的I/O请求列。
然后我们需要从这个完成队列中，取出调用后的结果(需要通过一个Overlapped结构来接收调用的结果)。怎么知道这个队列中已经有处理后的结果呢，调用GetQueuedCompletionStatus函数。
工作线程与完成端口：
和异步过程调用不同(在一个Overlapped I/O完成之后，系统调用该回调函数。OS在有信号状态下(设备句柄)，才会调用回调函数（可能有很多APCS等待处理了）)
GetQueuedCompletionStatus
在工作线程内调用GetQueuedCompletionStatus函数。
GetQueuedCompletionStatus(
    HANDLE CompletionPort,
    LPDWORD lpNumberOfBytesTransferred,
    LPDWORD lpCompletionKey,
    LPOVERLAPPED *lpOverlapped,
    DWORD dwMilliseconds
);
CompletionPort：指出了线程要监视哪一个完成端口。很多服务应用程序只是使用一个I/O完成端口，所有的I/O请求完成以后的通知都将发给该端口。
lpNumberOfBytesTransferred：传输的数据字节数
lpCompletionKey：
完成端口的单句柄数据指针，这个指针将可以得到我们在CreateIoCompletionPort中申请那片内存。
lpOverlapped:
重叠I/O请求结构，这个结构同样是指向我们在重叠请求时所申请的内存块，同时和lpCompletionKey,一样我们也可以利用这个内存块来存储我们要保存的任意数据。
dwMilliseconds:
等待的最长时间(毫秒),如果超时，lpOverlapped被设为NULL，函数返回False.
GetQueuedCompletionStatus功能及隐藏的秘密：
GetQueuedCompletionStatus使调用线程挂起，直到指定的端口的I/O完成队列中出现了一项或直到超时。（I/0完成队列中出现了记录）调用GetQueuedCompletionStatus时，调用线程的ID(cpu*2个线程，每个ServerWorkerThread的线程ID)就被放入该等待线程队列中。
     等待线程队列很简单，只是保存了这些线程的ID。完成端口会按照后进先出的原则将一个线程队列的ID放入到释放线程列表中。
这样，I/O完成端口内核对象就知道哪些线程正在等待处理完成的I/O请求。当端口的I/O完成队列出现一项时，完成端口就唤醒（睡眠状态中变为可调度状态）等待线程队列中的一个线程。线程将得到完成I/O项中的信息：传输的字节数，完成键(单句柄数据结构)和Overlapped结构地址，线程是通过GetQueuedCompletionStatus返回这些信息，等待CPU的调度。
GetQueuedCompletionStatus返回可能有多种原因，如果传递无效完成端口句柄，函数返回False，GetLastError返回一个错误(ERROR_INVALID_HANDLE),如果超时，返回False, GetLastError返回WAIT_TIMEOUT, i/o完成队列删除一项，该表项是一个成功完成的I/O请求，则返回True。
    调用GetQueuedCompletionStatus的线程是后进先出的方式唤醒的，比如有4个线程等待，如果有一个I/O，最后一个调用GetQueuedCompletionStatus的线程被唤醒来处理。处理完之后，再调用GetQueuedCompletionStatus进入等待线程队列中。
深入分析完成端口线程池调度原理：
    假设我们运行在2CPU的机器上。创建完成端口时指定2个并发，创建了4个工作线程加入线程池中等待完成I/O请求，且完成端口队列（先入先出）中有3个完成I/O的请求的情况：
工作线程运行, 创建了4个工作线程，调用GetQueuedCompletionStatus时，该调用线程就进入了睡眠状态，假设这个时候，I/O完成队列出现了三项，调用线程的ID就被放入该等待线程队列中。
I/O完成端口内核对象（第3个参数等级线程队列）,因此知道哪些线程正在等待处理完成的I/O请求。当端口的I/O完成队列出现一项时，完成端口就唤醒（睡眠状态中变为可调度状态）等待线程队列中的一个线程(前面讲过等待线程队列是后进先出)。所以线程D将得到完成I/O项中的信息：传输的字节数，完成键(单句柄数据结构)和Overlapped结构地址，线程是通过GetQueuedCompletionStatus返回这些信息。
在前面我们指定了并发线程的数目是2，所以I/O完成端口唤醒2个线程，线程D和线程C，另两个继续休眠（线程B，线程A），直到线程D处理完了，发现表项里还有要处理的，就唤醒同一线程继续处理。
线程并发量：

   并发量限制了与该完成端口相关联的可运行线程的数目, 它类似阀门的作用。 当与该完成端口相关联的可运行线程的总数目达到了该并发量，系统就会阻塞任何与该完成端口相关联的后续线程的执行， 直到与该完成端口相关联的可运行线程数目下降到小于该并发量为止。所以解释了线程池中的运行线程可能会比设置的并发线程多的原因。
    它的作用:
最有效的假想是发生在有完成包在队列中等待，而没有等待被满足，因为此时完成端口达到了其并发量的极限。此时，一个正在运行中的线程调用 GetQueuedCompletionStatus时，它就会立刻从队列中取走该完成包。这样就不存在着环境的切换，因为该处于运行中的线程就会连续不断地从队列中取走完成包，而其他的线程就不能运行了。
注意：如果池中的所有线程都在忙，客户请求就可能拒绝，所以要适当调整这个参数，获得最佳性能。
线程并发：D线程挂起，加入暂停线程，醒来后又加入释放线程队列。
线程的安全退出：
PostQueudCompletionStatus函数，我们可以用它发送一个自定义的包含了OVERLAPPED成员变量的结构地址，里面包含一个状态变量，当状态变量为退出标志时，线程就执行清除动作然后退出。
完成端口使用需要注意的地方：
1．在执行wsasend和wsarecv操作前，请先将overlapped结构体使用memset进行清零。
2. 关于完成端口的内部实现, 微软的文档如下说:http://technet.microsoft.com/en-us/sysinternals/bb963891.aspx 
Completion Port Internals
A call to the Win32 API CreateIoCompletionPort with a NULL completion port handle results in the execution of the native API function NtCreateIoCompletion, which invokes the corresponding kernel-mode system service of the same name. Internally, completion ports are based on an undocumented executive synchronization object called a Queue. Thus, the system service creates a completion port object and initializes a queue object in the port's allocated memory (a pointer to the port also points to the queue object since the queue is at the start of the port memory). A queue object has (coincidentally) a concurrency value that is specified when a thread initializes one, and in this case the value that is used is the one that was passed to CreateIoCompletionPort. KeInitializeQueue is the function that NtCreateIoCompletion calls to initialize a port's queue object.

When an application calls CreateIoCompletionPort to associate a file handle with a port the Win32 API invokes the native function NtSetInformationFile with the file handle as the primary parameter. The information class that is set is FileCompletionInformation and the completion port's handle and the CompletionKey parameter from CreateIoCompletionPort are the data values. NtSetInformationFile dereferences the file handle to obtain the file object and allocates a completion context data structure, which is defined in NTDDK.H as:

typedef struct _IO_COMPLETION_CONTEXT {  PVOIDPort;  ULONGKey;} IO_COMPLETION_CONTEXT, *PIO_COMPLETION_CONTEXT;

Finally, NtSetInformationFile sets the CompletionContext field in the file object to point at the context structure. When an I/O operation completes on a file object the internal I/O manager function IopCompleteRequest executes and, if the I/O was asynchronous, checks to see if the CompletionContext field in the file object is non-NULL. If its non-NULL the I/O Manager allocates a completion packet and queues it to the completion port by calling KeInsertQueue with the port as the queue on which to insert the packet (remember that the completion port object and queue object are synonymous).

When GetQueuedCompletionStatus is invoked by a server thread, it calls the native API function NtRemoveIoCompletion, which transfers control to the NtRemoveIoCompletion system service. After validating parameters and translating the completion port handle to a pointer to the port, NtRemoveIoCompletion calls KeRemoveQueue.

As you can see, KeRemoveQueue and KeInsertQueue are the engine behind completion ports and are the functions that determine whether a thread waiting for an I/O completion packet should be activated or not. Internally, a queue object maintains a count of the current number of active threads and the maximum active threads. If the current number equals or exceeds the maximum when a thread calls KeRemoveQueue, the thread will be put (in LIFO order) onto a list of threads waiting for a turn to process a completion packet. The list of threads hangs off the queue object. A thread's control block data structure has a pointer in it that references the queue object of a queue that it is associated with; if the pointer is NULL then the thread is not associated with a queue.

So how does NT keep track of threads that become inactive because they block on something other than the completion port" The answer lies in the queue pointer in a thread's control block. The scheduler routines that are executed in response to a thread blocking (KeWaitForSingleObject, KeDelayExecutionThread, etc.) check the thread's queue pointer and if its not NULL they will call KiActivateWaiterQueue, a queue-related function. KiActivateWaiterQueue decrements the count of active threads associated with the queue, and if the result is less than the maximum and there is at least one completion packet in the queue then the thread at the front of the queue's thread list is woken and given the oldest packet. Conversely, whenever a thread that is associated with a queue wakes up after blocking the scheduler executes the function KiUnwaitThread, which increments the queue's active count.

Finally, the PostQueuedCompletionStatus Win32 API calls upon the native function NtSetIoCompletion. As with the other native APIs in the completion port group, this one invokes a system service bearing the same name, which simply inserts that packet onto the completion port's queue using KeInsertQueue.

Top of page
Not Exported
Windows NT's completion port API provides an easy-to-use and efficient way to maximize a server's performance by minimizing context switches while obtaining high-degrees of parallelism. The API is made possible with support in the I/O Manager, Kernel, and system services. While the Queue object is exported for use by device drivers (it is undocumented but its interfaces are relatively easy to figure out), the completion port APIs are not. However, if the queue interfaces are derived it is possible to mimick the completion port interfaces by simply using the queue routines and manually associating file objects with queues by setting the CompletionContext entry.

IOCP和EPOLL的异同
但凡作过比较深入的网络编程的人，都会知道，在win平台下，高效的IO模型是IOCP，而在linux底下则是epoll。那么，epoll与iocp之间到底有哪些异同之处呢？

首先，我们看一下它们相同的地方。两者都是处理异步IO的高效模型，这种高效，除了“异步处理”这个共同的特征之外，二者都可以通过指针携带应用层数据：在IOCP里，应用层数据可以通过单句柄数据和单IO数据来与IOCP底层通信；而在epoll里，可以通过epoll_data里的"void *ptr"来传递。这是一种很重要的思想，也是它们高效的原因所在：当事件的通知到来时，它不仅告诉你发生了什么样的事件，还同时告诉这次事件所操作的数据是哪些。

那么，epoll和iocp到底又有什么不同呢？以我目前粗浅的使用经验来看，至少可以得到以下结论：

1.iocp是在IO操作完成之后，才通过get函数返回这个完成通知的；而epoll则不是在IO操作完成之后才通知你，它的工作原理是，你如果想进行IO操作时，先向epoll查询是否可读或可写，如果处于可读或可写状态后，epoll会通过epoll_wait函数通知你，此时你再进行进一步的recv或send操作。

2. 在1的基础上，我们其实可以看到，epoll仅仅是一个异步事件的通知机制，其本身并不作任何的IO读写操作，它只负责告诉你是不是可以读或可以写了，而具体的读写操作，还要应用层自己来作；但iocp的封装就要多一些，它不仅会有完成之后的事件通知，更重要的是，它同时封装了一部分的IO控制逻辑。从这一点上来看，iocp的封装似乎更全面一点，但是，换个角度看，epoll仅提供这种机制也是非常好的，它保持了事件通知与IO操作之间彼此的独立性，使得epoll的使用更加灵活。

这只是我初步使用epoll开发过程中的体会，以后有更深的体会时还会发上来跟大家分享。
来自微软的例子
来自微软的完成端口例子，就讲解一下它的使用套路吧
反正编程这个玩意，只要用过，自然就知道什么回事，一次不会再看一次，学习这个玩意，无他，勤奋而已。
奢谈效率等等，那只是孰能生巧上的功夫。


  这个例子是在console下的例子,算是一个echo服务器吧,
  跑起来后将在5150端口监听,一旦有个端口连接上来,发个数据给服务端口,它就echo回数据给那个端口. 直到那个连接中断.
 
完成端口,其实理解成一个通道或管子就可以了,和管道也差不了多少,不过可以实现异步处理罢了,
你这边往管子里丢数据,通过GetQueuedCompletionStatus来查管子那头出数据没,出了就处理,这个管子就是通过一个自定义有点特殊的结构来写入或读出数据而已.
那个完成端口,其实就相当是标识那个数据块的句柄,

//下面请看例子
＃i nclude <winsock2.h>
＃i nclude <windows.h>
＃i nclude <stdio.h>

#define PORT 5150
#define DATA_BUFSIZE 8192

#pragma comment(lib, "Ws2_32")

typedef struct                        //这个玩意就是灌数据，取数据的一个自定义数据结构

                                              //和那个wm_data差不了多少，不过就是老要塞一个OverLapped结构，
{
   OVERLAPPED Overlapped;
   WSABUF DataBuf;
   CHAR Buffer[DATA_BUFSIZE];                    
   DWORD BytesSEND;                                 //发送字节数
   DWORD BytesRECV;                                
} PER_IO_OPERATION_DATA, * LPPER_IO_OPERATION_DATA;


typedef struct
{
   SOCKET Socket;
} PER_HANDLE_DATA, * LPPER_HANDLE_DATA;


DWORD WINAPI ServerWorkerThread(LPVOID CompletionPortID);


void main(void)
{
   SOCKADDR_IN InternetAddr;
   SOCKET Listen;
   SOCKET Accept;
   HANDLE CompletionPort;
   SYSTEM_INFO SystemInfo;
   LPPER_HANDLE_DATA PerHandleData;
   LPPER_IO_OPERATION_DATA PerIoData;
   int i;
   DWORD RecvBytes;
   DWORD Flags;
   DWORD ThreadID;
   WSADATA wsaData;
   DWORD Ret;

   if ((Ret = WSAStartup(0x0202, &wsaData)) != 0)
   {
      printf("WSAStartup failed with error %d\n", Ret);
      return;
   }

   //
   //完成端口的建立得搞2次，这是第一次调用，至于为什么？我问问你
   //
   if ((CompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0)) == NULL)
   {
      printf( "CreateIoCompletionPort failed with error: %d\n", GetLastError());
      return;
   }
   //老套子api，不谈也罢
   GetSystemInfo(&SystemInfo);
  
   //发现2个CPU，那就开个双倍的线程跑吧
   for(i = 0; i < SystemInfo.dwNumberOfProcessors * 2; i++)
   {
      HANDLE ThreadHandle;
     
      //
      //完成端口挂到线程上面来了，就像管子把灌数据的和读数据的两头都连上了,           
     //
      if ((ThreadHandle = CreateThread(NULL, 0, ServerWorkerThread, CompletionPort,
         0, &ThreadID)) == NULL)
      {
         printf("CreateThread() failed with error %d\n", GetLastError());
         return;
      }     
      CloseHandle(ThreadHandle);
   }

   //
   //启动一个监听socket ，以下都是长长的交代
   //
   if ((Listen = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0,
      WSA_FLAG_OVERLAPPED)) == INVALID_SOCKET)
   {
      printf("WSASocket() failed with error %d\n", WSAGetLastError());
      return;
   }

   InternetAddr.sin_family = AF_INET;
   InternetAddr.sin_addr.s_addr = htonl(INADDR_ANY);
   InternetAddr.sin_port = htons(PORT);

   if (bind(Listen, (PSOCKADDR) &InternetAddr, sizeof(InternetAddr)) == SOCKET_ERROR)
   {
      printf("bind() failed with error %d\n", WSAGetLastError());
      return;
   }  

   if (listen(Listen, 5) == SOCKET_ERROR)
   {
      printf("listen() failed with error %d\n", WSAGetLastError());
      return;
   }

   //
   // 监听端口打开，就开始在这里循环，一有socket连上，WSAAccept就创建一个socket，
   // 这个socket 又和完成端口联上,
   //
   // 嘿嘿，完成端口第二次调用那个createxxx函数，为什么，留给人思考思考可能更深刻，
   // 反正这套路得来2次，
   // 完成端口completionport和accept socket挂起来了，
   //
   while(TRUE)
   {

    //主线程跑到这里就等啊等啊，但是线程却开工了，
      if ((Accept = WSAAccept(Listen, NULL, NULL, NULL, 0)) == SOCKET_ERROR)
      {
         printf("WSAAccept() failed with error %d\n", WSAGetLastError());
         return;
      }
     
      if ((PerHandleData = (LPPER_HANDLE_DATA) GlobalAlloc(GPTR, sizeof(PER_HANDLE_DATA))) == NULL)
      {
         printf("GlobalAlloc() failed with error %d\n", GetLastError());
         return;
      }     
     
      PerHandleData->Socket = Accept;
     
      //
     //把这头和完成端口completionPort连起来
     //就像你把漏斗接到管子口上,开始要灌数据了
     //
      if (CreateIoCompletionPort((HANDLE) Accept, CompletionPort, (DWORD) PerHandleData,
         0) == NULL)
      {
         printf("CreateIoCompletionPort failed with error %d\n", GetLastError());
         return;
      }
     
      //
      //清管子的数据结构,准备往里面灌数据
      //
      if ((PerIoData = (LPPER_IO_OPERATION_DATA) GlobalAlloc(GPTR,sizeof(PER_IO_OPERATION_DATA))) == NULL)
      {
         printf("GlobalAlloc() failed with error %d\n", GetLastError());
         return;
      }

      ZeroMemory(&(PerIoData->Overlapped), sizeof(OVERLAPPED));
      PerIoData->BytesSEND = 0;
      PerIoData->BytesRECV = 0;
      PerIoData->DataBuf.len = DATA_BUFSIZE;
      PerIoData->DataBuf.buf = PerIoData->Buffer;

      Flags = 0;
     
      //
      //  accept接到了数据，就放到PerIoData中,而perIoData又通过线程中的函数取出,
     //
      if (WSARecv(Accept, &(PerIoData->DataBuf), 1, &RecvBytes, &Flags,
         &(PerIoData->Overlapped), NULL) == SOCKET_ERROR)
      {
         if (WSAGetLastError() != ERROR_IO_PENDING)
         {
            printf("WSARecv() failed with error %d\n", WSAGetLastError());
            return;
         }
      }
   }
}

//
//线程一但调用，就老在里面循环，
// 注意，传入的可是完成端口啊，就是靠它去取出管子中的数据
//
DWORD WINAPI ServerWorkerThread(LPVOID CompletionPortID)
{
   HANDLE CompletionPort = (HANDLE) CompletionPortID;
  
   DWORD BytesTransferred;
   LPOVERLAPPED Overlapped;
   LPPER_HANDLE_DATA PerHandleData;
   LPPER_IO_OPERATION_DATA PerIoData;        
   DWORD SendBytes, RecvBytes;
   DWORD Flags;
 
   while(TRUE)
   {
      //
      //在这里检查完成端口部分的数据buf区，数据来了吗？
      // 这个函数参数要看说明，
      // PerIoData 就是从管子流出来的数据,
      //PerHandleData 也是从管子里取出的，是何时塞进来的，
     //就是在建立第2次createIocompletionPort时
    //

      if (GetQueuedCompletionStatus(CompletionPort, &BytesTransferred,
         (LPDWORD)&PerHandleData, (LPOVERLAPPED *) &PerIoData, INFINITE) == 0)
      {
         printf("GetQueuedCompletionStatus failed with error %d\n", GetLastError());
         return 0;
      }

      // 检查数据传送完了吗
      if (BytesTransferred == 0)
      {
         printf("Closing socket %d\n", PerHandleData->Socket);

         if (closesocket(PerHandleData->Socket) == SOCKET_ERROR)
         {
            printf("closesocket() failed with error %d\n", WSAGetLastError());
            return 0;
         }

         GlobalFree(PerHandleData);
         GlobalFree(PerIoData);
         continue;
      }    
     //
    //看看管子里面有数据来了吗？=0，那是刚收到数据
    //
      if (PerIoData->BytesRECV == 0)
      {
         PerIoData->BytesRECV = BytesTransferred;
         PerIoData->BytesSEND = 0;
      }
      else   //来了，
      {
         PerIoData->BytesSEND += BytesTransferred;
      }
  
      //
      // 数据没发完？继续send出去
     //
     if (PerIoData->BytesRECV > PerIoData->BytesSEND)
      {

         ZeroMemory(&(PerIoData->Overlapped), sizeof(OVERLAPPED)); //清0为发送准备
         PerIoData->DataBuf.buf = PerIoData->Buffer + PerIoData->BytesSEND;
         PerIoData->DataBuf.len = PerIoData->BytesRECV - PerIoData->BytesSEND;

       //1个字节一个字节发送发送数据出去
         if (WSASend(PerHandleData->Socket, &(PerIoData->DataBuf), 1, &SendBytes, 0,
            &(PerIoData->Overlapped), NULL) == SOCKET_ERROR)
         {
            if (WSAGetLastError() != ERROR_IO_PENDING)
            {
               printf("WSASend() failed with error %d\n", WSAGetLastError());
               return 0;
            }
         }
      }
      else
      {
         PerIoData->BytesRECV = 0;

         Flags = 0;
         ZeroMemory(&(PerIoData->Overlapped), sizeof(OVERLAPPED));

         PerIoData->DataBuf.len = DATA_BUFSIZE;
         PerIoData->DataBuf.buf = PerIoData->Buffer;

         if (WSARecv(PerHandleData->Socket, &(PerIoData->DataBuf), 1, &RecvBytes, &Flags,
            &(PerIoData->Overlapped), NULL) == SOCKET_ERROR)
         {
            if (WSAGetLastError() != ERROR_IO_PENDING)
            {
               printf("WSARecv() failed with error %d\n", WSAGetLastError());
               return 0;
            }
         }
      }
   }
}