Completion Port

Hey. Now I will tell you about the IO Completion Ports mechanism in Windows. Developers describe the port of completion as "a tool that improves the performance of applications that often use I / O operations." In general, they do not lie, so IOCP is often used when writing scalable server applications. However, it is believed that the port of completion is a tricky and difficult topic to understand.

Theory.

The “port” object, in essence, is a kernel event queue from which I / O operations are extracted and added. Naturally, not all current operations are added there, but only those that we have indicated to the port. This is done by associating the handle (handle) of the file (not necessarily the file on the disk, it can be a socket, pipe, mailslot, etc.) with a port handle. When an asynchronous I / O operation is initiated on a file, after its completion, the corresponding entry is added to the port.
For processing the results, a pool of threads is used, the number of which is selected by the user. When a thread is attached to a pool, it retrieves one result from the queue and processes it. If at the time of joining the queue is empty, then the thread falls asleep until a message for processing appears. An interesting feature of the port of completion is the fact that a message can be put into the queue with a “hand” in order to extract it later.
Looks confusing? In practice, somewhat easier.

Implementation.
')
After the description of the scheme of work we proceed to more specific things. Namely, the implementation of an application that uses IOCP. As an example, I will use a server that simply accepts incoming connections and packets from clients. Used language - C.

So, for a start it would be nice to create this very port. This is done by API

HANDLE CreateIOCompletionPort(
HANDLE FileHandle,
HANDLE ExistingCompletionPort,
ULONG_PTR CompletionKey,
DWORD NumberOfConcurrentThreads);

Remarkably, the same call is used to associate the file handle with an already existing port. Why this was done - is unknown.
Line
HANDLE iocp=CreateIoCompletionPort(INVALID_HANDLE_VALUE,0,0,0);
will create a new port object of completion and return us its handle. Here, the first argument is the value INVALID_HANDLE_VALUE, which means that we need a new port. The following two arguments must be set to 0. When creating, you can specify how many threads can simultaneously work for this port using the last argument. If you specify 0, the value equal to the number of processors in the system will be used.

The next step is to create threads that will be involved in the pool. You can not give universal advice. Some say that there should be twice the number of processors in the system, others, that their number should be equal, others dynamically change the size of the pool. What to do here depends on the application and configuration of the computer. I have an old stump with HyperThreading, so the system sees my processor as two. For this reason, in my example there will be 2 worker threads in the pool.

for(int i=1;i<=2;i++)
{
HANDLE hWorking=CreateThread(0,0,(LPTHREAD_START_ROUTINE)&WorkingThread,iocp,0,0);
CloseHandle(hWorking);
}

I draw your attention: we pass to the workflows the handle of the completion port as a parameter. They will need it when the threads declare their willingness to work. The function itself WorkingThread () will be shown below.

Now that the threads have been created, you can begin receiving clients and their messages. I will not give Winsock initialization code here (but it is in the source text of this article), so I’ll just write:

while(1)
{
SOCKET clientsock=WSAAccept(listensock,(sockaddr *)&clientaddr,&clientsize,0,0);
...
}

The accept call returns us the socket of the next client, to which we can write and from which we can read as from a regular file. In your case, there may be a file on disk or any other IO object.
Now we need to notify the port of completion that we want it to monitor this socket. To do this, bind the socket and port descriptors:

CreateIoCompletionPort((HANDLE)clientsock,iocp,(ULONG_PTR)key,0);

The last argument in this case is ignored, since the port has already been created. But the penultimate requires a special investigation. In the prototype, it is listed as CompletionKey (“completion key”). In fact, the key is a pointer to any data, i.e. on the structure or instance of the class you defined. It is used so that within a stream you can distinguish one operation from another, or to store the status of a particular client. At a minimum, you will have to store there a byte indicating which operation has completed - sending or receiving (reading or writing).

After binding the descriptors, you can initiate asynchronous I / O. For sockets, Winsock2 functions are used - WSASend () and WSARecv () with the pointer to the OVERLAPPED structure passed there, which actually marks an asynchronous operation. For files, you can use the WriteFile () and ReadFile () functions, respectively.
In addition to the OVERLAPPED structure, you need to pass some other IO information to the stream — for example, the address of the buffer, its length, or even the buffer itself. This can be done either through the completion key, or you can create a structure containing OVERLAPPED in the first field and pass a pointer to it in WSASend () / WSARecv ().

Now consider the API function that attaches the thread that calls it to the pool:

BOOL GetQueuedCompletionStatus(
HANDLE CompletionPort,
LPDWORD lpNumberOfBytes,
PULONG_PTR lpCompletionKey,
LPOVERLAPPED *lpOverlapped,
DWORD dwMilliseconds);

