#23 before_command() wait for ongoing rollbacks leaks

Storing background rollbacker thread's ID in client state.
Tis can be used for detecting if there is ongoing background rollback,
and client should keep waiting in before_command() entry to avoid conflicts
in accessing client state during background rollbacking.

There is new public method for client_state: prepare_for_background_rollback()
Background rollbacker thread should call this, before backround rollbacking
starts, to assign himself as the rollbacker for the client.

sync_rollback_complete() method has been modified to release the backround
rollbacker thread from client_state

include/wsrep/client_state.hpp

@@ -425,15 +425,29 @@ namespace wsrep
             return transaction_.after_rollback();
         }
 
+        /**
+         * This marks the client with external rollbacker thread ID
+         *
+         */
+        void prepare_for_background_rollback()
+        {
+            wsrep::unique_lock<wsrep::mutex> lock(mutex_);
+            assert(state_ == s_idle && mode_ == m_local);
+            set_rollbacker(wsrep::this_thread::get_id());
+        }
+
         /**
          * This method should be called by the background rollbacker
          * thread after the rollback is complete. This will allow
          * the client to proceed with command execution.
          */
         void sync_rollback_complete()
         {
+            wsrep::unique_lock<wsrep::mutex> lock(mutex_);
+
             assert(state_ == s_idle && mode_ == m_local &&
                    transaction_.state() == wsrep::transaction::s_aborted);
+           set_rollbacker(owning_thread_id_);
             cond_.notify_all();
         }
         /** @} */
@@ -707,6 +721,7 @@ namespace wsrep
                      enum mode mode)
             : owning_thread_id_(wsrep::this_thread::get_id())
             , current_thread_id_(owning_thread_id_)
+            , rollbacker_thread_id_(owning_thread_id_)
             , mutex_(mutex)
             , cond_(cond)
             , server_state_(server_state)
@@ -749,6 +764,7 @@ namespace wsrep
 
         wsrep::thread::id owning_thread_id_;
         wsrep::thread::id current_thread_id_;
+        wsrep::thread::id rollbacker_thread_id_;
         wsrep::mutex& mutex_;
         wsrep::condition_variable& cond_;
         wsrep::server_state& server_state_;
@@ -766,6 +782,21 @@ namespace wsrep
         int debug_log_level_;
         enum wsrep::client_error current_error_;
         enum wsrep::provider::status current_error_status_;
+
+        /**
+         * Assigns external rollbacker thread for the client
+         * this will block client in before_command(), until
+         * rolbacker has released the client
+         */
+        void set_rollbacker(wsrep::thread::id id)
+        {
+            rollbacker_thread_id_ = id;
+        }
+
+        bool has_rollbacker()
+        {
+            return(!(rollbacker_thread_id_ == owning_thread_id_));
+        }
     };
 
     static inline const char* to_c_string(

src/client_state.cpp

@@ -37,6 +37,7 @@ void wsrep::client_state::open(wsrep::client_id id)
     debug_log_state("open: enter");
     owning_thread_id_ = wsrep::this_thread::get_id();
     current_thread_id_ = owning_thread_id_;
+    rollbacker_thread_id_ = owning_thread_id_;
     state(lock, s_idle);
     id_ = id;
     debug_log_state("open: leave");
@@ -86,7 +87,7 @@ int wsrep::client_state::before_command()
     if (transaction_.active() &&
         server_state_.rollback_mode() == wsrep::server_state::rm_sync)
     {
-        while (transaction_.state() == wsrep::transaction::s_aborting)
+        while (has_rollbacker())
         {
             cond_.wait(lock);
         }