Fix potential memleak in Pregel conductor garbage collection. (#14466)

jsteemann · maierlars · commit 254986b5bd68 · 2021-07-08T13:20:59.000+02:00
diff --git a/CHANGELOG b/CHANGELOG
@@ -1,6 +1,8 @@
 devel
 -----
 
+* Fix potential memleak in Pregel conductor garbage collection.
+
 * Added a retry loop for arangorestore during the initial connection phase. The
   number of retries defaults to 3 and can be configured using
   --initial-connect-retries.
diff --git a/arangod/Pregel/PregelFeature.cpp b/arangod/Pregel/PregelFeature.cpp
@@ -261,6 +261,7 @@ void PregelFeature::scheduleGarbageCollection() {
                                    "No thread available to queue Pregel garbage collection");
   }
 
+  MUTEX_LOCKER(guard, _mutex);
   _gcHandle = std::move(handle);
 }
 
@@ -278,9 +279,10 @@ void PregelFeature::start() {
 
 void PregelFeature::beginShutdown() {
   TRI_ASSERT(isStopping());
-  _gcHandle.reset();
   
   MUTEX_LOCKER(guard, _mutex);
+  _gcHandle.reset();
+
   // cancel all conductors and workers
   for (auto& it : _conductors) {
     it.second.conductor->cancel();
@@ -291,6 +293,8 @@ void PregelFeature::beginShutdown() {
 }
 
 void PregelFeature::unprepare() {
+  garbageCollectConductors();
+  
   MUTEX_LOCKER(guard, _mutex);
   decltype(_conductors) cs = std::move(_conductors);
   decltype(_workers) ws = std::move(_workers);
@@ -336,6 +340,7 @@ void PregelFeature::garbageCollectConductors() try {
     if (it->second.conductor->canBeGarbageCollected()) {
       uint64_t executionNumber = it->first;
 
+      it->second.conductor->cancel();
       it = _conductors.erase(it);
       
       _workers.erase(executionNumber);
diff --git a/arangod/Scheduler/Scheduler.h b/arangod/Scheduler/Scheduler.h
@@ -154,27 +154,27 @@ class Scheduler {
   [[nodiscard]] virtual bool queueItem(RequestLane lane, std::unique_ptr<WorkItemBase> item) = 0;
 
  public:
-    // delay Future returns a future that will be fulfilled after the given duration
-    // requires scheduler
-    // If d is zero, the future is fulfilled immediately. Throws a logic error
-    // if delay was cancelled.
-    futures::Future<futures::Unit> delay(clock::duration d) {
-      if (d == clock::duration::zero()) {
-        return futures::makeFuture();
-      }
+  // delay Future returns a future that will be fulfilled after the given duration
+  // requires scheduler
+  // If d is zero, the future is fulfilled immediately. Throws a logic error
+  // if delay was cancelled.
+  futures::Future<futures::Unit> delay(clock::duration d) {
+    if (d == clock::duration::zero()) {
+      return futures::makeFuture();
+    }
 
-      futures::Promise<bool> p;
-      futures::Future<bool> f = p.getFuture();
+    futures::Promise<bool> p;
+    futures::Future<bool> f = p.getFuture();
 
-      auto item = queueDelay(RequestLane::DELAYED_FUTURE, d,
-        [pr = std::move(p)](bool cancelled) mutable { pr.setValue(cancelled); });
+    auto item = queueDelay(RequestLane::DELAYED_FUTURE, d,
+      [pr = std::move(p)](bool cancelled) mutable { pr.setValue(cancelled); });
 
-      return std::move(f).thenValue([item = std::move(item)](bool cancelled) {
-        if (cancelled) {
-          throw std::logic_error("delay was cancelled");
-        }
-      });
-    }
+    return std::move(f).thenValue([item = std::move(item)](bool cancelled) {
+      if (cancelled) {
+        throw std::logic_error("delay was cancelled");
+      }
+    });
+  }
 
   // ---------------------------------------------------------------------------
   // CronThread and delayed tasks
