Merge lp:~vlad-lesin/percona-playback/connections_pool into lp:percona-playback

[Vlad Lesin]

A bunch of changes for implementing this https://blueprints.launchpad.net/percona-playback/+spec/pool-of-threads blueprint as well as fixes for some critical bugs.

[lp:~vlad-lesin/percona-playback/connections_pool]

171. By Vlad Lesin on 2012-11-19

Fixes for bug #1070824 and bug #1080654

172. By Vlad Lesin on 2012-11-19

merged with trunk

173. By Vlad Lesin on 2012-11-20

Send init session query on reconnect

174. By Vlad Lesin on 2012-11-20

Initialize session init query with the value from cmd line options.

175. By Vlad Lesin on 2012-11-23

Thread-safety fixes for mysql client plugin.

176. By Vlad Lesin on 2012-11-23

Don't push empty query log queries for execution

177. By Vlad Lesin on 2012-11-23

Convert multiline query to single line to avoid syntax errors from server.

178. By Vlad Lesin on 2012-11-23

Allow multi-statement queries.

[Stewart Smith (stewart)]

review: Approve

=== modified file 'Makefile.am'
--- Makefile.am	2012-11-08 22:17:07 +0000
+++ Makefile.am	2012-11-23 20:15:24 +0000
@@ -59,13 +59,11 @@
 libpercona_playback_la_SOURCES = \
   percona_playback/percona_playback.cc \
   percona_playback/plugin.cc \
-  percona_playback/db_thread.cc \
-  percona_playback/dispatcher.cc
+  percona_playback/db_thread.cc
 
 nobase_include_HEADERS += \
   percona_playback/percona_playback.h \
   percona_playback/db_thread.h \
-  percona_playback/dispatcher.h \
   percona_playback/query_result.h \
   percona_playback/plugin.h \
   percona_playback/tokenize.h \
@@ -89,6 +87,7 @@
   percona_playback/test/crashme-slow \
   percona_playback/test/sqlbench-transactions-slow \
   percona_playback/test/sysbench-slow \
+  percona_playback/test/thread-pool-sysbench-slow \
   percona_playback/test/preserve_query_time \
   percona_playback/test/tcpdump_without_handshake \
   percona_playback/test/tcpdump_fragmented_packet \
@@ -104,6 +103,7 @@
 percona_playback_test_crashme_slow_CXXFLAGS= $(AM_CXXFLAGS) -DSRCDIR=\"${srcdir}\"
 percona_playback_test_sqlbench_transactions_slow_CXXFLAGS= $(AM_CXXFLAGS) -DSRCDIR=\"${srcdir}\"
 percona_playback_test_sysbench_slow_CXXFLAGS= $(AM_CXXFLAGS) -DSRCDIR=\"${srcdir}\"
+percona_playback_test_thread_pool_sysbench_slow_CXXFLAGS= $(AM_CXXFLAGS) -DSRCDIR=\"${srcdir}\"
 percona_playback_test_preserve_query_time_CXXFLAGS= $(AM_CXXFLAGS) -DSRCDIR=\"${srcdir}\"
 percona_playback_test_tcpdump_without_handshake_CXXFLAGS= $(AM_CXXFLAGS) -DSRCDIR=\"${srcdir}\"
 percona_playback_test_tcpdump_fragmented_packet_CXXFLAGS= $(AM_CXXFLAGS) -DSRCDIR=\"${srcdir}\"
@@ -135,6 +135,9 @@
 percona_playback_test_sysbench_slow_SOURCES= percona_playback/test/sysbench-slow.cc
 percona_playback_test_sysbench_slow_LDADD= $(LDADD) $(MYSQL_LIBS)
 
+percona_playback_test_thread_pool_sysbench_slow_SOURCES= percona_playback/test/sysbench-slow.cc
+percona_playback_test_thread_pool_sysbench_slow_LDADD= $(LDADD) $(MYSQL_LIBS)
+
 percona_playback_test_preserve_query_time_SOURCES= percona_playback/test/preserve_query_time.cc
 percona_playback_test_preserve_query_time_LDADD= $(LDADD) $(MYSQL_LIBS)
 

=== modified file 'percona_playback/db_thread.cc'
--- percona_playback/db_thread.cc	2012-11-14 11:44:59 +0000
+++ percona_playback/db_thread.cc	2012-11-23 20:15:24 +0000
@@ -24,25 +24,23 @@
 
 extern std::string g_session_init_query;
 
