Represents a pub/sub publisher which can publish messages of type T Supports blocking and non-blocking publishing. More...

#include <pub_sub.hpp>

Collaboration diagram for mtcore::thread::Publisher< T >:

[legend]

Public Types
using	Message = T

Public Member Functions
Result< void, AllocationError >	init (Allocator &alloc, size_t queueSize=10)
	Initializes a publisher.

void	close ()
	Closes a pub/sub channel All subscribers will sbe notified of closure.

void	deinit (Allocator &alloc)
	Closes a pub/sub channel and dereferences broker Broker will be cleaned up once all subscribers are cleaned up Note: all threads that are publishing should be joined prior to calling this method.

Result< Subscription< T >, AllocationError >	subscribe ()
	Creates a new Subscription to pub/sub channel.

Result< void, ChannelSendBlockErrors >	send_block (const T &message)
	Sends a message to all subscribers Will block if needed to ensure all subscribers get Queue Guarantees a message is delivered to all or blocks indefinitely.

Result< void, PubSubSendBlockGrowErrors >	send_block (Allocator &alloc, const T &message)
	Sends a message to all subscribers Will try to increase subscriber Queue sizes if there is not enough room If a Queue cannot be increased, then the message will be dropped for that subscriber.

template<typename Clock, typename Duration>
Result< void, ChannelSendBeforeErrors >	send_before (const T &message, const std::chrono::time_point< Clock, Duration > &timePoint)
	Sends a message to all subscribers Will block if needed to ensure all subscribers get Queue Guarantees a message is delivered to all or blocks It will timeout it the current time exceeds the provided time point.

template<typename Clock, typename Duration>
Result< void, PubSubSendBeforeGrowsError >	send_before (Allocator &alloc, const T &message, const std::chrono::time_point< Clock, Duration > &timePoint)
	Sends a message to all subscribers Will try to increase subscriber Queue sizes if there is not enough room If a Queue cannot be increased, then the message will be dropped for that subscriber Will timeout if cannot send in time.

Result< void, PubSubTrySendGrowsError >	try_send (const T &message)
	Sends a message to all subscribers Will not block.

Result< void, PubSubTrySendGrowsError >	try_send (Allocator &alloc, const T &message)
	Sends a message to all subscribers Will not block.

Public Attributes
Arc< impl::Broker< T > >	broker

Detailed Description

template<typename T>
struct mtcore::thread::Publisher< T >

Represents a pub/sub publisher which can publish messages of type T Supports blocking and non-blocking publishing.

Locks:

1 for entire data structure and all subscriptions (single lock)

Condition Variables:

2 condition variables (one to send, one to receive)

Non-Blocking Availability:

With Failure

Template Parameters

T	Message Type

Definition at line 253 of file pub_sub.hpp.

Member Typedef Documentation

◆ Message

template<typename T>

using mtcore::thread::Publisher< T >::Message = T

Definition at line 254 of file pub_sub.hpp.

Member Function Documentation

◆ close()

template<typename T>

void mtcore::thread::Publisher< T >::close ( )

inline

Closes a pub/sub channel All subscribers will sbe notified of closure.

Definition at line 280 of file pub_sub.hpp.

                 {
      auto lock = std::unique_lock(broker->mux);
 
      // close out all our inboxes
      broker->closed = true;
      broker->senderSignal.notify_all();
      broker->readSignal.notify_all();
    }

Here is the caller graph for this function:

◆ deinit()

template<typename T>

void mtcore::thread::Publisher< T >::deinit ( Allocator & alloc )

inline

Closes a pub/sub channel and dereferences broker Broker will be cleaned up once all subscribers are cleaned up Note: all threads that are publishing should be joined prior to calling this method.

Definition at line 295 of file pub_sub.hpp.

                                  {
      if (!broker.valid()) {
        return;
      }
      close();
      broker.deinit(alloc);
    }

Here is the call graph for this function:

◆ init()

template<typename T>

Result< void, AllocationError > mtcore::thread::Publisher< T >::init	(	Allocator &	alloc,
		size_t	queueSize = 10 )

inline

Initializes a publisher.

Parameters

alloc	Allocates a publisher
queueSize	Initial Queue size

Returns: success or failure

