🔒 Locked

iocp and packet reordering problem

Started by Kitt3n Aug 31, 2008 at 10:14 AM 4 replies 4.7k views

Original Post

Aug 31, 2008 at 10:14 AM

I'm trying to get my head around IO-completion ports and out-of-order packets (due to scheduling of the threads). My current implementation is working (blissfully ignoring out-of-order packets or even fragmented packets). The key part is the thread::run-function in the ::OPCODE_RECV_COMPLETED case-part, where I just call a client-callback passing the bytestream which was received. Now I'd like to make sure the packet-order is correct - I looked for some code/examples (found some on codeproject), but they are either cryptic/hard to follow or without any documentation. Assume I post two writes, I have two iocp-threads and both get one of the writes. Now it depends on the OS which one goes first (same for read) So what now? Add a mutex/critical-section? - Wouldn't that beat the purpose of having several iocp-threads? - And what about the network-card? I image at some point the driver will write something into it's buffers - at least there should be some kind of lock because 2 threads writing are probably bad? Can anyone explain some background on how it's supposed to work (or maybe someone knows a recent book discussing this problem?) My iocp-thread::run function looks like this



  bool TCPIocpThread::run (float _dtInSeconds)
  {
    TCPSocketIocp* iocp_socket;
    OVERLAPPED*    overlapped_intern;
    Context*       overlapped_ext;
    DWORD          dwIoSize;

    // 1st parameter is the value we passed when creating the completion port 
    // so we can cast the dword pointer to whatever type we used on creation
    bool iocpStatus = iocp->getStatus((PDWORD_PTR)&iocp_socket, &dwIoSize, &overlapped_intern);

    if (!iocpStatus)
      return true;

    // A completion key of 0 is posted to the iocp-thread to request shutdown
    if (!iocp_socket)
      return false;

    // cast it to our extended overlapped structure
    overlapped_ext = (Context*) overlapped_intern;

    switch (overlapped_ext->opcode)
    {
      case Context::OPCODE_SEND_REQUEST:
      {
        // start transmitting, reuse original overlapped structure
        DWORD dwNumBytesSent = 0, ret;
        // when iocp returns, status is 'send_completed'
        overlapped_ext->opcode = Context::OPCODE_SEND_COMPLETED; 
        ret = WSASend( iocp_socket->socket, &overlapped_ext->wsabuf, 1, &dwNumBytesSent, 0, overlapped_ext, NULL );

        DWORD dwLastError = WSAGetLastError();
        if ( ret == 0 )
          return true;
        else if (ret == SOCKET_ERROR && dwLastError == WSA_IO_PENDING )
          return true;
        else 
        {
          iocp_socket->onConnectionError (dwLastError, "TCPIocpThread::run(OPCODE_SEND_REQUEST)");
          iocp_socket->freeContext (overlapped_ext);
          iocp_socket->closeSocket(false);
        }	
      }
      break;

      case Context::OPCODE_RECV_HDR_REQUEST:
      case Context::OPCODE_RECV_REQUEST:
      {
        // when iocp returns, status will be 'OPCODE_RECV_(HDR_)COMPLETED'
        if (overlapped_ext->opcode==Context::OPCODE_RECV_HDR_REQUEST)
          overlapped_ext->opcode = Context::OPCODE_RECV_HDR_COMPLETED; 
        else
          overlapped_ext->opcode = Context::OPCODE_RECV_COMPLETED; 

        DWORD dwNumBytes = 0, dwFlags = 0;

        if (WSARecv( iocp_socket->socket, &overlapped_ext->wsabuf, 1, &dwNumBytes, &dwFlags, overlapped_ext, NULL)==SOCKET_ERROR)
        {
          DWORD dwLastError = ::WSAGetLastError();
          if (ERROR_IO_PENDING != dwLastError)
            iocp_socket->onConnectionError (dwLastError, "TCPIocpThread::run(OPCODE_RECV_REQUEST)");
          else
          {
            //iocp_socket->onConnectionError (dwLastError, "TCPIocpThread::run(OPCODE_RECV_REQUEST)");
            //iocp_socket->freeContext (overlapped_ext);
            //iocp_socket->disconnectSocket(false);
          }
        }
      }
      break; 

      case Context::OPCODE_RECV_HDR_COMPLETED:
      {

        iNetCallback* handler = iocp_socket->pHandler;
        // we received the header - interpret it to determine how much data is following
        U32 datalen = handler->getProtocol()->getDataLength ((const U8*)overlapped_ext->wsabuf.buf, overlapped_ext->wsabuf.len);
        NetId netSrc = handler->getProtocol()->getNetDest((const U8*)overlapped_ext->wsabuf.buf, overlapped_ext->wsabuf.len);
        iocp_socket->freeContext (overlapped_ext);
        // post read-request for the actual data
        iocp_socket->setInitialReadRequest (datalen, netSrc);
      }

      break;

      case Context::OPCODE_RECV_COMPLETED:
      {

        iNetCallback* handler = iocp_socket->pHandler;
        // we received the data which followed the header
        <b>handler->onReceivedData (0,overlapped_ext->netSrc, overlapped_ext->wsabuf.buf, overlapped_ext->wsabuf.len);</b>
        iocp_socket->freeContext (overlapped_ext);
        // post next read request (read next header)
        iocp_socket->setInitialReadRequest();
      }
      break;

      case Context::OPCODE_SEND_COMPLETED:
      {
        iocp_socket->freeContext (overlapped_ext);
      }
      break;

      case Context::OPCODE_DISCONNECT:
      {
      //::shutdown (iocp_socket->socket, SD_BOTH);
      //closesocket( iocp_socket->socket);
      //iocp_socket->freeContext (overlapped_ext);
      }
      break;

      /*case OPCODE_CONNECT:
      {
      }
      break;*/

      default: 
      {
      }
      break;
    }
    return true;
  }

