DPL: AsyncQueue without std::function

ktf · ktf · commit 2a45b16c6ad1 · 2024-09-19T21:54:32.000+02:00
diff --git a/Framework/Core/include/Framework/AsyncQueue.h b/Framework/Core/include/Framework/AsyncQueue.h
@@ -12,49 +12,91 @@
 #define O2_FRAMEWORK_ASYNCQUUE_H_
 
 #include "Framework/TimesliceSlot.h"
-#include <functional>
 #include <string>
 #include <vector>
 
 namespace o2::framework
 {
 
-struct AsyncTaskSpec {
-  std::string name;
-  // Its priority compared to the other tasks
-  int score = 0;
-};
-
 /// The position of the TaskSpec in the prototypes
+/// Up to 127 different kind of tasks are allowed per queue
 struct AsyncTaskId {
-  int value = -1;
+  int16_t value = -1;
 };
 
 /// An actuatual task to be executed
 struct AsyncTask {
-  // The task to be executed. Id can be used as unique
-  // id for the signpost in the async_queue stream.
-  std::function<void(size_t id)> task;
-  // The associated task spec
-  AsyncTaskId id = {-1};
+  // The timeslice after which this callback should be executed.
   TimesliceId timeslice = {TimesliceId::INVALID};
   // Only the task with the highest debounce value will be executed
-  int debounce = 0;
+  // The associated task spec. Notice that the context
+  // is stored separately so that we do not need to
+  // manage lifetimes of std::functions
+  AsyncTaskId id = {-1};
+  // Debounce value.
+  int8_t debounce = 0;
   bool runnable = false;
+  // Some unuser integer
+  int32_t unused = 0;
+  // The callback to be executed. id can be used as unique
+  // id for the signpost in the async_queue stream.
+  // Context is provided by the userdata attached to the
+  // task the moment we post it.
+  // Notice that we do not store the task in the prototype
+  // because we do support at the moment coalescing two
+  // different callbaks with the same id.
+  void (*callback)(AsyncTask& task, size_t id);
+  // Some user data e.g. to decode what comes next
+  // This can either be used via the .data pointer
+  // or by asking a cast to the appropriate type via
+  // the user() method.
+  void* data[5] = {nullptr, nullptr, nullptr, nullptr, nullptr};
+
+  // Helper to return userdata
+  template <typename T>
+  T& user()
+  {
+    static_assert(sizeof(T) <= 5 * sizeof(void*), "User object does not fit user data");
+    return *(T*)data;
+  }
+
+  // Helper to set userdata
+  // @return a reference to this task modified. This is meant to be used like:
+  //
+  // AsyncQueueHelpers::post(queue, AsyncTask{.id = myTask}.user(Context{.contextValue}));
+  //
+  // Coupled with the other one:
+  //
+  // task.user<Context>().contextValue;
+  //
+  // it can be used to mimick capturing lambdas
+  template <typename T>
+  AsyncTask& user(T&& value)
+  {
+    static_assert(sizeof(T) <= 5 * sizeof(void*), "User object does not fit user data");
+    new (&data[0])(T){value};
+    return *this;
+  }
+};
+
+struct AsyncTaskSpec {
+  std::string name;
+  // Its priority compared to the other tasks
+  int score = 0;
 };
 
 struct AsyncQueue {
   std::vector<AsyncTaskSpec> prototypes;
   std::vector<AsyncTask> tasks;
   size_t iteration = 0;
+  AsyncQueue();
 };
 
 struct AsyncQueueHelpers {
-  using AsyncCallback = std::function<void(size_t)>;
   static AsyncTaskId create(AsyncQueue& queue, AsyncTaskSpec spec);
   // Schedule a task with @a taskId to be executed whenever the timeslice
   // is past timeslice. If debounce is provided, only execute the task
-  static void post(AsyncQueue& queue, AsyncTaskId taskId, AsyncCallback task, TimesliceId timeslice, int64_t debounce = 0);
+  static void post(AsyncQueue& queue, AsyncTask const& task);
   /// Run all the tasks which are older than the oldestPossible timeslice
   /// executing them by:
   /// 1. sorting the tasks by timeslice
diff --git a/Framework/Core/src/AsyncQueue.cxx b/Framework/Core/src/AsyncQueue.cxx
@@ -17,6 +17,10 @@ O2_DECLARE_DYNAMIC_LOG(async_queue);
 
 namespace o2::framework
 {
+AsyncQueue::AsyncQueue()
+{
+}
+
 auto AsyncQueueHelpers::create(AsyncQueue& queue, AsyncTaskSpec spec) -> AsyncTaskId
 {
   AsyncTaskId id;
@@ -25,14 +29,9 @@ auto AsyncQueueHelpers::create(AsyncQueue& queue, AsyncTaskSpec spec) -> AsyncTa
   return id;
 }
 
-auto AsyncQueueHelpers::post(AsyncQueue& queue, AsyncTaskId id, AsyncCallback task, TimesliceId timeslice, int64_t debounce) -> void
+auto AsyncQueueHelpers::post(AsyncQueue& queue, AsyncTask const& task) -> void
 {
-  AsyncTask taskToPost;
-  taskToPost.task = task;
-  taskToPost.id = id;
-  taskToPost.timeslice = timeslice;
-  taskToPost.debounce = debounce;
-  queue.tasks.push_back(taskToPost);
+  queue.tasks.push_back(task);
 }
 
 auto AsyncQueueHelpers::run(AsyncQueue& queue, TimesliceId oldestPossible) -> void
@@ -85,6 +84,8 @@ auto AsyncQueueHelpers::run(AsyncQueue& queue, TimesliceId oldestPossible) -> vo
     }
   }
   // Keep only the tasks with the highest debounce value for a given id
+  // For this reason I need to keep the callback in the task itself, because
+  // two different callbacks with the same id will be coalesced.
   auto newEnd = std::unique(order.begin(), order.end(), [&queue](int a, int b) {
     return queue.tasks[a].runnable == queue.tasks[b].runnable && queue.tasks[a].id.value == queue.tasks[b].id.value && queue.tasks[a].debounce >= 0 && queue.tasks[b].debounce >= 0;
   });
@@ -111,7 +112,7 @@ auto AsyncQueueHelpers::run(AsyncQueue& queue, TimesliceId oldestPossible) -> vo
       O2_SIGNPOST_EVENT_EMIT(async_queue, opid, "run", "Running task %{public}s (%d) for timeslice %zu",
                              queue.prototypes[queue.tasks[i].id.value].name.c_str(), i,
                              queue.tasks[i].timeslice.value);
-      queue.tasks[i].task(opid.value);
+      queue.tasks[i].callback(queue.tasks[i], opid.value);
       O2_SIGNPOST_EVENT_EMIT(async_queue, opid, "run", "Done running %d", i);
     }
   }
