Merge pull request ethereum#3605 from fjl/event-feed

event: add new Subscription type and related utilities

core/tx_pool.go

@@ -89,7 +89,7 @@ type TxPool struct {
 	gasLimit     func() *big.Int // The current gas limit function callback
 	minGasPrice  *big.Int
 	eventMux     *event.TypeMux
-	events       event.Subscription
+	events       *event.TypeMuxSubscription
 	localTx      *txSet
 	signer       types.Signer
 	mu           sync.RWMutex

eth/filters/filter_system.go

@@ -74,7 +74,7 @@ type subscription struct {
 // subscription which match the subscription criteria.
 type EventSystem struct {
 	mux       *event.TypeMux
-	sub       event.Subscription
+	sub       *event.TypeMuxSubscription
 	backend   Backend
 	lightMode bool
 	lastHead  *types.Header
@@ -277,7 +277,7 @@ func (es *EventSystem) SubscribePendingTxEvents(hashes chan common.Hash) *Subscr
 type filterIndex map[Type]map[rpc.ID]*subscription
 
 // broadcast event to filters that match criteria.
-func (es *EventSystem) broadcast(filters filterIndex, ev *event.Event) {
+func (es *EventSystem) broadcast(filters filterIndex, ev *event.TypeMuxEvent) {
 	if ev == nil {
 		return
 	}

eth/handler.go

@@ -78,8 +78,8 @@ type ProtocolManager struct {
 	SubProtocols []p2p.Protocol
 
 	eventMux      *event.TypeMux
-	txSub         event.Subscription
-	minedBlockSub event.Subscription
+	txSub         *event.TypeMuxSubscription
+	minedBlockSub *event.TypeMuxSubscription
 
 	// channels for fetcher, syncer, txsyncLoop
 	newPeerCh   chan *peer

event/event.go

@@ -14,7 +14,7 @@
 // You should have received a copy of the GNU Lesser General Public License
 // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
 
-// Package event implements an event multiplexer.
+// Package event deals with subscriptions to real-time events.
 package event
 
 import (
@@ -25,33 +25,22 @@ import (
 	"time"
 )
 
-// Event is a time-tagged notification pushed to subscribers.
-type Event struct {
+// TypeMuxEvent is a time-tagged notification pushed to subscribers.
+type TypeMuxEvent struct {
 	Time time.Time
 	Data interface{}
 }
 
-// Subscription is implemented by event subscriptions.
-type Subscription interface {
-	// Chan returns a channel that carries events.
-	// Implementations should return the same channel
-	// for any subsequent calls to Chan.
-	Chan() <-chan *Event
-
-	// Unsubscribe stops delivery of events to a subscription.
-	// The event channel is closed.
-	// Unsubscribe can be called more than once.
-	Unsubscribe()
-}
-
 // A TypeMux dispatches events to registered receivers. Receivers can be
 // registered to handle events of certain type. Any operation
 // called after mux is stopped will return ErrMuxClosed.
 //
 // The zero value is ready to use.
+//
+// Deprecated: use Feed
 type TypeMux struct {
 	mutex   sync.RWMutex
-	subm    map[reflect.Type][]*muxsub
+	subm    map[reflect.Type][]*TypeMuxSubscription
 	stopped bool
 }
 
@@ -61,7 +50,7 @@ var ErrMuxClosed = errors.New("event: mux closed")
 // Subscribe creates a subscription for events of the given types. The
 // subscription's channel is closed when it is unsubscribed
 // or the mux is closed.
-func (mux *TypeMux) Subscribe(types ...interface{}) Subscription {
+func (mux *TypeMux) Subscribe(types ...interface{}) *TypeMuxSubscription {
 	sub := newsub(mux)
 	mux.mutex.Lock()
 	defer mux.mutex.Unlock()
@@ -72,15 +61,15 @@ func (mux *TypeMux) Subscribe(types ...interface{}) Subscription {
 		close(sub.postC)
 	} else {
 		if mux.subm == nil {
-			mux.subm = make(map[reflect.Type][]*muxsub)
+			mux.subm = make(map[reflect.Type][]*TypeMuxSubscription)
 		}
 		for _, t := range types {
 			rtyp := reflect.TypeOf(t)
 			oldsubs := mux.subm[rtyp]
 			if find(oldsubs, sub) != -1 {
 				panic(fmt.Sprintf("event: duplicate type %s in Subscribe", rtyp))
 			}
-			subs := make([]*muxsub, len(oldsubs)+1)
+			subs := make([]*TypeMuxSubscription, len(oldsubs)+1)
 			copy(subs, oldsubs)
 			subs[len(oldsubs)] = sub
 			mux.subm[rtyp] = subs
@@ -92,7 +81,7 @@ func (mux *TypeMux) Subscribe(types ...interface{}) Subscription {
 // Post sends an event to all receivers registered for the given type.
 // It returns ErrMuxClosed if the mux has been stopped.
 func (mux *TypeMux) Post(ev interface{}) error {
-	event := &Event{
+	event := &TypeMuxEvent{
 		Time: time.Now(),
 		Data: ev,
 	}
@@ -125,7 +114,7 @@ func (mux *TypeMux) Stop() {
 	mux.mutex.Unlock()
 }
 
-func (mux *TypeMux) del(s *muxsub) {
+func (mux *TypeMux) del(s *TypeMuxSubscription) {
 	mux.mutex.Lock()
 	for typ, subs := range mux.subm {
 		if pos := find(subs, s); pos >= 0 {
@@ -139,7 +128,7 @@ func (mux *TypeMux) del(s *muxsub) {
 	s.mux.mutex.Unlock()
 }
 
-func find(slice []*muxsub, item *muxsub) int {
+func find(slice []*TypeMuxSubscription, item *TypeMuxSubscription) int {
 	for i, v := range slice {
 		if v == item {
 			return i
@@ -148,14 +137,15 @@ func find(slice []*muxsub, item *muxsub) int {
 	return -1
 }
 
-func posdelete(slice []*muxsub, pos int) []*muxsub {
-	news := make([]*muxsub, len(slice)-1)
+func posdelete(slice []*TypeMuxSubscription, pos int) []*TypeMuxSubscription {
+	news := make([]*TypeMuxSubscription, len(slice)-1)
 	copy(news[:pos], slice[:pos])
 	copy(news[pos:], slice[pos+1:])
 	return news
 }
 
-type muxsub struct {
+// TypeMuxSubscription is a subscription established through TypeMux.
+type TypeMuxSubscription struct {
 	mux     *TypeMux
 	created time.Time
 	closeMu sync.Mutex
@@ -166,13 +156,13 @@ type muxsub struct {
 	// postC can be set to nil without affecting the return value of
 	// Chan.
 	postMu sync.RWMutex
-	readC  <-chan *Event
-	postC  chan<- *Event
+	readC  <-chan *TypeMuxEvent
+	postC  chan<- *TypeMuxEvent
 }
 
-func newsub(mux *TypeMux) *muxsub {
-	c := make(chan *Event)
-	return &muxsub{
+func newsub(mux *TypeMux) *TypeMuxSubscription {
+	c := make(chan *TypeMuxEvent)
+	return &TypeMuxSubscription{
 		mux:     mux,
 		created: time.Now(),
 		readC:   c,
@@ -181,16 +171,16 @@ func newsub(mux *TypeMux) *muxsub {
 	}
 }
 
-func (s *muxsub) Chan() <-chan *Event {
+func (s *TypeMuxSubscription) Chan() <-chan *TypeMuxEvent {
 	return s.readC
 }
 
-func (s *muxsub) Unsubscribe() {
+func (s *TypeMuxSubscription) Unsubscribe() {
 	s.mux.del(s)
 	s.closewait()
 }
 
-func (s *muxsub) closewait() {
+func (s *TypeMuxSubscription) closewait() {
 	s.closeMu.Lock()
 	defer s.closeMu.Unlock()
 	if s.closed {
@@ -205,7 +195,7 @@ func (s *muxsub) closewait() {
 	s.postMu.Unlock()
 }
 
-func (s *muxsub) deliver(event *Event) {
+func (s *TypeMuxSubscription) deliver(event *TypeMuxEvent) {
 	// Short circuit delivery if stale event
 	if s.created.After(event.Time) {
 		return

event/event_test.go

@@ -149,16 +149,34 @@ func emptySubscriber(mux *TypeMux, types ...interface{}) {
 	}()
 }
 
-func BenchmarkPost3(b *testing.B) {
-	var mux = new(TypeMux)
-	defer mux.Stop()
-	emptySubscriber(mux, testEvent(0))
-	emptySubscriber(mux, testEvent(0))
-	emptySubscriber(mux, testEvent(0))
+func BenchmarkPost1000(b *testing.B) {
+	var (
+		mux              = new(TypeMux)
+		subscribed, done sync.WaitGroup
+		nsubs            = 1000
+	)
+	subscribed.Add(nsubs)
+	done.Add(nsubs)
+	for i := 0; i < nsubs; i++ {
+		go func() {
+			s := mux.Subscribe(testEvent(0))
+			subscribed.Done()
+			for range s.Chan() {
+			}
+			done.Done()
+		}()
+	}
+	subscribed.Wait()
 
+	// The actual benchmark.
+	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		mux.Post(testEvent(0))
 	}
+
+	b.StopTimer()
+	mux.Stop()
+	done.Wait()
 }
 
 func BenchmarkPostConcurrent(b *testing.B) {

event/example_feed_test.go

@@ -0,0 +1,73 @@
+// Copyright 2017 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package event_test
+
+import (
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/event"
+)
+
+func ExampleFeed_acknowledgedEvents() {
+	// This example shows how the return value of Send can be used for request/reply
+	// interaction between event consumers and producers.
+	var feed event.Feed
+	type ackedEvent struct {
+		i   int
+		ack chan<- struct{}
+	}
+
+	// Consumers wait for events on the feed and acknowledge processing.
+	done := make(chan struct{})
+	defer close(done)
+	for i := 0; i < 3; i++ {
+		ch := make(chan ackedEvent, 100)
+		sub := feed.Subscribe(ch)
+		go func() {
+			defer sub.Unsubscribe()
+			for {
+				select {
+				case ev := <-ch:
+					fmt.Println(ev.i) // "process" the event
+					ev.ack <- struct{}{}
+				case <-done:
+					return
+				}
+			}
+		}()
+	}
+
+	// The producer sends values of type ackedEvent with increasing values of i.
+	// It waits for all consumers to acknowledge before sending the next event.
+	for i := 0; i < 3; i++ {
+		acksignal := make(chan struct{})
+		n := feed.Send(ackedEvent{i, acksignal})
+		for ack := 0; ack < n; ack++ {
+			<-acksignal
+		}
+	}
+	// Output:
+	// 0
+	// 0
+	// 0
+	// 1
+	// 1
+	// 1
+	// 2
+	// 2
+	// 2
+}