Antheus

216

Antheus

Aug 31, 2008 at 11:31 AM

But you have control over access to resources.

TCP is connection oriented stream protocol. As such, you issue read requests in context of connection. It will look something like this:

void onReceive(Connection *, data *) {  // append read data and read it  // issue read request(onReceive, this, buffer);}

So for each connection, you will have one read request pending. And as long as you don't issue any other read requests for same connection, you should be thread-safe.

For writing, the problem lies in your application. Assuming 2 logic threads write some data. This data (given it's small enough) will be sent atomically, but the order in which they will be issued is arbitrary.

Sending is deterministic, and it documentation specifies that it's performed atomically (given various documented constraints).

If however each of these sends cannot be completed in single send, then there is no other way than to manage concurrency yourself, by keeping ordered sequence of data to be sent atomically.

Manually tracking data to be sent is commonly done anyway for throttling purposes.

Problems however may occur if you wish to respond immediately:

onReceive(Connection *, data *) {  // append read data and read it  // issue write request(onSend, this, buffer);}onSend(Connection *, data *) {  // if send failed, try again  // otherwise, issue read request(onReceive, this, buffer);}

This way, you have two functions, with one event active per connection. Again, no threading problems, since only one of these will be active at any given time in some arbitrary thread.

Kitt3n

Author

Kitt3n

Aug 31, 2008 at 12:52 PM

The code you gave with "void onReceive(Connection *, data *) {...}
is comparable to mine (where Connection is my 'handler'->onXxx)

Since I only will write from one thread into the iocp-system, that
means that everything should be okay like it is, then?

However it makes me wonder what this guy is doing then (he seems to be
going into an awful lot of trouble to fix this reordering problem...)

from http://www.codeproject.com/KB/IP/iocp_server_client.aspx

quote:
"3.6.2 The package reordering problem
This problem is also being discussed by [3]. Although committed operations using the IO completion port will always be completed in the order they were submitted, thread scheduling issues may mean that the actual work associated with the completion is processed in an undefined order. For example, if you have two I/O worker threads and you should receive "byte chunk 1, byte chunk 2, byte chunk 3", you may process the byte chunks in the wrong order, namely, "byte chunk 2, byte chunk 1, byte chunk 3". This also means that when you are sending the data by posting a send request on the I/O completion port, the data can actually be sent in a reordered way.

This can be solved by only using one worker thread, and committing only one I/O call and waiting for it to finish, but if we do this, we lose all the benefits of IOCP.

A simple practical solution to this problem is to add a sequence number to our buffer class, and process the data in the buffer if the buffer sequence number is in order. This means that the buffers that have incorrect numbers have to be saved for later use, and because of performance reasons, we will save the buffers in a hash map object (e.g., m_SendBufferMap and m_ReadBufferMap)."

Antheus

216

Antheus

Aug 31, 2008 at 1:02 PM

Quote:
Original post by Kitt3n For example, if you have two I/O worker threads and you should receive "byte chunk 1, byte chunk 2, byte chunk 3", you may process the byte chunks in the wrong order, namely, "byte chunk 2, byte chunk 1, byte chunk 3". This also means that when you are sending the data by posting a send request on the I/O completion port, the data can actually be sent in a reordered way.

Yes, that means issuing multiple requests for each shared resource (socket in this case). The solution I presented above avoids that.

Typically, the above scenario can be worked around. If not, then you do need proper synchronization.

hplus0603

Mod

3,014

hplus0603

Aug 31, 2008 at 2:10 PM

There is no need to have more than one read request (or write request) outstanding for a single socket. The point of IOCP is that you can have many read requests outstanding for many sockets, and efficiently receive data from the sockets that do have data.

When it comes to writing, unless you plan on saturating the outgoing stream buffer per socket (usually a bad idea when your data will travel across the internet to a poorly connected client), you should just do a non-blocking write/send without IOCP, and let the kernel-side outgoing buffer take care of the asynchronizing. The reason you should still be asynchronous, though, (and thus do application-level buffering) is that some clients may be very laggy, or there may be a network packet storm or something, and the kernel-side send buffer could fill up. Trying to send then without async sending, would block your server.

If you're using UDP, there's very little reason to use IOCP, because all I/O is on a single socket. You'd be better off with just a single loop polling the socket.

phear-

Sep 15, 2008 at 6:39 PM

Quote:
Original post by Kitt3n

Can anyone explain some background on how it's supposed to work (or
maybe someone knows a recent book discussing this problem?)

A great book on IOCP that I have read is "Windows via C/C++ 5th Edition." It goes through the background/theory behind how it works before giving you any code.

Topic Locked

This topic has been locked by a moderator. New replies are not allowed.