Merge pull request ceph#3142 from ceph/wip-10220

mon: suspend proposing paxos values when flushing pending writes

Reviewed-by: Sage Weil <[email protected]>

src/common/config_opts.h

@@ -226,6 +226,8 @@ OPTION(mon_mds_force_trim_to, OPT_INT, 0)   // force mon to trim mdsmaps to this
 // dump transactions
 OPTION(mon_debug_dump_transactions, OPT_BOOL, false)
 OPTION(mon_debug_dump_location, OPT_STR, "/var/log/ceph/$cluster-$name.tdump")
+OPTION(mon_inject_transaction_delay_max, OPT_DOUBLE, 10.0)      // seconds
+OPTION(mon_inject_transaction_delay_probability, OPT_DOUBLE, 0) // range [0, 1]
 
 OPTION(mon_sync_provider_kill_at, OPT_INT, 0)  // kill the sync provider at a specific point in the work flow
 OPTION(mon_sync_requester_kill_at, OPT_INT, 0) // kill the sync requester at a specific point in the work flow

src/mon/MonitorDBStore.h

@@ -304,6 +304,24 @@ class MonitorDBStore
       : store(s), t(t), oncommit(f)
     {}
     void finish(int r) {
+      /* The store serializes writes.  Each transaction is handled
+       * sequentially by the io_work Finisher.  If a transaction takes longer
+       * to apply its state to permanent storage, then no other transaction
+       * will be handled meanwhile.
+       *
+       * We will now randomly inject random delays.  We can safely sleep prior
+       * to applying the transaction as it won't break the model.
+       */
+      double delay_prob = g_conf->mon_inject_transaction_delay_probability;
+      if (delay_prob && (rand() % 10000 < delay_prob * 10000.0)) {
+        utime_t delay;
+        double delay_max = g_conf->mon_inject_transaction_delay_max;
+        delay.set_from_double(delay_max * (double)(rand() % 10000) / 10000.0);
+        lsubdout(g_ceph_context, mon, 1)
+          << "apply_transaction will be delayed for " << delay
+          << " seconds" << dendl;
+        delay.sleep();
+      }
       int ret = store->apply_transaction(t);
       oncommit->complete(ret);
     }

src/mon/Paxos.cc

@@ -846,6 +846,12 @@ void Paxos::commit_start()
     state = STATE_WRITING;
   else
     assert(0);
+
+  if (mon->get_quorum().size() > 1) {
+    // cancel timeout event
+    mon->timer.cancel_event(accept_timeout_event);
+    accept_timeout_event = 0;
+  }
 }
 
 void Paxos::commit_finish()
@@ -900,16 +906,12 @@ void Paxos::commit_finish()
 
   if (do_refresh()) {
     commit_proposal();
-
-    finish_contexts(g_ceph_context, waiting_for_commit);
-
     if (mon->get_quorum().size() > 1) {
-      // cancel timeout event
-      mon->timer.cancel_event(accept_timeout_event);
-      accept_timeout_event = 0;
       extend_lease();
     }
 
+    finish_contexts(g_ceph_context, waiting_for_commit);
+
     assert(g_conf->paxos_kill_at != 10);
 
     finish_round();