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

@@ -104,7 +104,7 @@ namespace wsrep
             m_nbo
         };
 
-        static const int n_modes_ = m_rsu + 1;
+        static const int n_modes_ = m_nbo + 1;
         /**
          * Client state enumeration.
          *
@@ -638,16 +638,44 @@ 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_phase_one(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_phase_one();
 
-        int begin_nbo_phase_two();
+        /**
+         * 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();
 
         /**

src/client_state.cpp

@@ -24,7 +24,6 @@
 #include <sstream>
 #include <iostream>
 
-
 wsrep::provider& wsrep::client_state::provider() const
 {
     return server_state_.provider();
@@ -443,8 +442,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                                      //
 ///////////////////////////////////////////////////////////////////////////////
@@ -518,7 +614,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(
@@ -540,8 +635,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_);
@@ -550,6 +647,7 @@ void wsrep::client_state::state(
     {
         state_hist_.erase(state_hist_.begin());
     }
+
 }
 
 void wsrep::client_state::mode(
@@ -559,17 +657,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/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

@@ -33,7 +33,11 @@ BOOST_FIXTURE_TEST_CASE(test_local_nbo,
     key.append_key_part("k1", 2);
     key.append_key_part("k2", 2);
     wsrep::key_array keys{key};
-    BOOST_REQUIRE(cc.begin_nbo_phase_one(keys) == 0);
+    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);
@@ -45,7 +49,7 @@ BOOST_FIXTURE_TEST_CASE(test_local_nbo,
     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() == 0);
+    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);
@@ -55,4 +59,10 @@ BOOST_FIXTURE_TEST_CASE(test_local_nbo,
     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

@@ -42,6 +42,9 @@ BOOST_FIXTURE_TEST_CASE(test_toi_mode,
     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,