Add 2pc switch to dbsim, release critical section in callback

dbsim/db_client.cpp

@@ -93,6 +93,15 @@ int db::client::client_command(F f)
     return err;
 }
 
+static void release_commit_critical_section(void* ptr)
+{
+    auto* crit = static_cast<db::server::commit_critical_section*>(ptr);
+    if (crit->lock.owns_lock())
+    {
+        crit->lock.unlock();
+    }
+}
+
 void db::client::run_one_transaction()
 {
     if (params_.sync_wait)
@@ -152,10 +161,15 @@ void db::client::run_one_transaction()
                 commit_crit.lock.unlock();
             }
 
+            wsrep::provider::seq_cb seq_cb {
+                &commit_crit,
+                release_commit_critical_section
+            };
+
             assert(err == 0);
-            if (do_2pc())
+            if (params_.do_2pc)
             {
-                err = err || client_state_.before_prepare();
+                err = err || client_state_.before_prepare(&seq_cb);
                 err = err || client_state_.after_prepare();
             }
             err = err || client_state_.before_commit();

dbsim/db_client.hpp

@@ -62,7 +62,6 @@ namespace db
         void start();
         wsrep::client_state& client_state() { return client_state_; }
         wsrep::client_service& client_service() { return client_service_; }
-        bool do_2pc() const { return false; }
     private:
         friend class db::server_state;
         friend class db::client_service;

dbsim/db_params.cpp

@@ -98,6 +98,9 @@ db::params db::parse_args(int argc, char** argv)
         ("check-sequential-consistency",
          po::value<bool>(&params.check_sequential_consistency),
          "Check if the provider provides sequential consistency")
+        ("do-2pc",
+         po::value<bool>(&params.do_2pc),
+         "Run commits in 2pc")
         ;
     try
     {

dbsim/db_params.hpp

@@ -47,6 +47,7 @@ namespace db
         bool cond_checks{false};
         int tls_service{0};
         bool check_sequential_consistency{false};
+        bool do_2pc{false};
     };
 
     params parse_args(int argc, char** argv);