MDEV-17565: Sporadic Galera failures when testing MariaDB with mtr

galera/src/certification.cpp

@@ -808,14 +808,15 @@ galera::Certification::do_test_preordered(TrxHandle* trx)
 }
 
 
-galera::Certification::Certification(gu::Config& conf, ServiceThd& thd)
+galera::Certification::Certification(gu::Config& conf, ServiceThd& thd, gcache::GCache& gcache)
     :
     version_               (-1),
     trx_map_               (),
     cert_index_            (),
     cert_index_ng_         (),
     deps_set_              (),
     service_thd_           (thd),
+    gcache_                (gcache),
     mutex_                 (),
     trx_size_warn_count_   (0),
     initial_position_      (-1),
@@ -1066,7 +1067,7 @@ wsrep_seqno_t galera::Certification::set_trx_committed(TrxHandle* trx)
             deps_set_.erase(i);
         }
 
-        if (gu_unlikely(index_purge_required()))
+        if (gu_unlikely(gcache_.cleanup_required() || index_purge_required()))
         {
             ret = get_safe_to_discard_seqno_();
         }

galera/src/certification.hpp

@@ -48,7 +48,7 @@ namespace galera
             TEST_FAILED
         } TestResult;
 
-        Certification(gu::Config& conf, ServiceThd& thd);
+        Certification(gu::Config& conf, ServiceThd& thd, gcache::GCache& gcache);
         ~Certification();
 
         void assign_initial_position(wsrep_seqno_t seqno, int versiono);
@@ -183,6 +183,7 @@ namespace galera
         CertIndexNG   cert_index_ng_;
         DepsSet       deps_set_;
         ServiceThd&   service_thd_;
+        gcache::GCache& gcache_;
         gu::Mutex     mutex_;
         size_t        trx_size_warn_count_;
         wsrep_seqno_t initial_position_;

galera/src/galera_gcs.hpp

@@ -43,7 +43,8 @@ namespace galera
         virtual ssize_t replv(const WriteSetVector&,
                               gcs_action& act, bool) = 0;
         virtual ssize_t repl (gcs_action& act, bool) = 0;
-        virtual gcs_seqno_t caused() = 0;
+        virtual void    caused(gcs_seqno_t& seqno,
+                               gu::datetime::Date& wait_until) = 0;
         virtual ssize_t schedule() = 0;
         virtual ssize_t interrupt(ssize_t) = 0;
         virtual ssize_t resume_recv() = 0;
@@ -141,7 +142,23 @@ namespace galera
             return gcs_repl(conn_, &act, scheduled);
         }
 
-        gcs_seqno_t caused() { return gcs_caused(conn_);   }
+        void caused(gcs_seqno_t& seqno, gu::datetime::Date& wait_until)
+        {
+            long err;
+
+            while ((err = gcs_caused(conn_, seqno)) == -EAGAIN &&
+                   gu::datetime::Date::calendar() < wait_until)
+            {
+                usleep(1000);
+            }
+
+            if (err == -EAGAIN) err = -ETIMEDOUT;
+
+            if (err < 0)
+            {
+                gu_throw_error(-err);
+            }
+        }
 
         ssize_t schedule()   { return gcs_schedule(conn_); }
 
@@ -233,6 +250,11 @@ namespace galera
 
         size_t  max_action_size() const { return GCS_MAX_ACT_SIZE; }
 
+        void join_notification()
+        {
+            gcs_join_notification(conn_);
+        }
+
     private:
 
         Gcs(const Gcs&);
@@ -311,7 +333,10 @@ namespace galera
             return ret;
         }
 