Definition at line 264 of file pub_sub.hpp.

                                                                                {
      ensure(!broker.valid(), "ALREADY INITIALIZED");
      if (auto brokerRes = broker.init(alloc, &alloc, queueSize); brokerRes.is_error()) {
        return brokerRes.error();
      }
      if (auto res = broker->inboxes.init(alloc); res.is_error()) {
        broker.deinit(alloc);
        return res.error();
      }
      return success();
    }

Here is the call graph for this function:

◆ send_before() [1/2]

template<typename T>

template<typename Clock, typename Duration>

Result< void, PubSubSendBeforeGrowsError > mtcore::thread::Publisher< T >::send_before	(	Allocator &	alloc,
		const T &	message,
		const std::chrono::time_point< Clock, Duration > &	timePoint )

inline

Sends a message to all subscribers Will try to increase subscriber Queue sizes if there is not enough room If a Queue cannot be increased, then the message will be dropped for that subscriber Will timeout if cannot send in time.

Template Parameters

Clock	Clock to get time from
Duration	Duration of time point

Parameters

alloc	Allocator for increasing Queue size
message	Message to send
timePoint	When to time out

Returns: success (sent to all) or failure (sent to none or some)

Definition at line 484 of file pub_sub.hpp.

                                                                                                                {
      ensure(broker.valid(), "NOT INITIALIZED");
      auto lock = std::unique_lock(broker->mux, timePoint);
      if (!lock.owns_lock()) {
        return error(PubSubSendBeforeGrowsError::TIMEOUT);
      }
 
      if (broker->closed) {
        return error(PubSubSendBeforeGrowsError::CHANNEL_CLOSED);
      }
 
      auto iter = broker->inboxes.ptr_iter();
      Queue<T> *cur;
      bool allocated                               = false;
      bool sentSomething                           = false;
      Result<void, PubSubSendBeforeGrowsError> res = success();
      while (iter.next().copy_if_present(cur)) {
        allocated |= cur->is_full();
        if (!cur->push(alloc, message)) {
          res = error(PubSubSendBeforeGrowsError::ALLOCATION_FAILED);
        }
        else {
          sentSomething = true;
        }
      }
 
      if (res.is_error() && sentSomething) {
        res = error(PubSubSendBeforeGrowsError::PARTIAL_SEND);
      }
 
      // We've most likely at least partially unblocked someone, let people know
      if (sentSomething) {
        broker->readSignal.notify_all();
 
        if (allocated) {
          broker->senderSignal.notify_all();
        }
      }
      return res;
    }

Here is the call graph for this function:

◆ send_before() [2/2]

template<typename T>

template<typename Clock, typename Duration>

Result< void, ChannelSendBeforeErrors > mtcore::thread::Publisher< T >::send_before	(	const T &	message,
		const std::chrono::time_point< Clock, Duration > &	timePoint )

inline

Sends a message to all subscribers Will block if needed to ensure all subscribers get Queue Guarantees a message is delivered to all or blocks It will timeout it the current time exceeds the provided time point.

Template Parameters

Clock	Clock to get time from
Duration	Duration of time point

Parameters

message	Message to send
timePoint	When to time out

Returns: success or faulre

Definition at line 430 of file pub_sub.hpp.

                                                                                                             {
      ensure(broker.valid(), "NOT INITIALIZED");
      auto lock = std::unique_lock(broker->mux, timePoint);
      if (!lock.owns_lock()) {
        return error(ChannelSendBeforeErrors::TIMEOUT);
      }
 
      if (broker->closed) {
        return error(ChannelSendBeforeErrors::CHANNEL_CLOSED);
      }
 
      bool ready;
      do {
        ready = true;
        // pre-check that we won't error out when adding
        auto iter = broker->inboxes.ptr_iter();
        Queue<T> *cur;
        while (iter.next().copy_if_present(cur)) {
          if (cur->is_full()) {
            ready = false;
          }
        }
        if (!ready) {
          if (broker->senderSignal.wait_until(lock, timePoint) == std::cv_status::timeout) {
            return error(ChannelSendBeforeErrors::TIMEOUT);
          }
        }
      } while (!ready);
      {
        // Now that we're good, send
        auto iter = broker->inboxes.ptr_iter();
        Queue<T> *cur;
        while (iter.next().copy_if_present(cur)) {
          ensure(cur->push(message).is_success(), "PRE CHECK FAILED");
        }
        broker->readSignal.notify_all();
      }
      return success();
    }