diff --git a/Framework/Core/src/CommonServices.cxx b/Framework/Core/src/CommonServices.cxx
@@ -529,12 +529,72 @@ o2::framework::ServiceSpec CommonServices::ccdbSupportSpec()
       } },
     .kind = ServiceKind::Global};
 }
+struct DecongestionContext {
+  ServiceRegistryRef ref;
+  TimesliceIndex::OldestOutputInfo oldestPossibleOutput;
+};
+
+auto decongestionCallback = [](AsyncTask& task, size_t id) -> void {
+  auto& oldestPossibleOutput = task.user<DecongestionContext>().oldestPossibleOutput;
+  auto& ref = task.user<DecongestionContext>().ref;
+
+  auto& decongestion = ref.get<DecongestionService>();
+  auto& proxy = ref.get<FairMQDeviceProxy>();
+
+  O2_SIGNPOST_ID_GENERATE(cid, async_queue);
+  cid.value = id;
+  if (decongestion.lastTimeslice >= oldestPossibleOutput.timeslice.value) {
+    O2_SIGNPOST_EVENT_EMIT(async_queue, cid, "oldest_possible_timeslice", "Not sending already sent value: %" PRIu64 "> %" PRIu64,
+                           decongestion.lastTimeslice, (uint64_t)oldestPossibleOutput.timeslice.value);
+    return;
+  }
+  O2_SIGNPOST_EVENT_EMIT(async_queue, cid, "oldest_possible_timeslice", "Running oldest possible timeslice %" PRIu64 " propagation.",
+                         (uint64_t)oldestPossibleOutput.timeslice.value);
+  DataProcessingHelpers::broadcastOldestPossibleTimeslice(ref, oldestPossibleOutput.timeslice.value);
+
+  for (int fi = 0; fi < proxy.getNumForwardChannels(); fi++) {
+    auto& info = proxy.getForwardChannelInfo(ChannelIndex{fi});
+    auto& state = proxy.getForwardChannelState(ChannelIndex{fi});
+    // TODO: this we could cache in the proxy at the bind moment.
+    if (info.channelType != ChannelAccountingType::DPL) {
+      O2_SIGNPOST_EVENT_EMIT(async_queue, cid, "oldest_possible_timeslice", "Skipping channel %{public}s", info.name.c_str());
+      continue;
+    }
+    if (DataProcessingHelpers::sendOldestPossibleTimeframe(ref, info, state, oldestPossibleOutput.timeslice.value)) {
+      O2_SIGNPOST_EVENT_EMIT(async_queue, cid, "oldest_possible_timeslice",
+                             "Forwarding to channel %{public}s oldest possible timeslice %" PRIu64 ", priority %d",
+                             info.name.c_str(), (uint64_t)oldestPossibleOutput.timeslice.value, 20);
+    }
+  }
+  decongestion.lastTimeslice = oldestPossibleOutput.timeslice.value;
+};
+
+auto decongestionCallbackOrdered = [](AsyncTask& task, size_t id) -> void {
+  auto& oldestPossibleOutput = task.user<DecongestionContext>().oldestPossibleOutput;
+  auto& ref = task.user<DecongestionContext>().ref;
+
+  auto& decongestion = ref.get<DecongestionService>();
+  auto& state = ref.get<DeviceState>();
+  auto& timesliceIndex = ref.get<TimesliceIndex>();
+  O2_SIGNPOST_ID_GENERATE(cid, async_queue);
+  int64_t oldNextTimeslice = decongestion.nextTimeslice;
+  decongestion.nextTimeslice = std::max(decongestion.nextTimeslice, (int64_t)oldestPossibleOutput.timeslice.value);
+  if (oldNextTimeslice != decongestion.nextTimeslice) {
+    if (state.transitionHandling != TransitionHandlingState::NoTransition && DefaultsHelpers::onlineDeploymentMode()) {
+      O2_SIGNPOST_EVENT_EMIT_WARN(async_queue, cid, "oldest_possible_timeslice", "Stop transition requested. Some Lifetime::Timeframe data got dropped starting at %" PRIi64, oldNextTimeslice);
+    } else {
+      O2_SIGNPOST_EVENT_EMIT_ERROR(async_queue, cid, "oldest_possible_timeslice", "Some Lifetime::Timeframe data got dropped starting at %" PRIi64, oldNextTimeslice);
+    }
+    timesliceIndex.rescan();
+  }
+};
 
 // Decongestion service
 // If we do not have any Timeframe input, it means we must be creating timeslices
 // in order and that we should propagate the oldest possible timeslice at the end
 // of each processing step.
