timer: support join inflight task

src/braft/log.cpp

@@ -33,8 +33,8 @@
 #include "braft/protobuf_file.h"
 #include "braft/util.h"
 
-//#define BRAFT_SEGMENT_OPEN_PATTERN "log_inprogress_%020ld"
-//#define BRAFT_SEGMENT_CLOSED_PATTERN "log_%020ld_%020ld"
+// #define BRAFT_SEGMENT_OPEN_PATTERN "log_inprogress_%020ld"
+// #define BRAFT_SEGMENT_CLOSED_PATTERN "log_%020ld_%020ld"
 #define BRAFT_SEGMENT_OPEN_PATTERN "log_inprogress_%020" PRId64
 #define BRAFT_SEGMENT_CLOSED_PATTERN "log_%020" PRId64 "_%020" PRId64
 #define BRAFT_SEGMENT_META_FILE "log_meta"

src/braft/macros.h

@@ -27,7 +27,7 @@
 #define BRAFT_VLOG_IF(cond) VLOG_IF(89, (cond))
 #define BRAFT_VPLOG_IF(cond) VPLOG_IF(89, (cond))
 
-//#define USE_BTHREAD_MUTEX
+// #define USE_BTHREAD_MUTEX
 
 #ifdef USE_BTHREAD_MUTEX
 

src/braft/node.cpp

@@ -1030,10 +1030,10 @@ void NodeImpl::shutdown(Closure* done) {
             _state = STATE_SHUTTING;
 
             // Destroy all the timer
-            _election_timer.destroy();
-            _vote_timer.destroy();
-            _stepdown_timer.destroy();
-            _snapshot_timer.destroy();
+            _election_timer.destroy(true /*wait*/);
+            _vote_timer.destroy(true /*wait*/);
+            _stepdown_timer.destroy(true /*wait*/);
+            _snapshot_timer.destroy(true /*wait*/);
 
             // stop replicator and fsm_caller wait
             if (_log_manager) {

src/braft/repeated_timer_task.cpp

@@ -26,7 +26,8 @@ RepeatedTimerTask::RepeatedTimerTask()
       _stopped(true),
       _running(false),
       _destroyed(true),
-      _invoking(false) {}
+      _invoking(false),
+      _finish_event(0) {}
 
 RepeatedTimerTask::~RepeatedTimerTask() {
     CHECK(!_running) << "Is still running";
@@ -71,6 +72,7 @@ void RepeatedTimerTask::on_timedout() {
             lck.unlock();
             on_destroy();
         }
+        _finish_event.signal();
         return;
     }
     return schedule(lck);
@@ -94,6 +96,7 @@ void RepeatedTimerTask::start() {
     // is still running, in which case on_timedout would invoke
     // schedule as it would not see _stopped
     _running = true;
+    _finish_event.reset(1);
     schedule(lck);
 }
 
@@ -156,7 +159,7 @@ void RepeatedTimerTask::reset(int timeout_ms) {
     // else on_timedout would invoke schdule
 }
 
-void RepeatedTimerTask::destroy() {
+void RepeatedTimerTask::destroy(bool wait_infight_task) {
     std::unique_lock<raft_mutex_t> lck(_mutex);
     BRAFT_RETURN_IF(_destroyed);
     _destroyed = true;
@@ -175,6 +178,13 @@ void RepeatedTimerTask::destroy() {
         on_destroy();
         return;
     }
+
+    if (wait_infight_task) {
+        _finish_event.wait();
+        CHECK(!_running);
+        return;
+    }
+
     CHECK(_running);
 }
 

src/braft/repeated_timer_task.h

@@ -21,6 +21,7 @@
 #include <bthread/unstable.h>
 
 #include "braft/macros.h"
+#include "bthread/countdown_event.h"
 
 namespace braft {
 
@@ -50,7 +51,7 @@ class RepeatedTimerTask {
     void reset(int timeout_ms);
 
     // Destroy the timer
-    void destroy();
+    void destroy(bool wait_inflight_task = false);
 
     // Describe the current status of timer
     void describe(std::ostream& os, bool use_html);
@@ -78,6 +79,7 @@ class RepeatedTimerTask {
     bool _running;
     bool _destroyed;
     bool _invoking;
+    bthread::CountdownEvent _finish_event;
 };
 
 }  //  namespace braft

src/braft/snapshot.cpp

@@ -30,7 +30,7 @@
 #include "braft/remote_file_copier.h"
 #include "braft/util.h"
 
-//#define BRAFT_SNAPSHOT_PATTERN "snapshot_%020ld"
+// #define BRAFT_SNAPSHOT_PATTERN "snapshot_%020ld"
 #define BRAFT_SNAPSHOT_PATTERN "snapshot_%020" PRId64
 #define BRAFT_SNAPSHOT_META_FILE "__raft_snapshot_meta"