《Windows核心编程系列》十异步IO之IO完成端口

              http://blog.csdn.net/ithzhang/article/details/8508161转载请注明出处！！                                                   

                                                                 IO完成端口

 

    为了将Windows打造成一个出色的服务器环境，Microsoft开发出了IO完成端口。完成端口需要与线程池配合使用。

 

    完成端口背后的理论是并发运行的线程数量必须有一个上限。由于太多的线程将会导致系统花费很大的代价在各个线程cpu上下文进行切换。

 

   使用并发模型与创建进程相比开销要低很多，但是也需要为每个客户请求创建一个新的线程。这开销仍然很大。通过使用线程池可以是性能有很大的提高。IO完成端口需要配合线程池配合使用。

 

    IO完成端口也是一个内核对象。调用以下函数创建IO完成端口内核对象。

 

 

[cpp] view plaincopy

1. HANDLE CreateIoCompletionPort(  
2.   
3.       HANDLE hFile,  
4.   
5.       HANDLE hExistingCompletionPort,  
6.   
7.       ULONG_PTR CompletionKey,  
8.   
9.       DWORD dwNumberOfConcurrentThreads);  

 

这个函数会完成两个任务：

     一是创建一个IO完成端口对象。

 

     二是将一个设备与一个IO完成端口关联起来。

 

     hFile就是设备句柄。

 

     hExistingCompletionPort是与设备关联的IO完成端口句柄。为NULL时，系统会创建新的完成端口。

 

     dwCompletionKey是一个对我们有意义的值，但是操作系统并不关心我们传入的值。一般用它来区分各个设备。

 

     dwNumberOfConcurrentThreads告诉IO完成端口在同一时间最多能有多少进程处于可运行状态。如果传入0，那么将使用默认值（并发的线程数量等于cpu数量）。在第二章我们曾介绍说几乎所有的内核对象都需要安全属性参数。那时的几乎就是因为IO完成端口这个例外。它是唯一一个不需要安全属性的内核对象。 这是因为IO完成端口在设计时就是只在一个进程中使用。

 

      每次调用CreateIoCompletionPort时，函数会判断hExistingCompletionKey是否为NULL，如果为NULL，会创建新的完成端口内核对象。并为此完成端口创建设备列表然后将设备加入到此完成端口设备列表中（先入先出）。

 

    设备列表存储与该完成端口相关联的所有设备。

 

    设备列表只是调用CreateIoCompletionPort函数时的一个数据结构。除此之外还有四个结构。

 

    第二个结构是IO完成队列。当设备的一个异步IO请求完成时，系统会检查该设备是否与一个完成端口相关联，如果关联，系统会将这个已完成的IO请求添加到完成端口的IO完成队列中。每一项包括已传输字节数，完成键（dwCompletionKey）值，以及一个指向IO请求的OVERLAPPED结构指针和错误码。

 

     Windows为IO完成端口提供了一个函数，可以将线程切换到睡眠状态，来等待设备IO请求完成并进入完成端口。

 

 

[cpp] view plaincopy

1. BOOL GetQueuedCompletionStatus(  
2.   
3.      HANDLE hCompletionPort,  
4.   
5.      PDWORD pdwNumberOfBytesTransferred,  
6.   
7.      ULONG_PTR pCompletionKey,  
8.   
9.      OVERLAPPED** ppOverlapped,  
10.   
11.      DWORD dwMilliSeconds);  

 

     hCompletionPort表示线程希望对哪个完成端口进行监视，GetQueuedCompletionStatus的任务就是将调用线程切换到睡眠状态，也就是阻塞在此函数上，直到指定的IO完成端口出现一项或者超时。

        pdwNumberOfBytesTransferred返回在异步IO完成时传输的字节数。

      pCompletionKey返回完成键。

      ppOverlapped返回异步IO开始时传入的OVERLAPPED结构地址。

      dwMillisecond指定等待时间。

           函数执行成功则返回true，否则返回false。

     第三个结构是等待线程队列。当线程池中的每个线程调用GetQueuedCompletionStatus时，调用线程的线程标识符会被添加到这个等待线程队列，这使得IO完成端口对象能知道，有哪些线程当前正在等待对已完成的IO请求进行处理。当IO完成端口的IO完成队列中出现一项时，完成端口会唤醒等待线程队列中的一个线程。这个线程会得到已完成IO项的所有信息：已传输字节数，完成键以及OVERLAPPED结构地址。这些信息是通过传给GetQueuedCompletionStatus的参数来返回的。

 

     IO完成队列中的各项是以先入先出方式来进行的。但是唤醒等待队列中的线程是按照后入先出的方式进行。假设有四个线程正在等待队列中等待，如果出现了一个已完成的IO项，那么最后一个由于调用GetQueuedCompletionStatus而被挂起的线程会被唤醒来处理这一项。当处理完该项后，线程会由于再次调用GetQueuedCompletionStatus而进入等待线程队列。使用这种算法，系统可以将哪些长时间睡眠的线程换出到磁盘。

 

     作者一直在推崇IO完成端口。接下来我们来讨论下为什么IO完成端口这么有用！！！

 

      前面我们提到过IO完成端口只有配合线程池才能发挥更大的作用。当我们创建并关联设备时，需要指定有多少个线程并发运行。一般将这个值设置为cpu的数量。当已完成的IO项被添加到完成队列中时，IO完成端口会唤醒正在等待的线程，但是唤醒的线程数最多不会超过我们指定的数量。如果有四个IO请求已完成，且有四个线程等待GetQueuedCompletionStatus而被挂起，那么IO完成端口只唤醒两个线程处理，另外两个线程继续睡眠。此时读者可能会疑问：既然完成端口只允许唤醒指定数量的线程，那么为什么还指定更多的线程在线程池中呢？这就涉及到IO完成端口的第四个数据结构：已释放线程列表。它存储已被唤醒的线程句柄。这使得IO完成端口能够直到哪些线程已经被唤醒并监视它们的执行情况。如果此时已释放线程由于调用某些函数将线程切换到了等待状态，完成端口会将其从已释放队列中移除，并将其添加到已暂停线程列表。

 

     已暂停队列是IO完成端口第五个数据结构。

 

     完成端口的目标是根据创建完成端口时指定的并发线程数量，将尽可能多的线程保持在已释放线程列表中。如果一个已暂停的线程被唤醒，它会离开已暂停线程列表并重新进入已释放线程列表。这意味着已释放列表中的线程数量将大于最大允许的并发线程数量。这句话什么意思呢？正在运行的线程数量加上从暂停线程列表中被释放的线程数量使总数大于最大允许的数量。这可以使线程数量在短时间内超过指定数量。

 

    IO完成端口并不一定要用于设备IO，它还可以进行线程间通信。在可提醒IO中我们介绍了QueueUserAPC。该函数允许线程将一个APC项添加到另一个线程的队列中。IO完成端口也存在一个类似的函数：

