Since all of our IO operations are operating aynchronously they return imediately to the calling code. The actual implementation of these operations is made slightly more complex by the fact that any outstanding IO requests are terminated when the thread that issued those requests exits. Since we wish to ensure that our IO requests are not terminated inappropriately we marshal these calls into our socket server's IO thread pool rather than issuing them from the calling thread. This is done by posting an IO completion packet to the socket server's IO Completion Port. The server's worker threads know how to handle 4 kinds of operation: Read requests, read completions, write requests and write completions. The request operations are generated by calls to PostQueuedCompletionStatus and the completions are generated when calls to WSARecv and WSASend complete asyncronously.
To be able to read and write data we need somewhere to put it, so we need some kind of memory buffer. To reduce memory allocations we could pool these buffers so that we don't delete them once they're done with but instead maintain them in a list for reused. Our data buffers are managed by an allocator which is configured by passing arguments to the construtor of our socket server. This allows the user to set the size of the IO buffers used as well as being able to control how many buffers are retained in the list for reuse. The CIOBuffer class serves as our data buffer follows the standard IO Completion Port pattern of being an extended "overlapped" structure.
As all good references on IO Completion Ports tell you, calling GetQueuedCompletionStatus blocks your thread until a completion packet is available and, when it is, returns you a completion key, the number of bytes transferred and an "overlapped" structure. The completion key represents 'per device' data and the overlapped structure represents 'per call' data. In our server we use the completion key to pass our Socket class around and the overlapped structure to pass our data buffer. Both our Socket class and our data buffer class allow the user to associate 'user data' with them. This is in the form of a single unsigned long value (which could always be used to store a pointer to a larger structure).
The socket server's worker threads loop continuously, blocking on their completion port until work is available and then extracting the Socket and CIOBuffer from the completion data and processing the IO request. The loop looks something like this:
int CSocketServer::WorkerThread::Run()
{
while (true)
{
DWORD dwIoSize = 0;
Socket *pSocket = 0;
OVERLAPPED *pOverlapped = 0;
m_iocp.GetStatus((PDWORD_PTR)&pSocket, &dwIoSize, &pOverlapped);
CIOBuffer *pBuffer = CIOBuffer::FromOverlapped(pOverlapped);
switch pBuffer->GetUserData()
{
case IO_Read_Request :
Read(pSocket, pBuffer);
break;
case IO_Read_Completed :
ReadCompleted(pSocket, pBuffer);
break;
case IO_Write_Request :
Write(pSocket, pBuffer);
break;
case IO_Write_Completed :
WriteCompleted(pSocket, pBuffer);
break;
}
}
}
Read and write requests cause a read or write to be performed on the socket. Note that the actual read/write is being performed by our IO threads so that they cannot be terminated early due to the thread exiting. The ReadCompleted() and WriteCompleted() methods are called when the read or write actually completes. The worker thread provides two virtual functions to allow the caller's derived class to handle these situations. Most of the time the user will not be interested in the write completion, but the derived class is the only place that read completion can be handled.
virtual void ReadCompleted(
Socket *pSocket,
CIOBuffer *pBuffer) = 0;
virtual void WriteCompleted(
Socket *pSocket,
CIOBuffer *pBuffer);
Since our client must provide their own worker thread that derives from our socket server's worker thread we need to have a way for the server to be configured to use this derived worker thread. Whenever the server creates a worker thread (and this only occurs when the server first starts as the threads run for the life time of the server) it calls the following pure virtual function:
virtual WorkerThread *CreateWorkerThread(
CIOCompletionPort &iocp) = 0;
Comments