arangodb
diff --git a/‎lib/Async/Registry/CMakeLists.txt
Lines changed: 1 addition & 0 deletions b/‎lib/Async/Registry/CMakeLists.txt
Lines changed: 1 addition & 0 deletions
diff --git a/‎lib/Async/Registry/promise.cpp
Lines changed: 18 additions & 0 deletions b/‎lib/Async/Registry/promise.cpp
Lines changed: 18 additions & 0 deletions
diff --git a/‎lib/Async/Registry/promise.h
Lines changed: 22 additions & 18 deletions b/‎lib/Async/Registry/promise.h
Lines changed: 22 additions & 18 deletions
diff --git a/‎lib/Async/Registry/thread_registry.cpp
Lines changed: 8 additions & 6 deletions b/‎lib/Async/Registry/thread_registry.cpp
Lines changed: 8 additions & 6 deletions
diff --git a/‎lib/Async/Registry/thread_registry.h
Lines changed: 2 additions & 1 deletion b/‎lib/Async/Registry/thread_registry.h
Lines changed: 2 additions & 1 deletion
diff --git a/‎lib/Async/include/Async/async.h
Lines changed: 10 additions & 18 deletions b/‎lib/Async/include/Async/async.h
Lines changed: 10 additions & 18 deletions
diff --git a/‎lib/Futures/include/Futures/SharedState.h
Lines changed: 10 additions & 20 deletions b/‎lib/Futures/include/Futures/SharedState.h
Lines changed: 10 additions & 20 deletions
diff --git a/‎tests/Async/AsyncTest.cpp
Lines changed: 1 addition & 1 deletion b/‎tests/Async/AsyncTest.cpp
Lines changed: 1 addition & 1 deletion
diff --git a/‎tests/Async/Registry/CMakeLists.txt
Lines changed: 1 addition & 1 deletion b/‎tests/Async/Registry/CMakeLists.txt
Lines changed: 1 addition & 1 deletion
@@ -1,4 +1,5 @@
 add_library(arango_async_registry STATIC
+  promise.cpp
   registry.cpp
   registry_variable.cpp
   thread_registry.cpp)
 
@@ -0,0 +1,18 @@
+#include "promise.h"
+
+#include "Async/Registry/thread_registry.h"
+
+using namespace arangodb::async_registry;
+
+PromiseInList::PromiseInList(PromiseInList* next,
+                             std::shared_ptr<ThreadRegistry> registry,
+                             std::source_location entry_point)
+    : registry{registry},
+      thread{
+          Thread{.name = registry->thread_name, .id = registry->owning_thread}},
+      next{next},
+      entry_point{std::move(entry_point)} {}
+
+auto PromiseInList::mark_for_deletion() noexcept -> void {
+  registry->mark_for_deletion(this);
+}
@@ -12,40 +12,44 @@ namespace arangodb::async_registry {
 
 struct ThreadRegistry;
 
-struct Observables {
-  Observables(std::source_location loc) : _where(std::move(loc)) {}
-
-  std::source_location _where;
+struct Thread {
+  std::string name;
+  std::thread::id id;
 };
+template<typename Inspector>
+auto inspect(Inspector& f, Thread& x) {
+  return f.object(x).fields(f.field("name", x.name),
+                            f.field("id", fmt::format("{}", x.id)));
+}
 
-struct PromiseInList : Observables {
-  PromiseInList(std::source_location loc) : Observables(std::move(loc)) {}
+struct PromiseInList {
+  PromiseInList(PromiseInList* next, std::shared_ptr<ThreadRegistry> registry,
+                std::source_location location);
+  ~PromiseInList() = default;
 
-  virtual auto destroy() noexcept -> void = 0;
-  virtual ~PromiseInList() = default;
+  auto mark_for_deletion() noexcept -> void;
 
   // identifies the promise list it belongs to
-  std::shared_ptr<ThreadRegistry> registry = nullptr;
-
-  std::string thread_name;
-  std::thread::id thread_id;
+  std::shared_ptr<ThreadRegistry> registry;
+  Thread thread;
 
   PromiseInList* next = nullptr;
   // only needed to remove an item
   PromiseInList* previous = nullptr;
   // only needed to garbage collect promises
   PromiseInList* next_to_free = nullptr;
-};
 
+  std::source_location entry_point;
+};
 template<typename Inspector>
 auto inspect(Inspector& f, PromiseInList& x) {
   // perhaps just use for saving
   return f.object(x).fields(
-      f.field("source_location",
-              fmt::format("{}:{} {}", x._where.file_name(), x._where.line(),
-                          x._where.function_name())),
-      f.field("thread_name", x.thread_name),
-      f.field("thread_id", fmt::format("{}", x.thread_id)));
+      f.field("thread", x.thread),
+      f.field(
+          "source_location",
+          fmt::format("{}:{} {}", x.entry_point.file_name(),
+                      x.entry_point.line(), x.entry_point.function_name())));
 }
 
 }  // namespace arangodb::async_registry
