Re: [boost] Asio multithreaded server

From: Christopher Kohlhoff (chris_at_[hidden])
Date: 2006-11-17 08:29:55

Hi Giovanni,

Giovanni Piero Deretta <gpderetta_at_[hidden]> wrote:
> I'm curious and interested.

I don't have time right now to expand on it in great detail
(hopefully later), but here's the problem in a nutshell.

First, use the very latest asio in boost CVS, with the following
diff applied (to undo my workaround):


--- boost/asio/detail/win_iocp_io_service.hpp    17 Nov 2006 11:40:48 -0000    1.6
+++ boost/asio/detail/win_iocp_io_service.hpp    17 Nov 2006 12:45:01 -0000
@@ -260,7 +260,7 @@
       LPOVERLAPPED overlapped = 0;
       ::SetLastError(0);
       BOOL ok = ::GetQueuedCompletionStatus(iocp_.handle, &bytes_transferred,
-          &completion_key, &overlapped, block ? 1000 : 0);
+          &completion_key, &overlapped, block ? INFINITE : 0);
       DWORD last_error = ::GetLastError();
 
       if (!ok && overlapped == 0)


And here's a simplified test program:

//--------------------------------------------------------------------
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/thread.hpp>
#include <iostream>
#include <ostream>
#include <windows.h>

void handle_read()
{
  while (true)
    ;
}

void handle_accept_2nd()
{
  std::cout << "handle_accept_2nd\n";
}

void handle_accept_1st(
    boost::asio::ip::tcp::socket* socket1,
    boost::asio::ip::tcp::acceptor* acceptor,
    boost::asio::ip::tcp::socket* socket2)
{
  std::cout << "handle_accept_1st\n";

  static char buffer[1024];
  socket1->async_read_some(
      boost::asio::buffer(buffer),
      boost::bind(handle_read));

  acceptor->async_accept(*socket2,
      boost::bind(handle_accept_2nd));
}

void do_nothing()
{
}

void post_nothing(boost::asio::io_service* io_service)
{
  for (;;)
  {
    ::Sleep(10000);
    io_service->post(do_nothing);
  }
}

int main()
{
  try
  {
    boost::asio::io_service io_service;

    boost::asio::ip::tcp::endpoint endpoint(
        boost::asio::ip::tcp::v4(), 0);
    boost::asio::ip::tcp::acceptor acceptor(
        io_service, endpoint);

    std::cout << "Listening on ";
    std::cout << acceptor.local_endpoint();
    std::cout << "\n";

    boost::asio::ip::tcp::socket socket1(io_service);
    boost::asio::ip::tcp::socket socket2(io_service);

    acceptor.async_accept(socket1,
        boost::bind(handle_accept_1st,
          &socket1, &acceptor, &socket2));

    boost::thread run_thread1(
        boost::bind(&boost::asio::io_service::run,
          &io_service));

    boost::thread run_thread2(
        boost::bind(&boost::asio::io_service::run,
          &io_service));

    //boost::thread post_nothing_thread(
    //    boost::bind(post_nothing, &io_service));

    run_thread1.join();
    run_thread2.join();
    //post_nothing_thread.join();
  }
  catch (std::exception& e)
  {
    std::cout << "Exception: " << e.what() << "\n";
  }

  return 0;
}
//--------------------------------------------------------------------

I ran the test on a uniprocessor system with Windows XP SP 2,
using the following steps:

- Run the server
- Use telnet to connect one client
- Observe "handle_accept_1st" in server output
- Type one character into telnet
- Observe that CPU usage is now 100%
- Start a second telnet to connect another client
- The "handle_accept_2nd" never happens

Now uncomment the lines to start the post_nothing thread and run
the test again. This time you should see the "handle_accept_2nd"
message (probably after a delay).

What seems to be happening is that the completion event for an
asynchronous operation is "queued" to the thread that started
the operation (i.e. the one that called AcceptEx in this case).
If that thread is busy (the infinite loop in handle_read) then
the completion event is never delivered, even if another thread
is blocked on GetQueuedCompletionStatus.

However, if you wake up the second thread using a timeout (which
is my workaround) or by using PostQueuedCompletionStatus (which
happens when you uncomment the post_nothing thread) then the
second thread picks up the accept completion event off the queue
and delivers it.

(Note 1: I came to the conclusion about a completion event being
queued to the thread that started the operation by observing the
behaviour of a more complex test program than the one above.)

(Note 2: The I/O completion port is being created with
NumberOfConcurrentThreads set to 0xFFFFFFFF, which means that it
allows multiple active threads. A simple test using a thread
pool calling io_service::run(), and posting multiple CPU bound
tasks into it confirms that this is working correctly.)

(Note 3: The handle_read function runs on the same thread as
handle_accept_1st because I/O completion ports use LIFO for
choosing the thread to run an available completion event.)

Cheers,
Chris