Here is the call graph for this function:

◆ send_block() [1/2]

template<typename T>

Result< void, PubSubSendBlockGrowErrors > mtcore::thread::Publisher< T >::send_block	(	Allocator &	alloc,
		const T &	message )

inline

Sends a message to all subscribers Will try to increase subscriber Queue sizes if there is not enough room If a Queue cannot be increased, then the message will be dropped for that subscriber.

Parameters

alloc	Allocator for increasing Queue size
message	Message to send

Returns: success (sent to all) or failure (sent to none or some)

Definition at line 382 of file pub_sub.hpp.

                                                                                           {
      ensure(broker.valid(), "NOT INITIALIZED");
      auto lock = std::unique_lock(broker->mux);
 
      if (broker->closed) {
        return error(PubSubSendBlockGrowErrors::CHANNEL_CLOSED);
      }
 
      auto iter = broker->inboxes.ptr_iter();
      Queue<T> *cur;
      bool allocated                              = false;
      bool sentSomething                          = false;
      Result<void, PubSubSendBlockGrowErrors> res = success();
      while (iter.next().copy_if_present(cur)) {
        allocated |= cur->is_full();
        if (!cur->push(alloc, message)) {
          res = error(PubSubSendBlockGrowErrors::ALLOCATION_FAILED);
        }
        else {
          sentSomething = true;
        }
      }
 
      if (res.is_error() && sentSomething) {
        res = error(PubSubSendBlockGrowErrors::PARTIAL_SEND);
      }
 
      if (sentSomething) {
        broker->readSignal.notify_all();
        if (allocated) {
          broker->senderSignal.notify_all();
        }
      }
      return res;
    }

Here is the call graph for this function:

◆ send_block() [2/2]

template<typename T>

Result< void, ChannelSendBlockErrors > mtcore::thread::Publisher< T >::send_block ( const T & message )

inline

Sends a message to all subscribers Will block if needed to ensure all subscribers get Queue Guarantees a message is delivered to all or blocks indefinitely.

Parameters

message Message to send

Returns: success or failure

Definition at line 337 of file pub_sub.hpp.

                                                                      {
      ensure(broker.valid(), "NOT INITIALIZED");
      auto lock = std::unique_lock(broker->mux);
 
      if (broker->closed) {
        return error(ChannelSendBlockErrors::CHANNEL_CLOSED);
      }
 
      bool ready;
      do {
        ready = true;
        // pre-check that we won't error out when adding
        auto iter = broker->inboxes.ptr_iter();
        Queue<T> *cur;
        while (iter.next().copy_if_present(cur)) {
          if (cur->is_full()) {
            ready = false;
          }
        }
        if (!ready) {
          broker->senderSignal.wait(lock);
        }
      } while (!ready);
 
      {
        // Now that we're good, send
        auto iter = broker->inboxes.ptr_iter();
        Queue<T> *cur;
        while (iter.next().copy_if_present(cur)) {
          ensure(cur->push(message).is_success(), "PRE CHECK FAILED");
        }
        broker->readSignal.notify_all();
      }
      return success();
    }

Here is the call graph for this function:

◆ subscribe()

template<typename T>

Result< Subscription< T >, AllocationError > mtcore::thread::Publisher< T >::subscribe ( )

inline

Creates a new Subscription to pub/sub channel.

Returns: New subscription (on success) or error (on failure)

Definition at line 307 of file pub_sub.hpp.

                                                         {
      ensure(broker.valid(), "NOT INITIALIZED");
      auto lock = std::unique_lock(broker->mux);
      Queue<T> q;
      if (auto res = q.init(*broker->alloc, broker->queueSize); res.is_error()) {
        return res.error();
      }
 
      auto handleRes = broker->inboxes.add(q);
      if (handleRes.is_error()) {
        q.deinit(*broker->alloc);
        return error(AllocationError::ALLOCATION_FAILED);
      }
 
      auto subRef = broker.acquire(*broker->alloc);
      if (subRef.is_error()) {
        broker->inboxes.remove(handleRes.value());
        q.deinit(*broker->alloc);
        return error(AllocationError::ALLOCATION_FAILED);
      }
      return success(Subscription<T>{subRef.value(), *handleRes});
    }