-void DBThread::connect_and_init_session()
+void DBThread::init_session()
 {
-  connect();
-  if (!g_session_init_query.empty())
-  {
-    QueryResult r;
-    QueryResult er;
-    execute_query(g_session_init_query, &r, er);
-  }
+  if (g_session_init_query.empty())
+    return;
+  QueryResult r;
+  QueryResult er;
+  execute_query(g_session_init_query, &r, er);
 }
 
 void DBThread::run()
 {
   connect_and_init_session();
 
-  QueryEntryPtr query;
   while (true)
   {
-    queries.pop(query);
+    QueryEntryPtr query;
+    queries->pop(query);
 
     if (query->is_shutdown())
       break;

=== modified file 'percona_playback/db_thread.h'
--- percona_playback/db_thread.h	2012-11-14 11:44:59 +0000
+++ percona_playback/db_thread.h	2012-11-23 20:15:24 +0000
@@ -16,6 +16,7 @@
 #ifndef PERCONA_PLAYBACK_DB_THREAD_H
 #define PERCONA_PLAYBACK_DB_THREAD_H
 
+#include <memory>
 #include "percona_playback/visibility.h"
 #include <boost/thread.hpp>
 #include <boost/shared_ptr.hpp>
@@ -33,55 +34,49 @@
 
 class QueryResult;
 
-class DBThreadState
-{
-public:
-  virtual ~DBThreadState(){}
-};
-
 class DBThread
 {
 
 private:
   boost::thread thread;
   uint64_t thread_id;
-  boost::shared_ptr<DBThreadState> state;
 
 public:
   typedef tbb::concurrent_bounded_queue<QueryEntryPtr> Queries;
-  Queries queries;
+  boost::shared_ptr<Queries> queries;
 
-  DBThread(uint64_t _thread_id) : thread_id(_thread_id)  {
-    queries.set_capacity(g_db_thread_queue_depth);
+  DBThread(uint64_t _thread_id,
+	   boost::shared_ptr<Queries> _queries) :
+	  thread_id(_thread_id), queries(_queries)  {
+    queries->set_capacity(g_db_thread_queue_depth);
   }
 
   virtual ~DBThread() {}
 
-  void set_state(boost::shared_ptr<DBThreadState> new_state)
-  {
-    state= new_state;
-  }
-
-  boost::shared_ptr<DBThreadState> get_state() const
-  {
-    return state;
-  }
-
   void join()
   {
     thread.join();
   }
 
-  void connect_and_init_session();
+  bool connect_and_init_session()
+  {
+    if (connect())
+    {
+      init_session();
+      return true;
+    }
+    return false;
+  }
 
-  virtual void connect()= 0;
+  void    init_session();
+  virtual bool connect()= 0;
 
   virtual void disconnect()= 0;
   virtual void execute_query(const std::string &query,
 			     QueryResult *r,
 			     const QueryResult &expected_result)= 0;
 
-  void run();
+  virtual void run();
 
   void start_thread();
 };

=== removed file 'percona_playback/dispatcher.cc'
--- percona_playback/dispatcher.cc	2012-07-03 07:41:49 +0000
+++ percona_playback/dispatcher.cc	1970-01-01 00:00:00 +0000
@@ -1,105 +0,0 @@
-/* BEGIN LICENSE
- * Copyright (C) 2011-2012 Percona Inc.
- * This program is free software: you can redistribute it and/or modify it
- * under the terms of the GNU General Public License version 2, as published
- * by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but
- * WITHOUT ANY WARRANTY; without even the implied warranties of
- * MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
- * PURPOSE.  See the GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License along
- * with this program.  If not, see <http://www.gnu.org/licenses/>.
- * END LICENSE */
-#include "percona_playback/dispatcher.h"
-#include "percona_playback/db_thread.h"
-#include "percona_playback/plugin.h"
-
-#include <assert.h>
-
-Dispatcher g_dispatcher;
-
-extern percona_playback::DBClientPlugin *g_dbclient_plugin;
-
-boost::shared_ptr<DBThreadState>
-Dispatcher::get_thread_state(
-  uint64_t thread_id,
-  boost::function1<void, DBThread *> run_on_db_thread_create)
-{
-  DBExecutorsTable::accessor a;
-
-  if (executors.insert(a, thread_id))
-  {
-    DBThread *db_thread= g_dbclient_plugin->create(thread_id);
-    assert(db_thread);
-    a->second= db_thread;
-    if (!run_on_db_thread_create.empty())
-      run_on_db_thread_create(db_thread);
-    db_thread->start_thread();
-  }
-
-  return a->second->get_state();
-}
-
-void
-Dispatcher::dispatch(QueryEntryPtr query_entry)
-{
-  uint64_t thread_id= query_entry->getThreadId();
-  {
-    DBExecutorsTable::accessor a;
-    if (executors.insert(a, thread_id))
-    {
-      DBThread *db_thread= g_dbclient_plugin->create(thread_id);
-      a->second= db_thread;
-      db_thread->start_thread();
-    }
-    a->second->queries.push(query_entry);
-  }
-}
-
-bool
-Dispatcher::finish_and_wait(uint64_t thread_id)
-{
-  DBThread *db_thread= NULL;
-  {
-    DBExecutorsTable::accessor a;
-    if (executors.find(a, thread_id))
-    {
-      db_thread= a->second;
-      executors.erase(a);
-    }
-  }
-
-  if (!db_thread)
-    return false;
-
-  db_thread->queries.push(QueryEntryPtr(new FinishEntry()));
-  db_thread->join();
-
-  delete db_thread;
-
-  return true;
-}
-
-void
-Dispatcher::finish_all_and_wait()
-{
-  QueryEntryPtr shutdown_command(new FinishEntry());
-  while(executors.size())
-  {
-    uint64_t thread_id;
-    DBThread *t;
-    {
-      DBExecutorsTable::const_iterator iter= executors.begin();
-      thread_id= (*iter).first;
-      t= (*iter).second;
-    }
-    executors.erase(thread_id);
-
-    t->queries.push(shutdown_command);
-    t->join();
-
-    delete t;
-  }
-}

=== removed file 'percona_playback/dispatcher.h'
--- percona_playback/dispatcher.h	2012-07-06 07:15:28 +0000
+++ percona_playback/dispatcher.h	1970-01-01 00:00:00 +0000
@@ -1,46 +0,0 @@
-/* BEGIN LICENSE
- * Copyright (C) 2011 Percona Inc.
- * This program is free software: you can redistribute it and/or modify it
- * under the terms of the GNU General Public License version 2, as published
- * by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but
- * WITHOUT ANY WARRANTY; without even the implied warranties of
- * MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
- * PURPOSE.  See the GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License along
- * with this program.  If not, see <http://www.gnu.org/licenses/>.
- * END LICENSE */
-
-#ifndef PERCONA_PLAYBACK_DISPATCHER_H
-#define PERCONA_PLAYBACK_DISPATCHER_H
-
-#include <tbb/concurrent_hash_map.h>
-#include <boost/function.hpp>
-
-#include "percona_playback/query_entry.h"
-
-class DBThread;
-class DBThreadState;
-
-class Dispatcher
-{
-  typedef tbb::concurrent_hash_map<uint64_t, DBThread*> DBExecutorsTable;
-  DBExecutorsTable  executors;
-  void db_thread_func(DBThread *thread);
-  void start_thread(DBThread *thread);
-
-public:
-  boost::shared_ptr<DBThreadState>
-    get_thread_state(uint64_t thread_id,
-                     boost::function1<void, DBThread *>
-                      run_on_db_thread_create);
-  void dispatch(QueryEntryPtr query_entry);
-  bool finish_and_wait(uint64_t thread_id);
-  void finish_all_and_wait();
-};
-
-extern Dispatcher g_dispatcher;
-
-#endif /* PERCONA_PLAYBACK_DISPATCHER_H */

=== modified file 'percona_playback/libdrizzle_client/libdrizzle_client.cc'
--- percona_playback/libdrizzle_client/libdrizzle_client.cc	2012-07-23 07:08:23 +0000
+++ percona_playback/libdrizzle_client/libdrizzle_client.cc	2012-11-23 20:15:24 +0000
@@ -39,13 +39,13 @@
 
  public:
   LibDrizzleDBThread(uint64_t _thread_id, LibDrizzleOptions *opt) :
-    DBThread(_thread_id),
+    DBThread(_thread_id, boost::shared_ptr<Queries>(new Queries())),
     driz(NULL),
     options(opt)
   {
   }
 
-  void connect();
+  bool connect();
   void disconnect();
   void execute_query(const std::string &query, QueryResult *r,
 		     const QueryResult &expected_result);
@@ -70,7 +70,7 @@
   unsigned int port;
 };
 
-void LibDrizzleDBThread::connect()
+bool LibDrizzleDBThread::connect()
 {
   driz= drizzle_create(NULL);
   assert(driz != NULL);
@@ -81,6 +81,7 @@
 			      options->user.c_str(),
 			      options->password.c_str(),
 			      options->schema.c_str(), DRIZZLE_CON_MYSQL);
+  return true;
 }
 
 void LibDrizzleDBThread::disconnect()

=== modified file 'percona_playback/mysql_client/mysql_client.cc'
--- percona_playback/mysql_client/mysql_client.cc	2012-07-04 09:10:16 +0000
+++ percona_playback/mysql_client/mysql_client.cc	2012-11-23 20:15:24 +0000
@@ -42,17 +42,23 @@
   unsigned int port;
 };
 
