event: proposal for improving the API

[fjl]

Package event provides a mechanism for pushing change notifications to an unknown number of
interested parties. This issue is a collection of ideas for improving package event. We've lived with the current API for almost a year now and while it has held up well, there are a few things that can be improved.

Posting events should not be synchronous by default.

We've learned this the hard way, with several hard-to-see deadlocks being caused by a blocking call to Post while a mutex is held.

Posting should return the number of recipients.

This allows for response gathering, e.g.

resp := make(chan Reply)
n := mux.Post(SomeEvent{resp})
for i := 0; i < n; i++ {
    <-resp
}

Implementation improvements

The current implementation of TypeMux assumes that event subscriptions are mostly static (set up once, then used). Adding and removing subscriptions has non-trivial overhead, halting all event processing while the internal state is updated.

Delivery of each event of a certain type is sequential, waiting for each subscriber to receive the event before attempting to send to the next. This can delay reception if subscribers are blocked.

Provide helpers for shutdown coordination

Package event might also be the right place to contain a generic solution for shutdown signaling. Many components in go-ethereum have boilerplate that coordinates shutdown of one or more 'worker loop'
goroutines when the Stop method is called. Since quite a lot of boilerplate is involved to get this right, components usually cheat and implement a simple solution that doesn't wait for workers to terminate and
panics with "close of closed channel" if Stop is called multiple times.

I propose that we add a generic solution for shutdown coordination to package event.

[obscuren]

This does introduce tight coupling again. The purpose of the venter is to reduce the tight coupling to some amount while depending on Go's type system and package dependencies.

I do not like this suggestion.

[fjl]

@obscuren I have updated the proposal and removed the offending section.
The new part is the one about gathering responses, which we'll need for #2254.

[fjl]

Having thought about events in the context of #2808 and #2798, I'd like us to reconsider the removal of TypeMux as originally suggested above. In the linked PRs, we are introducing a model for event subscriptions that looks like this:

func (x *X) SubscribeSomething(chan Y, ...) (Subscription, error) { ... }

Where Subscription is an object that implements the interface below.

type Subscription interface{
    // Unsubscribe cancels the sending of events to the data channel
    // and closes the error channel.
    Unsubscribe()

    // Err returns the subscription error channel. The error channel receives
    // a value if there is an issue with the subscription (e.g. the network connection
    // delivering the events has been closed). Only one value will ever be sent.
    // The error channel is closed by Unsubscribe.
    Err() <-chan error
}

To help with this model, package event would provide a reduced version of TypeMux that handles a single type of event, e.g.

type Feed struct { ... } 

// Subscribe adds a channel to the feed. Future sends will be
// delivered on the channel until the subscription is canceled.
// All channels added must have the same element type.
func (Feed) Subscribe(channel interface{}) Subscription { ... }

// Send delivers to all subscribed channels simultaneously.
func (Feed) Send(v interface{}) { ... } 

Components which send events would use this type to manage their subscriptions, e.g.

package core

type HeaderChain struct {
    ...
    newHeadFeed event.Feed
}

func (hc *HeaderChain) InsertHeaderChain(chain []*types.Header) {
    ...
    hc.newHeadFeed.Send(lastHeader)
}

func (hc *HeaderChain) SubscribeNewHead(c chan<- *types.Header) (event.Subscription, error) {
    return hc.newHeadEvent.Subscribe(c), nil
}

If you wonder about the error, consider that the same method can also be implemented by
the typed RPC API client:

package ethclient

func (c *Client) SubscribeNewHead(c chan<- *types.Header) (event.Subscription, error) {
    return c.rpcClient.EthSubscribe(c, "newHead")
}

@obscuren You previously said that you don't like this approach because it couples the subscriber
to a particular source of events. Please consider that the SubscribeSomething method can be put
into an interface. This is a huge advantage.

[fjl]

Removed 1.5.0 milestone because the scope of this issue is too broad. I would like to start adding code to support the model outlined above, but we won't switch everything at once.

[fjl]

The other ideas listed in the issue description still sound nice, but seem less important now.

[fjl]

I have done an initial implementation of Feed in https://github.com/fjl/go-ethereum/blob/event-feed/event/feed.go