Here is the call graph for this function:

◆ try_send() [1/2]

template<typename T>

Result< void, PubSubTrySendGrowsError > mtcore::thread::Publisher< T >::try_send	(	Allocator &	alloc,
		const T &	message )

inline

Sends a message to all subscribers Will not block.

If blocking is required, will simply fail If there is not enough room, will try to grow a subscriber's Queue If growing fails, will not send to that subscriber

Parameters

alloc	Allocator to use to grow (if needed)
message	Message to send

Returns: success or failure (no send or partial send)

Definition at line 574 of file pub_sub.hpp.

                                                                                       {
      ensure(broker.valid(), "NOT INITIALIZED");
      auto lock = std::unique_lock(broker->mux, std::try_to_lock_t{});
      if (!lock.owns_lock()) {
        return error(PubSubTrySendGrowsError::COULD_NOT_LOCK);
      }
 
      if (broker->closed) {
        return error(PubSubTrySendGrowsError::CHANNEL_CLOSED);
      }
 
      auto iter                                 = broker->inboxes.ptr_iter();
      bool allocated                            = false;
      bool sentSomething                        = false;
      Result<void, PubSubTrySendGrowsError> res = success();
      if (Queue<T> cur; iter.next().copy_if_present(cur)) {
        allocated |= cur.is_full();
        if (auto push = cur.push(alloc, message); push.is_error()) {
          res = error(PubSubTrySendGrowsError::ALLOCATION_FAILED);
        }
        else {
          sentSomething = true;
        }
      }
 
      if (res.is_error() && sentSomething) {
        res = error(PubSubTrySendGrowsError::PARTIAL_SEND);
      }
 
      if (sentSomething) {
        broker->readSignal.notify_all();
        if (allocated) {
          broker->senderSignal.notify_all();
        }
      }
      return res;
    }

Here is the call graph for this function:

◆ try_send() [2/2]

template<typename T>

Result< void, PubSubTrySendGrowsError > mtcore::thread::Publisher< T >::try_send ( const T & message )

inline

Sends a message to all subscribers Will not block.

If blocking is required, will simply fail Guarantees a message is delivered to all or not at all

Parameters

message Message to send

Returns: success or failure

Definition at line 533 of file pub_sub.hpp.

                                                                     {
      ensure(broker.valid(), "NOT INITIALIZED");
      auto lock = std::unique_lock(broker->mux, std::try_to_lock_t{});
      if (!lock.owns_lock()) {
        return error(PubSubTrySendGrowsError::COULD_NOT_LOCK);
      }
 
      if (broker->closed) {
        return error(PubSubTrySendGrowsError::CHANNEL_CLOSED);
      }
      {
        // pre-check that we won't block
        auto iter = broker->inboxes.ptr_iter();
        Queue<T> *cur;
        while (iter.next().copy_if_present(cur)) {
          if (cur->is_full()) {
            return error(PubSubTrySendGrowsError::NO_ROOM);
          }
        }
      }
      {
        // Now that we're good, send
        auto iter = broker->inboxes.ptr_iter();
        Queue<T> *cur;
        while (iter.next().copy_if_present(cur)) {
          ensure(cur->push(message).is_success(), "PRE CHECK FAILED");
        }
        broker->readSignal.notify_all();
      }
      return success();
    }

Here is the call graph for this function:

Member Data Documentation

◆ broker

template<typename T>

Arc<impl::Broker<T> > mtcore::thread::Publisher< T >::broker

Definition at line 256 of file pub_sub.hpp.

The documentation for this struct was generated from the following file:

pub_sub.hpp

Public Types

Public Member Functions

Public Attributes

Detailed Description

Member Typedef Documentation

◆ Message

Member Function Documentation

◆ close()

◆ deinit()

◆ init()

◆ send_before() [1/2]

◆ send_before() [2/2]

◆ send_block() [1/2]

◆ send_block() [2/2]

◆ subscribe()

◆ try_send() [1/2]

◆ try_send() [2/2]

Member Data Documentation

◆ broker