>IO通知追加到IO完成端口?size:14pt"> 

[cpp] view plaincopy

1. BOOL PostQueuedCompletionStatus(  
2.   
3.      HANDLE hCompletionPort,  
4.   
5.      DWORD dwNumBytes,  
6.   
7.      ULONG_PTR CompletionKey,  
8.   
9.      OVERLAPPED*pOverlapped);  

 

    这个函数用来将已完成的IO通知追加到IO完成端口的队列中。

 

     hCompletionPort表示我们要将已完成的IO项添加到哪个完成端口的队列中。

 

    剩下的三个参数表示应该返回给主调线程的值。

 

    每个线程都调用一次GetQueuedCompletionStatus，将它们都唤?E7??程通信。例如：当用户终止服务程序时，我们想要所有线程退出。如果各个线程还在等待完成端口但有没有已完成的IO请求，那么无法将它们唤醒。我们可以为线程池中的每个线程都调用一次GetQueuedCompletionStatus，将它们都唤醒。各线程对GetQueuedCompletionStatus函数返回值进行检查，如果发现应用程序正在终止，就会正常退出。（由于线程等待队列是以栈方式唤醒各线程，为了保证线程池中每个线程都有机会得到模拟IO项，我们还必须在程序中采用其他线程同步机制）

 

    当对完成端口调用CloseHandle时，系统会将所有正在等待GetQueuedCompletionStatus返回的线程唤醒，并返回false。GetLastError返回ERROR_INVALID_HANDLE，此时线程就可以知道应该退出了。

 

            下面的例子使用异步IO 实现了文件复制工作。首先选择要复制的文件，并取得源文件大小。点击复制按钮创建一个线程。新线程将完成创建IO完成端口和关联设备及执行文件复制工作。将源文件和目标文件都关联到同一个完成端口，根据GetQueuedCompletionStatus返回时的完成键来区分到底是属于谁的异步IO返回。在启动循环时采用了一个小伎俩：使用PostQueuedCompletionStatus向IO完成端口发送一个模拟的异步IO请求。完成键设置为WRITE_KEY。此时程序将执行从源文件读数据操作。这样就开动了引擎。直到文件复制完成。注意源文件和目标文件以及GetQueuedCompletionStatus使用的OVERLAPPED结构不要使用同一个。

 

选择文件函数：

[cpp] view plaincopy

1. void CIOCompletionPortDlg::OnBnClickedBtnChoosefile()  
2. {  
3.     // TODO: 在此添加控件通知处理程序代码  
4.     CFileDialog dlg(true);  
5.     dlg.DoModal();  
6.     m_fileName=dlg.GetPathName();  
7.     SetDlgItemText(IDC_EDIT_FILENAME,m_fileName);  
8.   
9.     m_hSrcFile=CreateFile(m_fileName,GENERIC_READ,FILE_SHARE_READ,NULL,OPEN_ALWAYS,FILE_FLAG_OVERLAPPED,NULL);  
10.     if(m_hSrcFile==INVALID_HANDLE_VALUE)  
11.     {  
12.         return ;  
13.     }  
14.     DWORD filesize;  
15.     DWORD filesizeHigh;  
16.     m_SrcFileSize=GetFileSize(m_hSrcFile,&filesizeHigh);  
17.     DWORD t=m_SrcFileSize/1024.0;  
18.     //filesize/=1024.0;  
19.     CString temp;  
20.     temp.Format(TEXT("%d KB"),t);  
21.     SetDlgItemText(IDC_EDIT_FILESIZE,temp);  
22. }  

