9
\$\begingroup\$

I've written an event dispatcher for a private project which task is to collect events from different threads and dispatch them in a single thread:

namespace events
{
namespace
{
// unfortunately make_unique is not part of c++11 so we use our own
template <typename T, typename... Args>
std::unique_ptr<T> make_unique(Args &&... args)
{
  return std::unique_ptr<T>(new T(std::forward<Args>(args)...));
}
}

// Dispatcher for events posted from arbitrary threads
// Users can register functions that listen for certain events. When such events are posted from
// arbitary threads, a single dispatcher thread will sequentially call the registered callbacks.
template <typename BaseEvent>
class dispatcher
{
public:
  dispatcher() : running_(false), ready_(false) {}
  // non copyable
  dispatcher(dispatcher const &) = delete;
  dispatcher &operator=(dispatcher) = delete;
  // but movable
  dispatcher(dispatcher &&) = default;

  // Needs to be called once for dispatching to start
  // It is possible to register callbacks and post events before this function has been called
  void start()
  {
    std::unique_lock<std::mutex> lock(mutex_);
    running_ = true;
    thread_ = make_unique<std::thread>(&dispatcher<BaseEvent>::worker, this);
    // do not return until thread is ready for processing events
    condition_.wait(lock, [this]() { return !ready_; });
  }

  // Stops processing of events and joins dispatching thread
  // Depending on timing, events already queued may still be executed. The queue is not cleared
  // after stopping, so it is possible to use stop()/start() sequences to pause execution
  // After stop() returns it is ensured that no callbacks will be executed
  void stop()
  {
    {
      std::unique_lock<std::mutex> lock(mutex_);
      running_ = false;
    }
    condition_.notify_one();
    thread_->join();
  }

  using callback_t = std::function<void(BaseEvent const &)>;

  // Register a function as an event listener
  // It is allowed to register multiple callbacks for the same event id
  void register_event(int event_id, callback_t function)
  {
    std::unique_lock<std::mutex> lock(mutex_);
    callbacks_.insert(std::make_pair(event_id, std::move(function)));
  }

  // Post event for immediate execution
  template <typename Event>
  void post(int event_id, Event ev)
  {
    post_delayed_until(event_id, std::move(ev), std::chrono::high_resolution_clock::now());
  }

  // Post event for delayed execution
  template <typename Event, typename Duration>
  void post_delayed_for(int event_id, Event ev, Duration const &duration)
  {
    post_delayed_until(event_id, std::move(ev),
                       std::chrono::high_resolution_clock::now() + duration);
  }

  // Post event for delayed execution
  template <typename Event, typename TimePoint>
  void post_delayed_until(int event_id, Event ev, TimePoint time_point)
  {
    {
      std::unique_lock<std::mutex> lock(mutex_);
      events_.insert(std::make_pair(std::move(time_point),
                                    std::bind(&dispatcher<BaseEvent>::defer<Event>, this, event_id,
                                              std::move(ev), std::placeholders::_1)));
    }
    condition_.notify_one();
  }

private:
  // dispatcher working thread
  void worker()
  {
    bool wait_delayed = false;
    std::chrono::time_point<std::chrono::high_resolution_clock> wait_until;

    std::unique_lock<std::mutex> lock(mutex_);

    // signal start() function that we are ready for processing
    ready_ = true;
    condition_.notify_one();

    while (running_)
    {
      // step through the event queue, since we are giving up our lock during callback execution we
      // cannot rely on it being valid after that which makes traversing a little more complicated
      auto it = events_.begin();
      while (running_ && it != events_.end())
      {
        // our event queue is sorted for execution timestamps, so we can stop iterating once we hit
        // an event in the future
        if (it->first > std::chrono::high_resolution_clock::now())
        {
          wait_delayed = true;
          wait_until = it->first;
          break;
        }

        auto function = it->second;
        events_.erase(it);

        // giving up lock here!
        function(lock);

        it = events_.begin();
      }

      // spurious interrupts are okay here; if the queue is still empty we will just return to sleep
      if (wait_delayed)
      {
        condition_.wait_until(lock, wait_until);
        wait_delayed = false;
      }
      else
      {
        condition_.wait(lock);
      }
    }
  }

  // wrapper around callback function
  template <typename Event>
  void defer(int event_id, Event ev, std::unique_lock<std::mutex> &lock)
  {
    // because we unlock the mutex during callbacks and the callbacks map may change, we must make
    // a copy of the functions we need to call
    auto range = callbacks_.equal_range(event_id);
    std::vector<callback_t> functions;
    for (auto it = range.first; it != range.second; ++it)
    {
      functions.push_back(it->second);
    }

    // since we made a copy of the callback list we can now process all registered callbacks in one
    // batch without mutex lock
    lock.unlock();
    for (auto &function : functions)
    {
      function(ev);
    }
    lock.lock();
  }