-void MySQLDBThread::connect()
+bool MySQLDBThread::connect()
 {
   mysql_init(&handle);
-  mysql_real_connect(&handle,
-		     options->host.c_str(),
-		     options->user.c_str(),
-		     options->password.c_str(),
-		     options->schema.c_str(),
-		     options->port,
-		     "",
-		     0);
+  if (!mysql_real_connect(&handle,
+			  options->host.c_str(),
+			  options->user.c_str(),
+		          options->password.c_str(),
+		          options->schema.c_str(),
+		          options->port,
+		          "",
+		          CLIENT_MULTI_STATEMENTS))
+  {
+    fprintf(stderr, "Can't connect to server: %s\n",
+	    mysql_error(&handle));
+    return false;
+  }
+  return true;
 }
 
 void MySQLDBThread::disconnect()
@@ -63,29 +69,50 @@
 void MySQLDBThread::execute_query(const std::string &query, QueryResult *r,
 				  const QueryResult &)
 {
-  int mr= mysql_real_query(&handle, query.c_str(), query.length());
-  if(mr != 0)
-  {
-    r->setError(mr);
-  }
-  else
-  {
-    MYSQL_RES* mysql_res= NULL;
-
-    r->setError(mr);
-    r->setWarningCount(mysql_warning_count(&handle));
-
-    mysql_res= mysql_store_result(&handle);
-
-    if (mysql_res != NULL)
-      r->setRowsSent(mysql_num_rows(mysql_res));
+  int mr;
+  for(unsigned i = 0; i < max_err_num; ++i)
+  {
+    mr= mysql_real_query(&handle, query.c_str(), query.length());
+    r->setError(mr);
+    if(mr != 0)
+    {
+      fprintf(stderr,
+	      "Error during query: %s, number of tries %u\n",
+	      mysql_error(&handle),
+	      i);
+      disconnect();
+      connect_and_init_session();
+    }
     else
+    {
+      r->setWarningCount(mysql_warning_count(&handle));
       r->setRowsSent(0);
-
-    mysql_free_result(mysql_res);
+      do
+      {
+        MYSQL_RES* mysql_res= NULL;
+
+        mysql_res= mysql_store_result(&handle);
+
+        if (mysql_res != NULL)
+	{
+          r->setRowsSent(mysql_num_rows(mysql_res));
+          mysql_free_result(mysql_res);
+	}
+
+      } while(!mysql_next_result(&handle));
+
+      break;
+    }
   }
 }
 
+void MySQLDBThread::run()
+{
+  mysql_thread_init();
+  DBThread::run();
+  mysql_thread_end();
+}
+
 class MySQLDBClientPlugin : public percona_playback::DBClientPlugin
 {
 private:
@@ -126,6 +153,21 @@
       return -1;
     }
 
+    if (!active)
+      return 0;
+
+    if (!mysql_thread_safe())
+    {
+      fprintf(stderr, "libmysqlclient is not thread safe\n");
+      return -1;
+    }
+
+    if (mysql_library_init(0, NULL, NULL))
+    {
+      fprintf(stderr, "could not initialize mysql library\n");
+      return -1;
+    }
+
     if (vm.count("mysql-host"))
     {
       options.host= vm["mysql-host"].as<std::string>();

=== modified file 'percona_playback/mysql_client/mysql_client.h'
--- percona_playback/mysql_client/mysql_client.h	2012-06-15 10:53:23 +0000
+++ percona_playback/mysql_client/mysql_client.h	2012-11-23 20:15:24 +0000
@@ -10,16 +10,19 @@
  private:
   MYSQL handle;
   MySQLOptions *options;
+  static const unsigned max_err_num = 10;
 
  public:
   MySQLDBThread(uint64_t _thread_id, MySQLOptions *opt) :
-    DBThread(_thread_id),
+    DBThread(_thread_id,
+	     boost::shared_ptr<Queries>(new Queries())),
     options(opt)
   {
   }
 
-  void connect();
+  bool connect();
   void disconnect();
   void execute_query(const std::string &query, QueryResult *r,
 		     const QueryResult &expected_result);
+  void run();
 };

=== modified file 'percona_playback/null_dbclient/null_dbclient.cc'
--- percona_playback/null_dbclient/null_dbclient.cc	2012-04-05 04:47:09 +0000
+++ percona_playback/null_dbclient/null_dbclient.cc	2012-11-23 20:15:24 +0000
@@ -20,10 +20,11 @@
 class NULLDBThread : public DBThread
 {
  public:
-  NULLDBThread(uint64_t _thread_id) : DBThread(_thread_id) {
-  }
+  NULLDBThread(uint64_t _thread_id) :
+	  DBThread(_thread_id,
+		   boost::shared_ptr<Queries>(new Queries())) {}
 
-  void connect() {};
+  bool connect() { return true; };
   void disconnect() {};
   void execute_query(const std::string &, QueryResult *r,
 		     const QueryResult &expected_result) {

=== modified file 'percona_playback/percona_playback.cc'
--- percona_playback/percona_playback.cc	2012-11-14 11:44:59 +0000
+++ percona_playback/percona_playback.cc	2012-11-23 20:15:24 +0000
@@ -40,6 +40,7 @@
 
 percona_playback::DBClientPlugin *g_dbclient_plugin= NULL;
 percona_playback::InputPlugin *g_input_plugin= NULL;
+percona_playback::DispatcherPlugin *g_dispatcher_plugin= NULL;
 unsigned int g_db_thread_queue_depth;
 std::string g_session_init_query;
 
@@ -132,6 +133,22 @@
     std::cerr << _("Selected Input Plugin: ")
               << g_input_plugin->name
               << std::endl;
+
+    std::cerr << std::endl << _("Loaded Dispatcher Plugins: ");
+
+    BOOST_FOREACH(const PluginRegistry::DispatcherPluginPair &pp,
+		  PluginRegistry::singleton().dispatcher_plugins)
+    {
+      std::cerr << pp.first << " ";
+    }
+
+    std::cerr << std::endl;
+    std::cerr << std::endl;
+
+    assert(g_dispatcher_plugin);
+    std::cerr << _("Selected dispatcher Plugin: ")
+              << g_dispatcher_plugin->name
+              << std::endl;
 }
 
 int percona_playback_argv(percona_playback_st *the_percona_playback,
@@ -152,12 +169,13 @@
   db_options.add_options()
     ("db-plugin", po::value<std::string>(), _("Database plugin"))
     ("input-plugin", po::value<std::string>(), _("Input plugin"))
+    ("dispatcher-plugin", po::value<std::string>(), _("Dispatcher plugin"))
     ("queue-depth", po::value<unsigned int>(),
      _("Queue depth for DB executor (thread). The larger this number the"
      " greater the played-back workload can deviate from the original workload"
      " as some connections may be up to queue-depth behind. (default 1)"))
     ("session-init-query",
-     po::value<std::string>()->default_value(""),
+     po::value<std::string>(&g_session_init_query)->default_value(""),
      _("This query will be executed just after each connect to db"))
     ;
 
@@ -237,6 +255,32 @@
   }
   g_input_plugin->active= true;
 
+  if (vm.count("dispatcher-plugin"))
+  {
+    PluginRegistry::DispatcherPluginMap::iterator it;
+    it= PluginRegistry::singleton().dispatcher_plugins.find(
+      vm["dispatcher-plugin"].as<std::string>());
+    if (it == PluginRegistry::singleton().dispatcher_plugins.end())
+    {
+      fprintf(stderr, _("Invalid Dispatcher Plugin\n"));
+      return -1;
+    }
+    g_dispatcher_plugin= it->second;
+  }
+  else
+  {
+    PluginRegistry::DispatcherPluginMap::iterator it;
+    it= PluginRegistry::singleton().dispatcher_plugins.
+      find("thread-per-connection");
+    if (it == PluginRegistry::singleton().dispatcher_plugins.end())
+    {
+      fprintf(stderr, _("Invalid Dispatcher plugin\n"));
+      return -1;
+    }
+    g_dispatcher_plugin= it->second;
+  }
+  g_dispatcher_plugin->active= true;
+
   if (vm.count("help") || argc==1)
   {
     help(options_description);
@@ -299,6 +343,7 @@
   std::cerr << _("Database Plugin: ") << g_dbclient_plugin->name << std::endl;
   std::cerr << _(" Running...") << std::endl;
 
+  g_dispatcher_plugin->run();
   g_input_plugin->run(*r);
 
   BOOST_FOREACH(const PluginRegistry::ReportPluginPair pp,

=== modified file 'percona_playback/plugin.h'
--- percona_playback/plugin.h	2012-06-24 04:28:41 +0000
+++ percona_playback/plugin.h	2012-11-23 20:15:24 +0000
@@ -20,6 +20,9 @@
 #include <vector>
 #include <string>
 #include <map>
+#include <boost/shared_ptr.hpp>
+#include <boost/function.hpp>
+#include "percona_playback/query_entry.h"
 #include <percona_playback/visibility.h>
 #include <percona_playback/version.h>
 
@@ -31,6 +34,7 @@
 }
 
 class DBThread;
+class DBThreadState;
 class QueryResult;
 class percona_playback_run_result;
 
@@ -102,6 +106,21 @@
 
 };
 
+class DispatcherPlugin : public plugin
+{
+ public:
+  std::string name;
+
+  DispatcherPlugin(const std::string &_name) : name(_name) {}
+
+  virtual void dispatch(QueryEntryPtr query_entry)= 0;
+  virtual bool finish_and_wait(uint64_t thread_id)= 0;
+  virtual void finish_all_and_wait()= 0;
+
+  virtual void run() {};
+
+};
+
 class PluginRegistry
 {
  public:
@@ -122,6 +141,9 @@
   typedef std::map<std::string, InputPlugin*> InputPluginMap;
   typedef std::pair<std::string, InputPlugin*> InputPluginPair;
 
+  typedef std::map<std::string, DispatcherPlugin*> DispatcherPluginMap;
+  typedef std::pair<std::string, DispatcherPlugin*> DispatcherPluginPair;
+
   typedef std::map<std::string, plugin*> PluginMap;
   typedef std::pair<std::string, plugin*> PluginPair;
 
@@ -129,6 +151,7 @@
   DBClientPluginMap dbclient_plugins;
   ReportPluginMap report_plugins;
   InputPluginMap input_plugins;
+  DispatcherPluginMap dispatcher_plugins;
 
   void add(const std::string &name, plugin* plugin_object)
   {
@@ -154,6 +177,12 @@
     all_plugins.insert(PluginPair(name, input_plugin));
   }
 
+  void add(const std::string &name, DispatcherPlugin* dispatcher_plugin)
+  {
+    dispatcher_plugins.insert(DispatcherPluginPair(name, dispatcher_plugin));
+    all_plugins.insert(PluginPair(name, dispatcher_plugin));
+  }
+
 };
 
 

=== modified file 'percona_playback/query_entry.h'
--- percona_playback/query_entry.h	2012-07-03 02:37:51 +0000
+++ percona_playback/query_entry.h	2012-11-23 20:15:24 +0000
@@ -45,7 +45,8 @@
 class FinishEntry : public QueryEntry
 {
 public:
-  FinishEntry() { set_shutdown(); }
+  FinishEntry(uint64_t _thread_id) :
+    QueryEntry (_thread_id, true) {}
   bool is_quit() { return false; }
 
   void execute(DBThread *) {}

=== modified file 'percona_playback/query_log/query_log.cc'
--- percona_playback/query_log/query_log.cc	2012-11-08 22:17:07 +0000
+++ percona_playback/query_log/query_log.cc	2012-11-23 20:15:24 +0000
@@ -39,7 +39,6 @@
 #include <percona_playback/db_thread.h>
 #include <percona_playback/query_log/query_log.h>
 #include <percona_playback/query_result.h>
-#include "percona_playback/dispatcher.h"
 #include <percona_playback/gettext.h>
 
 #include <boost/foreach.hpp>
@@ -51,6 +50,8 @@
 static bool g_run_set_timestamp;
 static bool g_preserve_query_time;
 
+extern percona_playback::DispatcherPlugin *g_dispatcher_plugin;
+
 class ParseQueryLogFunc: public tbb::filter {
 public:
   ParseQueryLogFunc(FILE *input_file_,
@@ -83,7 +84,7 @@
     new std::vector<boost::shared_ptr<QueryLogEntry> >();
 
   boost::shared_ptr<QueryLogEntry> tmp_entry(new QueryLogEntry());
-  entries->push_back(tmp_entry);
+ // entries->push_back(tmp_entry);
 
   char *line= NULL;
   size_t buflen = 0;
@@ -130,9 +131,10 @@
 
     if (strncmp(p, "# User@Host", strlen("# User@Host")) == 0)
     {
+      if (!tmp_entry->getQuery().empty())
+        entries->push_back(tmp_entry);
       count++;
       tmp_entry.reset(new QueryLogEntry());
-      entries->push_back(tmp_entry);
       (*this->nr_entries)++;
     }
 
@@ -172,6 +174,9 @@
     p= line;
   }
 
+  if (!tmp_entry->getQuery().empty())
+    entries->push_back(tmp_entry);
+
   free(line);
   return entries;
 }
@@ -239,7 +244,19 @@
      && s.compare(0, timestamp_query.length(), timestamp_query) == 0)
     set_timestamp_query= s;
   else
-    query.append(s);
+  {
+    //Append space insead of \r\n
+    std::string::const_iterator end = s.end() - 1;
+    if (s.length() >= 2 && *(s.end() - 2) == '\r')
+      --end;
+    //Remove initial spaces for best query viewing in reports
+    std::string::const_iterator begin;
+    for (begin = s.begin(); begin != end; ++begin)
+      if (*begin != ' ' && *begin != '\t')
+        break;
+    query.append(begin, end);
+    query.append(" ");
+  }
 }
 
 bool QueryLogEntry::parse_metadata(const std::string &s)
@@ -281,13 +298,13 @@
       r= true;
     }
   }