//创建任务线程：

[cpp] view plaincopy

1. void CIOCompletionPortDlg::OnBnClickedBtnCopy()  
2. {  
3.     // TODO: 在此添加控件通知处理程序代码  
4.   
5.     m_hCopyThread=CreateThread(NULL,0,CopyThread,this,0,NULL);  
6.     //CloseHandle(m_hCopyThread);  
7.   
8. }  

//新线程入口函数：

[cpp] view plaincopy

1. <span style="font-size:12px;">DWORD WINAPI CIOCompletionPortDlg::CopyThread( PVOID ppram )  
2. {  
3.     CIOCompletionPortDlg *pdlg=(CIOCompletionPortDlg*)ppram;  
4.     pdlg->m_hDesFile=CreateFile(TEXT("备份.exe"),GENERIC_WRITE,0,NULL,CREATE_ALWAYS,FILE_FLAG_OVERLAPPED,pdlg->m_hSrcFile);  
5.     LARGE_INTEGER filesize;  
6.     filesize.HighPart=0;  
7.     filesize.LowPart=pdlg->m_SrcFileSize;  
8.     SetFilePointerEx(pdlg->m_hDesFile,filesize,NULL,FILE_BEGIN);  
9.     SetEndOfFile(pdlg->m_hDesFile);  
10.   
11.   
12.     //创建IO完成端口。创建一个完成端口，将两个设备将关联到此完成端口上。  
13.     HANDLE hIOCP=CreateIoCompletionPort(INVALID_HANDLE_VALUE,NULL,0,4);//创建完成端口，但不关联设备。  
14.     if(hIOCP==NULL)  
15.     {  
16.         return 0;  
17.     }  
18.     CreateIoCompletionPort(pdlg->m_hSrcFile,hIOCP,READ_KEY,0);//与IO完成端口关联。  
19.     CreateIoCompletionPort(pdlg->m_hDesFile,hIOCP,WRITE_KEY,0);//在关联时传入了完成键。可以根据完成键来区别从完成队列中取出的请求属于哪个设备。  
20.   
21.     OVERLAPPED ov={0};  
22.     PostQueuedCompletionStatus(hIOCP,0,WRITE_KEY,&ov);//发送模拟完成异步IO消息。  
23.     BYTE *pBuffer=new BYTE[BUFFERSIZE];  
24.     OVERLAPPED ovDes={0};  
25.     OVERLAPPED ovSrc={0};  
26.     while(true)  
27.     {  
28.         //memset(pBuffer,0,sizeof(pBuffer));  
29.         DWORD nTransfer;  
30.         OVERLAPPED *overlapped;  
31.         ULONG_PTR CompletionKey;  
32.         GetQueuedCompletionStatus(hIOCP,&nTransfer,&CompletionKey,(OVERLAPPED**)&overlapped,INFINITE);//IO完成队列没有请求项则挂起。否则从IO完成队列取出。  
33.         switch(CompletionKey)  
34.         {  
35.         case READ_KEY://从IO完成端口取出读完成。  
36.             {                                                                                       BOOL r=WriteFile(pdlg->m_hDesFile,pBuffer,overlapped->InternalHigh,NULL,&ovDes);  
37.   
38.                 ovDes.Offset+=BUFFERSIZE;  
39.             }  
40.             break;  
41.         case WRITE_KEY://从IO完成队列中取出写完成。  
42.             {  
43.                 memset(pBuffer,0,BUFFERSIZE);  
44.                 if(ovSrc.Offset<pdlg->m_SrcFileSize)  
45.                 {  
46.                     DWORD nBytes;  
47.                     if(ovSrc.Offset+BUFFERSIZE<pdlg->m_SrcFileSize)  
48.                         nBytes=BUFFERSIZE;  
49.                     else  
50.                         nBytes=pdlg->m_SrcFileSize-ovSrc.Offset;  
51.                     ReadFile(pdlg->m_hSrcFile,pBuffer,nBytes,NULL,&ovSrc);//异步IO忽略文件指针。所有对文件的定位操作由OVERLAPPED结构指定。  
52.                     //一定要注意为每次异步IO请求提供一个OVERLAPPED结构。刚才由于在接收和发送使用了  
53.                     //同一个OVERLAPPED结构，导致出现重叠 I/O 操作在进行中。错误代码：997  
54.                     ovSrc.Offset+=BUFFERSIZE;//OVERLAPPED的OffsetHigh结构必须每次都得设置。  
55.                 }  
56.                 else  
57.                 {  
58.                     ::MessageBox(NULL,TEXT("文件复制完成"),TEXT(""),MB_OK);  
59.                     return 0;  
60.                 }  
61.   
62.             }  
63.             break;  
64.         default:  
65.             break;  
66.         }  
67.     }  
68.     return 0;  
69. }</span>  

运行结果：