套接字select异步模型

Select 模型原理

利用 select 函数，判断套接字上是否存在数据，或者能否向一个套接字写入数据。目的是防止应用程序在套接字处于锁定模式时，调用 recv （或 send ）从没有数据的套接字上接收数据，被迫进入阻塞状态。

 

select 参数和返回值意义如下：

int  select   (

  IN  int   nfds,                              //0, 无意义

  IN OUT  fd_set*   readfds,       // 检查可读性

  IN OUT  fd_set*   writefds,       // 检查可写性

  IN OUT fd_set*  exceptfds,   // 例外数据

  IN const  struct timeval*   timeout);      // 函数的返回时间

 

struct   timeval   {

         long     tv_sec;         // 秒

         long     tv_usec;      // 毫秒

};

select 返回 fd_set 中可用的套接字个数。

 

fd_set 是一个 SOCKET 队列 ，以下宏可以对该队列进行操作：

FD_CLR( s, *set)  从队列 set 删除句柄 s;

FD_ISSET( s, *set)   检查句柄 s 是否存在与队列 set 中 ;

FD_SET( s, *set ) 把句柄 s 添加到队列 set 中 ;

FD_ZERO( *set )   把 set 队列初始化成空队列 .

 

Select 工作流程

1 ：用 FD_ZERO 宏来初始化我们感兴趣的 fd_set 。

也就是 select 函数的第二三四个参数。

2 ：用 FD_SET 宏来将套接字句柄分配给相应的 fd_set 。

如果想要检查一个套接字是否有数据需要接收，可以用 FD_SET 宏把套接接字句柄加入可读性检查队列中

3 ：调用 select 函数。

如果该套接字没有数据需要接收， select 函数会把该套接字从可读性检查队列中删除掉，

4 ：用 FD_ISSET 对套接字句柄进行检查 。

如果我们所关注的那个套接字句柄仍然在开始分配的那个 fd_set 里，那么说明马上可以进行相应的 IO 操   作。比如一个分配给 select 第一个参数的套接字句柄在 select 返回后仍然在 select 第一个参数的 fd_set 里，那么说明当前数据已经来了，   马上可以读取成功而不会被阻塞。

 

 

/*************************************************************************************/

#include "stdafx.h"  

#include <winsock.h>  

#include <stdio.h>  

#define PORT   5150  

#define MSGSIZE   1024  

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

int g_iTotalConn = 0;  

SOCKET g_CliSocketArr[FD_SETSIZE];  

DWORD WINAPI WorkerThread(LPVOID lpParam);  

int main(int argc, char* argv[])  

{  

     WSADATA wsaData;  

     SOCKET sListen, sClient;  

     SOCKADDR_IN local, client;  

     int iAddrSize = sizeof(SOCKADDR_IN);  

     DWORD dwThreadId;  

      // Initialize windows socket library  

     WSAStartup(0x0202, &wsaData);  

     // Create listening socket  

     sListen = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);  

      // Bind  

     local.sin_family = AF_INET;  

     local.sin_addr.S_un.S_addr = htonl(INADDR_ANY);  

     local.sin_port = htons(PORT);  

     bind(sListen, (sockaddr*)&local, sizeof(SOCKADDR_IN));  

     // Listen  

     listen(sListen, 3);  

     // Create worker thread  

     CreateThread(NULL, 0, WorkerThread, NULL, 0, &dwThreadId);   

     while (TRUE)   

     {  

          // Accept a connection  

         sClient = accept(sListen, (sockaddr*)&client, &iAddrSize);  

         printf("Accepted client:%s:%d\n", inet_ntoa(client.sin_addr), ntohs(client.sin_port));  

          // Add socket to g_CliSocketArr  

         g_CliSocketArr[g_iTotalConn++] = sClient;   

     }  

     return 0;  

}  

DWORD WINAPI WorkerThread(LPVOID lpParam)  