-o2::framework::ServiceSpec CommonServices::decongestionSpec()
+o2::framework::ServiceSpec
+  CommonServices::decongestionSpec()
 {
   return ServiceSpec{
     .name = "decongestion",
@@ -548,7 +608,7 @@ o2::framework::ServiceSpec CommonServices::decongestionSpec()
         }
       }
       auto& queue = services.get<AsyncQueue>();
-      decongestion->oldestPossibleTimesliceTask = AsyncQueueHelpers::create(queue, {"oldest-possible-timeslice", 100});
+      decongestion->oldestPossibleTimesliceTask = AsyncQueueHelpers::create(queue, {.name = "oldest-possible-timeslice", .score = 100});
       return ServiceHandle{TypeIdHelpers::uniqueId<DecongestionService>(), decongestion, ServiceKind::Serial};
     },
     .postForwarding = [](ProcessingContext& ctx, void* service) {
@@ -632,7 +692,6 @@ o2::framework::ServiceSpec CommonServices::decongestionSpec()
       auto& decongestion = services.get<DecongestionService>();
       auto& relayer = services.get<DataRelayer>();
       auto& timesliceIndex = services.get<TimesliceIndex>();
-      auto& proxy = services.get<FairMQDeviceProxy>();
       O2_SIGNPOST_ID_FROM_POINTER(cid, data_processor_context, &decongestion);
       O2_SIGNPOST_EVENT_EMIT(data_processor_context, cid, "oldest_possible_timeslice", "Received oldest possible timeframe %" PRIu64 " from channel %d",
                              (uint64_t)oldestPossibleTimeslice, channel.value);
@@ -650,59 +709,22 @@ o2::framework::ServiceSpec CommonServices::decongestionSpec()
         return;
       }
       auto& queue = services.get<AsyncQueue>();