  bool running_;
  bool ready_;

  std::unique_ptr<std::thread> thread_;
  std::mutex mutex_;
  std::condition_variable condition_;

  std::multimap<int, callback_t> callbacks_;

  using defer_t = std::function<void(std::unique_lock<std::mutex> &)>;
  std::multimap<std::chrono::time_point<std::chrono::high_resolution_clock>, defer_t> events_;
};
}

I'm trying to make the dispatcher as user friendly as possible. What I am still unhappy with so far, is that you have to use a base event structure for passing data. By far cooler would be to pass in arbitrary argument lists in post, but I have not found a way so far to make it possible.

Anyway, all kind of feedback is appreciated!

Here is a usage example:

struct event
{
  events::dispatcher<event> &dispatcher;
  static int const id = 0;
};

int main()
{
  events::dispatcher<event> dispatcher;

  dispatcher.register_event(event::id, [](event const &e) {
    std::cout << "timer event occured\n";
    e.dispatcher.post_delayed_for(event::id, e, std::chrono::seconds(1));
  });

  dispatcher.start();

  event e{dispatcher};
  dispatcher.post_delayed_for(event::id, std::move(e), std::chrono::seconds(1));

  std::this_thread::sleep_for(std::chrono::seconds(10));

  dispatcher.stop();

  return 0;
}

And here is a snippet of what I'd imagine the usage to be ideally:

int main()
{
  events::dispatcher dispatcher;

  dispatcher.register_event(EVENT_SAMPLE, [](int a) {
    std::cout << "received event: " << a << "\n";
  });

  dispatcher.start();
  dispatcher.post(EVENT_SAMPLE, 42);
  dispatcher.stop();

  return 0;
}

But I cannot find a way to allow different signatures for events while remaining type safe. I have posted a sample trying something like that on ideone, but it has some serious problems: dependent on RTTI, detects validity of posts at runtime instead of at compile time, and it does not properly detect lambdas, since there is no implicit conversion when looking for template matches.

When further overthinking this, it makes sense though, since people could register events with different function signatures, and I wouldn't know how to handle that, but allowing multiple functions to be registered to one event is important to me. So I guess I'll drop the idea and roll with the event structure.

\$\endgroup\$
9
  • \$\begingroup\$ Welcome to CR. Interesting question. \$\endgroup\$ – RubberDuck Mar 19 '16 at 11:55
  • 1
    \$\begingroup\$ Note: the arguments to start() were originally there to allow different thread implementations (std::thread was a template argument then) but I dropped that for simplicity and the arguments were forgotten. I'll leave them in here to not mutilate the original post :-) \$\endgroup\$ – nijansen Mar 19 '16 at 17:37
  • \$\begingroup\$ it's ok to modify the code before you've received an answer. We just don't want good answers invalidated. Sometimes it's useful to know why communities have arcane rules. =) \$\endgroup\$ – RubberDuck Mar 19 '16 at 17:44
  • \$\begingroup\$ @RubberDuck thanks for the hint; i've corrected the code accordingly \$\endgroup\$ – nijansen Mar 19 '16 at 17:48
  • \$\begingroup\$ Why is the thread behind a unique_ptr? \$\endgroup\$ – Veedrac Mar 23 '16 at 9:24
8
+100
\$\begingroup\$

Start/Stop

You should consider calling these from your constructor/destructor so that the thread is ready when constructed and automatically stopped when destroyed. It would make is easier to use. Ask yourself this, would you ever create a dispatcher without the intent of calling start()/stop() on it? If not, they should be called from your ctor/dtor.

ready_ is redundant

In start you can simply wait for running_ to be true and remove ready_. That said I don't really think that there is a need to wait for the thread to have started... You could just return without waiting as the events will be executed as soon as the thread is up. Yes, sure you may have a very slight delay before your events start executing, but you may have that even if the thread is already up because the OS doesn't guarantee that your thread will be scheduled immediately.

Skip unique_ptr on thread

As has already been mentioned as you have disallowed copy/move I would recommend that you just have the thread as a member in the class without the unique_ptr. The thread is default constructed as an empty thread object without an associated thread.

Better naming

As mutex_ is protecting events_ it should probably be called events_lock_ or something similar. Same goes for condition_.

Simplify loop logic

The loop in your worker seems a bit convoluted I would probably structure it something like this (caveat-emptor: I might have changed behaviour on shutdown):

