thread safe scheduler (#241)

Signed-off-by: turuslan <[email protected]>

Author: turuslan

housekeeping/clang-tidy-diff.sh

@@ -8,5 +8,5 @@
 BUILD_DIR="${BUILD_DIR:?}"
 
 cd $(dirname $0)/..
-git diff -U0 origin/master | clang-tidy-diff.py -p1 -path $BUILD_DIR -iregex '(include|src)\/.*\.(h|c|hpp|cpp)' | tee clang-tidy.log
+git diff -U0 origin/master | clang-tidy-diff.py -p1 -path $BUILD_DIR -iregex '(include|src)\/.*\.(h|c|hpp|cpp)' -clang-tidy-binary clang-tidy-15 | tee clang-tidy.log
 ! grep ': error:' clang-tidy.log

include/libp2p/basic/cancel.hpp

@@ -0,0 +1,48 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <memory>
+
+namespace libp2p {
+  /**
+   * Interface with destructor for `std::unique_ptr`.
+   */
+  // NOLINTNEXTLINE(cppcoreguidelines-special-member-functions)
+  class CancelDtor {
+   public:
+    inline virtual ~CancelDtor() = 0;
+  };
+  CancelDtor::~CancelDtor() = default;
+
+  /**
+   * RAII object to cancel operation.
+   */
+  using Cancel = std::unique_ptr<CancelDtor>;
+
+  template <typename F>
+  class CancelDtorFn final : public CancelDtor {
+    F f;
+
+   public:
+    explicit CancelDtorFn(F f) : f{std::move(f)} {}
+
+    ~CancelDtorFn() override {
+      f();
+    }
+
+    // clang-tidy cppcoreguidelines-special-member-functions
+    CancelDtorFn(const CancelDtorFn &) = delete;
+    void operator=(const CancelDtorFn &) = delete;
+    CancelDtorFn(CancelDtorFn &&) = delete;
+    void operator=(CancelDtorFn &&) = delete;
+  };
+
+  Cancel cancelFn(auto fn) {
+    return std::make_unique<CancelDtorFn<decltype(fn)>>(std::move(fn));
+  }
+}  // namespace libp2p

include/libp2p/basic/scheduler.hpp

@@ -10,64 +10,13 @@
 #include <functional>
 #include <memory>
 
-#include <libp2p/outcome/outcome.hpp>
+#include <libp2p/basic/cancel.hpp>
 
 namespace libp2p::basic {
-
-  constexpr std::chrono::milliseconds kZeroTime =
-      std::chrono::milliseconds::zero();
-
-  /**
-   * Feedback from scheduler backend to Scheduler implementation.
-   */
-  class SchedulerBackendFeedback {
-   public:
-    virtual ~SchedulerBackendFeedback() = default;
-
-    /**
-     * Called from backend to fire callbacks
-     * @param current_clock
-     * For timers: non-zero current async ticks in milliseconds since async's
-     * epoch;
-     * For callbacks deferred to next IO loop cycle: zero
-     */
-    virtual void pulse(std::chrono::milliseconds current_clock) noexcept = 0;
-  };
-
-  /**
-   * Scheduling engine that depends on implementation:
-   * 1) AsioSchedulerBackend (for asio-based asynchrony, it uses
-   * std::chrono::steady_clock)
-   * 2) ManualSchedulerBackend (for testing purposes,
-   * uses manual time shifts as a async)
-   */
-  class SchedulerBackend {
-   public:
-    virtual ~SchedulerBackend() = default;
-
-    /**
-     * Current async
-     * @return Milliseconds elapsed from async's epoch
-     */
-    virtual std::chrono::milliseconds now() const noexcept = 0;
-
-    /**
-     * Implementation-defined defer or delay function.
-     * @param abs_time Time since async's epoch.
-     * If abs_time == 0 then SchedulerBackendFeedback::pulse()
-     * will be called on the next IO loop cycle with zero argument
-     * @param scheduler Weak reference to scheduler (as it may expire)
-     */
-    virtual void setTimer(
-        std::chrono::milliseconds abs_time,
-        std::weak_ptr<SchedulerBackendFeedback> scheduler) = 0;
-  };
-
   /**
    * Scheduler API. Provides callback deferring facilities and low-res timer
    */
-  class Scheduler : public SchedulerBackendFeedback,
-                    public std::enable_shared_from_this<Scheduler> {
+  class Scheduler {
    public:
     struct Config {
       static constexpr std::chrono::milliseconds kMaxTimerThreshold =
@@ -80,79 +29,19 @@ namespace libp2p::basic {
       std::chrono::milliseconds max_timer_threshold = kMaxTimerThreshold;
     };
 
-    enum class Error {
-      // Invalid argument passed
-      kInvalidArgument = 1,
-
-      // Scheduler handle detached, cannot reschedule
-      kHandleDetached,
-
-      // Scheduler item not found, cannot reschedule
-      kItemNotFound,
-    };
-
-    /**
-     * Handle provides scoped callbacks lifetime, allows for canceling and
-     * rescheduling
-     */
-    class Handle {
-     public:
-      /**
-       * { async-time, seq-number } structure for proper ordering and uniqueness
-       * within Scheduler
-       */
-      using Ticket = std::pair<std::chrono::milliseconds, uint64_t>;
-
-      Handle(const Handle &) = delete;
-      Handle &operator=(const Handle &) = delete;
-
-      Handle() = default;
-      Handle(Handle &&) = default;
-
-      /**
-       * Ctor. Called from SchedulerImpl
-       * @param ticket unique ticket, used for cancelling and rescheduling
-       * @param scheduler handle's owner object
-       */
-      Handle(Ticket ticket, std::weak_ptr<Scheduler> scheduler);
-
-      /**
-       *  Move assignment cancels existing ticket, if any
-       */
-      Handle &operator=(Handle &&r) noexcept;
-
-      /**
-       *  Non-trivial dtor cancels existing ticket, if any
-       */
-      ~Handle();
-
-      /**
-       * Cancels existing ticket, if any
-       */
-      void cancel() noexcept;
-
-      /**
-       * Reschedules existing ticket, if it is still active.
-       * Allows reentrancy (can be called from inside the callback)
-       * @param delay_from_now Relative time
-       * @return success or error value from Scheduler::Error
-       */
-      outcome::result<void> reschedule(
-          std::chrono::milliseconds delay_from_now) noexcept;
-
-     private:
-      Ticket ticket_;
-      std::weak_ptr<Scheduler> scheduler_;
-    };
+    using Handle = Cancel;
 
     using Callback = std::function<void()>;
+    using Time = std::chrono::milliseconds;
+
+    virtual ~Scheduler() = default;
 
     /**
      * Defers callback to be executed during the next IO loop cycle
      * @param cb callback
      */
     void schedule(Callback &&cb) noexcept {
-      std::ignore = scheduleImpl(std::move(cb), kZeroTime, false);
+      std::ignore = scheduleImpl(std::move(cb), Time::zero(), false);
     }
 
     /**
@@ -172,7 +61,7 @@ namespace libp2p::basic {
      * lifetime
      */
     [[nodiscard]] Handle scheduleWithHandle(Callback &&cb) noexcept {
-      return scheduleImpl(std::move(cb), kZeroTime, true);
+      return scheduleImpl(std::move(cb), Time::zero(), true);
     }
 
     /**
@@ -201,9 +90,6 @@ namespace libp2p::basic {
                                          bool) = delete;
 
    protected:
-    /// Handle calls cancel() and reschedule()
-    friend class Handle;
-
     /**
      * Called from schedule() and scheduleWithHandle() functions
      * @param cb callback
@@ -214,24 +100,5 @@ namespace libp2p::basic {
     virtual Handle scheduleImpl(Callback &&cb,
                                 std::chrono::milliseconds delay_from_now,
                                 bool make_handle) noexcept = 0;
-
-    /**
-     * Called from Handle (from move assignment, destructor, or manually)
-     * @param ticket handle's existing ticket
-     */
-    virtual void cancel(Handle::Ticket ticket) noexcept = 0;
-
-    /**
-     * Called from Handle.
-     * Reschedules existing ticket.
-     * Allows reentrancy (can be called from inside the callback)
-     * @param delay_from_now Relative time
-     * @return success or error value from Scheduler::Error
-     */
-    virtual outcome::result<Handle::Ticket> reschedule(
-        Handle::Ticket ticket,
-        std::chrono::milliseconds delay_from_now) noexcept = 0;
   };
 }  // namespace libp2p::basic
-
-OUTCOME_HPP_DECLARE_ERROR(libp2p::basic, Scheduler::Error)

include/libp2p/basic/scheduler/asio_scheduler_backend.hpp

@@ -9,7 +9,7 @@
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/steady_timer.hpp>
 
-#include <libp2p/basic/scheduler.hpp>
+#include <libp2p/basic/scheduler/backend.hpp>
 
 namespace libp2p::basic {
 
@@ -27,6 +27,8 @@ namespace libp2p::basic {
     explicit AsioSchedulerBackend(
         std::shared_ptr<boost::asio::io_context> io_context);
 
+    void post(std::function<void()> &&) override;
+
     /**
      * @return Milliseconds since steady clock's epoch
      */

include/libp2p/basic/scheduler/backend.hpp

@@ -0,0 +1,60 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <chrono>
+#include <functional>
+#include <memory>
+
+namespace libp2p::basic {
+  /**
+   * Feedback from scheduler backend to Scheduler implementation.
+   */
+  class SchedulerBackendFeedback {
+   public:
+    virtual ~SchedulerBackendFeedback() = default;
+
+    /**
+     * Called from backend to fire callbacks
+     */
+    virtual void pulse() noexcept = 0;
+  };
+
+  /**
+   * Scheduling engine that depends on implementation:
+   * 1) AsioSchedulerBackend (for asio-based asynchrony, it uses
+   * std::chrono::steady_clock)
+   * 2) ManualSchedulerBackend (for testing purposes,
+   * uses manual time shifts as a async)
+   */
+  class SchedulerBackend {
+   public:
+    virtual ~SchedulerBackend() = default;
+
+    /**
+     * boost::asio::io_context::post
+     */
+    virtual void post(std::function<void()> &&) = 0;
+
+    /**
+     * Current async
+     * @return Milliseconds elapsed from async's epoch
+     */
+    virtual std::chrono::milliseconds now() const noexcept = 0;
+
+    /**
+     * Implementation-defined defer or delay function.
+     * @param abs_time Time since async's epoch.
+     * If abs_time == 0 then SchedulerBackendFeedback::pulse()
+     * will be called on the next IO loop cycle with zero argument
+     * @param scheduler Weak reference to scheduler (as it may expire)
+     */
+    virtual void setTimer(
+        std::chrono::milliseconds abs_time,
+        std::weak_ptr<SchedulerBackendFeedback> scheduler) = 0;
+  };
+}  // namespace libp2p::basic

include/libp2p/basic/scheduler/manual_scheduler_backend.hpp

@@ -6,9 +6,10 @@
 
 #pragma once
 
-#include <vector>
+#include <deque>
+#include <optional>
 
-#include <libp2p/basic/scheduler.hpp>
+#include <libp2p/basic/scheduler/backend.hpp>
 
 namespace libp2p::basic {
 
@@ -22,6 +23,8 @@ namespace libp2p::basic {
    public:
     ManualSchedulerBackend() : current_clock_(1) {}
 
+    void post(std::function<void()> &&) override;
+
     /**
      * @return Milliseconds since clock's epoch. Clock is set manually
      */
@@ -54,24 +57,29 @@ namespace libp2p::basic {
      * @return true if no more events scheduled
      */
     bool empty() const {
-      return deferred_callbacks_.empty() && !timer_callback_;
+      return deferred_callbacks_.empty() and not timer_expires_;
+    }
+
+    void run() {
+      while (not empty()) {
+        shiftToTimer();
+      }
     }
 
    private:
+    void callDeferred();
+
     /// Current time, set manually
     std::chrono::milliseconds current_clock_;
 
     /// Callbacks deferred for the next cycle
-    std::vector<Callback> deferred_callbacks_;
-
-    /// Currently processed callback (reentrancy + rescheduling reasons here)
-    std::vector<Callback> in_process_;
+    std::deque<Callback> deferred_callbacks_;
 
     /// Timer callback
-    Callback timer_callback_;
+    std::weak_ptr<SchedulerBackendFeedback> scheduler_;
 
     /// Expiry of timer event
-    std::chrono::milliseconds timer_expires_{};
+    std::optional<std::chrono::milliseconds> timer_expires_;
   };
 
 }  // namespace libp2p::basic