-      const auto& spec = services.get<DeviceSpec const>();
       const auto& state = services.get<DeviceState>();
-      auto *device = services.get<RawDeviceService>().device();
       /// We use the oldest possible timeslice to debounce, so that only the latest one
       /// at the end of one iteration is sent.
       O2_SIGNPOST_EVENT_EMIT(data_processor_context, cid, "oldest_possible_timeslice", "Queueing oldest possible timeslice %" PRIu64 " propagation for execution.",
                              (uint64_t)oldestPossibleOutput.timeslice.value);
       AsyncQueueHelpers::post(
-        queue, decongestion.oldestPossibleTimesliceTask, [ref = services, oldestPossibleOutput, &decongestion, &proxy, &spec, device, &timesliceIndex](size_t id) {
-          O2_SIGNPOST_ID_GENERATE(cid, async_queue);
-          cid.value = id;
-          if (decongestion.lastTimeslice >= oldestPossibleOutput.timeslice.value) {
-            O2_SIGNPOST_EVENT_EMIT(async_queue, cid, "oldest_possible_timeslice", "Not sending already sent value: %" PRIu64 "> %" PRIu64,
-                decongestion.lastTimeslice, (uint64_t)oldestPossibleOutput.timeslice.value);
-            return;
-          }
-          O2_SIGNPOST_EVENT_EMIT(async_queue, cid, "oldest_possible_timeslice", "Running oldest possible timeslice %" PRIu64 " propagation.",
-                                 (uint64_t)oldestPossibleOutput.timeslice.value);
-          DataProcessingHelpers::broadcastOldestPossibleTimeslice(ref, oldestPossibleOutput.timeslice.value);
-
-          for (int fi = 0; fi < proxy.getNumForwardChannels(); fi++) {
-            auto& info = proxy.getForwardChannelInfo(ChannelIndex{fi});
-            auto& state = proxy.getForwardChannelState(ChannelIndex{fi});
-            // TODO: this we could cache in the proxy at the bind moment.
-            if (info.channelType != ChannelAccountingType::DPL) {
-              O2_SIGNPOST_EVENT_EMIT(async_queue, cid, "oldest_possible_timeslice", "Skipping channel %{public}s", info.name.c_str());
-              continue;
-            }
-            if (DataProcessingHelpers::sendOldestPossibleTimeframe(ref, info, state, oldestPossibleOutput.timeslice.value)) {
-              O2_SIGNPOST_EVENT_EMIT(async_queue, cid, "oldest_possible_timeslice",
-                                     "Forwarding to channel %{public}s oldest possible timeslice %" PRIu64 ", priority %d",
-                                     info.name.c_str(), (uint64_t)oldestPossibleOutput.timeslice.value, 20);
-            }
-          }
-          decongestion.lastTimeslice = oldestPossibleOutput.timeslice.value;
-        },
-        TimesliceId{oldestPossibleTimeslice}, -1);
+        queue, AsyncTask{ .timeslice = TimesliceId{oldestPossibleTimeslice}, 
+                .id = decongestion.oldestPossibleTimesliceTask, 
+                .debounce = -1, .callback = decongestionCallback}
+            .user<DecongestionContext>(DecongestionContext{.ref = services, .oldestPossibleOutput = oldestPossibleOutput}));
+
       if (decongestion.orderedCompletionPolicyActive) {
         AsyncQueueHelpers::post(
-          queue, decongestion.oldestPossibleTimesliceTask, [ref = services, oldestPossibleOutput, &decongestion, &proxy, &spec, &state, device, &timesliceIndex](size_t id) {
-            O2_SIGNPOST_ID_GENERATE(cid, async_queue);
-            int64_t oldNextTimeslice = decongestion.nextTimeslice;
-            decongestion.nextTimeslice = std::max(decongestion.nextTimeslice, (int64_t)oldestPossibleOutput.timeslice.value);
-            if (oldNextTimeslice != decongestion.nextTimeslice) {
-              if (state.transitionHandling != TransitionHandlingState::NoTransition && DefaultsHelpers::onlineDeploymentMode()) {
-                O2_SIGNPOST_EVENT_EMIT_WARN(async_queue, cid, "oldest_possible_timeslice", "Stop transition requested. Some Lifetime::Timeframe data got dropped starting at %" PRIi64, oldNextTimeslice);
-              } else {
-                O2_SIGNPOST_EVENT_EMIT_ERROR(async_queue, cid, "oldest_possible_timeslice", "Some Lifetime::Timeframe data got dropped starting at %" PRIi64, oldNextTimeslice);
-              }
-              timesliceIndex.rescan();
-            }
-          },
-          TimesliceId{oldestPossibleOutput.timeslice.value}, -1);
+          queue, AsyncTask{.timeslice = TimesliceId{oldestPossibleOutput.timeslice.value},.id = decongestion.oldestPossibleTimesliceTask,  .debounce = -1, 
+          .callback = decongestionCallbackOrdered}
+          .user<DecongestionContext>({.ref = services, .oldestPossibleOutput = oldestPossibleOutput}));
       } },
     .kind = ServiceKind::Serial};
 }
diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx
@@ -636,6 +636,39 @@ static auto toBeforwardedMessageSet = [](std::vector<ChannelIndex>& cachedForwar
   return cachedForwardingChoices.empty() == false;
 };
 
+struct DecongestionContext {
+  ServiceRegistryRef ref;
+  TimesliceIndex::OldestOutputInfo oldestTimeslice;
+};
+
+auto decongestionCallbackLate = [](AsyncTask& task, size_t aid) -> void {
+  auto& oldestTimeslice = task.user<DecongestionContext>().oldestTimeslice;
+  auto& ref = task.user<DecongestionContext>().ref;
+
+  auto& decongestion = ref.get<DecongestionService>();
+  auto& proxy = ref.get<FairMQDeviceProxy>();
+  if (oldestTimeslice.timeslice.value <= decongestion.lastTimeslice) {
+    LOG(debug) << "Not sending already sent oldest possible timeslice " << oldestTimeslice.timeslice.value;
+    return;
+  }
+  for (int fi = 0; fi < proxy.getNumForwardChannels(); fi++) {
+    auto& info = proxy.getForwardChannelInfo(ChannelIndex{fi});
+    auto& state = proxy.getForwardChannelState(ChannelIndex{fi});
+    O2_SIGNPOST_ID_GENERATE(aid, async_queue);
+    // TODO: this we could cache in the proxy at the bind moment.
+    if (info.channelType != ChannelAccountingType::DPL) {
+      O2_SIGNPOST_EVENT_EMIT(async_queue, aid, "forwardInputsCallback", "Skipping channel %{public}s because it's not a DPL channel",
+                             info.name.c_str());
+
+      continue;
+    }
+    if (DataProcessingHelpers::sendOldestPossibleTimeframe(ref, info, state, oldestTimeslice.timeslice.value)) {
+      O2_SIGNPOST_EVENT_EMIT(async_queue, aid, "forwardInputsCallback", "Forwarding to channel %{public}s oldest possible timeslice %zu, prio 20",
+                             info.name.c_str(), oldestTimeslice.timeslice.value);
+    }
+  }
+};
+
 // This is how we do the forwarding, i.e. we push
 // the inputs which are shared between this device and others
 // to the next one in the daisy chain.
