现在的位置: 首页 > 综合 > 正文

IOCP的程序

2017年12月11日 ⁄ 综合 ⁄ 共 14682字 ⁄ 字号 小 中 大 ⁄ 评论关闭
C代码  收藏代码
  1. #include <winsock2.h>  
  2. #include <mswsock.h>  
  3. #include <windows.h>  
  4. #include <stdio.h>  
  5. #include <stdlib.h>  
  6. #include <assert.h>  
  7. #include "vld.h"  
  8.   
  9. #pragma message("automatic link to ws2_32.lib and mswsock.lib")  
  10. #pragma comment(lib, "ws2_32.lib")  
  11. #pragma comment(lib, "mswsock.lib")  
  12.   
  13.   
  14.   
  15.   
  16. #define RETURN_SUCESS (0)  
  17. #define RETURN_FAILED (-1)  
  18.   
  19. #define PORT 5150//端口  
  20. #define IOCP_THREAD_MAX 16//最大的线程数  
  21. #define DATA_BUFSIZE 8192  
  22.   
  23.   
  24.   
  25. //区别是何种完成事件  
  26. enum IO_EVENT  
  27. {  
  28.     IO_EVENT_ACCEPT,  
  29.     IO_EVENT_WSARECV,  
  30.     IO_EVENT_WSARECVFROM,  
  31.     //不支持异步发送  
  32.     //IO_EVENT_WSASEND,  
  33.     //IO_EVENT_WSASENDTO  
  34. };  
  35.   
  36.   
  37.   
  38.   
  39. typedef struct overlapped_wrapper  
  40. {  
  41.     //OVERLAPPED要放在第一个  
  42.     OVERLAPPED overlapped;  
  43.     int io_type;//指示是何种IO操作  
  44. }overlapped_wrapper;  
  45.   
  46.   
  47.   
  48. typedef struct acceptex_block  
  49. {  
  50.     //OVERLAPPED要放在第一个  
  51.     OVERLAPPED overlapped;  
  52.     int io_type;//指示是何种IO操作  
  53.   
  54.     char buffer[DATA_BUFSIZE];  
  55.     SOCKET listen_socket;  
  56.     SOCKET accept_socket;  
  57. }acceptex_block;  
  58.   
  59.   
  60.   
  61. typedef struct recv_block  
  62. {  
  63.     //OVERLAPPED要放在第一个  
  64.     OVERLAPPED overlapped;  
  65.     int io_type;//指示是何种IO操作  
  66.   
  67.     char buffer[DATA_BUFSIZE];  
  68.     SOCKET socket;  
  69.     WSABUF wsa_recv_buf;  
  70.     DWORD bytes_recveived;  
  71. }recv_block;  
  72.   
  73.   
  74.   
  75. typedef struct recvfrom_block  
  76. {  
  77.     //OVERLAPPED要放在第一个  
  78.     OVERLAPPED overlapped;  
  79.     int io_type;//指示是何种IO操作  
  80.   
  81.     char buffer[DATA_BUFSIZE];  
  82.     SOCKET socket;  
  83.     WSABUF wsa_recv_buf;  
  84.     DWORD bytes_recveived;  
  85.   
  86.     //UDP包的源地址  
  87.     struct sockaddr_in from_address;  
  88.     int from_address_len;  
  89. }recvfrom_block;  
  90.   
  91.   
  92.   
  93.   
  94.   
  95.   
  96. int get_cpu_number();  
  97.   
  98. int async_AcceptEx(acceptex_block* block);  
  99. int async_WSARecv(recv_block* block);  
  100. int async_WSARecvFrom(recvfrom_block* block);  
  101.   
  102. void on_acceptex(acceptex_block* block);  
  103. void on_recv(recv_block* block);  
  104. void on_recvfrom(recvfrom_block* block);  
  105. void on_tcp_listen_close(acceptex_block* block);  
  106. void on_tcp_close(recv_block* block);  
  107. void on_udp_close(recvfrom_block* block);  
  108.   
  109. int init(void);  
  110. void uninit(void);  
  111. DWORD WINAPI worker_thread(LPVOID CompletionPortID);  
  112. void exit_error();  
  113.   
  114.   
  115.   
  116.   
  117.   
  118. //完成端口的句柄  
  119. HANDLE g_completion_port = INVALID_HANDLE_VALUE;  
  120. //工作线程句柄  
  121. HANDLE g_threads[IOCP_THREAD_MAX];  
  122. //工作线程数量  
  123. int g_threads_number = 0;  
  124.   
  125.   
  126. int main(void)  
  127. {  
  128.     /************************************************************************/  
  129.     /*TCP的例子*/  
  130.     /************************************************************************/  
  131.     SOCKADDR_IN internet_address;  
  132.     SOCKET listen_socket;  
  133.     acceptex_block* block;  
  134.   
  135.     if(RETURN_FAILED == init())  
  136.         exit_error();  
  137.   
  138.     if ((listen_socket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) == INVALID_SOCKET)  
  139.         exit_error();  
  140.   
  141.     internet_address.sin_family = AF_INET;  
  142.     internet_address.sin_addr.s_addr = htonl(INADDR_ANY);  
  143.     internet_address.sin_port = htons(PORT);  
  144.   
  145.     if (bind(listen_socket, (PSOCKADDR) &internet_address, sizeof(internet_address)) == SOCKET_ERROR)  
  146.         exit_error();  
  147.   
  148.     if (listen(listen_socket, SOMAXCONN) == SOCKET_ERROR)  
  149.         exit_error();  
  150.   
  151.     printf("listening socket %d\n", PORT);  
  152.   
  153.     //把监听的SOCKET和完成端口绑定  
  154.     if(NULL == CreateIoCompletionPort((HANDLE)listen_socket, g_completion_port, (u_long)listen_socket, 0))  
  155.         exit_error();  
  156.   
  157.     block = (acceptex_block*)malloc(sizeof(acceptex_block));  
  158.     block->listen_socket = listen_socket;  
  159.     async_AcceptEx(block);  
  160.     getchar();  
  161.     closesocket(listen_socket);  
  162.     getchar();  
  163.     uninit();  
  164.     return 0;  
  165.   
  166.   
  167.   
  168.     /************************************************************************/  
  169.     /*UDP的例子*/  
  170.     /************************************************************************/  
  171.     //SOCKADDR_IN internet_address;  
  172.     //SOCKET sock;  
  173.     //recvfrom_block* block = (recvfrom_block*)malloc(sizeof(recvfrom_block));  
  174.   
  175.     //if(RETURN_FAILED == init())  
  176.     //  exit_error();  
  177.   
  178.     //if ((sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) == INVALID_SOCKET)  
  179.     //  exit_error();  
  180.   
  181.     //internet_address.sin_family = AF_INET;  
  182.     //internet_address.sin_addr.s_addr = htonl(INADDR_ANY);  
  183.     //internet_address.sin_port = htons(PORT);  
  184.   
  185.     //if (bind(sock, (PSOCKADDR) &internet_address, sizeof(internet_address)) == SOCKET_ERROR)  
  186.     //  exit_error();  
  187.   
  188.     //if(NULL == CreateIoCompletionPort((HANDLE)sock, g_completion_port, (u_long)sock, 0))  
  189.     //  exit_error();  
  190.   
  191.     //block->socket = sock;  
  192.     //async_WSARecvFrom(block);  
  193.   
  194.   
  195.     //getchar();  
  196.     //closesocket(sock);  
  197.     //getchar();  
  198.     //uninit();  
  199.     //return 0;  
  200. }  
  201.   
  202.   
  203.   
  204.   
  205. int init(void)  
  206. {  
  207.     WSADATA wsa_data;  
  208.     int i;  
  209. #if defined _DEBUG || defined DEBUG  
  210.     //调试时用一个线程方便  
  211.     int threads = 1;  
  212. #else  
  213.     int threads = get_cpu_number();  
  214. #endif  
  215.   
  216.   
  217.     if (WSAStartup(0x0202, &wsa_data) != 0)  
  218.         return RETURN_FAILED;  
  219.   
  220.     //建立完成端口  
  221.     if ((g_completion_port = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0)) == NULL)  
  222.         return RETURN_FAILED;  
  223.   
  224.     if(threads > IOCP_THREAD_MAX)  
  225.         threads = IOCP_THREAD_MAX;  
  226.     for(i = 0; i < threads; i++)  
  227.     {  
  228.         //创建工作线程  
  229.         g_threads[g_threads_number++] = CreateThread(NULL, 0, worker_thread, NULL, 0, 0);  
  230.     }  
  231.     return RETURN_SUCESS;  
  232. }  
  233.   
  234.   
  235.   
  236. void uninit(void)  
  237. {  
  238.     //自定义的退出协议,三个参数全为0时退出  
  239.     PostQueuedCompletionStatus(g_completion_port, 0, 0, NULL);  
  240.   
  241.     WaitForMultipleObjects(g_threads_number, g_threads, TRUE, INFINITE);  
  242.   
  243.     CloseHandle(g_completion_port);  
  244.   
  245.     WSACleanup();  
  246. }  
  247.   
  248.   
  249. int get_cpu_number()  
  250. {  
  251.     SYSTEM_INFO system_info;  
  252.     GetSystemInfo(&system_info);  
  253.     return system_info.dwNumberOfProcessors;  
  254. }  
  255.   
  256.   
  257. void exit_error()  
  258. {  
  259.     int error = GetLastError();  
  260.     if (error == 0)  
  261.     {  
  262.         exit(RETURN_SUCESS);  
  263.     }  
  264.     else  
  265.     {  
  266.         fprintf(stderr, "error:%d\n", error);  
  267.         exit(RETURN_FAILED);  
  268.     }  
  269. }  
  270.   
  271.   
  272.   
  273. /* 
  274. 投递一次AcceptEx请求 
  275. 返回TRUE，成功 
  276. 返回FALSE，失败，WSAGetLastError()获取进一步信息 
  277. */  
  278. int async_AcceptEx(acceptex_block* block)  
  279. {  
  280.     DWORD address_length;  
  281.     DWORD bytes_received;  
  282.   
  283.     //准备投递一个异步接受请求  
  284.     SOCKET accept_socket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);  
  285.     if(accept_socket == INVALID_SOCKET)  
  286.         return RETURN_FAILED;  
  287.   
  288.     block->io_type = IO_EVENT_ACCEPT;  
  289.     block->accept_socket = accept_socket;  
  290.     memset(&block->overlapped, 0,   
  291.         sizeof(block->overlapped));  
  292.   
  293.   
  294.     address_length = sizeof(struct sockaddr_in) + 16;  
  295.     if(!AcceptEx(  
  296.         block->listen_socket,   
  297.         accept_socket,  
  298.         block->buffer,//这个参数会传递给完成端口  
  299.         0,//DATA_BUFSIZE - i * 2,       //传0进来，接受的连接的时候不接受数据  
  300.         address_length, address_length,  
  301.         &bytes_received,   
  302.         &block->overlapped))  
  303.     {  
  304.         if(WSAGetLastError() != ERROR_IO_PENDING)  
  305.             goto fail;  
  306.     }  
  307.     return RETURN_SUCESS;  
  308.   
  309. fail:  
  310.     closesocket(accept_socket);  
  311.     return RETURN_FAILED;  
  312. }  
  313.   
  314.   
  315.   
  316.   
  317.   
  318.   
  319.   
  320.   
  321. /* 
  322. 投递一次WSARecv请求 
  323. 返回TRUE，表示接受请求投递成功 
  324. 返回FALSE，表示连接已经断开，WSAGetLastError()获取进一步信息 
  325. */  
  326. int async_WSARecv(recv_block* block)  
  327. {  
  328.     int ret;  
  329.     DWORD recv_bytes, flags = 0;  
  330.   
  331.     block->io_type = IO_EVENT_WSARECV;  
  332.     block->bytes_recveived = 0;  
  333.     block->wsa_recv_buf.len = DATA_BUFSIZE;  
  334.     block->wsa_recv_buf.buf = block->buffer;  
  335.     memset(&(block->overlapped), 0, sizeof(block->overlapped));  
  336.   
  337.   
  338.     //投递一次接受请求  
  339.     ret = WSARecv(block->socket, &block->wsa_recv_buf, 1, &recv_bytes, &flags,  
  340.         &(block->overlapped), NULL);  
  341.   
  342.     if(ret == -1 && WSAGetLastError() != ERROR_IO_PENDING)  
  343.     {  
  344.         printf("WSARecv() error returns %d\n", ret);  
  345.         on_tcp_close(block);  
  346.         return RETURN_FAILED;  
  347.     }  
  348.     else if((ret == 0) || (ret == -1 && WSAGetLastError() == ERROR_IO_PENDING))  
  349.     {  
  350.         //waiting... for next turn  
  351.     }  
  352.     else  
  353.     {  
  354.         //如果直接收到了数据  
  355.         block->bytes_recveived = ret;  
  356.         on_recv(block);  
  357.         //递归  
  358.         return async_WSARecv(block);  
  359.     }  
  360.   
  361.     return RETURN_SUCESS;  
  362. }  
  363.   
  364.   
  365.   
  366.   
  367.   
  368. /* 
  369. 投递一次WSARecvFrom请求 
  370. 返回TRUE，表示接受请求投递成功 
  371. 返回FALSE，表示接受请求投递失败，WSAGetLastError()获取进一步信息 
  372. */  
  373. int async_WSARecvFrom(recvfrom_block* block)  
  374. {  
  375.     int ret;  
  376.     DWORD recv_bytes = 0, flags = 0;  
  377.   
  378.     block->io_type = IO_EVENT_WSARECVFROM;  
  379.     block->bytes_recveived = 0;  
  380.     block->wsa_recv_buf.len = DATA_BUFSIZE;  
  381.     block->wsa_recv_buf.buf = block->buffer;  
  382.     memset(&block->from_address, 0, sizeof(block->from_address));  
  383.     block->from_address_len = sizeof(block->from_address);  
  384.     memset(&(block->overlapped), 0, sizeof(block->overlapped));  
  385.   
  386.   
  387.     //投递一次接受请求  
  388.     ret = WSARecvFrom(block->socket, &block->wsa_recv_buf, 1, &recv_bytes, &flags,  
  389.         (struct sockaddr*)&block->from_address, &block->from_address_len,  
  390.         &(block->overlapped), NULL);  
  391.   
  392.     if(ret == -1 && WSAGetLastError() != ERROR_IO_PENDING)  
  393.     {  
  394.         printf("WSARecvFrom() error returns %d %d\n", ret, WSAGetLastError());  
  395.         on_udp_close(block);  
  396.         return RETURN_FAILED;  
  397.     }  
  398.     else if((ret == 0) || (ret == -1 && WSAGetLastError() == ERROR_IO_PENDING))  
  399.     {  
  400.         //waiting... for next turn  
  401.     }  
  402.     else  
  403.     {  
  404.         //如果直接收到了数据  
  405.         block->bytes_recveived = ret;  
  406.         on_recvfrom(block);  
  407.         //递归  
  408.         return async_WSARecvFrom(block);  
  409.     }  
  410.   
  411.     return RETURN_SUCESS;  
  412. }  
  413.   
  414.   
  415. void on_acceptex(acceptex_block* block)  
  416. {  
  417.     DWORD i;  
  418.     struct sockaddr *p_local_addr;  
  419.     int local_addr_len = sizeof(struct sockaddr_in);  
  420.     struct sockaddr *p_remote_addr;  
  421.     int remote_addr_len = sizeof(struct sockaddr_in);  
  422.     struct sockaddr_in *p_v4_addr;  
  423.   
  424.     recv_block* r_block;  
  425.   
  426.   
  427.     printf("on_acceptex %d\n", block->accept_socket);  
  428.   
  429.     i = sizeof(struct sockaddr_in) + 16;  
  430.     GetAcceptExSockaddrs(  
  431.         block->buffer,  
  432.         0,//DATA_BUFSIZE - i * 2,  
  433.         i, i,  
  434.         &p_local_addr,  
  435.         &local_addr_len,  
  436.         &p_remote_addr,  
  437.         &remote_addr_len  
  438.         );  
  439.   
  440.     p_v4_addr = (struct sockaddr_in *)p_local_addr;  
  441.     printf("\t本地地址%s:%d\n",   
  442.         inet_ntoa(p_v4_addr->sin_addr), ntohs(p_v4_addr->sin_port));  
  443.     p_v4_addr = (struct sockaddr_in *)p_remote_addr;  
  444.     printf("\t远程地址%s:%d\n",   
  445.         inet_ntoa(p_v4_addr->sin_addr), ntohs(p_v4_addr->sin_port));  
  446.   
  447.   
  448.   
  449.     //准备投递一次WSARecv请求  
  450.     r_block = (recv_block*)malloc(sizeof(recv_block));  
  451.     r_block->socket = block->accept_socket;  
  452.   
  453.     //绑定  
  454.     CreateIoCompletionPort((HANDLE)r_block->socket,   
  455.         g_completion_port, (u_long)r_block->socket, 0);  
  456.   
  457.     //投递一次接受请求  
  458.     async_WSARecv(r_block);  
  459.   
  460.     //继续投递AcceptEx请求  
  461.     async_AcceptEx(block);  
  462. }  
  463.   
  464. void on_recv(recv_block* block)  
  465. {  
  466.     printf("on_recv %d, 收到%d bytes数据\n", block->socket, block->bytes_recveived);  
  467.   
  468.     async_WSARecv(block);  
  469. }  
  470.   
  471.   
  472. void on_recvfrom(recvfrom_block* block)  
  473. {  
  474.     printf("on_recvfrom %d, 收到%d bytes数据, 来自%s:%d\n",  
  475.         block->socket,  
  476.         block->bytes_recveived,  
  477.         inet_ntoa(block->from_address.sin_addr),  
  478.         ntohs(block->from_address.sin_port));  
  479.   
  480.     async_WSARecvFrom(block);  
  481. }  
  482.   
  483.   
  484. void on_tcp_listen_close(acceptex_block* block)  
  485. {  
  486.     printf("on_tcp_listen_close %d\n", block->accept_socket);  
  487.     free(block);  
  488.     closesocket(block->accept_socket);  
  489. }  
  490.   
  491.   
  492. void on_tcp_close(recv_block* block)  
  493. {  
  494.     printf("on_tcp_close %d\n", block->socket);  
  495.     free(block);  
  496.     closesocket(block->socket);  
  497. }  
  498.   
  499.   
  500. void on_udp_close(recvfrom_block* block)  
  501. {  
  502.     printf("on_udp_close %d\n", block->socket);  
  503.     free(block);  
  504.     closesocket(block->socket);  
  505. }  
  506.   
  507.   
  508.   
  509. DWORD WINAPI worker_thread(LPVOID nothing)  
  510. {  
  511.     DWORD bytes;  
  512.     overlapped_wrapper* over_type;  
  513.     BOOL close_socket = FALSE;  
  514.     BOOL ret;  
  515.   
  516.     UNREFERENCED_PARAMETER(nothing);  
  517.   
  518.     for(;;)  
  519.     {  
  520.         SOCKET socket;  
  521.         //注意第三个参数，他是CreateIoCompletionPort时传入的，直接传入的是一个SOCKET  
  522.         //注意第四个参数，他可能是一个recv_block或acceptex_block结构的指针  
  523.         //因为OVERLAPPED是PER_IO_OPERATION_DATA的第一个成员，所以可以安全的进行转换  
  524.         ret = GetQueuedCompletionStatus(g_completion_port, &bytes,  
  525.             (LPDWORD)&socket, (LPOVERLAPPED *) &over_type, INFINITE);  
  526.   
  527.   
  528.         if(ret == ERROR_SUCCESS)  
  529.         {  
  530.             DWORD last_error = GetLastError();  
  531.   
  532.             if(ERROR_INVALID_HANDLE == last_error)  
  533.             {  
  534.                 printf("完成端口被关闭,退出\n");  
  535.                 return 0;  
  536.             }  
  537.             else if(ERROR_NETNAME_DELETED == last_error  
  538.                 || ERROR_OPERATION_ABORTED == last_error)  
  539.             {  
  540.                 printf("socket被关闭 或者 操作被取消\n");  
  541.                 close_socket = TRUE;  
  542.             }  
  543.             else  
  544.             {  
  545.                 printf("GetLastError %d\n", last_error);  
  546.                 continue;  
  547.             }  
  548.         }  
  549.         //自定义的退出协议,三个参数全为0时退出(见uninit中的PostQueuedCompletionStatus)  
  550.         else if(bytes == 0 && socket == 0 && over_type == NULL)  
  551.         {  
  552.             return 0;  
  553.         }  
  554.   
  555.         assert(over_type);  
  556.   
  557.         switch(over_type->io_type)  
  558.         {  
  559.         case IO_EVENT_ACCEPT:  
  560.             {  
  561.                 acceptex_block* a_block = (acceptex_block*)over_type;  
  562.   
  563.                 if(close_socket)  
  564.                 {  
  565.                     on_tcp_listen_close(a_block);  
  566.                 }  
  567.                 else  
  568.                 {  
  569.                     on_acceptex(a_block);  
  570.                 }  
  571.             }  
  572.             break;  
  573.   
  574.         case IO_EVENT_WSARECV:  
  575.             {  
  576.                 recv_block* r_block = (recv_block*)over_type;  
  577.                 //连接断开  
  578.                 if (close_socket || bytes == 0 || bytes == -1)  
  579.                 {  
  580.                     //测试一下，确定对方肯定关闭连接了  
  581.                     char test_close;  
  582.                     int r = recv(r_block->socket, &test_close, sizeof(test_close), MSG_PEEK);  
  583.                     if(r == 0 || r == -1)  
  584.                     {  
  585.                         on_tcp_close(r_block);  
  586.                     }  
  587.                 }  
  588.                 //收到了bytes字节的数据  
  589.                 else  
  590.                 {  
  591.                     //处理数据  
  592.                     r_block->bytes_recveived = bytes;  
  593.                     on_recv(r_block);  
  594.                 }  
  595.             }  
  596.             break;  
  597.   
  598.   
  599.         case IO_EVENT_WSARECVFROM:  
  600.             {  
  601.                 recvfrom_block* rf_block = (recvfrom_block*)over_type;  
  602.   
  603.                 if(close_socket || bytes == -1 || bytes == 0)  
  604.                 {  
  605.                     on_udp_close(rf_block);  
  606.                 }  
  607.                 else  
  608.                 {  
  609.                     //处理数据  
  610.                     rf_block->bytes_recveived = bytes;  
  611.                     on_recvfrom(rf_block);  
  612.                 }  
  613.             }  
  614.             break;  
  615.   
  616.   
  617.         default:  
  618.             break;  
  619.         }  
  620.     }  
  621.     return 0;  
  622. } 

返回
【上篇】
【下篇】

作者:

抱歉!评论已关闭.

返回首页

Copyright © 2013-2018 学步园  保留所有权利.
软文销售 QQ客服：2265327166 点击这里给我发消息（其他合作也可洽谈）
必威体育
必威电竞