A broadcaster is able to send messages to multiple channels, with each channel having its own lock Sending messages requires locking the entire broadcaster. More...

#include <broadcaster.hpp>

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

[legend]

Public Member Functions
Result< void, AllocationError >	init (Allocator &alloc, size_t initCapacity=64)
	Initializes a broadcaster.

Result< void, ChannelSendBlockErrors >	send_block (Allocator &alloc, const T &msg)
	Blocks while sending data.

Result< BroadcastSubscription< T >, BroadcastSubscribeError >	subscribe (Allocator &alloc)
	Subscribes to a broadcaster Will create a new inbox for the subscription.

void	close ()
	Closes the broadcaster which prevents messages from being sent (which aren't already being sent) and it also prevents new subscriptions from being created It will also close all existing subscriptions, which can result in partial writes (e.g.

void	deinit (Allocator &alloc)
	Will close the broadcaster and will also clean up the broadcaster's references to each subscription (a single Arc.release()) Will also close any subscriptions that didn't get cleaned up.

Public Attributes
GenList< BroadcastSubscription< T > >	channels

std::mutex	mux

bool	closed = false

Detailed Description

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

A broadcaster is able to send messages to multiple channels, with each channel having its own lock Sending messages requires locking the entire broadcaster.

However, reading messages allows for many separate locks. This makes it ideal for scenarios with one publisher and many subscribers.

Note that there is less insight into the state of each channel as they are locked separately. Because of this, only blocking sends and blocking receives are allowed, as acquiring the lock and waiting is the only way to really prevent partial sends.

So, while we should have better read throughput than the Publisher/Subscription flow, we have less flexibility in the API and only have blocking operations. This also means we cannot use Broadcaster with select()

Locks:

1 lock for entire structure
1 lock per channel

Atomic Reference Counting:

1 per channel

Condition Variables:

2 condition variables per channel

Non-Blocking Availability:

None

Known Race Conditions (aka. designed in):

If a send is started, the broadcaster may get closed or denitialized
- Planned Behavior: Channels will get closed and may prevent the send from completing successfully. The send may fail and enter a partial send state

Template Parameters

T	Message Type

Definition at line 103 of file broadcaster.hpp.

Member Function Documentation

◆ close()

template<typename T>

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

inline

Closes the broadcaster which prevents messages from being sent (which aren't already being sent) and it also prevents new subscriptions from being created It will also close all existing subscriptions, which can result in partial writes (e.g.

a message is blocked after sending to half of the subscriptions)

Definition at line 201 of file broadcaster.hpp.

                 {
      auto l = std::unique_lock{mux};
      closed = true;
      BroadcastSubscription<T> cur;
      auto iter = channels.iter();
      while (iter.next().copy_if_present(cur)) {
        if (cur.inbox.valid()) {
          cur.inbox->close();
        }
      }
    }

◆ deinit()

template<typename T>

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

inline

Will close the broadcaster and will also clean up the broadcaster's references to each subscription (a single Arc.release()) Will also close any subscriptions that didn't get cleaned up.

Parameters

alloc Allocator to free memory with

Definition at line 219 of file broadcaster.hpp.

                                  {
      auto l = std::unique_lock{mux};
      closed = true;
      BroadcastSubscription<T> cur;
      auto iter = channels.iter();
      while (iter.next().copy_if_present(cur)) {
        cur.deinit(alloc);
      }
      channels.deinit(alloc);
    }

Here is the call graph for this function:

◆ init()

template<typename T>

Result< void, AllocationError > mtcore::thread::broadcaster< T >::init	(	Allocator &	alloc,
		size_t	initCapacity = 64 )

inline

Initializes a broadcaster.

Parameters

alloc	Allocator for allocating memory
initCapacity	Initial capacity

Returns

Definition at line 114 of file broadcaster.hpp.

                                                                                   {
      return channels.init(alloc, initCapacity);
    }

◆ send_block()

template<typename T>

Result< void, ChannelSendBlockErrors > mtcore::thread::broadcaster< T >::send_block	(	Allocator &	alloc,
		const T &	msg )

inline

Blocks while sending data.

Will grow inboxes as needed, and will free closed subscriptions as needed

Parameters

alloc	Allocator to use for growing and closing
msg	Message to send

Returns: Errors if they happened

Definition at line 124 of file broadcaster.hpp.

                                                                                    {
      // We need the lock since we'll be iterating and removing stuff
      auto l = std::unique_lock{mux};
 
      if (closed) {
        return error(ChannelSendBlockErrors::CHANNEL_CLOSED);
      }
 
      Handle handle;
      auto iter = channels.handles();
      while (iter.next().copy_if_present(handle)) {
        auto &e = channels[handle];
        if (e.inbox.valid()) {
          if (auto res = e.inbox->send_block(alloc, msg); res.is_error()) {
            // If we failed allocation, try a non-allocating send
            if (res.error().code == ChannelGrowSendBlockErrors::ALLOCATION_FAILED) {
              // We can still fail if our channel is closed. In this case, remove the channel
              if (auto res2 = e.inbox->send_block(msg); res2.is_error()) {
                e.inbox.release(alloc);
                channels.remove(handle);
              }
            }
            // otherwise, our channel is closed, just remove the channel
            else {
              e.inbox.release(alloc);
              channels.remove(handle);
            }
          }
        }
        else {
          // clean up and remove the channel
          e.deinit(alloc);
          channels.remove(handle);
        }
      }
      return success();
    }

Here is the call graph for this function:

◆ subscribe()

template<typename T>

Result< BroadcastSubscription< T >, BroadcastSubscribeError > mtcore::thread::broadcaster< T >::subscribe ( Allocator & alloc )

inline

Subscribes to a broadcaster Will create a new inbox for the subscription.

Parameters

alloc Allocator to use for creating the subscription

Returns: the subscription on success or an error on failure

Definition at line 168 of file broadcaster.hpp.

                                                                                          {
      BroadcastSubscription<T> sub;
      if (auto subRes = sub.init(alloc); subRes.is_error()) {
        return error(BroadcastSubscribeError::ALLOCATION_FAILED);
      }
 
      // We need the lock since we'll be removing stuff
      {
        auto l = std::unique_lock{mux};
        if (closed) {
          sub.deinit(alloc);
          return error(BroadcastSubscribeError::CHANNEL_CLOSED);
        }
 
        if (auto addRes = channels.add(alloc, sub); addRes.is_error()) {
          sub.deinit(alloc);
          return error(BroadcastSubscribeError::ALLOCATION_FAILED);
        }
      }
      auto acquireRes = sub.inbox.acquire(alloc);
      if (acquireRes.is_error()) {
        sub.deinit(alloc);
        return error(BroadcastSubscribeError::CANNOT_ACQUIRE);
      }
      return BroadcastSubscription<T>{acquireRes.value()};
    }