@@ -721,36 +754,13 @@ static auto forwardInputs = [](ServiceRegistryRef registry, TimesliceSlot slot,
 
   auto& asyncQueue = registry.get<AsyncQueue>();
   auto& decongestion = registry.get<DecongestionService>();
-  auto& timesliceIndex = registry.get<TimesliceIndex>();
   O2_SIGNPOST_ID_GENERATE(aid, async_queue);
   O2_SIGNPOST_EVENT_EMIT(async_queue, aid, "forwardInputs", "Queuing forwarding oldestPossible %zu", oldestTimeslice.timeslice.value);
-  AsyncQueueHelpers::post(
-    asyncQueue, decongestion.oldestPossibleTimesliceTask, [&proxy, &decongestion, registry, oldestTimeslice, &timesliceIndex](size_t aid) {
-      // DataProcessingHelpers::broadcastOldestPossibleTimeslice(proxy, oldestTimeslice.timeslice.value);
-      if (oldestTimeslice.timeslice.value <= decongestion.lastTimeslice) {
-        LOG(debug) << "Not sending already sent oldest possible timeslice " << oldestTimeslice.timeslice.value;
-        return;
-      }
-      for (int fi = 0; fi < proxy.getNumForwardChannels(); fi++) {
-        auto& info = proxy.getForwardChannelInfo(ChannelIndex{fi});
-        auto& state = proxy.getForwardChannelState(ChannelIndex{fi});
-        O2_SIGNPOST_ID_GENERATE(aid, async_queue);
-        // TODO: this we could cache in the proxy at the bind moment.
-        if (info.channelType != ChannelAccountingType::DPL) {
-          O2_SIGNPOST_EVENT_EMIT(async_queue, aid, "forwardInputsCallback", "Skipping channel %{public}s because it's not a DPL channel",
-                                 info.name.c_str());
-
-          continue;
-        }
-        if (DataProcessingHelpers::sendOldestPossibleTimeframe(registry, info, state, oldestTimeslice.timeslice.value)) {
-          O2_SIGNPOST_EVENT_EMIT(async_queue, aid, "forwardInputsCallback", "Forwarding to channel %{public}s oldest possible timeslice %zu, prio 20",
-                                 info.name.c_str(), oldestTimeslice.timeslice.value);
-        }
-      }
-    },
-    oldestTimeslice.timeslice, -1);
+  AsyncQueueHelpers::post(asyncQueue, AsyncTask{.timeslice = oldestTimeslice.timeslice, .id = decongestion.oldestPossibleTimesliceTask, .debounce = -1, .callback = decongestionCallbackLate}
+                                        .user<DecongestionContext>({.ref = registry, .oldestTimeslice = oldestTimeslice}));
   O2_SIGNPOST_END(forwarding, sid, "forwardInputs", "Forwarding done");
 };
