Initial implementation of new task system

CMakeLists.txt

@@ -280,6 +280,14 @@ add_ccf_static_library(
   LINK_LIBS qcbor t_cose http_parser ccfcrypto ccf_kv
 )
 
+# CCF task system library
+add_ccf_static_library(
+  ccf_tasks
+  SRCS ${CCF_DIR}/src/tasks/task_system.cpp ${CCF_DIR}/src/tasks/job_board.cpp
+       ${CCF_DIR}/src/tasks/ordered_tasks.cpp
+       ${CCF_DIR}/src/tasks/fan_in_tasks.cpp
+)
+
 # Common test args for Python scripts starting up CCF networks
 set(WORKER_THREADS
     0
@@ -529,6 +537,20 @@ if(BUILD_TESTS)
     )
     target_link_libraries(ds_test PRIVATE ${CMAKE_THREAD_LIBS_INIT})
 
+    add_unit_test(
+      task_system_test
+      ${CMAKE_CURRENT_SOURCE_DIR}/src/tasks/test/basic_tasks.cpp
+      ${CMAKE_CURRENT_SOURCE_DIR}/src/tasks/test/ordered_tasks.cpp
+      ${CMAKE_CURRENT_SOURCE_DIR}/src/tasks/test/delayed_tasks.cpp
+      ${CMAKE_CURRENT_SOURCE_DIR}/src/tasks/test/fan_in_tasks.cpp
+    )
+    target_link_libraries(task_system_test PRIVATE ccf_tasks)
+
+    add_unit_test(
+      task_system_demo ${CMAKE_CURRENT_SOURCE_DIR}/src/tasks/test/demo/main.cpp
+    )
+    target_link_libraries(task_system_demo PRIVATE ccf_tasks)
+
     add_unit_test(
       ledger_test ${CMAKE_CURRENT_SOURCE_DIR}/src/host/test/ledger.cpp
     )
@@ -775,6 +797,14 @@ if(BUILD_TESTS)
     )
     add_picobench(merkle_bench SRCS src/node/test/merkle_bench.cpp)
     add_picobench(hash_bench SRCS src/ds/test/hash_bench.cpp)
+
+    add_picobench(
+      task_bench
+      SRCS ${CMAKE_CURRENT_SOURCE_DIR}/src/tasks/test/bench/merge_bench.cpp
+           ${CMAKE_CURRENT_SOURCE_DIR}/src/tasks/test/bench/sleep_bench.cpp
+           ${CMAKE_CURRENT_SOURCE_DIR}/src/tasks/test/bench/contention_bench.cpp
+      LINK_LIBS ccf_tasks
+    )
   endif()
 
   if(LONG_TESTS)

src/tasks/basic_task.h

@@ -0,0 +1,35 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the Apache 2.0 License.
+#pragma once
+
+#include "tasks/task.h"
+
+namespace ccf::tasks
+{
+  struct BasicTask : public BaseTask
+  {
+    using Fn = std::function<void()>;
+
+    Fn fn;
+    const std::string name;
+
+    BasicTask(const Fn& _fn, const std::string& s = "[Anon]") : fn(_fn), name(s)
+    {}
+
+    void do_task_implementation() override
+    {
+      fn();
+    }
+
+    std::string get_name() const override
+    {
+      return name;
+    }
+  };
+
+  template <typename... Ts>
+  Task make_basic_task(Ts&&... ts)
+  {
+    return std::make_shared<BasicTask>(std::forward<Ts>(ts)...);
+  }
+}

src/tasks/fan_in_tasks.cpp

@@ -0,0 +1,138 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the Apache 2.0 License.
+
+#include "tasks/fan_in_tasks.h"
+
+#include <map>
+#include <mutex>
+#include <stdexcept>
+
+#define FMT_HEADER_ONLY
+#include <fmt/format.h>
+
+namespace ccf::tasks
+{
+  struct FanInTasks::PImpl
+  {
+    std::string name;
+    IJobBoard& job_board;
+
+    std::mutex mutex;
+    std::map<size_t, Task> pending_tasks;
+    size_t next_expected_task_index = 0;
+    std::atomic<bool> active = false;
+  };
+
+  void FanInTasks::enqueue_on_board()
+  {
+    pimpl->job_board.add_task(shared_from_this());
+  }
+
+  void FanInTasks::do_task_implementation()
+  {
+    std::vector<Task> current_batch;
+
+    {
+      std::lock_guard<std::mutex> lock(pimpl->mutex);
+      pimpl->active.store(true);
+
+      auto it = pimpl->pending_tasks.find(pimpl->next_expected_task_index);
+      while (it != pimpl->pending_tasks.end())
+      {
+        current_batch.push_back(it->second);
+        pimpl->pending_tasks.erase(it);
+
+        ++pimpl->next_expected_task_index;
+        it = pimpl->pending_tasks.find(pimpl->next_expected_task_index);
+      }
+    }
+
+    for (auto& task : current_batch)
+    {
+      task->do_task();
+    }
+
+    {
+      std::lock_guard<std::mutex> lock(pimpl->mutex);
+      pimpl->active.store(false);
+
+      auto it = pimpl->pending_tasks.find(pimpl->next_expected_task_index);
+      if (it != pimpl->pending_tasks.end())
+      {
+        // While we were executing the previous batch, a call to fan_in_task
+        // provided the _next_ contiguous task. We're now responsible for
+        // re-enqueuing this task
+        enqueue_on_board();
+      }
+    }
+  }
+
+  FanInTasks::FanInTasks(IJobBoard& jb, const std::string& s) :
+    pimpl(std::make_unique<FanInTasks::PImpl>(s, jb))
+  {}
+
+  FanInTasks::~FanInTasks() = default;
+
+  std::string FanInTasks::get_name() const
+  {
+    return pimpl->name;
+  }
+
+  void FanInTasks::add_task(size_t task_index, Task task)
+  {
+    {
+      std::lock_guard<std::mutex> lock(pimpl->mutex);
+
+      if (task_index < pimpl->next_expected_task_index)
+      {
+        throw std::runtime_error(fmt::format(
+          "[{}] Received task {} ({}) out-of-order - already advanced next "
+          "expected "
+          "to {}",
+          get_name(),
+          task_index,
+          task->get_name(),
+          pimpl->next_expected_task_index));
+      }
+
+      auto it = pimpl->pending_tasks.find(task_index);
+      if (it != pimpl->pending_tasks.end())
+      {
+        throw std::runtime_error(fmt::format(
+          "[{}] Received duplicate task {} ({}) - already have pending task {}",
+          get_name(),
+          task_index,
+          task->get_name(),
+          it->second == nullptr ? std::string("nullptr") :
+                                  it->second->get_name()));
+      }
+
+      pimpl->pending_tasks.emplace(task_index, task);
+
+      if (!pimpl->active.load())
+      {
+        if (task_index == pimpl->next_expected_task_index)
+        {
+          enqueue_on_board();
+        }
+      }
+    }
+  }
+
+  namespace
+  {
+    struct ConcreteFanInTasks : public FanInTasks
+    {
+    public:
+      ConcreteFanInTasks(IJobBoard& jb, const std::string& s) :
+        FanInTasks(jb, s)
+      {}
+    };
+  }
+
+  std::shared_ptr<FanInTasks> make_fan_in_tasks(
+    IJobBoard& jb, const std::string& s)
+  {
+    return std::make_shared<ConcreteFanInTasks>(jb, s);
+  }
+}

