* Renamed client service rollback() to bf_rollback() to better

describe its purpose.
* Raise deadlock error for BF aborted transaction in after_statement()
  call if the error is not set yet.

dbsim/db_client_service.cpp

@@ -18,7 +18,7 @@ int db::client_service::commit(const wsrep::ws_handle&,
     return ret;
 }
 
-int db::client_service::rollback()
+int db::client_service::bf_rollback()
 {
     db::client* client(client_state_.client());
     int ret(client_state_.before_rollback());

dbsim/db_client_service.hpp

@@ -75,7 +75,7 @@ namespace db
 
         int commit(const wsrep::ws_handle&, const wsrep::ws_meta&) override;
 
-        int rollback() override;
+        int bf_rollback() override;
 
         void will_replay() override
         { }

include/wsrep/client_service.hpp

@@ -81,10 +81,16 @@ namespace wsrep
         // Rollback
         //
         /**
-         * Roll back all transactions which are currently active
+         * Perform brute force rollback.
-         * in the client session.
+         *
+         * This method may be called from two contexts, either from
+         * client state methods when the BF abort condition is detected,
+         * or from the background rollbacker thread. The task for this
+         * method is to release all reasources held by the client
+         * after BF abort so that the high priority thread can continue
+         * applying.
          */
-        virtual int rollback() = 0;
+        virtual int bf_rollback() = 0;
 
         //
         // Interface to global server state

include/wsrep/client_state.hpp

@@ -267,6 +267,7 @@ namespace wsrep
          */
         int start_transaction(const wsrep::transaction_id& id)
         {
+            wsrep::unique_lock<wsrep::mutex> lock(mutex_);
             assert(state_ == s_exec);
             return transaction_.start_transaction(id);
         }
@@ -327,6 +328,7 @@ namespace wsrep
         int start_transaction(const wsrep::ws_handle& wsh,
                               const wsrep::ws_meta& meta)
         {
+            wsrep::unique_lock<wsrep::mutex> lock(mutex_);
             assert(mode_ == m_high_priority);
             return transaction_.start_transaction(wsh, meta);
         }

src/client_state.cpp

@@ -35,7 +35,7 @@ void wsrep::client_state::close()
     lock.unlock();
     if (transaction_.active())
     {
-        client_service_.rollback();
+        client_service_.bf_rollback();
     }
     debug_log_state("close: leave");
 }
@@ -90,7 +90,7 @@ int wsrep::client_state::before_command()
                    wsrep::server_state::rm_async);
             override_error(wsrep::e_deadlock_error);
             lock.unlock();
-            client_service_.rollback();
+            client_service_.bf_rollback();
             (void)transaction_.after_statement();
             lock.lock();
             assert(transaction_.state() ==
@@ -128,7 +128,7 @@ void wsrep::client_state::after_command_before_result()
     {
         override_error(wsrep::e_deadlock_error);
         lock.unlock();
-        client_service_.rollback();
+        client_service_.bf_rollback();
         (void)transaction_.after_statement();
         lock.lock();
         assert(transaction_.state() == wsrep::transaction::s_aborted);
@@ -148,7 +148,7 @@ void wsrep::client_state::after_command_after_result()
         transaction_.state() == wsrep::transaction::s_must_abort)
     {
         lock.unlock();
-        client_service_.rollback();
+        client_service_.bf_rollback();
         lock.lock();
         assert(transaction_.state() == wsrep::transaction::s_aborted);
         override_error(wsrep::e_deadlock_error);
@@ -203,7 +203,7 @@ wsrep::client_state::after_statement()
     (void)transaction_.after_statement();
     if (current_error() == wsrep::e_deadlock_error)
     {
-        if (mode_ == m_replicating && client_service_.is_autocommit())
+        if (mode_ == m_replicating)
         {
             debug_log_state("after_statement: may_retry");
             return asr_may_retry;

src/transaction.cpp

@@ -49,6 +49,7 @@ wsrep::transaction::~transaction()
 int wsrep::transaction::start_transaction(
     const wsrep::transaction_id& id)
 {
+    debug_log_state("start_transaction enter");
     assert(active() == false);
     id_ = id;
     state_ = s_executing;
@@ -59,10 +60,13 @@ int wsrep::transaction::start_transaction(
     {
     case wsrep::client_state::m_local:
     case wsrep::client_state::m_high_priority:
+        debug_log_state("start_transaction success");
         return 0;
     case wsrep::client_state::m_replicating:
+        debug_log_state("start_transaction success");
         return provider().start_transaction(ws_handle_);
     default:
+        debug_log_state("start_transaction error");
         assert(0);
         return 1;
     }
@@ -518,7 +522,7 @@ int wsrep::transaction::after_statement()
     case s_cert_failed:
         client_state_.override_error(wsrep::e_deadlock_error);
         lock.unlock();
-        ret = client_service_.rollback();
+        ret = client_service_.bf_rollback();
         lock.lock();
         if (state() != s_must_replay)
         {
@@ -554,6 +558,10 @@ int wsrep::transaction::after_statement()
         break;
     }
     case s_aborted:
+        if (bf_aborted() && client_state_.current_error() == wsrep::e_success)
+        {
+            client_state_.override_error(wsrep::e_deadlock_error);
+        }
         break;
     default:
         assert(0);
@@ -799,7 +807,7 @@ int wsrep::transaction::certify_fragment(
         sr_lock.lock();
         sr_transaction.state(sr_lock, s_must_abort);
         sr_lock.unlock();
-        sr_client_state.client_service().rollback();
+        sr_client_state.client_service().bf_rollback();
         ret = 1;
         break;
     }

test/mock_client_state.cpp

@@ -31,7 +31,7 @@ int wsrep::mock_client_service::commit(
     return ret;
 }
 
-int wsrep::mock_client_service::rollback()
+int wsrep::mock_client_service::bf_rollback()
 {
     int ret(0);
     if (client_state_.before_rollback())

test/mock_client_state.hpp

@@ -28,7 +28,7 @@ namespace wsrep
         {
             if (transaction().active())
             {
-                (void)client_service().rollback();
+                (void)client_service().bf_rollback();
             }
         }
     private:
@@ -66,7 +66,7 @@ namespace wsrep
         int commit(const wsrep::ws_handle&, const wsrep::ws_meta&)
             WSREP_OVERRIDE;
 
-        int rollback() WSREP_OVERRIDE;
+        int bf_rollback() WSREP_OVERRIDE;
 
         bool is_autocommit() const WSREP_OVERRIDE
         { return is_autocommit_; }