+
 extern volatile int region_read_global_dummy_variable;
 volatile int region_read_global_dummy_variable;
 
diff --git a/Framework/Core/test/test_AsyncQueue.cxx b/Framework/Core/test/test_AsyncQueue.cxx

Original file line number	Diff line number	Diff line change
`@@ -17,6 +17,10 @@ O2_DECLARE_DYNAMIC_LOG(async_queue);`
`17`	`17`
`18`	`18`	`namespace o2::framework`
`19`	`19`	`{`
	`20`	`+AsyncQueue::AsyncQueue()`
	`21`	`+{`
	`22`	`+}`
	`23`	`+`
`20`	`24`	`auto AsyncQueueHelpers::create(AsyncQueue& queue, AsyncTaskSpec spec) -> AsyncTaskId`
`21`	`25`	`{`
`22`	`26`	`AsyncTaskId id;`
`@@ -25,14 +29,9 @@ auto AsyncQueueHelpers::create(AsyncQueue& queue, AsyncTaskSpec spec) -> AsyncTa`
`25`	`29`	`return id;`
`26`	`30`	`}`
`27`	`31`
`28`		`-auto AsyncQueueHelpers::post(AsyncQueue& queue, AsyncTaskId id, AsyncCallback task, TimesliceId timeslice, int64_t debounce) -> void`
	`32`	`+auto AsyncQueueHelpers::post(AsyncQueue& queue, AsyncTask const& task) -> void`
`29`	`33`	`{`
`30`		`- AsyncTask taskToPost;`
`31`		`- taskToPost.task = task;`
`32`		`- taskToPost.id = id;`
`33`		`- taskToPost.timeslice = timeslice;`
`34`		`- taskToPost.debounce = debounce;`
`35`		`- queue.tasks.push_back(taskToPost);`
	`34`	`+ queue.tasks.push_back(task);`
`36`	`35`	`}`
`37`	`36`
`38`	`37`	`auto AsyncQueueHelpers::run(AsyncQueue& queue, TimesliceId oldestPossible) -> void`
`@@ -85,6 +84,8 @@ auto AsyncQueueHelpers::run(AsyncQueue& queue, TimesliceId oldestPossible) -> vo`
`85`	`84`	`}`
`86`	`85`	`}`
`87`	`86`	`// Keep only the tasks with the highest debounce value for a given id`
	`87`	`+ // For this reason I need to keep the callback in the task itself, because`
	`88`	`+ // two different callbacks with the same id will be coalesced.`
`88`	`89`	`auto newEnd = std::unique(order.begin(), order.end(), [&queue](int a, int b) {`
`89`	`90`	`return queue.tasks[a].runnable == queue.tasks[b].runnable && queue.tasks[a].id.value == queue.tasks[b].id.value && queue.tasks[a].debounce >= 0 && queue.tasks[b].debounce >= 0;`
`90`	`91`	`});`
`@@ -111,7 +112,7 @@ auto AsyncQueueHelpers::run(AsyncQueue& queue, TimesliceId oldestPossible) -> vo`
`111`	`112`	`O2_SIGNPOST_EVENT_EMIT(async_queue, opid, "run", "Running task %{public}s (%d) for timeslice %zu",`
`112`	`113`	`queue.prototypes[queue.tasks[i].id.value].name.c_str(), i,`
`113`	`114`	`queue.tasks[i].timeslice.value);`
`114`		`- queue.tasks[i].task(opid.value);`
	`115`	`+ queue.tasks[i].callback(queue.tasks[i], opid.value);`
`115`	`116`	`O2_SIGNPOST_EVENT_EMIT(async_queue, opid, "run", "Done running %d", i);`
`116`	`117`	`}`
`117`	`118`	`}`