DPL: Use X9 to make AsyncQueue atomic

Framework/Core/CMakeLists.txt

@@ -163,6 +163,7 @@ o2_add_library(Framework
  O2::Headers
  O2::MemoryResources
  O2::PCG
+ O2::X9
  RapidJSON::RapidJSON
  Arrow::arrow_shared
  Microsoft.GSL::GSL

Framework/Core/include/Framework/AsyncQueue.h

@@ -14,6 +14,10 @@
 #include "Framework/TimesliceSlot.h"
 #include <string>
 #include <vector>
+#include <atomic>
+
+typedef struct x9_inbox_internal x9_inbox;
+typedef struct x9_node_internal x9_node;
 
 namespace o2::framework
 {
@@ -89,6 +93,12 @@ struct AsyncQueue {
  std::vector<AsyncTaskSpec> prototypes;
  std::vector<AsyncTask> tasks;
  size_t iteration = 0;
+
+ std::atomic<bool> first = true;
+
+ // Inbox for the message queue used to append
+ // tasks to this queue.
+ x9_inbox* inbox = nullptr;
  AsyncQueue();
 };
 
@@ -104,6 +114,8 @@ struct AsyncQueueHelpers {
  /// 3. only execute the highest (timeslice, debounce) value
  static void run(AsyncQueue& queue, TimesliceId oldestPossibleTimeslice);
 
+ // Flush tasks which were posted but not yet committed to the queue
+ static void flushPending(AsyncQueue& queue);
  /// Reset the queue to its initial state
  static void reset(AsyncQueue& queue);
 };

Framework/Core/src/AsyncQueue.cxx

@@ -11,14 +11,17 @@
 
 #include "Framework/AsyncQueue.h"
 #include "Framework/Signpost.h"
+#include "x9.h"
 #include <numeric>
 
 O2_DECLARE_DYNAMIC_LOG(async_queue);
 
 namespace o2::framework
 {
 AsyncQueue::AsyncQueue()
+ : inbox(x9_create_inbox(16, "async_queue", sizeof(AsyncTask)))
 {
+ this->inbox = x9_create_inbox(16, "async_queue", sizeof(AsyncTask));
 }
 
 auto AsyncQueueHelpers::create(AsyncQueue& queue, AsyncTaskSpec spec) -> AsyncTaskId
@@ -31,11 +34,39 @@ auto AsyncQueueHelpers::create(AsyncQueue& queue, AsyncTaskSpec spec) -> AsyncTa
 
 auto AsyncQueueHelpers::post(AsyncQueue& queue, AsyncTask const& task) -> void
 {
- queue.tasks.push_back(task);
+ // Until we do not manage to write to the inbox, keep removing
+ // items from the queue if you are the first one which fails to
+ // write.
+ while (!x9_write_to_inbox(queue.inbox, sizeof(AsyncTask), &task)) {
+ AsyncQueueHelpers::flushPending(queue);
+ }
+}
+
+auto AsyncQueueHelpers::flushPending(AsyncQueue& queue) -> void
+{
+ bool isFirst = true;
+ if (!std::atomic_compare_exchange_strong(&queue.first, &isFirst, false)) {
+ // Not the first, try again.
+ return;
+ }
+ // First thread which does not manage to write to the queue.
+ // Flush it a bit before we try again.
+ AsyncTask toFlush;
+ // This potentially stalls if the inserting tasks are faster to insert
+ // than we are to retrieve. We should probably have a cut-off
+ while (x9_read_from_inbox(queue.inbox, sizeof(AsyncTask), &toFlush)) {
+ queue.tasks.push_back(toFlush);
+ }
+ queue.first = true;
 }
 
 auto AsyncQueueHelpers::run(AsyncQueue& queue, TimesliceId oldestPossible) -> void
 {
+ // We synchronize right before we run to get as many
+ // tasks as possible. Notice we might still miss some
+ // which will have to handled on a subsequent iteration.
+ AsyncQueueHelpers::flushPending(queue);
+
  if (queue.tasks.empty()) {
  return;
  }