@@ -1,7 +1,9 @@
 #include "thread_registry.h"
+#include <source_location>
 
 #include "Assertions/ProdAssert.h"
 #include "Async/Registry/Metrics.h"
+#include "Async/Registry/promise.h"
 #include "Basics/Thread.h"
 #include "Inspection/Format.h"
 #include "Logger/LogMacros.h"
@@ -34,7 +36,8 @@ ThreadRegistry::~ThreadRegistry() noexcept {
   cleanup();
 }
 
-auto ThreadRegistry::add(PromiseInList* promise) noexcept -> void {
+auto ThreadRegistry::add_promise(std::source_location location) noexcept
+    -> PromiseInList* {
   // promise needs to live on the same thread as this registry
   ADB_PROD_ASSERT(std::this_thread::get_id() == owning_thread)
       << "ThreadRegistry::add was called from thread "
@@ -45,10 +48,8 @@ auto ThreadRegistry::add(PromiseInList* promise) noexcept -> void {
     metrics->total_functions->count();
   }
   auto current_head = promise_head.load(std::memory_order_relaxed);
-  promise->next = current_head;
-  promise->registry = shared_from_this();
-  promise->thread_name = thread_name;
-  promise->thread_id = owning_thread;
+  auto promise =
+      new PromiseInList{current_head, shared_from_this(), std::move(location)};
   if (current_head != nullptr) {
     current_head->previous = promise;
   }
@@ -57,6 +58,7 @@ auto ThreadRegistry::add(PromiseInList* promise) noexcept -> void {
   if (metrics->active_functions != nullptr) {
     metrics->active_functions->fetch_ad
7E71
d(1);
   }
+  return promise;
 }
 
 auto ThreadRegistry::mark_for_deletion(PromiseInList* promise) noexcept
@@ -98,7 +100,7 @@ auto ThreadRegistry::cleanup() noexcept -> void {
       metrics->ready_for_deletion_functions->fetch_sub(1);
     }
     remove(current);
-    current->destroy();
+    delete current;
   }
 }
 
 
@@ -6,6 +6,7 @@
 #include <cstdint>
 #include <memory>
 #include <mutex>
+#include <source_location>
 #include <thread>
 #include "fmt/format.h"
 #include "fmt/std.h"
