对 IOCP Content 大家是怎么管理的？

Question

希望听听行家的建议。

Answer 1

struct _IO_CONTEXT
{
WSAOVERLAPPED ol; // 重叠结构
WSABUF wsaBuffer; // WSASend/WSARecv使用的缓冲区
SOCKET sClient; // 客户端套接字
DWORD dwBytesSent; // 已经发送的字节数
DWORD dwBytesRecv; // 已经接收的字节数
DWORD dwIoFlag; // IO标志
char szBuffer[BUFFER_SIZE]; // 接收数据的缓冲区
};
int nResult = 0;
DWORD dwIosize = 0;
DWORD dwFlags = 0;
ZeroMemory(&lpIoContext->ol, sizeof(WSAOVERLAPPED));
nResult = WSARecv(
lpIoContext->sClient,
&lpIoContext->wsaBuffer,
1,
&dwIosize,
&dwFlags,
&lpIoContext->ol,
NULL
);
if((nResult==SOCKET_ERROR) && (WSAGetLastError()!=WSA_IO_PENDING))
{ 
return FALSE;
}return TRUE; 
lpIoContext就是一个new出来的_IO_CONTEXT， 当你要同时提交多个WSASend或者多个WSARecv的时候，为每一个WSASend或者WSARecv新new一个lpIoContext就行了。提供比较完整的代码：bool CIOCPServerTCP::InitializeIOCP()
{
DWORD i;
DWORD nThreadID;
SYSTEM_INFO systemInfo;// 创建完成端口
SOCKET s;s = socket(AF_INET, SOCK_STREAM, IPPROTO_IP);
if ( s == INVALID_SOCKET ) 
return false;m_hIOCP = CreateIoCompletionPort( (HANDLE)s, NULL, 0, 0 );
if ( m_hIOCP == NULL ) 
{
closesocket( s );
return false;
}
// Close the socket, we don't need it any longer.
closesocket( s );// create the work thread
// Determine how many processors are on the system.
GetSystemInfo( &systemInfo );
m_ProcessorNum = (short) systemInfo.dwNumberOfProcessors;
UINT nWorkerCnt = m_ProcessorNum * 2;// We need to save the Handles for Later Termination
HANDLE hWorker = NULL;for ( i = 0; i < nWorkerCnt; i++ ) 
{
hWorker = (HANDLE)CreateThread( NULL, 0,// Stack size - use default
ThreadPoolFunc, // Thread fn entry point
(void*) this, // Param for thread
0, // Init flag
&nThreadID); // Thread address
if (hWorker == NULL ) 
{
return false;
}
CloseHandle(hWorker);
}return true;
}bool CIOCPServerTCP::CreateListenSocket()
{
int nRet;m_SocketListen = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
if (m_SocketListen == INVALID_SOCKET)
{
return false;
}// bind the server
SOCKADDR_IN saServer; saServer.sin_family = AF_INET;
saServer.sin_port = htons(m_nPort);
saServer.sin_addr.s_addr = INADDR_ANY;// bind our name to the socket
nRet = bind( m_SocketListen, 
(LPSOCKADDR)&saServer, 
sizeof(struct sockaddr) );
if (nRet == SOCKET_ERROR)
{
closesocket(m_SocketListen);
return false;
}// Set the socket to listen
nRet = listen(m_SocketListen, m_MaxConnections);
if (nRet == SOCKET_ERROR)
{
closesocket(m_SocketListen);
return false;
}// beging the listening thread
DWORD dwThreadId = 0;m_hThread = (HANDLE)CreateThread( NULL, // Security
0, // Stack size - use default
ListenThreadProc, // Thread fn entry point
(void*) this, 
0, // Init flag
&dwThreadId); // Thread address
if (m_hThread == INVALID_HANDLE_VALUE)
{
return false;
}return true;
}DWORD WINAPI CIOCPServerTCP::ListenThreadProc( LPVOID lpVoid )
{
int nRet=0;
CIOCPServerTCP* pThis = reinterpret_cast<CIOCPServerTCP*>(lpVoid);SIOCPPerSocketTCPContext* lpPerSocketContext = NULL;
SOCKET SocketAccept = INVALID_SOCKET;DWORD dwRecvNumBytes, dwFlags=MSG_PARTIAL;while( TRUE ) 
{
// Loop forever accepting connections from clients 
// until console shuts down.
SocketAccept = WSAAccept(pThis->m_SocketListen, NULL, 
NULL, NULL, 0);
if( SocketAccept == SOCKET_ERROR ) 
{
return -1;
}// we add the just returned socket descriptor to 
// the IOCP along with its associated key data.
// also added context structures to a global list.
lpPerSocketContext = pThis->UpdateCompletionPort(SocketAccept, 
IOCPIORead, TRUE);
if( lpPerSocketContext == NULL )
return -1;// post initial receive on this socket
nRet = WSARecv( SocketAccept, 
&(lpPerSocketContext->IOContext.wsaInBuf), 
1, &dwRecvNumBytes, &dwFlags,
&(lpPerSocketContext->IOContext.Overlapped),
NULL );int error = WSAGetLastError();
if( nRet == SOCKET_ERROR 
&& (ERROR_IO_PENDING != WSAGetLastError()) ) 
{
pThis->ClientContentClose( lpPerSocketContext, FALSE );
}
} //whilereturn 0;
}DWORD WINAPI CIOCPServerTCP::ThreadPoolFunc( LPVOID WorkContext )
{
CIOCPServerTCP* pThis = reinterpret_cast<CIOCPServerTCP*>(WorkContext);
HANDLE hIOCP = pThis->m_hIOCP;SIOCPPerSocketTCPContext* lpPerSocketContext = NULL;
SIOCPPerIoTCPContext* llpIOContext = NULL;
LPWSAOVERLAPPED lpOverlapped = NULL;SPacketServerRecvTCP* lpPacketTCPTemp = NULL;BOOL bSuccess = FALSE;
int nRet = 0;DWORD dwRecvNumBytes = 0;
DWORD dwSendNumBytes = 0;
DWORD dwFlags = 0;
DWORD dwIoSize = 0;uint4 nDataSize=0;
SPacketServerRecvTCP *lpPacketTCP=NULL;
while( true ) 
{
// continually loop to service io completion packets
bSuccess = GetQueuedCompletionStatus( hIOCP, &dwIoSize,
(LPDWORD) &lpPerSocketContext,
if( !lpPerSocketContext ) 
continue;// determine what type of IO packet has completed
llpIOContext = (SIOCPPerIoTCPContext*)lpOverlapped;
if( !llpIOContext ) 
continue;switch( llpIOContext->IOOperation ) 
{
// a read operation has completed
case IOCPIORead:
// 注意，连接突然中断时要从等待接收队列中去除该等待项
if( dwIoSize == 0 )
{
// 去除该等待项并关闭连接
pThis->m_pTCPSocketIDToPacketRecvTCPMapWait->RemoveKey( lpPacketTCPTemp->nSocketID );
pThis->ClientContentClose( lpPerSocketContext, FALSE );
continue;
}

Answer 2

// 接收数据处理// continue to recv
nRet = WSARecv( lpPerSocketContext->Socket, 
&(lpPerSocketContext->IOContext.wsaInBuf), 
1, &dwRecvNumBytes, &dwFlags,
&(lpPerSocketContext->IOContext.Overlapped),
NULL );
if( nRet == SOCKET_ERROR 
&& (ERROR_IO_PENDING != WSAGetLastError()) ) 
{
// 关闭连接
pThis->ClientContentClose(lpPerSocketContext, FALSE);
}
break;
// a write operation has completed
case IOCPIOWrite:if( dwIoSize <= 0 )
{
// 释放内存
if( lpPerSocketContext->IOContext.wsaOutBuf.buf )
{
// 释放发送空间
pThis->m_pMemoryPoolLinearMul->PutMem( lpPerSocketContext->IOContext.wsaOutBuf.buf, 
lpPerSocketContext->IOContext.wsaOutBuf.len );
}
if( nRet != 0 )
{
// 去除该等待项
pThis->m_pTCPSocketIDToPacketRecvTCPMapWait->RemoveKey( lpPacketTCPTemp->nSocketID );
}
// 关闭连接
pThis->ClientContentClose( lpPerSocketContext, FALSE );
continue;
}
else
{
// 发送完成}break;default:
break;
} //switch
} //whilereturn 0;
}SIOCPPerSocketTCPContext* CIOCPServerTCP::UpdateCompletionPort(SOCKET sd, 
IOCP_IO_OPERATION ClientIo, BOOL bAddToList) 
{
SIOCPPerSocketTCPContext* lpPerSocketTCPContext = NULL;lpPerSocketTCPContext = ClientContentAlloc(sd, ClientIo);
if( lpPerSocketTCPContext == NULL )
return(NULL);// 初始化设备上下文
ZeroMemory( lpPerSocketTCPContext, sizeof(SIOCPPerSocketTCPContext) );// 将连接的socket与对应的设备上下文与完成端口句柄关联起来。
HANDLE hIOCPTemp = NULL;
hIOCPTemp = CreateIoCompletionPort( (HANDLE)sd, 
m_hIOCP, (DWORD )lpPerSocketTCPContext, 0 );
if( !hIOCPTemp ) 
{
ClientContentFree( lpPerSocketTCPContext );
return (NULL);
}// 获得TCP连接的唯一编号
bool bContinue = true;
int nSocketID = 0;
SIOCPPerSocketTCPContext* lpTempContext;
while( bContinue )
{
nSocketID = m_rSysCoreServerStatus.GetTCPSocketID();
bContinue = m_pTCPSocketIDToTCPContextMap->Lookup( nSocketID, lpTempContext );
}lpPerSocketTCPContext->nSocketID = nSocketID;
lpPerSocketTCPContext->Socket = sd;
lpPerSocketTCPContext->IOContext.wsaInBuf.buf = lpPerSocketTCPContext->IOContext.szBuffer;
lpPerSocketTCPContext->IOContext.wsaInBuf.len = sizeof( lpPerSocketTCPContext->IOContext.szBuffer );
lpPerSocketTCPContext->IOContext.IOOperation = IOCPIORead;// 将其加入TCP连接编号与设备上下文映射
if( bAddToList ) 
m_pTCPSocketIDToTCPContextMap->SetAt( lpPerSocketTCPContext->nSocketID, lpPerSocketTCPContext );return lpPerSocketTCPContext;
}// Close down a connection with a client.
bool CIOCPServerTCP::ClientContentClose ( SIOCPPerSocketTCPContext* lpPerSocketContext,
BOOL bGraceful ) 
{
// 取消所有IO操作
CancelIo( (HANDLE)lpPerSocketContext->Socket);// 从客户连接链表中删除该项
m_pTCPSocketIDToTCPContextMap->RemoveKey( lpPerSocketContext->nSocketID );// 关闭连接
closesocket( lpPerSocketContext->Socket );
lpPerSocketContext->Socket = INVALID_SOCKET;// 释放内存
ClientContentFree( lpPerSocketContext );return true;
} // 客户信息结构分配
SIOCPPerSocketTCPContext* CIOCPServerTCP::ClientContentAlloc(SOCKET sd, 
IOCP_IO_OPERATION ClientIO)
{
SIOCPPerSocketTCPContext* lpTemp = new SIOCPPerSocketTCPContext(); 
memset( lpTemp, 0, sizeof(SIOCPPerSocketTCPContext) );
return lpTemp;
}