Implementation of client_state NBO operations.

- Implemented calls to enter and leave NBO phase one and two
- Extended client_state mode checking to include m_nbo
- Changed client_state state and mode change sanity checks to
  print a warning and assert() instead of throwing exceptions
  to be more graceful in release builds.

include/wsrep/client_state.hpp

@@ -100,10 +100,12 @@ namespace wsrep
             /** Client is in total order isolation mode */
             m_toi,
             /** Client is executing rolling schema upgrade */
-            m_rsu
+            m_rsu,
+            /** Client is executing NBO */
+            m_nbo
         };
 
-        static const int n_modes_ = m_rsu + 1;
+        static const int n_modes_ = m_nbo + 1;
         /**
          * Client state enumeration.
          *
@@ -669,6 +671,9 @@ namespace wsrep
          * @param buffer Buffer containing the action to execute inside
          *               total order isolation section
          * @param flags  Provider flags for TOI operation
+         * @todo Flags argument is not necessary for TOI operation,
+         *       they should always have start_transaction | commit
+         *       when passed to provider.
          *
          * @return Zero on success, non-zero otherwise.
          */
@@ -729,13 +734,45 @@ namespace wsrep
 
         /**
          * Begin non-blocking operation.
+         *
+         * The NBO operation is started by grabbing TOI critical
+         * section. The keys and buffer are certifed as in TOI
+         * operation. If the call fails due to error returned by
+         * the provider, the provider error code can be retrieved
+         * by current_error_status() call.
+         *
+         * @param keys Array of keys for NBO operation.
+         * @param buffer NBO write set
+         *
+         * @return Zero in case of success, non-zero in case of failure.
          */
-        int begin_nbo(const wsrep::key_array&);
+        int begin_nbo_phase_one(const wsrep::key_array& keys,
+                                const wsrep::const_buffer& buffer);
 
         /**
-         * End non-blocking operation
+         * End non-blocking operation phase after aquiring required
+         * resources for operation.
          */
-        int end_nbo();
+        int end_nbo_phase_one();
+
+        /**
+         * Begin non-blocking operation phase two. The keys argument
+         * passed to this call must contain the same keys which were
+         * passed to begin_nbo_phase_one().
+         *
+         * @todo Keys should be stored internally in client_state
+         *       so that they would not be required as an argument
+         *       here.
+         *
+         * @param keys Array of keys for non-blocking operation.
+         */
+        int begin_nbo_phase_two(const wsrep::key_array& keys);
+
+        /**
+         * End non-blocking operation phase two. This call will
+         * release TOI critical section and set the mode to m_local.
+         */
+        int end_nbo_phase_two();
 
         /**
          * Get reference to the client mutex.
@@ -1011,6 +1048,7 @@ namespace wsrep
         case wsrep::client_state::m_high_priority: return "high priority";
         case wsrep::client_state::m_toi: return "toi";
         case wsrep::client_state::m_rsu: return "rsu";
+        case wsrep::client_state::m_nbo: return "nbo";
         }
         return "unknown";
     }

src/client_state.cpp

@@ -24,7 +24,6 @@
 #include <sstream>
 #include <iostream>
 
-
 wsrep::provider& wsrep::client_state::provider() const
 {
     return server_state_.provider();
@@ -444,8 +443,105 @@ int wsrep::client_state::end_rsu()
     return ret;
 }
 
+///////////////////////////////////////////////////////////////////////////////
+//                                 NBO                                       //
+///////////////////////////////////////////////////////////////////////////////
 
+int wsrep::client_state::begin_nbo_phase_one(const wsrep::key_array& keys,
+                                             const wsrep::const_buffer& buffer)
+{
+    wsrep::unique_lock<wsrep::mutex> lock(mutex_);
+    assert(state_ == s_exec);
+    assert(mode_ == m_local);
+    assert(toi_mode_ == m_local);
 
+    /**
+     * @todo Implement retrying if the call fails due to certification
+     * failure.
+     */
+    lock.unlock();
+    enum wsrep::provider::status status(
+        provider().enter_toi(
+            id_, keys, buffer, toi_meta_,
+            wsrep::provider::flag::start_transaction));
+    lock.lock();
+    switch (status)
+    {
+    case wsrep::provider::success:
+        toi_mode_ = mode_;
+        mode(lock, m_nbo);
+        return 0;
+    default:
+        current_error_status_ = status;
+        return 1;
+    }
+}
+
+int wsrep::client_state::end_nbo_phase_one()
+{
+    assert(state_ == s_exec);
+    assert(mode_ == m_nbo);
+    assert(in_toi());
+
+    enum wsrep::provider::status status(provider().leave_toi(id_));
+    wsrep::unique_lock<wsrep::mutex> lock(mutex_);
+    int ret;
+    switch (status)
+    {
+    case wsrep::provider::success:
+        ret = 0;
+        break;
+    default:
+        current_error_status_ = status;
+        ret = 1;
+        break;
+    }
+    toi_meta_ = wsrep::ws_meta();
+    return ret;
+}
+
+int wsrep::client_state::begin_nbo_phase_two(const wsrep::key_array& keys)
+{
+    assert(state_ == s_exec);
+    assert(mode_ == m_nbo);
+
+    wsrep::unique_lock<wsrep::mutex> lock(mutex_);
+    enum wsrep::provider::status status(
+        provider().enter_toi(id_, keys, wsrep::const_buffer(), toi_meta_,
+                             wsrep::provider::flag::commit));
+    switch (status)
+    {
+    case wsrep::provider::success:
+        return 0;
+    default:
+        current_error_status_ = status;
+        return 1;
+    }
+}
+
+int wsrep::client_state::end_nbo_phase_two()
+{
+    assert(state_ == s_exec);
+    assert(mode_ == m_nbo);
+    assert(in_toi());
+    enum wsrep::provider::status status(
+        provider().leave_toi(id_));
+    wsrep::unique_lock<wsrep::mutex> lock(mutex_);
+    int ret;
+    switch (status)
+    {
+    case wsrep::provider::success:
+        ret = 0;
+        break;
+    default:
+        current_error_status_ = status;
+        ret = 1;
+        break;
+    }
+    toi_meta_ = wsrep::ws_meta();
+    mode(lock, m_local);
+    return ret;
+}
 ///////////////////////////////////////////////////////////////////////////////
 //                                 Misc                                      //
 ///////////////////////////////////////////////////////////////////////////////