-        gcs_seqno_t caused() { return global_seqno_; }
+        void caused(gcs_seqno_t& seqno, gu::datetime::Date& wait_until)
+        {
+            seqno = global_seqno_;
+        }
 
         ssize_t schedule()
         {

galera/src/galera_service_thd.cpp

@@ -52,21 +52,43 @@ galera::ServiceThd::thd_func (void* arg)
         {
             if (data.act_ & A_LAST_COMMITTED)
             {
-                ssize_t const ret
-                    (st->gcs_.set_last_applied(data.last_committed_));
+                static const size_t max_set_attempts(4);
+                size_t attempts = 0;
+                ssize_t ret;
 
-                if (gu_unlikely(ret < 0))
+                do
                 {
-                    log_warn << "Failed to report last committed "
-                             << data.last_committed_ << ", " << ret
-                             << " (" << strerror (-ret) << ')';
-                    // @todo: figure out what to do in this case
+                    ret = st->gcs_.set_last_applied(data.last_committed_);
+
+                    if (gu_likely(ret != -EINTR))
+                    {
+                        break;
+                    }
+
+                    attempts++;
+
+                    // gcs_set_last_applied() may return EINTR if the send
+                    // monitor was interruped, this is not an error and
+                    // in this case there is no need to display a warning:
+                    log_debug << "Reporting of last committed was "
+                                 "interrupted: "
+                              << data.last_committed_
+                              << "\nRetrying " << attempts << "th time";
                 }
-                else
+                while (attempts != max_set_attempts);
+
+                if (gu_likely(ret >= 0))
                 {
                     log_debug << "Reported last committed: "
                               << data.last_committed_;
                 }
+                else
+                {
+                    log_warn << "Failed to report last committed "
+                             << data.last_committed_ << ", " << ret
+                             << " (" << strerror (-ret) << ')';
+                    // @todo: figure out what to do in this case
+                }
             }
 
             if (data.act_ & A_RELEASE_SEQNO)

galera/src/replicator_smm.cpp

@@ -164,7 +164,7 @@ galera::ReplicatorSMM::ReplicatorSMM(const struct wsrep_init_args* args)
     ist_receiver_       (config_, slave_pool_, args->node_address),
     ist_senders_        (gcs_, gcache_),
     wsdb_               (),
-    cert_               (config_, service_thd_),
+    cert_               (config_, service_thd_, gcache_),
     local_monitor_      (),
     apply_monitor_      (),
     commit_monitor_     (),
@@ -859,8 +859,8 @@ wsrep_status_t galera::ReplicatorSMM::replay_trx(TrxHandle* trx, void* trx_ctx)
         ApplyOrder ao(*trx);
         gu_trace(apply_monitor_.enter(ao));
         trx->set_state(TrxHandle::S_MUST_REPLAY_CM);
-        // fall through
     }
+    // fall through
     case TrxHandle::S_MUST_REPLAY_CM:
         if (co_mode_ != CommitOrder::BYPASS)
         {
@@ -895,6 +895,13 @@ wsrep_status_t galera::ReplicatorSMM::replay_trx(TrxHandle* trx, void* trx_ctx)
         catch (gu::Exception& e)
         {
             st_.mark_corrupt();
+
+            /* Before doing a graceful exit ensure that node isolate itself
+            from the cluster. This will cause the quorum to re-evaluate
+            and if minority nodes are left with different set of data
+            they can turn non-Primary to avoid further data consistency issue. */
+            param_set("gmcast.isolate", "1");
+
             throw;
         }
 
@@ -967,12 +974,19 @@ wsrep_status_t galera::ReplicatorSMM::post_rollback(TrxHandle* trx)
 
 wsrep_status_t galera::ReplicatorSMM::causal_read(wsrep_gtid_t* gtid)
 {
-    wsrep_seqno_t cseq(static_cast<wsrep_seqno_t>(gcs_.caused()));
+    wsrep_seqno_t cseq;
+    gu::datetime::Date wait_until(gu::datetime::Date::calendar() +
+                                  causal_read_timeout_);
 
-    if (cseq < 0)
+    try
     {
-        log_warn << "gcs_caused() returned " << cseq << " (" << strerror(-cseq)
-                 << ')';
+        gcs_.caused(cseq, wait_until);
+        assert(cseq >= 0);
+    }
+    catch (gu::Exception& e)
+    {
+        log_warn << "gcs_caused() returned " << -e.get_errno()
+                 << " (" << strerror(e.get_errno()) << ")";
         return WSREP_TRX_FAIL;
     }
 
@@ -985,8 +999,6 @@ wsrep_status_t galera::ReplicatorSMM::causal_read(wsrep_gtid_t* gtid)
         // at monitor drain and disallowing further waits until
         // configuration change related operations (SST etc) have been
         // finished.
-        gu::datetime::Date wait_until(gu::datetime::Date::calendar()
-                                      + causal_read_timeout_);
         if (gu_likely(co_mode_ != CommitOrder::BYPASS))
         {
             commit_monitor_.wait(cseq, wait_until);
@@ -1198,6 +1210,13 @@ galera::ReplicatorSMM::sst_sent(const wsrep_gtid_t& state_id, int const rcode)
     if (state_() != S_DONOR)
     {
         log_error << "sst sent called when not SST donor, state " << state_();
+        /* If sst_sent() fails node should restore itself back to the joined
+        state. The sst_sent function can fail. commonly due to network errors,
+        where DONOR may lose connectivity to JOINER (or existing cluster).
+        But on re-join it should restore the original state without waiting
+        for transition to JOINER state (DONOR->JOINER->JOINED->SYNCED).
+        SST failure on JOINER will gracefully shutdown the joiner.*/
+        gcs_.join_notification();
         return WSREP_CONN_FAIL;
     }
 
@@ -1247,7 +1266,14 @@ void galera::ReplicatorSMM::process_trx(void* recv_ctx, TrxHandle* trx)
 
             log_fatal << "Failed to apply trx: " << *trx;
             log_fatal << e.what();
-            log_fatal << "Node consistency compromized, aborting...";
+            log_fatal << "Node consistency compromised, aborting...";
+
+            /* Before doing a graceful exit ensure that node isolate itself
+            from the cluster. This will cause the quorum to re-evaluate
+            and if minority nodes are left with different set of data
+            they can turn non-Primary to avoid further data consistency issue. */
+            param_set("gmcast.isolate", "1");
+
             abort();
         }
         break;
@@ -1412,6 +1438,12 @@ galera::ReplicatorSMM::process_conf_change(void*                    recv_ctx,
     size_t app_req_len(0);
 
     const_cast<wsrep_view_info_t&>(view_info).state_gap = st_required;
+
+    // We need to set the protocol version BEFORE the view callback, so that
+    // any version-dependent code is run using the correct version instead of -1.
+    if (view_info.view >= 0) // Primary configuration
+        establish_protocol_versions (repl_proto);
+
     wsrep_cb_status_t const rcode(
         view_cb_(app_ctx_, recv_ctx, &view_info, 0, 0, &app_req, &app_req_len));
 
@@ -1420,6 +1452,7 @@ galera::ReplicatorSMM::process_conf_change(void*                    recv_ctx,
         assert(app_req_len <= 0);
         log_fatal << "View callback failed. This is unrecoverable, "
                   << "restart required.";
+        local_monitor_.leave(lo);
         close();
         abort();
     }
@@ -1428,14 +1461,13 @@ galera::ReplicatorSMM::process_conf_change(void*                    recv_ctx,
         log_fatal << "Local state UUID " << state_uuid_
                   << " is different from group state UUID " << group_uuid
                   << ", and SST request is null: restart required.";
+        local_monitor_.leave(lo);
         close();
         abort();
     }
 
     if (view_info.view >= 0) // Primary configuration
     {
-        establish_protocol_versions (repl_proto);
-
         // we have to reset cert initial position here, SST does not contain
         // cert index yet (see #197).
         // Also this must be done before releasing GCache buffers.
@@ -1528,6 +1560,7 @@ galera::ReplicatorSMM::process_conf_change(void*                    recv_ctx,
         {
             log_fatal << "Internal error: unexpected next state for "
                       << "non-prim: " << next_state << ". Restart required.";
+            local_monitor_.leave(lo);
             close();
             abort();
         }
@@ -1870,6 +1903,6 @@ galera::ReplicatorSMM::update_state_uuid (const wsrep_uuid_t& uuid)
 void
 galera::ReplicatorSMM::abort()
 {
-    gcs_.close();
+    close();
     gu_abort();
 }

galera/src/replicator_str.cpp

@@ -536,7 +536,9 @@ ReplicatorSMM::prepare_state_request (const void* const   sst_req,
             }
             catch (gu::Exception& e)
             {
-                log_warn
+                log_info << "State gap can't be serviced using IST."
+                            " Switching to SST";
+                log_info
                     << "Failed to prepare for incremental state transfer: "
                     << e.what() << ". IST will be unavailable.";
             }

galera/tests/write_set_check.cpp

@@ -51,6 +51,7 @@ namespace
 
         gu::Config&         conf() { return conf_; }
         galera::ServiceThd& thd()  { return thd_;  }
+        gcache::GCache&     gcache() { return gcache_; }
 
     private:
 
@@ -408,7 +409,7 @@ START_TEST(test_cert_hierarchical_v1)
     size_t nws(sizeof(wsi)/sizeof(wsi[0]));
 
     TestEnv env;
-    galera::Certification cert(env.conf(), env.thd());
+    galera::Certification cert(env.conf(), env.thd(), env.gcache());
     int const version(1);
     cert.assign_initial_position(0, version);
     galera::TrxHandle::Params const trx_params("", version,KeySet::MAX_VERSION);
@@ -530,7 +531,7 @@ START_TEST(test_cert_hierarchical_v2)
     size_t nws(sizeof(wsi)/sizeof(wsi[0]));
 
     TestEnv env;
-    galera::Certification cert(env.conf(), env.thd());
+    galera::Certification cert(env.conf(), env.thd(), env.gcache());
 
     cert.assign_initial_position(0, version);
     galera::TrxHandle::Params const trx_params("", version,KeySet::MAX_VERSION);
@@ -580,7 +581,8 @@ START_TEST(test_trac_726)
 
     const int version(2);
     TestEnv env;
-    galera::Certification cert(env.conf(), env.thd());
+    galera::Certification cert(env.conf(), env.thd(), env.gcache());
+
     galera::TrxHandle::Params const trx_params("", version,KeySet::MAX_VERSION);
     wsrep_uuid_t uuid1 = {{1, }};
     wsrep_uuid_t uuid2 = {{2, }};

gcache/src/GCache.hpp

@@ -99,6 +99,14 @@ namespace gcache
                                    int64_t& seqno_d,
                                    ssize_t& size);
 
+        /*!
+         * Implements the cleanup policy test.
+         */
+        bool cleanup_required()
+        {
+            return (params.keep_pages_size() && ps.total_size() > params.keep_pages_size());
+        }
+
         class Buffer
         {
         public:

gcs/src/SConscript

@@ -7,6 +7,7 @@ libgcs_env = env.Clone()
 
 # Include paths
 libgcs_env.Append(CPPPATH = Split('''
+                                     #/common
                                      #/galerautils/src
                                      #/gcomm/src
                                      #/gcache/src
@@ -15,18 +16,18 @@ libgcs_env.Append(CPPPATH = Split('''
 # Backends (TODO: Get from global options)
 libgcs_env.Append(CPPFLAGS = ' -DGCS_USE_GCOMM')
 # For C-style logging
-libgcs_env.Append(CPPFLAGS = ' -DGALERA_LOG_H_ENABLE_CXX -Wno-variadic-macros')
+libgcs_env.Append(CPPFLAGS = ' -DGALERA_LOG_H_ENABLE_CXX')
 # Disable old style cast warns until code is fixed
-libgcs_env.Append(CPPFLAGS = ' -Wno-old-style-cast')
+libgcs_env.Replace(CXXFLAGS = libgcs_env['CXXFLAGS'].replace('-Wold-style-cast', ''))
+libgcs_env.Replace(CXXFLAGS = libgcs_env['CXXFLAGS'].replace('-Weffc++', ''))
 # Allow zero sized arrays
 libgcs_env.Replace(CCFLAGS = libgcs_env['CCFLAGS'].replace('-pedantic', ''))
-libgcs_env.Append(CPPFLAGS = ' -Wno-missing-field-initializers')
-libgcs_env.Append(CPPFLAGS = ' -Wno-effc++')
+libgcs_env.Append(CCFLAGS = ' -Wno-missing-field-initializers')
+libgcs_env.Append(CCFLAGS = ' -Wno-variadic-macros')
 
-print libgcs_env['CFLAGS']
-print libgcs_env['CCFLAGS']
-print libgcs_env['CPPFLAGS']
-print libgcs_env['CXXFLAGS']
+print('gcs flags:')
+for f in ['CFLAGS', 'CXXFLAGS', 'CCFLAGS', 'CPPFLAGS']:
+    print(f + ': ' + libgcs_env[f].strip())
 
 gcs4garb_env = libgcs_env.Clone()