src/tasks/fan_in_tasks.h

@@ -0,0 +1,37 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the Apache 2.0 License.
+#pragma once
+
+#include "tasks/job_board_interface.h"
+#include "tasks/task.h"
+
+#include <memory>
+
+namespace ccf::tasks
+{
+  class FanInTasks : public BaseTask,
+                     public std::enable_shared_from_this<FanInTasks>
+  {
+  protected:
+    struct PImpl;
+    std::unique_ptr<PImpl> pimpl = nullptr;
+
+    void enqueue_on_board();
+    void do_task_implementation() override;
+
+    // Constructor is protected, to ensure this is only created via the
+    // make_fan_in_tasks factory function (ensuring this is always owned by
+    // a shared_ptr)
+    FanInTasks(IJobBoard& jb, const std::string& s);
+
+  public:
+    ~FanInTasks();
+
+    std::string get_name() const override;
+
+    void add_task(size_t task_index, Task task);
+  };
+
+  std::shared_ptr<FanInTasks> make_fan_in_tasks(
+    IJobBoard& jb, const std::string& s = "[FanIn]");
+}

src/tasks/job_board.cpp

@@ -0,0 +1,53 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the Apache 2.0 License.
+#include "tasks/job_board.h"
+
+namespace ccf::tasks
+{
+  void JobBoard::add_task(Task&& t)
+  {
+    {
+      std::lock_guard<std::mutex> lock(mutex);
+      queue.emplace(std::move(t));
+    }
+    work_beacon.notify_work_available();
+  }
+
+  Task JobBoard::get_task()
+  {
+    std::lock_guard<std::mutex> lock(mutex);
+    if (queue.empty())
+    {
+      return nullptr;
+    }
+
+    Task t = queue.front();
+    queue.pop();
+    return t;
+  }
+
+  bool JobBoard::empty()
+  {
+    std::lock_guard<std::mutex> lock(mutex);
+    return queue.empty();
+  }
+
+  Task JobBoard::wait_for_task(const std::chrono::milliseconds& timeout)
+  {
+    using TClock = std::chrono::system_clock;
+
+    const auto start = TClock::now();
+    const auto until = start + timeout;
+
+    while (true)
+    {
+      auto task = get_task();
+      if (task != nullptr || TClock::now() >= until)
+      {
+        return task;
+      }
+
+      work_beacon.wait_for_work_with_timeout(timeout);
+    }
+  }
+}

src/tasks/job_board.h

@@ -0,0 +1,25 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the Apache 2.0 License.
+#pragma once
+
+#include "ds/work_beacon.h"
+#include "tasks/job_board_interface.h"
+
+#include <mutex>
+#include <queue>
+
+namespace ccf::tasks
+{
+  struct JobBoard : public IJobBoard
+  {
+    std::mutex mutex;
+    std::queue<Task> queue;
+    ccf::ds::WorkBeacon work_beacon;
+
+    void add_task(Task&& t) override;
+    Task get_task() override;
+    bool empty() override;
+
+    Task wait_for_task(const std::chrono::milliseconds& timeout) override;
+  };
+}

src/tasks/job_board_interface.h

@@ -0,0 +1,21 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the Apache 2.0 License.
+#pragma once
+
+#include "tasks/task.h"
+
+#include <mutex>
+#include <queue>
+#include <thread>
+
+namespace ccf::tasks
+{
+  struct IJobBoard
+  {
+    virtual void add_task(Task&& t) = 0;
+    virtual Task get_task() = 0;
+    virtual bool empty() = 0;
+
+    virtual Task wait_for_task(const std::chrono::milliseconds& timeout) = 0;
+  };
+}