BOOKKEEPER-117: Support multi threads in hedwig cpp client to leverage multi-core hardware (Sijie Guo via ivank)

git-svn-id: https://svn.apache.org/repos/asf/zookeeper/bookkeeper/trunk@1207289 13f79535-47bb-0310-9956-ffa450edef68

Author: ivankelly

CHANGES.txt

@@ -110,6 +110,8 @@ BUGFIXES:
 
   BOOKKEEPER-118: Hedwig client doesn't kill and remove old subscription channel after redirection. (Sijie Guo via ivank)
 
+  BOOKKEEPER-117: Support multi threads in hedwig cpp client to leverage multi-core hardware (Sijie Guo via ivank)
+
 IMPROVEMENTS:
 
  BOOKKEEPER-28: Create useful startup scripts for bookkeeper and hedwig (ivank)

hedwig-client/src/main/cpp/inc/hedwig/client.h

@@ -46,6 +46,7 @@ namespace Hedwig {
     static const std::string RECONNECT_SUBSCRIBE_RETRY_WAIT_TIME;
     static const std::string SYNC_REQUEST_TIMEOUT;
     static const std::string SUBSCRIBER_AUTOCONSUME;
+    static const std::string NUM_DISPATCH_THREADS;
 
   public:
     Configuration() {};

hedwig-client/src/main/cpp/lib/channel.cpp

@@ -50,8 +50,8 @@ using namespace Hedwig;
 
 DuplexChannel::DuplexChannel(EventDispatcher& dispatcher, const HostAddress& addr, 
 			     const Configuration& cfg, const ChannelHandlerPtr& handler)
-  : dispatcher(dispatcher), address(addr), handler(handler), 
-    socket(dispatcher.getService()), instream(&in_buf), copy_buf(NULL), copy_buf_length(0),
+  : dispatcher(dispatcher), address(addr), handler(handler), service(dispatcher.getService()),
+    socket(service), instream(&in_buf), copy_buf(NULL), copy_buf_length(0),
     state(UNINITIALISED), receiving(false), reading(false), sending(false)
 {
   LOG4CXX_DEBUG(logger, "Creating DuplexChannel(" << this << ")");

hedwig-client/src/main/cpp/lib/channel.h

@@ -38,6 +38,7 @@
 #include <boost/shared_ptr.hpp>
 #include <boost/enable_shared_from_this.hpp>
 
+#include <boost/asio.hpp>
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/thread/mutex.hpp>
 #include <boost/thread/shared_mutex.hpp>
@@ -110,6 +111,10 @@ namespace Hedwig {
     void channelDisconnected(const std::exception& e);
     virtual void kill();
 
+    inline boost::asio::io_service & getService() {
+      return service;
+    }
+
     virtual ~DuplexChannel();
   private:
     enum State { UNINITIALISED, CONNECTING, CONNECTED, DEAD };
@@ -121,6 +126,7 @@ namespace Hedwig {
     HostAddress address;
     ChannelHandlerPtr handler;
 
+    boost::asio::io_service &service;
     boost::asio::ip::tcp::socket socket;
     boost::asio::streambuf in_buf;
     std::istream instream;

hedwig-client/src/main/cpp/lib/client.cpp

@@ -37,6 +37,7 @@ const std::string Configuration::MAX_MESSAGE_QUEUE_SIZE = "hedwig.cpp.max_msgque
 const std::string Configuration::RECONNECT_SUBSCRIBE_RETRY_WAIT_TIME = "hedwig.cpp.reconnect_subscribe_retry_wait_time";
 const std::string Configuration::SYNC_REQUEST_TIMEOUT = "hedwig.cpp.sync_request_timeout";
 const std::string Configuration::SUBSCRIBER_AUTOCONSUME = "hedwig.cpp.subscriber_autoconsume";
+const std::string Configuration::NUM_DISPATCH_THREADS = "hedwig.cpp.num_dispatch_threads";
 
 Client::Client(const Configuration& conf) {
   LOG4CXX_DEBUG(logger, "Client::Client (" << this << ")");

hedwig-client/src/main/cpp/lib/clientimpl.cpp

@@ -30,6 +30,7 @@ static log4cxx::LoggerPtr logger(log4cxx::Logger::getLogger("hedwig."__FILE__));
 using namespace Hedwig;
 
 const std::string DEFAULT_SERVER_DEFAULT_VAL = "";
+const int DEFAULT_NUM_DISPATCH_THREADS = 1;
 
 void SyncOperationCallback::wait() {
   boost::unique_lock<boost::mutex> lock(mut);
@@ -182,15 +183,15 @@ ClientImplPtr ClientImpl::Create(const Configuration& conf) {
   ClientImplPtr impl(new ClientImpl(conf));
   LOG4CXX_DEBUG(logger, "Creating Clientimpl " << impl);
 
-  impl->dispatcher.start();
+  impl->dispatcher->start();
 
   return impl;
 }
 
 void ClientImpl::Destroy() {
   LOG4CXX_DEBUG(logger, "destroying Clientimpl " << this);
 
-  dispatcher.stop();
+  dispatcher->stop();
   {
     boost::lock_guard<boost::shared_mutex> lock(allchannels_lock);
 
@@ -217,6 +218,7 @@ ClientImpl::ClientImpl(const Configuration& conf)
   : conf(conf), publisher(NULL), subscriber(NULL), counterobj(), shuttingDownFlag(false)
 {
   defaultHost = HostAddress::fromString(conf.get(Configuration::DEFAULT_SERVER, DEFAULT_SERVER_DEFAULT_VAL));
+  dispatcher = EventDispatcherPtr(new EventDispatcher(conf.getInt(Configuration::NUM_DISPATCH_THREADS, DEFAULT_NUM_DISPATCH_THREADS)));
 }
 
 Subscriber& ClientImpl::getSubscriber() {
@@ -312,7 +314,7 @@ DuplexChannelPtr ClientImpl::createChannel(const std::string& topic, const Chann
     setHostForTopic(topic, addr);
   }
 
-  DuplexChannelPtr channel(new DuplexChannel(dispatcher, addr, conf, handler));
+  DuplexChannelPtr channel(new DuplexChannel(*dispatcher, addr, conf, handler));
 
   boost::lock_guard<boost::shared_mutex> lock(allchannels_lock);
   if (shuttingDownFlag) {
@@ -392,5 +394,5 @@ const Configuration& ClientImpl::getConfiguration() {
 }
 
 boost::asio::io_service& ClientImpl::getService() {
-  return dispatcher.getService();
+  return dispatcher->getService();
 }

hedwig-client/src/main/cpp/lib/clientimpl.h

@@ -134,7 +134,8 @@ namespace Hedwig {
 
     ClientTxnCounter counterobj;
 
-    EventDispatcher dispatcher;
+    typedef boost::shared_ptr<EventDispatcher> EventDispatcherPtr;
+    EventDispatcherPtr dispatcher;
 
     typedef std::tr1::unordered_multimap<HostAddress, std::string, HostAddressHash > Host2TopicsMap;
     Host2TopicsMap host2topics;

hedwig-client/src/main/cpp/lib/eventdispatcher.cpp

@@ -27,49 +27,76 @@ static log4cxx::LoggerPtr logger(log4cxx::Logger::getLogger("hedwig."__FILE__));
 
 using namespace Hedwig;
 
-EventDispatcher::EventDispatcher() : service(), dummy_work(NULL), t(NULL) {
+EventDispatcher::EventDispatcher(int numThreads)
+  : num_threads(numThreads), running(false), next_io_service(0) {
+  if (0 == num_threads) {
+    throw std::runtime_error("number of threads in dispatcher is zero");
+  }
+  for (size_t i = 0; i < num_threads; i++) {
+    io_service_ptr service(new boost::asio::io_service);
+    services.push_back(service);
+  }
 }
 
-void EventDispatcher::run_forever() {
-  LOG4CXX_DEBUG(logger, "Starting event dispatcher");
+void EventDispatcher::run_forever(io_service_ptr service, size_t idx) {
+  LOG4CXX_DEBUG(logger, "Starting event dispatcher " << idx);
 
   while (true) {
     try {
-      service.run();
+      service->run();
       break;
     } catch (std::exception &e) {
-    LOG4CXX_ERROR(logger, "Exception in dispatch handler. " << e.what());
+    LOG4CXX_ERROR(logger, "Exception in dispatch handler " << idx << " : " << e.what());
     }
   }
-  LOG4CXX_DEBUG(logger, "Event dispatcher done");
+  LOG4CXX_DEBUG(logger, "Event dispatcher " << idx << " done");
 }
 
 void EventDispatcher::start() {
-  if (t) {
+  if (running) {
     return;
   }
-  dummy_work = new boost::asio::io_service::work(service);
-  t = new boost::thread(boost::bind(&EventDispatcher::run_forever, this));
+  for (size_t i = 0; i < num_threads; i++) {
+    io_service_ptr service = services[i];
+    work_ptr work(new boost::asio::io_service::work(*service));
+    works.push_back(work);
+    // new thread
+    thread_ptr t(new boost::thread(boost::bind(&EventDispatcher::run_forever, this, service, i)));
+    threads.push_back(t);
+  }
+  running = true;
 }
 
 void EventDispatcher::stop() {
-  if (!t) {
+  if (!running) {
     return;
   }
-  delete dummy_work;
-  dummy_work = NULL;
-  
-  service.stop();
-  
-  t->join();
-  delete t;
-  t = NULL;
+
+  works.clear();
+
+  for (size_t i = 0; i < num_threads; i++) {
+    services[i]->stop();
+  }
+
+  for (size_t i = 0; i < num_threads; i++) {
+    threads[i]->join();
+  }
+  threads.clear();
+
+  running = false;
 }
 
 EventDispatcher::~EventDispatcher() {
-  delete dummy_work;
+  services.clear();
 }
 
 boost::asio::io_service& EventDispatcher::getService() {
+  size_t next = 0;
+  {
+    boost::lock_guard<boost::mutex> lock(next_lock);
+    next = next_io_service;
+    next_io_service = (next_io_service + 1) % num_threads;
+  }
+  boost::asio::io_service& service = *services[next];
   return service;
 }

hedwig-client/src/main/cpp/lib/eventdispatcher.h

@@ -18,13 +18,21 @@
 #ifndef EVENTDISPATCHER_H
 #define EVENTDISPATCHER_H
 
+#include <vector>
+
 #include <boost/asio.hpp>
 #include <boost/thread.hpp>
+#include <boost/thread/mutex.hpp>
+#include <boost/shared_ptr.hpp>
 
 namespace Hedwig {
+  typedef boost::shared_ptr<boost::asio::io_service> io_service_ptr;
+  typedef boost::shared_ptr<boost::asio::io_service::work> work_ptr;
+  typedef boost::shared_ptr<boost::thread> thread_ptr;
+
   class EventDispatcher {
   public:  
-    EventDispatcher();
+    EventDispatcher(int numThreads = 1);
     ~EventDispatcher();
 
     void start();
@@ -33,11 +41,21 @@ namespace Hedwig {
     boost::asio::io_service& getService();
 
   private:
-    void run_forever();
+    void run_forever(io_service_ptr service, size_t idx);
 
-    boost::asio::io_service service;
-    boost::asio::io_service::work* dummy_work;
-    boost::thread* t;
+    // number of threads
+    size_t num_threads;
+    // running flag
+    bool running;
+    // pool of io_services.
+    std::vector<io_service_ptr> services;
+    // pool of works
+    std::vector<work_ptr> works;
+    // threads
+    std::vector<thread_ptr> threads;
+    // next io_service used for a connection
+    boost::mutex next_lock;
+    std::size_t next_io_service;
   };
 }
 

hedwig-client/src/main/cpp/lib/subscriberimpl.cpp

@@ -127,7 +127,7 @@ void SubscriberConsumeCallback::operationFailed(const std::exception& exception)
   LOG4CXX_ERROR(logger, "Error passing message to client transaction: " << data->getTxnId() << " error: " << exception.what() 
 		<< " retrying in " << retrywait << " Microseconds");
 
-  boost::asio::deadline_timer t(client->getService(), boost::posix_time::milliseconds(retrywait));
+  boost::asio::deadline_timer t(handler->getChannel()->getService(), boost::posix_time::milliseconds(retrywait));
 
   t.async_wait(boost::bind(&SubscriberConsumeCallback::timerComplete, handler, m, boost::asio::placeholders::error));  
 }
@@ -207,7 +207,7 @@ void SubscriberClientChannelHandler::channelDisconnected(const DuplexChannelPtr&
     int retrywait = client->getConfiguration().getInt(Configuration::RECONNECT_SUBSCRIBE_RETRY_WAIT_TIME,
 						      DEFAULT_RECONNECT_SUBSCRIBE_RETRY_WAIT_TIME);
 
-    boost::asio::deadline_timer t(client->getService(), boost::posix_time::milliseconds(retrywait));
+    boost::asio::deadline_timer t(channel->getService(), boost::posix_time::milliseconds(retrywait));
     t.async_wait(boost::bind(&SubscriberClientChannelHandler::reconnectTimerComplete, shared_from_this(), 
 			     channel, e, boost::asio::placeholders::error));  
     return;