Here, CompletionPort is the handle of the port to which the pool should be connected; lpNumberOfBytes - a pointer to a variable, in which the number of bytes transferred is written as a result of the completion of the operation (in fact, it is the return value of recv () and send () in synchronous mode); lpCompletionKey - a pointer to a variable, into which a pointer to the termination key will be written; lpOverlapped - pointer to OVERLAPPED associated with this IO transaction; Finally, dwMilliseconds is the time for which the thread can fall asleep while waiting for the completion of any request. If you specify INFINITE, it will wait forever.

Now that we are familiar with the extraction function from the queue, we can look at the function with which the worker threads start executing.

void WorkingThread(HANDLE iocp)
{
while(1)
{
if(!GetQueuedCompletionStatus(iocp,&bytes,&key,&overlapped,INFINITE))
//
break;
if(!bytes)
//0 , ..
switch(key->OpType)
{
...
}
}
}

Inside the switch, new asynchronous operations are invoked, which will be processed the next time the loop passes. If we do not want a definite final operation to be transferred to the port (for example, when the result is not important to us), we can use the following trick - set the first bit of the OVERLAPPED.hEvent field to 1. It is worth noting that, in terms of performance, put processing the incoming information in the same cycle is not the most reasonable solution, since this will slow down the server's reaction to incoming packets. To solve the problem, you can make the parsing of the read information into another separate thread, and here the third API function is useful to us:

BOOL PostQueuedCompletionStatus(
HANDLE CompletionPort,
DWORD dwNumberOfBytesTransferred,
ULONG_PTR dwCompletionKey,
LPOVERLAPPED lpOverlapped);

Its essence is clear from the name - it puts a message in the port queue. Actually all asynchronous functions upon completion of the operation imperceptibly call it. All the arguments listed here are immediately passed to one of the threads. Thanks to PostQueuedCompletionStatus, the termination port can be used not only for processing IO operations, but also simply for efficient queuing with a pool of threads.
In our example, it makes sense to create another port and, after completing some operation, call PostQueuedCompletionStatus (), transferring the received packet to another stream for processing in the key.

Internal organization.
The port of completion is the following structure:

typedef stuct _IO_COMPLETION
{
KQUEUE Queue;
} IO_COMPLETION;

As noted above, this is just a kernel event queue. Here is a description of the KQUEUE structure:

typedef stuct _KQUEUE
{
DISPATCHER_HEADER Header;
LIST_ENTRY EnrtyListHead; //
DWORD CurrentCount;
DWORD MaximumCount;
LIST_ENTRY ThreadListHead; //
} KQUEUE;

When creating the port, the CreateIoCompletionPort function calls the internal NtCreateIoCompletion service. It is then initialized using the KeInitializeQueue function. When the port is bound to the file object, the Win32 function CreateIoCompletionPort calls NtSetInformationFile.

NtSetInformationFile(
HANDLE FileHandle,
PIO_STATUS_BLOCK IoStatusBlock,
PVOID FileInformation,
ULONG Length,
FILE_INFORMATION_CLASS FileInformationClass);

For this function, FILE_INFORMATION_CLASS is set as FileCompletionInformation, and a pointer to the IO_COMPLETION_CONTEXT or FILE_COMPLETION_INFORMATION structure is passed as the FileInformation parameter.

typedef struct _IO_COMPLETION_CONTEXT
{
PVOID Port;
PVOID Key;
} IO_COMPLETION_CONTEXT;

typedef struct _FILE_COMPLETION_INFORMATION
{
HANDLE IoCompletionHandle;
ULONG CompletionKey;
} FILE_COMPLETION_INFORMATION, *PFILE_COMPLETION_INFORMATION;

After completing an asynchronous I / O operation for an associated file, the I / O manager creates a request packet from the OVERLAPPED structure and a completion key and places it in a queue by calling KeInsertQueue. When a thread calls the GetQueuedCompletionStatus function, the NtRemoveIoCompletion function is actually called. NtRemoveIoCompletion checks the parameters and calls the KeRemoveQueue function, which blocks the stream if there are no requests in the queue, or the CurrentCount field of the KQUEUE structure is greater than or equal to MaximumCount. If there are requests and the number of active threads is less than the maximum, KeRemoveQueue removes the thread that caused it from the queue of waiting threads and increases the number of active threads by 1. When the stream is put on the queue of waiting threads, the Queue field of the KTHREAD structure is set to the address of the final port. When a request is placed on the termination port with the PostQueuedCompletionStatus function, the NtSetIoCompletion function is actually called, which after checking the parameters and converting the port handle to the pointer, calls KeInsertQueue.

I have it all. Below is a link to the program with an example of use.
iocp.c