while (running_) {
    auto next_event = std::chrono::time_point::max(); 
    if(!events.empty()){
        next_event = events.begin()->first;
    }

    // Will wait until:
    // 1) the time of the next event OR 
    // 2) the thread was notified AND
    //    a) The thread is set to terminate OR
    //    b) A event with a closer time point was detected
    //       at which point we re-calculate the sleep time.
    condition_.wait_until(lock, next_event, [&](){
            return !running ||
                   (!events.empty() && events.begin()->first < next_event);
        });

    // Woken up either through notify of new events, or because it is time
    // for the next event.
    if(next_event <= std::chrono::high_resolution_clock::now()){
         // Note that here events is never empty because otherwise
         // next_event would be end of time and we would not enter here.
         // Only place where events are removed is below. 
         auto it = events.begin();
         auto func = it->second;
         events.erase(it);
         func(lock); // Temporarily releases lock
    }
}

Use standard functions/methods

I would change:

auto range = callbacks_.equal_range(event_id);
std::vector<callback_t> functions;
for (auto it = range.first; it != range.second; ++it)
{
  functions.push_back(it->second);
}

to:

auto range = callbacks_.equal_range(event_id);
std::vector<callback_t> functions();
std::transform(range.first, range.second, std::back_inserter(functions), 
    [](auto& p){ return p->second; });

Use std::atomic

Signalling variables such as running_ and ready_ typically are implemented using atomic types. This means that you do not need to take a mutex lock to read/write to them.

Suffix _t is reserved for use by POSIX.

If you're not on a POSIX system then you're safe. For more info, see here.

\$\endgroup\$
5
  • \$\begingroup\$ "Start/Stop" "You should consider just making these into your constructor/destructor." → But stop-start pairs are intended to be called to "pause" the thread, so that doesn't really work. \$\endgroup\$ – Veedrac Mar 25 '16 at 3:47
  • \$\begingroup\$ Very nice findings! Is type_t really also reserved when not in the global namespace? I doubt that Posix will ever place something in an events::dispatcher namespace. \$\endgroup\$ – nijansen Mar 25 '16 at 10:19
  • \$\begingroup\$ @nijansen to the best of my knowledge, POSIX is not aware of namespaces and the identifiers are just reserved anywhere (they may use them in macros for example). \$\endgroup\$ – Emily L. Mar 25 '16 at 12:11
  • \$\begingroup\$ While implementing the points you mention, some questions came up: while your worker loop is a lot simpler than mine, doesn't it lead to a busy loop when only events are in the queue with timestamps in the future? Since the predicate is checked first from my understanding, there will be no waiting if the queue is not empty, correct? Also I can't just use std::vector<callback_t>(range.first, range.second) since they are of type std::pair<int, callback_t>, so I do still need my loop? \$\endgroup\$ – nijansen Mar 25 '16 at 21:50
  • \$\begingroup\$ @nijansen I fixed the answer, hopefully fixed all the issues :) \$\endgroup\$ – Emily L. Mar 26 '16 at 10:24
2
\$\begingroup\$

Some small thoughts here.

thread_, as mentioned in the comments, should probably not be behind a pointer. This removes the need for the local make_unique entirely.

lock_ is a poor name for a lock; what is it locking? What logical invariant does it hold?

start and stop aren't seem thread-safe relative to each other. What is callable concurrently should be well documented, which is lacking here.

running_ seems like it could just be an atomic_flag. Those are normally simpler than locking, so I'd prefer it.

condition_.wait(lock, [this]() { return !ready_; }); seems like it should be condition_.wait(lock, [this]() { return ready_; }); as condition_.wait(lock, pred) acts like

while (!pred()) wait(lck);

and you seem to want

while (!ready_) wait(lck);

instead of

while (!!ready_) wait(lck);

I don't like that you're reusing condition_ for two things and haven't explained why they can't interfere. Heck, when dealing with locks very little matters except for the explanation of why you think things should work.

I don't get

if (wait_delayed)
{
  condition_.wait_until(lock, wait_until);
  wait_delayed = false;
}
else
{
  condition_.wait(lock);
}

as wait_until is weaker than wait, so it's hard to say you've delayed anything. Delaying would be more like

sleep_until(wait_until);
condition_.wait(lock);

This is also odd:

if (it->first > std::chrono::high_resolution_clock::now())
{
  wait_delayed = true;
  wait_until = it->first;
  break;
}

given you just use the last wait_until, rather than, say, the maximum.

\$\endgroup\$
2
  • \$\begingroup\$ Concerning the wait_until: The logic here is that we find the first event that is supposed to happen in the future and remember the time it should be executed. Since maps are sorted by key, it's sufficient to find the first one. Then, if we found an event that is supposed to be executed in the future, wait for further events, but at most until the event we found is due. If no event is queued for future execution we can wait until a new event is posted (no need for wait_until here). If I have a logic mistake here please let me know! Besides that you bring up some good points. \$\endgroup\$ – nijansen Mar 23 '16 at 22:48
  • \$\begingroup\$ @nijansen My problem isn't the logic, it's the fact the variable names suggest you're doing something else entirely. \$\endgroup\$ – Veedrac Mar 24 '16 at 1:08

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.