@@ -519,7 +615,6 @@ void wsrep::client_state::debug_log_state(const char* context) const
                     << "," << to_c_string(mode_)
                     << "," << wsrep::to_string(current_error_)
                     << ")");
-    
 }
 
 void wsrep::client_state::state(
@@ -541,8 +636,10 @@ void wsrep::client_state::state(
         };
     if (!allowed[state_][state])
     {
-        wsrep::log_debug() << "client_state: Unallowed state transition: "
+        std::ostringstream os;
-                           << state_ << " -> " << state;
+        os << "client_state: Unallowed state transition: "
+           << state_ << " -> " << state;
+        wsrep::log_warning() << os.str();
         assert(0);
     }
     state_hist_.push_back(state_);
@@ -551,6 +648,7 @@ void wsrep::client_state::state(
     {
         state_hist_.erase(state_hist_.begin());
     }
+
 }
 
 void wsrep::client_state::mode(
@@ -560,17 +658,20 @@ void wsrep::client_state::mode(
     assert(lock.owns_lock());
 
     static const char allowed[n_modes_][n_modes_] =
-        {   /* u  l  h  t  r */
+        {   /* u  l  h  t  r  n */
-            {  0, 0, 0, 0, 0 }, /* undefined */
+            {  0, 0, 0, 0, 0, 0 }, /* undefined */
-            {  0, 0, 1, 1, 1 }, /* local */
+            {  0, 0, 1, 1, 1, 1 }, /* local */
-            {  0, 1, 0, 1, 0 }, /* high prio */
+            {  0, 1, 0, 1, 0, 1 }, /* high prio */
-            {  0, 1, 1, 0, 0 }, /* toi */
+            {  0, 1, 1, 0, 0, 0 }, /* toi */
-            {  0, 1, 0, 0, 0 }  /* rsu */
+            {  0, 1, 0, 0, 0, 0 }, /* rsu */
+            {  0, 1, 1, 0, 0, 0 }  /* nbo */
         };
     if (!allowed[mode_][mode])
     {
-        wsrep::log_debug() << "client_state: Unallowed mode transition: "
+        std::ostringstream os;
-                           << mode_ << " -> " << mode;
+        os << "client_state: Unallowed mode transition: "
+           << mode_ << " -> " << mode;
+        wsrep::log_warning() << os.str();
         assert(0);
     }
     mode_ = mode;

test/CMakeLists.txt

@@ -10,6 +10,7 @@ add_executable(wsrep-lib_test
   buffer_test.cpp
   gtid_test.cpp
   id_test.cpp
+  nbo_test.cpp
   server_context_test.cpp
   toi_test.cpp
   transaction_test.cpp

test/mock_provider.hpp

@@ -53,6 +53,9 @@ namespace wsrep
             , fragments_()
             , commit_fragments_()
             , rollback_fragments_()
+            , toi_write_sets_()
+            , toi_start_transaction_()
+            , toi_commit_()
         { }
 
         enum wsrep::provider::status
@@ -238,19 +241,22 @@ namespace wsrep
                                                const wsrep::key_array&,
                                                const wsrep::const_buffer&,
                                                wsrep::ws_meta& toi_meta,
-                                               int)
+                                               int flags)
             WSREP_OVERRIDE
         {
-                ++group_seqno_;
+            ++group_seqno_;
-                wsrep::gtid gtid(group_id_, wsrep::seqno(group_seqno_));
+            wsrep::gtid gtid(group_id_, wsrep::seqno(group_seqno_));
-                wsrep::stid stid(server_id_,
+            wsrep::stid stid(server_id_,
-                                 wsrep::transaction_id::undefined(),
+                             wsrep::transaction_id::undefined(),
-                                 client_id);
+                             client_id);
-                toi_meta = wsrep::ws_meta(gtid, stid,
+            toi_meta = wsrep::ws_meta(gtid, stid,
-                                          wsrep::seqno(group_seqno_ - 1),
+                                      wsrep::seqno(group_seqno_ - 1),
-                                          wsrep::provider::flag::start_transaction |
+                                      flags);
-                                          wsrep::provider::flag::commit);
+            ++toi_write_sets_;
-
+            if (flags & wsrep::provider::flag::start_transaction)
+                ++toi_start_transaction_;
+            if (flags & wsrep::provider::flag::commit)
+                ++toi_commit_;
             return wsrep::provider::success;
         }
 
@@ -325,7 +331,9 @@ namespace wsrep
         size_t fragments() const { return fragments_; }
         size_t commit_fragments() const { return commit_fragments_; }
         size_t rollback_fragments() const { return rollback_fragments_; }
-
+        size_t toi_write_sets() const { return toi_write_sets_; }
+        size_t toi_start_transaction() const { return toi_start_transaction_; }
+        size_t toi_commit() const { return toi_commit_; }
     private:
         wsrep::id group_id_;
         wsrep::id server_id_;
@@ -335,6 +343,9 @@ namespace wsrep
         size_t fragments_;
         size_t commit_fragments_;
         size_t rollback_fragments_;
+        size_t toi_write_sets_;
+        size_t toi_start_transaction_;
+        size_t toi_commit_;
     };
 }
 

test/nbo_test.cpp

@@ -0,0 +1,68 @@
+/*
+ * Copyright (C) 2019 Codership Oy <info@codership.com>
+ *
+ * This file is part of wsrep-lib.
+ *
+ * Wsrep-lib is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * Wsrep-lib is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY 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 wsrep-lib.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "wsrep/client_state.hpp"
+
+#include "client_state_fixture.hpp"
+
+#include <boost/test/unit_test.hpp>
+
+BOOST_FIXTURE_TEST_CASE(test_local_nbo,
+                        replicating_client_fixture_sync_rm)
+{
+    // NBO is executed in two consecutive TOI operations
+    BOOST_REQUIRE(cc.mode() == wsrep::client_state::m_local);
+    // First phase certifies the write set and enters TOI
+    wsrep::key key(wsrep::key::exclusive);
+    key.append_key_part("k1", 2);
+    key.append_key_part("k2", 2);
+    wsrep::key_array keys{key};
+    std::string data("data");
+    BOOST_REQUIRE(cc.begin_nbo_phase_one(
+                      keys,
+                      wsrep::const_buffer(data.data(),
+                                          data.size())) == 0);
+    BOOST_REQUIRE(cc.mode() == wsrep::client_state::m_nbo);
+    BOOST_REQUIRE(cc.in_toi());
+    BOOST_REQUIRE(cc.toi_mode() == wsrep::client_state::m_local);
+    // After required resoureces have been grabbed, NBO leave should
+    // end TOI but leave the client state in m_nbo.
+    BOOST_REQUIRE(cc.end_nbo_phase_one() == 0);
+    BOOST_REQUIRE(cc.mode() == wsrep::client_state::m_nbo);
+    BOOST_REQUIRE(cc.in_toi() == false);
+    BOOST_REQUIRE(cc.toi_mode() == wsrep::client_state::m_local);
+    // Second phase replicates the NBO end event and grabs TOI
+    // again for finalizing the NBO.
+    BOOST_REQUIRE(cc.begin_nbo_phase_two(keys) == 0);
+    BOOST_REQUIRE(cc.mode() == wsrep::client_state::m_nbo);
+    BOOST_REQUIRE(cc.in_toi());
+    BOOST_REQUIRE(cc.toi_mode() == wsrep::client_state::m_local);
+    // End of NBO phase will leave TOI and put the client state back to
+    // m_local
+    BOOST_REQUIRE(cc.end_nbo_phase_two() == 0);
+    BOOST_REQUIRE(cc.mode() == wsrep::client_state::m_local);
+    BOOST_REQUIRE(cc.in_toi() == false);
+    BOOST_REQUIRE(cc.toi_mode() == wsrep::client_state::m_local);
+
+    // There must have been two toi write sets, one with
+    // start transaction flag, another with commit flag.
+    BOOST_REQUIRE(sc.provider().toi_write_sets() == 2);
+    BOOST_REQUIRE(sc.provider().toi_start_transaction() == 1);
+    BOOST_REQUIRE(sc.provider().toi_commit() == 1);
+}

test/toi_test.cpp

@@ -27,6 +27,7 @@ BOOST_FIXTURE_TEST_CASE(test_toi_mode,
                         replicating_client_fixture_sync_rm)
 {
     BOOST_REQUIRE(cc.mode() == wsrep::client_state::m_local);
+    BOOST_REQUIRE(cc.toi_mode() == wsrep::client_state::m_local);
     wsrep::key key(wsrep::key::exclusive);
     key.append_key_part("k1", 2);
     key.append_key_part("k2", 2);
@@ -40,6 +41,10 @@ BOOST_FIXTURE_TEST_CASE(test_toi_mode,
     BOOST_REQUIRE(cc.toi_mode() == wsrep::client_state::m_local);
     BOOST_REQUIRE(cc.leave_toi() == 0);
     BOOST_REQUIRE(cc.mode() == wsrep::client_state::m_local);
+    BOOST_REQUIRE(cc.toi_mode() == wsrep::client_state::m_local);
+    BOOST_REQUIRE(sc.provider().toi_write_sets() == 1);
+    BOOST_REQUIRE(sc.provider().toi_start_transaction() == 1);
+    BOOST_REQUIRE(sc.provider().toi_commit() == 1);
 }
 
 BOOST_FIXTURE_TEST_CASE(test_toi_applying,
@@ -50,6 +55,7 @@ BOOST_FIXTURE_TEST_CASE(test_toi_applying,
                    cc.after_commit()) == 0);
     cc.after_applying();
 
+    BOOST_REQUIRE(cc.toi_mode() == wsrep::client_state::m_local);
     wsrep::ws_meta ws_meta(wsrep::gtid(wsrep::id("1"), wsrep::seqno(2)),
                            wsrep::stid(sc.id(),
                                        wsrep::transaction_id::undefined(),
@@ -61,5 +67,6 @@ BOOST_FIXTURE_TEST_CASE(test_toi_applying,
     BOOST_REQUIRE(cc.in_toi());
     BOOST_REQUIRE(cc.toi_mode() == wsrep::client_state::m_high_priority);
     BOOST_REQUIRE(cc.leave_toi() == 0);
+    BOOST_REQUIRE(cc.toi_mode() == wsrep::client_state::m_local);
     cc.after_applying();
 }