timeoutQueue.H

Go to the documentation of this file.

/*
                          Aleph_w
 
  Data structures & Algorithms
  version 2.0.0b
  https://github.com/lrleon/Aleph-w
 
  This file is part of Aleph-w library
 
  Copyright (c) 2002-2026 Leandro Rabindranath Leon
 
  Permission is hereby granted, free of charge, to any person obtaining a copy
  of this software and associated documentation files (the "Software"), to deal
  in the Software without restriction, including without limitation the rights
  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  copies of the Software, and to permit persons to whom the Software is
  furnished to do so, subject to the following conditions:
 
  The above copyright notice and this permission notice shall be included in all
  copies or substantial portions of the Software.
 
  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  SOFTWARE.
*/
 
 
# ifndef TIMEOUTQUEUE_H
# define TIMEOUTQUEUE_H
 
# include <unistd.h>
# include <iostream>
# include <cassert>
# include <cstdlib>
# include <mutex>
# include <condition_variable>
# include <thread>
# include <chrono>
# include <functional>
# include <string>
# include <atomic>
# include <cstdint>
# include <ah-errors.H>
# include <tpl_dynMapTree.H>
# include <tpl_dynSetTree.H>
# include <tpl_binHeap.H>
# include <ah-time.H>
# include <utility>
 
class TimeoutQueue
{
public:
  class Event : private BinHeapVtl<Time>::Node
  {
    friend class TimeoutQueue;
 
  public:
    enum Execution_Status
    {
      Out_Queue, 
      In_Queue, 
      Canceled, 
      Executing, 
      Executed, 
      To_Delete, 
      Deleted 
    };
 
    using CompletionCallback = std::function<void(Event *, Execution_Status)>;
 
    using EventId = uint64_t;
 
    static constexpr EventId InvalidId = 0;
 
  private:
    static std::atomic<EventId> nextId;
 
    EventId event_id;
    std::atomic<Execution_Status> execution_status;
    std::string event_name;
    CompletionCallback on_completed;
 
    Time& time_key() { return this->get_key(); }
    [[nodiscard]] const Time& time_key() const { return this->get_key(); }
 
    void set_trigger_time(const Time _t)
    {
      assert(_t.tv_nsec >= 0 and _t.tv_nsec < NSEC);
      time_key() = _t;
    }
 
    void set_execution_status(Execution_Status status)
    {
      execution_status = status;
    }
 
    void invoke_completion_callback()
    {
      if (on_completed)
        on_completed(this, execution_status);
    }
 
  protected:
    Event(const Time & t, std::string name = "")
      : BinHeapVtl<Time>::Node(t), event_id(++nextId),
        execution_status(Out_Queue), event_name(std::move(name))
    {
      assert(time_key().tv_nsec >= 0 and time_key().tv_nsec < NSEC);
    }
 
    Event(const long sec, const long nsec, std::string name = "")
      : event_id(++nextId), execution_status(Out_Queue), event_name(std::move(name))
    {
      assert(nsec >= 0 and nsec < NSEC);
 
      time_key().tv_sec = sec;
      time_key().tv_nsec = nsec;
    }
 
  public:
    ~Event() noexcept override
    {
      if (execution_status == In_Queue) [[unlikely]]
        {
          ah_warning(std::cerr)
              << "FATAL: Deleting Event " << this << " (ID " << event_id << ", name: '"
              << event_name << "') that is still In_Queue. "
              << "This causes use-after-free when the worker thread accesses it. "
              << "Must call cancel_event() or use cancel_delete_event() before destroying.";
          abort();
        }
    }
 
    [[nodiscard]] Execution_Status get_execution_status() const { return execution_status; }
 
    [[nodiscard]] Time getAbsoluteTime() const { return time_key(); }
 
    void set_for_deletion() { execution_status = To_Delete; }
 
    [[nodiscard]] EventId get_id() const { return event_id; }
 
    [[nodiscard]] const std::string &get_name() const { return event_name; }
 
    void set_name(const std::string & name) { event_name = name; }
 
    void set_completion_callback(CompletionCallback cb) { on_completed = std::move(cb); }
 
    virtual void EventFct() = 0;
  }; /* end class Event */
 
private:
  BinHeapVtl<Time> prio_queue;
  DynMapTree<Event::EventId, Event *> event_map; // For O(log n) lookup by ID
  DynSetTree<Event *> event_registry; // Currently managed event pointers
 
  mutable std::mutex mtx;
  std::condition_variable cond;
  std::thread workerThread;
 
  bool isShutdown;
  bool isPaused = false;
 
  size_t executedCount = 0;
  size_t canceledCount = 0;
 
  std::condition_variable emptyCondition;
 
  void triggerEvent();
 
  // Set isShutdown and notify; caller must hold mtx
  void shutdown_locked()
  {
    isShutdown = true;
    cond.notify_all();
  }
 
public:
  TimeoutQueue();
 
  ~TimeoutQueue();
 
  void schedule_event(const Time & trigger_time, Event *);
 
  void schedule_event(Event * event);
 
  bool cancel_event(Event * event);
 
  void reschedule_event(const Time & trigger_time, Event *event);
 
  void cancel_delete_event(Event *& event);
 
  void shutdown();
 
  [[nodiscard]] std::thread::id getThreadId() const { return workerThread.get_id(); }
 
  [[nodiscard]] size_t size() const;
 
  [[nodiscard]] bool is_empty() const;
 
  [[nodiscard]] bool is_running() const;
 
  void schedule_after_ms(int ms_from_now, Event *event);
 
  [[nodiscard]] Time next_event_time() const;
 
  size_t clear_all();
 
  [[nodiscard]] size_t executed_count() const;
 
  [[nodiscard]] size_t canceled_count() const;
 
  void reset_stats();
 
  void pause();
 
  void resume();
 
  [[nodiscard]] bool is_paused() const;
 
  bool wait_until_empty(int timeout_ms = 0);
 
  [[nodiscard]] Event * find_by_id(const Event::EventId id) const;
 
  bool cancel_by_id(Event::EventId id);
};
 
 
# endif /* TIMEOUTQUEUE_H */