title: boost::asio::async_read_some run in parent thread
body: <p>I am writing an efficient socket server. The intention is good overall throughput. I use the main thread as the listener. It <code>async_accept</code> a client and add the socket to a queue. There's a dispatcher threads picks up a socket, who is ready to be read from, from the queue and added to one of the worker threads' queue. I keep a pool of worker threads. A worker thread will do actual read/write.</p>
  
  <p>I use <code>async_accept</code> in my listener. To find out which socket is ready for read, I use async_read_some in my dispatcher. This idea works, but with a problem. My <code>io_service.run()</code> is called in listener, so the handler of <code>async_read_some</code> in dispatcher, is actually run in the listener thread. </p>
  
  <p>Here's my code:</p>
  
  <pre><code>using boost::asio::ip::tcp;
  using namespace std;
  
  std::queue&lt;std::shared_ptr&lt;tcp::socket&gt;&gt; q_sock;
  boost::mutex m_log1;
  boost::condition_variable m_cond1;
  boost::mutex::scoped_lock m_lock1 = boost::mutex::scoped_lock(m_log1);
  sem_t _sem_sock;
  
  enum { max_length1 = 1024 };
  char data_1[max_length1];
  
  void handle_read1(std::shared_ptr&lt;tcp::socket&gt; sock, const boost::system::error_code&amp; error,
    size_t bytes_transferred)
  {
      printf("handle_read1 : error : %s : %d, thread id is: %ld, pid : %d \n", error.category().name(), error.value(), (long int)syscall(SYS_gettid), getpid());
  
      boost::asio::write(*(sock.get()), boost::asio::buffer(data_1, bytes_transferred));
  }
  
  
  void sock_dispatch() {
      int v_size = 0;
      std::shared_ptr&lt;tcp::socket&gt; curr_sock;
  
      printf("sock_dispatch started. The ID of this of this thread is: %ld, pid : %d \n", (long int)syscall(SYS_gettid), getpid());
  
      while(1) {
  
          while(1) {
              sem_wait(&amp;_sem_sock);
              v_size = q_sock.size();
              sem_post(&amp;_sem_sock);
  
              if(v_size &lt;= 0)
                  m_cond1.timed_wait(m_lock1,boost::posix_time::milliseconds(5000));
              else
                  break;
          }
  
          sem_wait(&amp;_sem_sock);
          curr_sock = q_sock.front();
          q_sock.pop();
          sem_post(&amp;_sem_sock);
  
          curr_sock-&gt;async_read_some(boost::asio::buffer(data_1, max_length1),
          boost::bind(handle_read1, curr_sock,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
      }
  
  }
  
  class session
  {
      public:
        session(boost::asio::io_service&amp; io_service)
          : sockptr(new tcp::socket(io_service)) {}
  
        void start()
        {
              printf("START NEW SESSION   The ID of this of this thread is: %ld, pid : %d \n", (long int)syscall(SYS_gettid), getpid());
  
              sem_wait(&amp;_sem_sock);
  
              q_sock.push(sockptr);
  
              sem_post(&amp;_sem_sock);
  
              m_cond1.notify_all();
        }
  
        std::shared_ptr&lt;tcp::socket&gt; sockptr;
  };
  
  class server
  {
      public:
        server(boost::asio::io_service&amp; io_service, short port)
          : io_service_(io_service),
            acceptor_(io_service, tcp::endpoint(tcp::v4(), port))
        {
          session* new_session = new session(io_service_);
          acceptor_.async_accept(*(new_session-&gt;sockptr.get()),
              boost::bind(&amp;server::handle_accept, this, new_session,
                boost::asio::placeholders::error));
  
          printf("WAITING TO ACCEPT: The ID of this of this thread is: %ld, pid : %d \n", (long int)syscall(SYS_gettid), getpid());
  
        }
  
        void handle_accept(session* new_session,
            const boost::system::error_code&amp; error)
        {
            new_session-&gt;start();
            new_session = new session(io_service_);
            acceptor_.async_accept(*(new_session-&gt;sockptr.get()),
                boost::bind(&amp;server::handle_accept, this, new_session,
                  boost::asio::placeholders::error));
        }
  
      private:
        boost::asio::io_service&amp; io_service_;
        tcp::acceptor acceptor_;
  };
  
  int main(int argc, char* argv[])
  {
      sem_init(&amp;_sem_sock, 0, 1);
  
      boost::asio::io_service io_service;
  
      using namespace std;
      server s(io_service, atoi(argv[1]));
  
      boost::thread t(boost::bind(sock_dispatch));
  
      io_service.run();
  
      return 0;
  }
  </code></pre>
  
  <p>This code is modified from a boost::asio example, <a href="http://www.boost.org/doc/libs/1_39_0/doc/html/boost_asio/example/echo/async_tcp_echo_server.cpp" rel="nofollow">http://www.boost.org/doc/libs/1_39_0/doc/html/boost_asio/example/echo/async_tcp_echo_server.cpp</a>. And the client code is <a href="http://www.boost.org/doc/libs/1_39_0/doc/html/boost_asio/example/echo/blocking_tcp_echo_client.cpp" rel="nofollow">http://www.boost.org/doc/libs/1_39_0/doc/html/boost_asio/example/echo/blocking_tcp_echo_client.cpp</a>.</p>
  
  <p>When a client connects, the output of server:</p>
  
  <pre><code>WAITING TO ACCEPT: The ID of this of this thread is: 3843, pid : 3843 
  sock_dispatch started. The ID of this of this thread is: 3844, pid : 3843 
  START NEW SESSION   The ID of this of this thread is: 3843, pid : 3843 
  handle_read1 : error : system : 0, thread id is: 3843, pid : 3843
  </code></pre>
  
  <p>In this case the dispatcher thread id is 3944, but the handle_read1 is run in thread 3843.
  Ideally, handle_read1 should run in dispatcher, so it won't block accept in listener.</p>
  
  <p>Any idea what I should do to achieve this? Or there's better design for the whole thing at all :)? </p>