Applied review comment. Also added ALERT. Also print if we are still …

…starting async tasks although we already stopped them
arangodb · neunhoef · May 2, 2025 · Apr 30, 2025 · Apr 30, 2025 · Apr 30, 2025
commit ea856464d788bfe6ba09bb04c551e806a0f70c4d
diff --git a/arangod/Aql/ExecutionBlock.cpp b/arangod/Aql/ExecutionBlock.cpp
@@ -211,7 +211,9 @@ auto ExecutionBlock::printBlockInfo() const -> std::string const {
 
 auto ExecutionBlock::stopAsyncTasks() -> void {}
 
-auto ExecutionBlock::isDependencyInList(std::unordered_set<ExecutionBlock*> const& seenBlocks) const noexcept -> ExecutionBlock* {
+auto ExecutionBlock::isDependencyInList(
+    std::unordered_set<ExecutionBlock*> const& seenBlocks) const noexcept
+    -> ExecutionBlock* {
   for (auto const& dependency : _dependencies) {
     if (seenBlocks.find(dependency) != seenBlocks.end()) {
       return dependency;
@@ -220,15 +222,20 @@ auto ExecutionBlock::isDependencyInList(std::unordered_set<ExecutionBlock*> cons
   return nullptr;
 }
 
-auto ExecutionBlock::printBlockAndDependenciesInfo() const noexcept -> std::string const {
-
+auto ExecutionBlock::printBlockAndDependenciesInfo() const noexcept
+    -> std::string const {
   std::stringstream ss;
   ss << printBlockInfo();
-  ss << " async prefetching type: " <<(int) getPlanNode()->canUseAsyncPrefetching();
+  ss << " async prefetching type: "
+     << (int)getPlanNode()->canUseAsyncPrefetching();
   ss << " calls: [";
   for (auto const& dependency : _dependencies) {
     ss << " " << dependency->printBlockInfo() << ",";
   }
   ss << " ]";
   return ss.str();
+}
+
+auto ExecutionBlock::hasStoppedAsyncTasks() const noexcept -> bool {
+  return _stoppedAsyncTasks;
 }
diff --git a/arangod/Aql/ExecutionBlock.h b/arangod/Aql/ExecutionBlock.h
@@ -36,7 +36,6 @@
 #include <string_view>
 #include <vector>
 
-
 namespace arangodb {
 namespace transaction {
 class Methods;
@@ -141,6 +140,8 @@ class ExecutionBlock {
   [[nodiscard]] auto isDependencyInList(
       std::unordered_set<ExecutionBlock*> const& seenBlocks) const noexcept
       -> ExecutionBlock*;
+
+  [[nodiscard]] auto hasStoppedAsyncTasks() const noexcept -> bool;
 
  protected:
   // Trace the start of a execute call
@@ -184,6 +185,9 @@ class ExecutionBlock {
   /// @brief if this is set, we are done, this is reset to false by execute()
   bool _done;
 
+  /// @brief if this is set, we have stopped async tasks, this is set to true by stopAsyncTasks()
+  bool _stoppedAsyncTasks{false};
+
 #ifdef ARANGODB_ENABLE_MAINTAINER_MODE
   /// @brief if this is set to true, one thread is using this block, so we can
   /// assert that no other thread can access this block at the same time - as

diff --git a/arangod/Aql/ExecutionBlockImpl.tpp b/arangod/Aql/ExecutionBlockImpl.tpp
@@ -354,6 +354,8 @@ ExecutionBlockImpl<Executor>::~ExecutionBlockImpl() {
 
 template<class Executor>
 void ExecutionBlockImpl<Executor>::stopAsyncTasks() {
+  TRI_ASSERT(!_stoppedAsyncTasks) << "Someone already stopped async tasks for " << printBlockInfo();
+  _stoppedAsyncTasks = true;
   if (_prefetchTask && !_prefetchTask->isConsumed() &&
       !_prefetchTask->tryClaim()) {
     // some thread is still working on our prefetch task
@@ -1047,6 +1049,29 @@ auto ExecutionBlockImpl<Executor>::executeFetcher(ExecutionContext& ctx,
                 return;
               }
 
+              // We are entering debugging land here.
+              // Someone hasasked as to stop async tasks.
+              // But here we are just starting to work on one.
+              // This is okay and can rarely happen. But if it happens
+              // We want to know if we start the task and finish it quickly
+
+              bool hasStoppedAsyncTasks = block->hasStoppedAsyncTasks();
+              if (hasStoppedAsyncTasks) {
+                LOG_TOPIC("14d20", WARN, Logger::AQL)
+                    << "[query#" << block->getQuery().id() << "] ALERT"
+                    << block->printBlockInfo()
+                    << " was asked to stop async task. We still start one. This is an allowed rare race.";
+              }
+
+              auto stopGuard = ScopeGuard([block, hasStoppedAsyncTasks]() noexcept {
+                if (hasStoppedAsyncTasks) {
+                  LOG_TOPIC("14d21", WARN, Logger::AQL)
+                      << "[query#" << block->getQuery().id() << "] CLEAR ALERT"
+                      << block->printBlockInfo()
+                      << " We completed the task of the aforementioned race. All is fine.";
+                }
+              });
+
               TRI_IF_FAILURE("AsyncPrefetch::blocksDestroyedOutOfOrder") {
                 using namespace std::chrono_literals;
                 std::this_thread::sleep_for(100ms);

diff --git a/arangod/Aql/ExecutionEngine.cpp b/arangod/Aql/ExecutionEngine.cpp
@@ -278,28 +278,33 @@ ExecutionEngine::~ExecutionEngine() {
     // vector. Also we are in the destructor here, so we have guaranteed,
     // that no one else is cleaning up the blocks.
     std::unordered_set<ExecutionBlock*> seenBlocks;
+    bool needToPrintViolation = false;
     for (auto it = _blocks.rbegin(); it != _blocks.rend(); ++it) {
       auto block = it->get();
       if (ExecutionBlock* seenDependency =
               block->isDependencyInList(seenBlocks);
-          seenDependency != nullptr && block->getPlanNode()->getType() != ExecutionNode::GATHER) {
+          seenDependency != nullptr &&
+          block->getPlanNode()->getType() != ExecutionNode::GATHER) {
         // We have a dependency that has already been seen, we need to log this
         // situation in theory this could lead to deadlocks. Some Blocks are
         // fine we just want to see those here.
         // Gather Nodes are known to violate this, but they are safe.
         LOG_TOPIC("a6c2b", WARN, Logger::AQL)
-            << "Stopping async tasks for " << block->printBlockInfo()
+            << "ALERT Stopping async tasks for " << block->printBlockInfo()
             << " but have already stopped dependency "
             << seenDependency->printBlockInfo();
-        LOG_TOPIC("a6c2c", WARN, Logger::AQL) << "Full list of blocks:";
-        for (auto it2 = _blocks.rbegin(); it2 != _blocks.rend(); ++it2) {
-          LOG_TOPIC("a6c2d", WARN, Logger::AQL) << (*it2)->printBlockAndDependenciesInfo();
-        }
-        TRI_ASSERT(false) << "Triggered violation in ExecutionBlock ordering";
+        needToPrintViolation = true;
       }
       block->stopAsyncTasks();
       seenBlocks.insert(block);
     }
+    if (needToPrintViolation) {
+      for (auto it2 = _blocks.rbegin(); it2 != _blocks.rend(); ++it2) {
+        LOG_TOPIC("a6c2d", WARN, Logger::AQL)
+            << (*it2)->printBlockAndDependenciesInfo();
+      }
+      TRI_ASSERT(false) << "Triggered violation in ExecutionBlock ordering";
+    }
   }
   if (_sharedState) {  // ensure no async task is working anymore
     _sharedState->invalidate();