EventBus.h

Go to the documentation of this file.

#ifndef LINTEL_EVENTBUS_H
#define LINTEL_EVENTBUS_H
 
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <optional>
#include <string>
 
#include "lintel/core/shm/ShmSegmentAccessor.h"
#include "lintel/features/base/events/Event.h"
 
namespace lintel {
struct EventBusLimits
{
  static constexpr std::size_t MAX_TOPICS = 16;
  static constexpr std::size_t MAX_SUBSCRIBERS = 32;
  static constexpr std::size_t MAX_CAPACITY = 65534;
};
 
struct EventBusConfig
{
  std::size_t busCapacity = 128;
  std::size_t subscriberCapacity = 128;
  std::size_t maxTopics = EventBusLimits::MAX_TOPICS;
  std::size_t maxSubscribers = EventBusLimits::MAX_SUBSCRIBERS;
 
  [[nodiscard]] bool isValid() const
  {
    return busCapacity > 0 && busCapacity <= EventBusLimits::MAX_CAPACITY && subscriberCapacity > 0
           && subscriberCapacity <= EventBusLimits::MAX_CAPACITY && maxTopics > 0
           && maxTopics <= EventBusLimits::MAX_TOPICS && maxSubscribers > 0
           && maxSubscribers <= EventBusLimits::MAX_SUBSCRIBERS;
  }
};
 
class EventBus
{
public:
  EventBus(const std::string &name, const EventBusConfig &config, ShmSegmentAccessor segment);
 
  ~EventBus();
 
  EventBus(const EventBus &) = delete;
  EventBus &operator=(const EventBus &) = delete;
 
  [[nodiscard]] const char *getName() const;
 
  [[nodiscard]] EventBusConfig getConfig() const;
 
  [[nodiscard]] bool isLockFree() const;
 
  bool registerTopic(const std::string &topic);
 
  [[nodiscard]] bool isTopicRegistered(const std::string &topic) const;
 
  [[nodiscard]] std::size_t topicCount() const;
 
  bool subscribe(const std::string &topic, const std::string &subscriber);
 
  bool unsubscribe(const std::string &topic, const std::string &subscriber);
 
  [[nodiscard]] bool isSubscribed(const std::string &topic, const std::string &subscriber) const;
 
  [[nodiscard]] std::size_t numberOfSubscribersOf(const std::string &topic) const;
 
  [[nodiscard]] std::size_t subscriberCount() const;
 
  bool publish(const Event &event);
 
  bool publishToSubscribers(const Event &event);
 
  [[nodiscard]] std::optional<Event> receive();
 
  [[nodiscard]] std::optional<Event> receiveOf(const std::string &subscriber);
 
  [[nodiscard]] bool isEmpty() const;
 
  [[nodiscard]] bool isEmptyOf(const std::string &subscriber) const;
 
  [[nodiscard]] std::int64_t sequenceOf() const;
 
  [[nodiscard]] std::uint64_t droppedEventsOf() const;
 
  void reset();
 
  static std::size_t requiredSize(const EventBusConfig &config);
 
private:
  class ShmRing
  {
  public:
    static constexpr std::size_t CACHE_LINE = 64;
 
    struct alignas(CACHE_LINE) Slot
    {
      std::atomic<std::uint64_t> sequence{ 0 };
      Event value;
    };
 
    ShmRing() = default;
 
    ShmRing(std::size_t capacity, ShmSegmentAccessor segment);
 
    ~ShmRing();
 
    ShmRing(const ShmRing &) = delete;
    ShmRing &operator=(const ShmRing &) = delete;
 
    bool push(const Event &event);
 
    bool pop(Event &out);
 
    [[nodiscard]] bool empty() const;
 
    [[nodiscard]] static bool isLockFree();
 
  private:
    std::atomic<std::uint64_t> m_head{ 0 };
    std::byte m_headPadding[CACHE_LINE - sizeof(std::uint64_t)]{};
    std::atomic<std::uint64_t> m_tail{ 0 };
    std::byte m_tailPadding[CACHE_LINE - sizeof(std::uint64_t)]{};
    std::size_t m_capacity = 0;
    ShmSegmentAccessor m_segment;
    std::byte *m_allocation = nullptr;
    std::int64_t m_bufferOffset = 0;
 
    [[nodiscard]] Slot *buffer();
    [[nodiscard]] const Slot *buffer() const;
  };
 
  struct TopicEntry
  {
    char name[Event::NAME_SIZE]{};
    std::atomic<std::size_t> subscriberCount{ 0 };
    std::size_t subscribers[EventBusLimits::MAX_SUBSCRIBERS]{};
  };
 
  struct SubscriberEntry
  {
    char name[Event::NAME_SIZE]{};
    std::atomic<bool> active{ false };
    std::atomic<std::size_t> topicRefCount{ 0 };
  };
 
  class RegistryLock
  {
  public:
    RegistryLock();
 
    void acquire() const;
 
    void release() const;
 
  private:
    static constexpr std::uint64_t STALE_THRESHOLD_MS = 500;
    static constexpr std::uint64_t TIMESTAMP_MASK = 0xFFFFFFFFu;
 
    mutable std::atomic<std::uint64_t> m_state;
 
    static std::uint64_t ownerId();
    static std::uint64_t nowMs();
    static bool isStale(std::uint64_t state);
    static std::uint64_t encode();
  };
 