-
+/*
   if (s[0] == '#' && strncmp(s.c_str(), "# administrator", strlen("# administrator")))
   {
     query.append(s);
     r= true;
   }
-
+*/
   return r;
 }
 
@@ -300,7 +317,7 @@
     for (unsigned int i=0; i< input->size(); i++)
     {
       //      usleep(10);
-      g_dispatcher.dispatch((*input)[i]);
+      g_dispatcher_plugin->dispatch((*input)[i]);
     }
     delete input;
     return NULL;
@@ -330,7 +347,7 @@
   p.add_filter(f4);
   p.run(2);
 
-  g_dispatcher.finish_all_and_wait();
+  g_dispatcher_plugin->finish_all_and_wait();
 
   r->n_log_entries= entries;
   r->n_queries= queries;

=== modified file 'percona_playback/tcpdump/connection_state.cc'
--- percona_playback/tcpdump/connection_state.cc	2012-07-02 17:51:47 +0000
+++ percona_playback/tcpdump/connection_state.cc	2012-11-23 20:15:24 +0000
@@ -25,9 +25,11 @@
 #include "percona_playback/plugin.h"
 
 extern percona_playback::DBClientPlugin *g_dbclient_plugin;
+extern percona_playback::DispatcherPlugin *g_dispatcher_plugin;
 
 void