@@ -42,7 +43,7 @@ struct ThreadRegistry : std::enable_shared_from_this<ThreadRegistry> {
      Can only be called on the owning thread, crashes
      otherwise.
    */
-  auto add(PromiseInList* promise) noexcept -> void;
+  auto add_promise(std::source_location location) noexcept -> PromiseInList*;
 
   /**
      Executes a function on each promise in the registry that is not deleted yet
 
@@ -21,11 +21,12 @@ template<typename T>
 struct async;
 
 template<typename T>
-struct async_promise_base : async_registry::PromiseInList {
+struct async_promise_base {
   using promise_type = async_promise<T>;
 
   async_promise_base(std::source_location loc)
-      : PromiseInList(std::move(loc)) {}
+      : promise_in_registry{async_registry::get_thread_registry().add_promise(
+            std::move(loc))} {}
 
   std::suspend_never initial_suspend() noexcept { return {}; }
   auto final_suspend() noexcept {
@@ -38,7 +39,8 @@ struct async_promise_base : async_registry::PromiseInList {
         if (addr == nullptr) {
           return std::noop_coroutine();
         } else if (addr == self.address()) {
-          _promise->registry->mark_for_deletion(_promise);
+          _promise->promise_in_registry->mark_for_deletion();
+          self.destroy();
           return std::noop_coroutine();
         } else {
           return std::coroutine_handle<>::from_address(addr);
@@ -51,25 +53,13 @@ struct async_promise_base : async_registry::PromiseInList {
   }
   void unhandled_exception() { _value.set_exception(std::current_exception()); }
   auto get_return_object() {
-    async_registry::get_thread_registry().add(this);
     return async<T>{std::coroutine_handle<promise_type>::from_promise(
         *static_cast<promise_type*>(this))};
   }
 
-  auto destroy() noexcept -> void override {
-    try {
-      std::coroutine_handle<promise_type>::from_promise(
-          *static_cast<promise_type*>(this))
-          .destroy();
-    } catch (std::exception const& ex) {
-      LOG_TOPIC("4b96e", WARN, Logger::CRASH)
-          << "caught exception when destorying coroutine promise: "
-          << ex.what();
-    }
-  }
-
   std::atomic<void*> _continuation = nullptr;
   expected<T> _value;
+  async_registry::PromiseInList* promise_in_registry = nullptr;
 };
 
 template<typename T>
@@ -106,7 +96,8 @@ struct async {
       auto await_resume() {
         auto& promise = _handle.promise();
         expected<T> r = std::move(promise._value);
-        promise.registry->mark_for_deletion(&promise);
+        promise.promise_in_registry->mark_for_deletion();
+        _handle.destroy();
         return std::move(r).get();
       }
       explicit awaitable(std::coroutine_handle<promise_type> handle)
@@ -126,7 +117,8 @@ struct async {
       auto& promise = _handle.promise();
       if (promise._continuation.exchange(
               _handle.address(), std::memory_order_release) != nullptr) {
-        promise.registry->mark_for_deletion(&promise);
+        promise.promise_in_registry->mark_for_deletion();
+        _handle.destroy();
       }
       _handle = nullptr;
     }
 
@@ -51,7 +51,7 @@ namespace detail {
 ///   |                    ---> OnlyCallback ---                    |
 ///   +-------------------------------------------------------------+
 template<typename T>
-class SharedState : async_registry::PromiseInList {
+class SharedState {
   enum class State : uint8_t {
     Start = 1 << 0,
     OnlyResult = 1 << 1,
@@ -236,19 +236,15 @@ class SharedState : async_registry::PromiseInList {
  private:
   /// empty shared state
   SharedState(std::source_location loc)
-      : async_registry::PromiseInList(std::move(loc)),
-        _state(State::Start),
+      : _state(State::Start),
         _attached(2),
-        _managed_by_registry{true} {
-    async_registry::get_thread_registry().add(this);
+        promise_in_registry{
+            async_registry::get_thread_registry().add_promise(std::move(loc))} {
   }
 
   /// use to construct a ready future
   explicit SharedState(Try<T>&& t)
-      : async_registry::PromiseInList(std::source_location::current()),
-        _result(std::move(t)),
-        _state(State::OnlyResult),
-        _attached(1) {
+      : _result(std::move(t)), _state(State::OnlyResult), _attached(1) {
     // is not added to coroutine thread registry because it is immediately
     // resolved
   }
@@ -257,8 +253,7 @@ class SharedState : async_registry::PromiseInList {
   template<typename... Args>
   explicit SharedState(std::in_place_t, Args&&... args) noexcept(
       std::is_nothrow_constructible<T, Args&&...>::value)
-      : async_registry::PromiseInList(std::source_location::current()),
-        _result(std::in_place, std::forward<Args>(args)...),
+      : _result(std::in_place, std::forward<Args>(args)...),
         _state(State::OnlyResult),
         _attached(1) {
     // is not added to coroutine thread registry because it is immediately
@@ -277,13 +272,10 @@ class SharedState : async_registry::PromiseInList {
     TRI_ASSERT(a >= 1);
     if (a == 1) {
       _callback = nullptr;
-      if (_managed_by_registry) {
-        if (registry != nullptr) {
-          registry->mark_for_deletion(this);
-        }
-      } else {
-        destroy();
+      if (promise_in_registry != nullptr) {
+        promise_in_registry->mark_for_deletion();
       }
+      delete this;
     }
   }
 
@@ -297,8 +289,6 @@ class SharedState : async_registry::PromiseInList {
     _callback(std::move(_result));
   }
 
-  auto destroy() noexcept -> void override { delete this; }
-
  private:
   using Callback = fu2::unique_function<void(Try<T>&&)>;
   Callback _callback;
@@ -307,7 +297,7 @@ class SharedState : async_registry::PromiseInList {
   };
   std::atomic<State> _state;
   std::atomic<uint8_t> _attached;
-  const bool _managed_by_registry = false;
+  async_registry::PromiseInList* promise_in_registry = nullptr;
 };
 
 }  // namespace detail
 
@@ -383,7 +383,7 @@ TEST(AsyncTest, promises_are_registered) {
     std::vector<std::string> names;
     arangodb::async_registry::registry.for_promise(
         [&](arangodb::async_registry::PromiseInList* promise) {
-          names.push_back(promise->_where.function_name());
+          names.push_back(promise->entry_point.function_name());
         });
     EXPECT_EQ(names.size(), 3);
     EXPECT_TRUE(names[0].find("foo") != std::string::npos);
 
@@ -1,4 +1,4 @@
 target_sources(arangodbtests
   PRIVATE
-  RegistryTest.cpp
+  # RegistryTest.cpp
   ThreadRegistryTest.cpp)
Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,5 @@`
`1`	`1`	`add_library(arango_async_registry STATIC`
	`2`	`+ promise.cpp`
`2`	`3`	`registry.cpp`
`3`	`4`	`registry_variable.cpp`
`4`	`5`	`thread_registry.cpp)`
Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,9 @@`
`1`	`1`	`#include "thread_registry.h"`
	`2`	`+#include <source_location>`
`2`	`3`
`3`	`4`	`#include "Assertions/ProdAssert.h"`
`4`	`5`	`#include "Async/Registry/Metrics.h"`
	`6`	`+#include "Async/Registry/promise.h"`
`5`	`7`	`#include "Basics/Thread.h"`
`6`	`8`	`#include "Inspection/Format.h"`
`7`	`9`	`#include "Logger/LogMacros.h"`
`@@ -34,7 +36,8 @@ ThreadRegistry::~ThreadRegistry() noexcept {`
`34`	`36`	`cleanup();`
`35`	`37`	`}`
`36`	`38`
`37`		`-auto ThreadRegistry::add(PromiseInList* promise) noexcept -> void {`
	`39`	`+auto ThreadRegistry::add_promise(std::source_location location) noexcept`
	`40`	`+ -> PromiseInList* {`
`38`	`41`	`// promise needs to live on the same thread as this registry`
`39`	`42`	`ADB_PROD_ASSERT(std::this_thread::get_id() == owning_thread)`
`40`	`43`	`<< "ThreadRegistry::add was called from thread "`
`@@ -45,10 +48,8 @@ auto ThreadRegistry::add(PromiseInList* promise) noexcept -> void {`
`45`	`48`	`metrics->total_functions->count();`
`46`	`49`	`}`
`47`	`50`	`auto current_head = promise_head.load(std::memory_order_relaxed);`
`48`		`- promise->next = current_head;`
`49`		`- promise->registry = shared_from_this();`
`50`		`- promise->thread_name = thread_name;`
`51`		`- promise->thread_id = owning_thread;`
	`51`	`+ auto promise =`
	`52`	`+ new PromiseInList{current_head, shared_from_this(), std::move(location)};`
`52`	`53`	`if (current_head != nullptr) {`
`53`	`54`	`current_head->previous = promise;`
`54`	`55`	`}`
`@@ -57,6 +58,7 @@ auto ThreadRegistry::add(PromiseInList* promise) noexcept -> void {`
`57`	`58`	`if (metrics->active_functions != nullptr) {`
`58`	`59`	`metrics->active_functions->fetch_ad 7E71 d(1);`
`59`	`60`	`}`
	`61`	`+ return promise;`
`60`	`62`	`}`
`61`	`63`
`62`	`64`	`auto ThreadRegistry::mark_for_deletion(PromiseInList* promise) noexcept`
`@@ -98,7 +100,7 @@ auto ThreadRegistry::cleanup() noexcept -> void {`
`98`	`100`	`metrics->ready_for_deletion_functions->fetch_sub(1);`
`99`	`101`	`}`
`100`	`102`	`remove(current);`
`101`		`- current->destroy();`
	`103`	`+ delete current;`
`102`	`104`	`}`
`103`	`105`	`}`
`104`	`106`