  char m_name[Event::NAME_SIZE]{};
  std::size_t m_busCapacity = 0;
  std::size_t m_subscriberCapacity = 0;
  std::size_t m_maxTopics = 0;
  std::size_t m_maxSubscribers = 0;
  std::atomic<std::int64_t> m_sequence{ 0 };
  std::atomic<std::uint64_t> m_droppedEvents{ 0 };
  std::atomic<std::size_t> m_topicCount{ 0 };
  std::atomic<std::size_t> m_subscriberCount{ 0 };
  RegistryLock m_registryLock;
  ShmRing m_bus;
  TopicEntry m_topics[EventBusLimits::MAX_TOPICS]{};
  SubscriberEntry m_subscribers[EventBusLimits::MAX_SUBSCRIBERS]{};
  alignas(ShmRing) std::byte m_subscriberQueues[EventBusLimits::MAX_SUBSCRIBERS][sizeof(ShmRing)]{};
 
  [[nodiscard]] int findTopicIndex(const char *topic) const;
  [[nodiscard]] int findSubscriberIndex(const char *subscriber) const;
  [[nodiscard]] ShmRing &subscriberQueueOf(std::size_t index);
  [[nodiscard]] const ShmRing &subscriberQueueOf(std::size_t index) const;
 
  static void copyInto(char (&destination)[Event::NAME_SIZE], const std::string &value);
  [[nodiscard]] static bool nameEquals(const char *lhs, const char *rhs);
};
 
class IEventBus
{
public:
  virtual ~IEventBus() = default;
 
  [[nodiscard]] virtual const char *getName() const = 0;
 
  [[nodiscard]] virtual EventBusConfig getConfig() const = 0;
 
  [[nodiscard]] virtual bool isLockFree() const = 0;
 
  virtual bool registerTopic(const std::string &topic) = 0;
 
  [[nodiscard]] virtual bool isTopicRegistered(const std::string &topic) const = 0;
 
  [[nodiscard]] virtual std::size_t topicCount() const = 0;
 
  virtual bool subscribe(const std::string &topic, const std::string &subscriber) = 0;
 
  virtual bool unsubscribe(const std::string &topic, const std::string &subscriber) = 0;
 
  [[nodiscard]] virtual bool isSubscribed(const std::string &topic, const std::string &subscriber) const = 0;
 
  [[nodiscard]] virtual std::size_t numberOfSubscribersOf(const std::string &topic) const = 0;
 
  [[nodiscard]] virtual std::size_t subscriberCount() const = 0;
 
  virtual bool publish(const Event &event) = 0;
 
  virtual bool publishToSubscribers(const Event &event) = 0;
 
  [[nodiscard]] virtual std::optional<Event> receive() = 0;
 
  [[nodiscard]] virtual std::optional<Event> receiveOf(const std::string &subscriber) = 0;
 
  [[nodiscard]] virtual bool isEmpty() const = 0;
 
  [[nodiscard]] virtual bool isEmptyOf(const std::string &subscriber) const = 0;
 
  [[nodiscard]] virtual std::int64_t sequenceOf() const = 0;
 
  [[nodiscard]] virtual std::uint64_t droppedEventsOf() const = 0;
 
  virtual void reset() = 0;
};
 
class EventBusView : public IEventBus
{
public:
  explicit EventBusView(EventBus &bus) : m_bus(bus) {}
 
  [[nodiscard]] const char *getName() const override { return m_bus.getName(); }
 
  [[nodiscard]] EventBusConfig getConfig() const override { return m_bus.getConfig(); }
 
  [[nodiscard]] bool isLockFree() const override { return m_bus.isLockFree(); }
 
  bool registerTopic(const std::string &topic) override { return m_bus.registerTopic(topic); }
 
  [[nodiscard]] bool isTopicRegistered(const std::string &topic) const override
  {
    return m_bus.isTopicRegistered(topic);
  }
 
  [[nodiscard]] std::size_t topicCount() const override { return m_bus.topicCount(); }
 
  bool subscribe(const std::string &topic, const std::string &subscriber) override
  {
    return m_bus.subscribe(topic, subscriber);
  }
 
  bool unsubscribe(const std::string &topic, const std::string &subscriber) override
  {
    return m_bus.unsubscribe(topic, subscriber);
  }
 
  [[nodiscard]] bool isSubscribed(const std::string &topic, const std::string &subscriber) const override
  {
    return m_bus.isSubscribed(topic, subscriber);
  }
 
  [[nodiscard]] std::size_t numberOfSubscribersOf(const std::string &topic) const override
  {
    return m_bus.numberOfSubscribersOf(topic);
  }
 
  [[nodiscard]] std::size_t subscriberCount() const override { return m_bus.subscriberCount(); }
 
  bool publish(const Event &event) override { return m_bus.publish(event); }
 
  bool publishToSubscribers(const Event &event) override { return m_bus.publishToSubscribers(event); }
 
  [[nodiscard]] std::optional<Event> receive() override { return m_bus.receive(); }
 
  [[nodiscard]] std::optional<Event> receiveOf(const std::string &subscriber) override
  {
    return m_bus.receiveOf(subscriber);
  }
 
  [[nodiscard]] bool isEmpty() const override { return m_bus.isEmpty(); }
 
  [[nodiscard]] bool isEmptyOf(const std::string &subscriber) const override { return m_bus.isEmptyOf(subscriber); }
 
  [[nodiscard]] std::int64_t sequenceOf() const override { return m_bus.sequenceOf(); }
 
  [[nodiscard]] std::uint64_t droppedEventsOf() const override { return m_bus.droppedEventsOf(); }
 
  void reset() override { m_bus.reset(); }
 
private:
  EventBus &m_bus;
};
 
}// namespace lintel
#endif// LINTEL_EVENTBUS_H