-ConnectionState::ProcessMysqlPkts(const u_char      *pkts,
+ConnectionState::ProcessMysqlPkts(boost::shared_ptr<ConnectionState> cs_ptr,
+				  const u_char      *pkts,
                                   u_int             pkts_len,
                                   const timeval     &ts,
                                   const AddrPort    &addr_port,
@@ -76,13 +78,13 @@
       {
       case PKT_QUERY:
         if (!query.empty())
-          DispatchQuery(ts, query, addr_port);
+          DispatchQuery(cs_ptr, ts, query, addr_port);
         ++stats.nr_of_parsed_queries;
         ++stats.nr_of_parsed_packets;
         break;
 
       case PKT_RESULT:
-        DispatchResult(ts, addr_port);
+        DispatchResult(cs_ptr, ts, addr_port);
         ++stats.nr_of_parsed_packets;
         break;
 
@@ -107,12 +109,6 @@
 
 }
 
-void
-ConnectionState::ProcessFinishConnection()
-{
-  db_thread->queries.push(QueryEntryPtr(new FinishEntry()));
-}
-
 PktResult
 ConnectionState::ParseMysqlPkt(IN UCharBuffer  &buff,
                                OUT std::string &query)
@@ -303,23 +299,32 @@
 }
 
 void
-ConnectionState::DispatchQuery(const timeval        &ts,
+ConnectionState::DispatchQuery(/* Having shared pointer to "this" as an argument
+				  is just temporary step. We need a shared pointer
+				  to connection state inside of query entry. The
+				  better way to do this is to move code that creates
+				  query entries to the upper layer.
+			       */
+			       boost::shared_ptr<ConnectionState> cs_ptr,
+			       const timeval        &ts,
                                const std::string    &query,
                                const AddrPort       &addr_port)
 {
   boost::shared_ptr<TcpdumpQueryEntry> 
-    query_entry(new TcpdumpQueryEntry(ts,
+    query_entry(new TcpdumpQueryEntry(cs_ptr,
+				      ts,
                                       query,
                                       addr_port));
-  db_thread->queries.push(query_entry);
+  g_dispatcher_plugin->dispatch(query_entry);
 }
 
 void
-ConnectionState::DispatchResult(const timeval    &ts,
+ConnectionState::DispatchResult(boost::shared_ptr<ConnectionState> cs_ptr,
+				const timeval    &ts,
                                 const AddrPort   &addr_port)
 {
   boost::shared_ptr<TcpdumpResultEntry> 
-    result_entry(new TcpdumpResultEntry(addr_port, ts, last_query_result));
-  db_thread->queries.push(result_entry);
+    result_entry(new TcpdumpResultEntry(cs_ptr, addr_port, ts, last_query_result));
+  g_dispatcher_plugin->dispatch(result_entry);
 }
 

=== modified file 'percona_playback/tcpdump/connection_state.h'
--- percona_playback/tcpdump/connection_state.h	2012-07-03 07:41:49 +0000
+++ percona_playback/tcpdump/connection_state.h	2012-11-23 20:15:24 +0000
@@ -94,8 +94,7 @@
 };
 
 
-class ConnectionState : public DBThreadState
-
+class ConnectionState
 {
 
 public:
@@ -107,7 +106,7 @@
     SERVER
   };
 
-  ConnectionState() :
+  ConnectionState(uint64_t _thread_id) :
     current_origin(UNDEF),
     fragmented(false),
     handshake_from_client(false),
@@ -126,7 +125,7 @@
                           (drizzle_con_options_t)
                             DRIZZLE_CON_MYSQL),
       drizzle_con_free),
-    db_thread(NULL)
+    thread_id(_thread_id)
    {
      drizzle_con->result= NULL;
    }
@@ -136,18 +135,15 @@
     drizzle_result_free_all(drizzle_con.get());
   }
 
-  void ProcessMysqlPkts(const u_char                 *pkts,
+  void ProcessMysqlPkts(boost::shared_ptr<ConnectionState> cs_ptr,
+			const u_char                 *pkts,
                         u_int                        total_len,
                         const timeval                &ts,
                         const AddrPort               &addr_port,
                         OUT TcpdumpMysqlParserStats  &stats);
 
-  void ProcessFinishConnection();
-
   void SetCurrentOrigin(Origin o) { current_origin = o; }
 
-  void SetDBThread(DBThread *t) { db_thread= t; }
-  
   LastExecutedQueryInfo   last_executed_query_info;
 
 private:
@@ -159,11 +155,13 @@
   PktResult ServerPacket(IN UCharBuffer &buff);
   PktResult ClientPacket(IN UCharBuffer &buff, OUT std::string &query);
 
-  void DispatchQuery(const timeval      &ts,
+  void DispatchQuery(boost::shared_ptr<ConnectionState> cs_ptr,
+		     const timeval      &ts,
                      const std::string  &query,
                      const AddrPort     &addr_port);
 
-  void DispatchResult(const timeval     &ts,
+  void DispatchResult(boost::shared_ptr<ConnectionState> cs_ptr,
+		      const timeval     &ts,
                       const AddrPort    &addr_port);
 
   Origin            current_origin;
@@ -179,7 +177,7 @@
 
   QueryResult       last_query_result;
 
-  DBThread          *db_thread;
+  uint64_t          thread_id;
 
 };
 

=== modified file 'percona_playback/tcpdump/pcap_packets_parser.cc'
--- percona_playback/tcpdump/pcap_packets_parser.cc	2012-07-03 07:41:49 +0000
+++ percona_playback/tcpdump/pcap_packets_parser.cc	2012-11-23 20:15:24 +0000
@@ -18,7 +18,6 @@
 
 #include "percona_playback/db_thread.h"
 #include "percona_playback/plugin.h"
-#include "percona_playback/dispatcher.h"
 
 #include <netinet/in.h>
 #include <arpa/inet.h>
@@ -29,6 +28,7 @@
 #define SNAP_LEN            16500  // pcap's max capture size
 
 extern percona_playback::DBClientPlugin *g_dbclient_plugin;
+extern percona_playback::DispatcherPlugin *g_dispatcher_plugin;
 
 void
 PcapPacketsParser::ParsePkt(const struct pcap_pkthdr *header,
@@ -81,7 +81,8 @@
   if (size_mysql == 0 &&
       ((tcp->th_flags & TH_FIN) || (tcp->th_flags & TH_RST)))
   {
-    g_dispatcher.finish_and_wait(addr_port.ThreadId());
+    g_dispatcher_plugin->finish_and_wait(addr_port.ThreadId());
+    RemoveConnectionState(addr_port.ThreadId());
     return;
   }
 
@@ -97,37 +98,50 @@
   }
 
   /* If there is no DBThread with such id create it */
-  boost::shared_ptr<DBThreadState>
-    state= g_dispatcher.get_thread_state(addr_port.ThreadId(),
-      boost::bind(&PcapPacketsParser::CreateConnectionState,
-                  this,
-                  _1));
+  boost::shared_ptr<ConnectionState>
+    state= GetConnectionState(addr_port.ThreadId());
 
   assert(state.get());
 
-  ((ConnectionState *)state.get())->SetCurrentOrigin(origin);
-  ((ConnectionState *)state.get())->ProcessMysqlPkts(mysql,
-                                                     size_mysql,
-                                                     header->ts,
-                                                     addr_port,
-                                                     stats);
+  state->SetCurrentOrigin(origin);
+  state->ProcessMysqlPkts(state,
+			  mysql,
+                          size_mysql,
+                          header->ts,
+                          addr_port,
+                          stats);
 }
 
 void
 PcapPacketsParser::WaitForUnfinishedTasks()
 {
-  g_dispatcher.finish_all_and_wait();
-}
-
-void
-PcapPacketsParser::CreateConnectionState(DBThread *db_thread)
-{
-  assert(db_thread);
-  boost::shared_ptr<ConnectionState> state(new ConnectionState());
+  g_dispatcher_plugin->finish_all_and_wait();
+}
+
+boost::shared_ptr<ConnectionState>
+PcapPacketsParser::GetConnectionState(uint64_t thread_id)
+{
+  Connections::iterator it = connections.find(thread_id);
+  if (it == connections.end())
+    return CreateConnectionState(thread_id);
+  return it->second;
+}
+
+boost::shared_ptr<ConnectionState>
+PcapPacketsParser::CreateConnectionState(uint64_t thread_id)
+{
+  boost::shared_ptr<ConnectionState> state(new ConnectionState(thread_id));
   state->last_executed_query_info.end_pcap_timestamp=
     first_packet_pcap_timestamp;
   state->last_executed_query_info.end_timestamp=
     first_packet_timestamp;
-  db_thread->set_state(state);
-  state->SetDBThread(db_thread);
-}
+  connections[thread_id] = state;
+  return state;
+}
+
+void
+PcapPacketsParser::RemoveConnectionState(uint64_t thread_id)
+{
+  connections.erase(thread_id);
+}
+

=== modified file 'percona_playback/tcpdump/pcap_packets_parser.h'
--- percona_playback/tcpdump/pcap_packets_parser.h	2012-06-19 15:26:53 +0000
+++ percona_playback/tcpdump/pcap_packets_parser.h	2012-11-23 20:15:24 +0000
@@ -23,6 +23,8 @@
 #include <tbb/concurrent_queue.h>
 #include <boost/thread.hpp>
 #include <boost/bind.hpp>
+#include <boost/unordered_map.hpp>
+#include <boost/shared_ptr.hpp>
 
 class DBThread;
 
@@ -32,10 +34,14 @@
   timeval                    first_packet_timestamp;
   timeval                    first_packet_pcap_timestamp;
   bool                       was_first_packet;
+  typedef boost::unordered_map<uint64_t,
+			       boost::shared_ptr<ConnectionState> >
+          Connections;
+  Connections		     connections;
 
 public:
 
-  PcapPacketsParser() : was_first_packet(false)
+  PcapPacketsParser() : was_first_packet(false), connections(10000)
   {
     first_packet_timestamp.tv_sec= first_packet_timestamp.tv_usec= 0;
     first_packet_pcap_timestamp.tv_sec= first_packet_pcap_timestamp.tv_usec= 0;
@@ -52,9 +58,11 @@
 
   void WaitForUnfinishedTasks();
 
-  void CreateConnectionState(DBThread *db_thread);
 
 private:
+  boost::shared_ptr<ConnectionState> CreateConnectionState(uint64_t thread_id);
+  boost::shared_ptr<ConnectionState> GetConnectionState(uint64_t thread_id);
+  void RemoveConnectionState(uint64_t thread_id);
 
   void ParsePkt(const struct pcap_pkthdr *header,
                 const u_char *packet);

=== modified file 'percona_playback/tcpdump/tcpdump_query_entries.cc'
--- percona_playback/tcpdump/tcpdump_query_entries.cc	2012-07-09 11:54:52 +0000
+++ percona_playback/tcpdump/tcpdump_query_entries.cc	2012-11-23 20:15:24 +0000
@@ -30,15 +30,10 @@
   QueryResult expected_result;
   timeval start_time;
   timeval end_time;
-  boost::shared_ptr<DBThreadState> thr_state;
 
   assert(t);
 
-  thr_state= t->get_state();
-
-  assert(thr_state.get());
-
-  ConnectionState &state= (ConnectionState &)*thr_state;
+  ConnectionState &state= *connection_state;
 
   LastExecutedQueryInfo &last_query_info= state.last_executed_query_info;
 
@@ -96,12 +91,7 @@
 
   assert(t);
 
-  thr_state= t->get_state();
-
-  assert(thr_state.get());
-
-
-  ConnectionState &state= (ConnectionState &)*thr_state;
+  ConnectionState &state= *connection_state;
   LastExecutedQueryInfo &last_query_info= state.last_executed_query_info;
 
   timersub(&pcap_timestamp,

=== modified file 'percona_playback/tcpdump/tcpdump_query_entries.h'
--- percona_playback/tcpdump/tcpdump_query_entries.h	2012-07-03 07:41:49 +0000
+++ percona_playback/tcpdump/tcpdump_query_entries.h	2012-11-23 20:15:24 +0000
@@ -21,9 +21,12 @@
 #include "connection_state.h"
 
 #include <sys/time.h>
+#include <boost/shared_ptr.hpp>
 
 class TcpdumpQueryEntry : public QueryEntry
 {
+
+  boost::shared_ptr<ConnectionState> connection_state;
   timeval       pcap_timestamp;
   std::string   query;
 
@@ -34,10 +37,12 @@
     pcap_timestamp.tv_sec= pcap_timestamp.tv_usec= 0;
   }
 
-  TcpdumpQueryEntry(const timeval       &_pcap_timestamp,
+  TcpdumpQueryEntry(boost::shared_ptr<ConnectionState> _connection_state,
+		    const timeval       &_pcap_timestamp,
                     const std::string   &_query,
                     const AddrPort      &addr_port) :
     QueryEntry(addr_port.ThreadId(), false),
+    connection_state(_connection_state),
     pcap_timestamp(_pcap_timestamp),
     query(_query)
   {}
@@ -50,15 +55,18 @@
 class TcpdumpResultEntry : public QueryEntry
 {
 
+  boost::shared_ptr<ConnectionState> connection_state;
   timeval       pcap_timestamp;
   QueryResult   expected_result;
 
 public:
   TcpdumpResultEntry() {}
-  TcpdumpResultEntry(const AddrPort     &addr_port,
+  TcpdumpResultEntry(boost::shared_ptr<ConnectionState> _connection_state,
+		     const AddrPort     &addr_port,
                      const timeval      &_pcap_timestamp,
                      const QueryResult  &_expected_result) :
     QueryEntry(addr_port.ThreadId(), false),
+    connection_state(_connection_state),
     pcap_timestamp(_pcap_timestamp),
     expected_result(_expected_result)
   {}

=== added file 'percona_playback/test/thread-pool-sysbench-slow.cc'
--- percona_playback/test/thread-pool-sysbench-slow.cc	1970-01-01 00:00:00 +0000
+++ percona_playback/test/thread-pool-sysbench-slow.cc	2012-11-23 20:15:24 +0000
@@ -0,0 +1,60 @@
+/* BEGIN LICENSE
+ * Copyright (C) 2011 Stewart Smith <[e-mail]>
+ * This program is free software: you can redistribute it and/or modify it 
+ * under the terms of the GNU General Public License version 3, as published 
+ * by the Free Software Foundation.
+ * 
+ * This program is distributed in the hope that it will be useful, but 
+ * WITHOUT ANY WARRANTY; without even the implied warranties of 
+ * MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR 
+ * PURPOSE.  See the GNU General Public License for more details.
+ * 
+ * You should have received a copy of the GNU General Public License along 
+ * with this program.  If not, see <http://www.gnu.org/licenses/>.
+ * END LICENSE */
+
+#include "config.h"
+
+#include <iostream>
+#include <cstdio>
+#include <string>
+#include <cstring>
+#include <assert.h>
+#include <unistd.h>
+
+#include <percona_playback/percona_playback.h>
+
+/**
+ * @TODO Actually write a real test suite here
+ */
+int main(int argc, char **argv)
+{
+  (void)argc; (void)argv;
+
+  fprintf(stderr, "Working dir: %s\n\n", get_current_dir_name());
+
+  percona_playback_st *the_percona_playback= percona_playback_create("test_Percona Playback");
+  assert(the_percona_playback);
+
+  char dbplugin[]="--db-plugin=null";
+  char querylog[]="--query-log-file="SRCDIR"/percona_playback/test/sysbench-slow.log";
+  char threadpool[]="--dispatcher-plugin=thread-pool"
+
+  char **dbplugin_argv= new char*[4];
+  dbplugin_argv[0]= argv[0];
+  dbplugin_argv[1]= dbplugin;
+  dbplugin_argv[2]= querylog;
+  dbplugin_argv[3]= threadpool;
+
+
+  assert(0 == percona_playback_argv(the_percona_playback, 3, dbplugin_argv));
+
+  struct percona_playback_run_result *r= percona_playback_run(the_percona_playback);
+
+  assert(r->err == 0);
+  assert(r->n_queries == 10150);
+  assert(r->n_log_entries = 10149);
+
+  percona_playback_destroy(&the_percona_playback);
+  return r->err;
+}

=== added directory 'percona_playback/thread_per_connection'
=== added file 'percona_playback/thread_per_connection/plugin.ini'
--- percona_playback/thread_per_connection/plugin.ini	1970-01-01 00:00:00 +0000
+++ percona_playback/thread_per_connection/plugin.ini	2012-11-23 20:15:24 +0000
@@ -0,0 +1,6 @@
+[plugin]
+title=Thread-per-connection dispatcher plugin
+description=Creates thread per each DB connection
+version=0.1
+static=yes
+load_by_default=yes

=== added file 'percona_playback/thread_per_connection/thread_per_connection.cc'
--- percona_playback/thread_per_connection/thread_per_connection.cc	1970-01-01 00:00:00 +0000
+++ percona_playback/thread_per_connection/thread_per_connection.cc	2012-11-23 20:15:24 +0000
@@ -0,0 +1,109 @@
+/* BEGIN LICENSE
+ * Copyright (C) 2011-2012 Percona Inc.
+ * This program is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2, as published
+ * by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranties of
+ * MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ * PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program.  If not, see <http://www.gnu.org/licenses/>.
+ * END LICENSE */
+
+#include "percona_playback/plugin.h"
+#include "percona_playback/db_thread.h"
+
+#include <tbb/concurrent_hash_map.h>
+
+class ThreadPerConnectionDispatcher :
+	public percona_playback::DispatcherPlugin
+{
+  typedef tbb::concurrent_hash_map<uint64_t, DBThread*> DBExecutorsTable;
+  DBExecutorsTable  executors;
+  void db_thread_func(DBThread *thread);
+  void start_thread(DBThread *thread);
+
+public:
+  ThreadPerConnectionDispatcher(std::string _name) :
+	  DispatcherPlugin(_name) {}
+
+  void dispatch(QueryEntryPtr query_entry);
+  bool finish_and_wait(uint64_t thread_id);
+  void finish_all_and_wait();
+};
+
+extern percona_playback::DBClientPlugin *g_dbclient_plugin;
+
+void
+ThreadPerConnectionDispatcher::dispatch(QueryEntryPtr query_entry)
+{
+  uint64_t thread_id= query_entry->getThreadId();
+  {
+    DBExecutorsTable::accessor a;
+    if (executors.insert(a, thread_id))
+    {
+      DBThread *db_thread= g_dbclient_plugin->create(thread_id);
+      a->second= db_thread;
+      db_thread->start_thread();
+    }
+    a->second->queries->push(query_entry);
+  }
+}
+
+bool
+ThreadPerConnectionDispatcher::finish_and_wait(uint64_t thread_id)
+{
+  DBThread *db_thread= NULL;
+  {
+    DBExecutorsTable::accessor a;
+    if (executors.find(a, thread_id))
+    {
+      db_thread= a->second;
+      executors.erase(a);
+    }
+  }
+
+  if (!db_thread)
+    return false;
+
+  db_thread->queries->push(QueryEntryPtr(new FinishEntry(thread_id)));
+  db_thread->join();
+
+  delete db_thread;
+
+  return true;
+}
+
+void
+ThreadPerConnectionDispatcher::finish_all_and_wait()
+{
+  QueryEntryPtr shutdown_command(new FinishEntry(0));
+
+  while(executors.size())
+  {
+    uint64_t thread_id;
+    DBThread *t;
+    {
+      DBExecutorsTable::const_iterator iter= executors.begin();
+      thread_id= (*iter).first;
+      t= (*iter).second;
+    }
+    executors.erase(thread_id);
+
+    t->queries->push(shutdown_command);
+    t->join();
+
+    delete t;
+  }
+}
+
+static void init_plugin(percona_playback::PluginRegistry &r)
+{
+  r.add("thread-per-connection",
+	new ThreadPerConnectionDispatcher("thread-per-connection"));
+}
+
+PERCONA_PLAYBACK_PLUGIN(init_plugin);

=== added directory 'percona_playback/thread_pool'
=== added file 'percona_playback/thread_pool/plugin.ini'
--- percona_playback/thread_pool/plugin.ini	1970-01-01 00:00:00 +0000
+++ percona_playback/thread_pool/plugin.ini	2012-11-23 20:15:24 +0000
@@ -0,0 +1,6 @@
+[plugin]
+title=Thread-pool dispatcher plugin
+description=Creates threads pool to process connections to server
+version=0.1
+static=yes
+load_by_default=yes

=== added file 'percona_playback/thread_pool/thread_pool.cc'
--- percona_playback/thread_pool/thread_pool.cc	1970-01-01 00:00:00 +0000
+++ percona_playback/thread_pool/thread_pool.cc	2012-11-23 20:15:24 +0000
@@ -0,0 +1,130 @@
+/* BEGIN LICENSE
+ * Copyright (C) 2011-2012 Percona Inc.
+ * This program is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2, as published
+ * by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranties of
+ * MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ * PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program.  If not, see <http://www.gnu.org/licenses/>.
+ * END LICENSE */
+
+#include "percona_playback/plugin.h"
+#include "percona_playback/db_thread.h"
+#include <percona_playback/gettext.h>
+#include <boost/program_options.hpp>
+#include <boost/shared_ptr.hpp>
+#include <boost/crc.hpp>
+#include <vector>
+
+#include <tbb/concurrent_hash_map.h>
+#include <stdio.h>
+
+extern percona_playback::DBClientPlugin *g_dbclient_plugin;
+
+class ThreadPoolDispatcher :
+	public percona_playback::DispatcherPlugin
+{
+  typedef std::vector<boost::shared_ptr<DBThread> > Workers;
+
+  unsigned			threads_count;
+  boost::
+    program_options::
+      options_description	options;
+  Workers			workers;
+
+public:
+  ThreadPoolDispatcher(std::string _name) :
+	  DispatcherPlugin(_name),
+	  threads_count(0),
+	  options("Threads-pool Options") {}
+
+  virtual void dispatch(QueryEntryPtr query_entry);
+  virtual bool finish_and_wait(uint64_t) { return true; }
+  virtual void finish_all_and_wait();
+  virtual void run();
+
+  boost::program_options::options_description* getProgramOptions();
+  int processOptions(boost::program_options::variables_map &vm);
+
+};
+
+void ThreadPoolDispatcher::run()
+{
+  for (unsigned i = 0; i < threads_count; ++i)
+  {
+    boost::shared_ptr<DBThread> db_thread(g_dbclient_plugin->create(i));
+    workers.push_back(db_thread);
+    db_thread->start_thread();
+  }
+}
+
+void ThreadPoolDispatcher::dispatch(QueryEntryPtr query_entry)
+{
+  /*
+    Each worker has its own queue. For some types of input plugins
+    it is important to execute query entries with the same thread id
+    by the same worker. That is why we choose worker by simple hash from
+    thread id.
+  */
+  uint64_t thread_id= query_entry->getThreadId();
+  boost::crc_32_type crc;
+  crc.process_bytes(&thread_id, sizeof(thread_id));
+  uint32_t worker_index = crc.checksum() % workers.size();
+  workers[worker_index]->queries->push(query_entry);
+}
+
+void ThreadPoolDispatcher::finish_all_and_wait()
+{
+  QueryEntryPtr shutdown_command(new FinishEntry(0));
+  for (Workers::iterator i = workers.begin(), end = workers.end(); i != end; ++i)
+    (*i)->queries->push(shutdown_command);
+  for (Workers::iterator i = workers.begin(), end = workers.end(); i != end; ++i)
+    (*i)->join();
+  workers.clear();
+}
+
+
+boost::program_options::options_description*
+ThreadPoolDispatcher::getProgramOptions()
+{
+  unsigned default_threads_count =
+    boost::thread::hardware_concurrency();
+  options.add_options()
+     ("thread-pool-threads-count",
+     boost::program_options::value<unsigned>(&threads_count)
+       ->default_value(default_threads_count),
+     _("The number of threads in thread pool. If this options is omitted "
+       "the number of threads equals to hardware concurency."))
+    ;
+
+  return &options;
+
+}
+
+int
+ThreadPoolDispatcher::processOptions(boost::program_options::variables_map &vm)
+{
+  if (!active &&
+      !vm["thread-pool-threads-count"].defaulted())
+  {
+    fprintf(stderr, _("thread-pool plugin is not selected, "
+                      "you shouldn't use this plugin-related "
+                      "command line options\n"));
+    return -1;
+  }
+
+  return 0;
+}
+
+static void init_plugin(percona_playback::PluginRegistry &r)
+{
+  r.add("thread-pool",
+	new ThreadPoolDispatcher("thread-pool"));
+}
+
+PERCONA_PLAYBACK_PLUGIN(init_plugin);