concurrent

Documentation

Index

• type OneShotPubSub
  • func NewOneShotPubSub() *OneShotPubSub
  • func (p *OneShotPubSub) Notify(uid uint64, v interface{}) bool
  • func (p *OneShotPubSub) Wait(uid uint64) <-chan interface{}
• type PublishSubscribe
  • func NewPublishSubscribe() *PublishSubscribe
  • func (p *PublishSubscribe) Publish(tag uint64, value interface{})
  • func (p *PublishSubscribe) Stop()
  • func (p *PublishSubscribe) Subscribe(tag uint64, ch chan<- interface{})
  • func (p *PublishSubscribe) Unsubscribe(tag uint64, ch chan<- interface{})
• type SequenceBroadcast
  • func NewSequenceBroadcast(nextIndex uint64) *SequenceBroadcast
  • func (sb *SequenceBroadcast) Receive(start, limit uint64) <-chan interface{}
  • func (sb *SequenceBroadcast) Send(m interface{})

Types

type OneShotPubSub

type OneShotPubSub struct {
	sync.Mutex
	// contains filtered or unexported fields
}

OneShotPubSub is used to return replies to the clients whose requests are handled through the replication log. It's a sort of one-shot asynchronous publish-subscribe system: Only one subscriber can wait on a particular uid, and only one message can be sent on any given uid. Once the message is sent, the subscriber is notified asynchronously and the channel associated with that uid closes. Any further Notify calls on the same uid will return false and have no effect. A uid should never be reused.

func NewOneShotPubSub

func NewOneShotPubSub() *OneShotPubSub

NewOneShotPubSub initializes a OneShotPubSub.

func (*OneShotPubSub) Notify

func (p *OneShotPubSub) Notify(uid uint64, v interface{}) bool

Notify tries to send v to the waiter uid and returns whether it did.

func (*OneShotPubSub) Wait

func (p *OneShotPubSub) Wait(uid uint64) <-chan interface{}

Wait waits for a value to be sent by Notify.

type PublishSubscribe

type PublishSubscribe struct {
	// contains filtered or unexported fields
}

PublishSubscribe implements the publish-subscribe pattern where messages have uint64 "tags" (which could be thought of as "channels"). At any time, messages can be published with Publish, and subscriptions to messages with a particular tag can be created and stopped with Subscribe and Unsubscribe. PublishSubscribe is somewhat asynchronous: Publish calls do not wait for the value to be sent to subscribers. However, if a send to a subscribed channel blocks, the entire state machine blocks; thus, subscribers should either buffer or guarantee quick processing. Because of this, PublishSubscribe can guarantee that subscriptions will receive all values published after the Subscribe call and before the Unsubscribe call (with "after" and "before" defined according to Go's memory model [https://golang.org/ref/mem]).

func NewPublishSubscribe

func NewPublishSubscribe() *PublishSubscribe

func (*PublishSubscribe) Publish

func (p *PublishSubscribe) Publish(tag uint64, value interface{})

func (*PublishSubscribe) Stop

func (p *PublishSubscribe) Stop()

No calls may be made on the broadcaster after Stop() is initiated.

func (*PublishSubscribe) Subscribe

func (p *PublishSubscribe) Subscribe(tag uint64, ch chan<- interface{})

Starts listening with the channel, which is guaranteed to receive all values published with that tag after the Subscribe() call.

func (*PublishSubscribe) Unsubscribe

func (p *PublishSubscribe) Unsubscribe(tag uint64, ch chan<- interface{})

Stops listening with the channel. At some point after the Unsubscribe() call begins, the channel will stop receiving values and be closed. Subscribers are guaranteed to receive all values published with that tag before the Unsubscribe() call.

type SequenceBroadcast

type SequenceBroadcast struct {
	sync.Mutex
	// contains filtered or unexported fields
}

SequenceBroadcast acts as an asynchronous message broker for verifiers waiting for updates. It manages an ordered stream of messages, where each message is assigned a uint64 index, but does not remember old messages. Subscribers request to listen for messages in a particular range of indices, but if those indices have already gone by, the subscription fails, and clients must find the messages elsewhere. Also, subscription channels can be closed at any time if the receivers block, so clients must be capable of retrying. This enables Send to be non-blocking: it never waits for receivers and is guaranteed not to block indefinitely. The keyserver main calls Send each time a new log entry is available for verifiers. A grpc handler would call Read with the verifier's request data and use the returned channel to keep the verifier up to date.

func NewSequenceBroadcast

func NewSequenceBroadcast(nextIndex uint64) *SequenceBroadcast

NewSequenceBroadcast initializes a sequenceBroadcast such that the next call to send will be interpreted as the value of sequenceLog[nextIndex].

func (*SequenceBroadcast) Receive

func (sb *SequenceBroadcast) Receive(start, limit uint64) <-chan interface{}

Receive requests access to broadcasts for indexes [start, limit). If the broadcast for start has already been sent, Receive returns nil. When there are no more broadcasts left in the subscription (after limit-1 in the common case), the channel returned by Receive is closed. The caller is expected to consume the values from the returned channel quickly, if it blocks, the channel may be closed before the limit is reached.

func (*SequenceBroadcast) Send

func (sb *SequenceBroadcast) Send(m interface{})

Send broadcasts m to all subscribers registered with sb. Send is a network boundary: all inputs that are required to reproduce m MUST be synced to stable storage before Send(m) is called.