{  

     int i;  

     fd_set fdread;  

     int ret;  

     struct timeval tv = {1, 0};  

     char szMessage[MSGSIZE];  

     while (TRUE)   

     {  

         FD_ZERO(&fdread);      //1 清空队列

         for (i = 0; i < g_iTotalConn; i++)   

         {  

             FD_SET(g_CliSocketArr[i], &fdread);      //2 将要检查的套接口加入队列

         }  

          // We only care read event  

          ret = select(0, &fdread, NULL, NULL, &tv);     //3 查询满足要求的套接字，不满足要求，出队

         if (ret == 0)   

         {  

              // Time expired  

             continue;  

         }  

         for (i = 0; i < g_iTotalConn; i++)   

         {  

             if (FD_ISSET(g_CliSocketArr[i], &fdread))       //4. 是否依然在队列

             {  

                  // A read event happened on g_CliSocketArr  

                 ret = recv(g_CliSocketArr[i], szMessage, MSGSIZE, 0);  

                 if (ret == 0 || (ret == SOCKET_ERROR && WSAGetLastError() == WSAECONNRESET))   

                 {  

                      // Client socket closed  

                     printf("Client socket %d closed.\n", g_CliSocketArr[i]);  

                     closesocket(g_CliSocketArr[i]);  

                     if (i < g_iTotalConn-1)   

                     {  

                         g_CliSocketArr[i--] = g_CliSocketArr[--g_iTotalConn];  

                     }  

                 }   

                 else   

                 {  

                      // We reveived a message from client  

                     szMessage[ret] = '\0';  

                     send(g_CliSocketArr[i], szMessage, strlen(szMessage), 0);  

                 }  

             }  

         }  

     }  

} 

服务器的几个主要动作如下：

1. 创建监听套接字，绑定，监听；

2. 创建工作者线程；

3. 创建一个套接字数组，用来存放当前所有活动的客户端套接字，每 accept 一个连接就更新一次数组；

4. 接受客户端的连接。

 

这里有一点需要注意的，就是我没有重新定义 FD_SETSIZE 宏，所以服务器最多支持的并发连接数为 64 。而且，这里决不能无条件的 accept, 服务器应该根据当前的连接数来决定是否接受来自某个客户端的连接。一种比较好的实现方案就是采用 WSAAccept 函数，而且让 WSAAccept 回调自己实现的 Condition Function 。如下所示：

 

int CALLBACK ConditionFunc(LPWSABUF lpCallerId,LPWSABUF lpCallerData, LPQOS lpSQOS,LPQOS lpGQOS,LPWSABUF lpCalleeId, LPWSABUF lpCalleeData,GROUP FAR * g,DWORD dwCallbackData)

{

if ( 当前连接数   < FD_SETSIZE)

   return CF_ACCEPT;

else

   return CF_REJECT;

}

 

工作者线程里面是一个死循环，一次循环完成的动作是：

1. 将当前所有的客户端套接字加入到读集 fdread 中；

2. 调用 select 函数；

3. 查看某个套接字是否仍然处于读集中，如果是，则接收数据。如果接收的数据长度为 0 ，或者发生 WSAECONNRESET 错误，则表示客户端套接字主动关闭，这时需要将服务器中对应的套接字所绑定的资源释放掉，然后调整我们的套接字数组（将数组中最后一个套接字挪到当前的位置上）

 

除了需要有条件接受客户端的连接外，还需要在连接数为 0 的情形下做特殊处理，因为如果读集中没有任何套接字， select 函数会立刻返回，这将导致工作者线程成为一个毫无停顿的死循环， CPU 的占用率马上达到 100% 。

 

关系到套接字列表的操作都需要使用循环 , 在轮询的时候 , 需要遍历一次 , 再新的一轮开始时 , 将列表加入队列又需要遍历一次 . 也就是说 ,Select 在工作一次时 , 需要至少遍历 2 次列表 , 这是它效率较低的原因之一 . 在大规模的网络连接方面 , 还是推荐使用 IOCP 或 EPOLL 模型 . 但是 Select 模型可以使用在诸如对战类游戏上 , 比如类似星际这种 , 因为它小巧易于实现 , 而且对战类游戏的网络连接量并不大 .

 

对于 Select 模型想要突破 Windows 64 个限制的话 , 可以采取分段轮询 , 一次轮询 64 个 . 例如套接字列表为 128 个 , 在第一次轮询时 , 将前 64 个放入队列中用 Select 进行状态查询 , 待本次操作全部结束后 . 将后 64 个再加入轮询队列中进行轮询处理 . 这样处理需要在非阻塞式下工作 . 以此类推 ,Select 也能支持无限多个 .