Use sync map

License: MIT
Signed-off-by: Jakub Sztandera <kubuxu@protonmail.ch>

basic.go

@@ -11,144 +11,99 @@ import (
 ///////////////////////
 // BUS
 
-type basicBus struct {
+type bus struct {
 	lk    sync.Mutex
-	nodes map[reflect.Type]*node
+	nodes map[string]*node
 }
 
-func NewBus() *basicBus {
-	return &basicBus{
-		nodes: map[reflect.Type]*node{},
+func NewBus() Bus {
+	return &bus{
+		nodes: map[string]*node{},
 	}
 }
 
-func (b *basicBus) withNode(typ reflect.Type, cb func(*node), async func(*node)) error {
+func (b *bus) withNode(evtType interface{}, cb func(*node)) error {
+	typ := reflect.TypeOf(evtType)
+	if typ.Kind() != reflect.Ptr {
+		return errors.New("subscribe called with non-pointer type")
+	}
+	typ = typ.Elem()
+	path := typePath(typ)
+
 	b.lk.Lock()
 
-	n, ok := b.nodes[typ]
+	n, ok := b.nodes[path]
 	if !ok {
 		n = newNode(typ)
-		b.nodes[typ] = n
+		b.nodes[path] = n
 	}
 
 	n.lk.Lock()
 	b.lk.Unlock()
-
+	defer n.lk.Unlock()
 	cb(n)
-
-	go func() {
-		defer n.lk.Unlock()
-		async(n)
-	}()
-
 	return nil
 }
 
-func (b *basicBus) tryDropNode(typ reflect.Type) {
+func (b *bus) tryDropNode(evtType interface{}) {
+	path := typePath(reflect.TypeOf(evtType).Elem())
+
 	b.lk.Lock()
-	n, ok := b.nodes[typ]
+	n, ok := b.nodes[path]
 	if !ok { // already dropped
 		b.lk.Unlock()
 		return
 	}
 
 	n.lk.Lock()
-	if atomic.LoadInt32(&n.nEmitters) > 0 || len(n.sinks) > 0 {
+	if n.nEmitters > 0 || n.sinkLen() > 0 {
 		n.lk.Unlock()
 		b.lk.Unlock()
 		return // still in use
 	}
 	n.lk.Unlock()
 
-	delete(b.nodes, typ)
+	delete(b.nodes, path)
 	b.lk.Unlock()
 }
 
-func (b *basicBus) Subscribe(typedChan interface{}, opts ...SubOption) (c CancelFunc, err error) {
-	var settings subSettings
-	for _, opt := range opts {
-		if err := opt(&settings); err != nil {
-			return nil, err
-		}
-	}
-
-	refCh := reflect.ValueOf(typedChan)
-	typ := refCh.Type()
-	if typ.Kind() != reflect.Chan {
-		return nil, errors.New("expected a channel")
-	}
-	if typ.ChanDir()&reflect.SendDir == 0 {
-		return nil, errors.New("channel doesn't allow send")
-	}
-
-	if settings.forcedType != nil {
-		if settings.forcedType.Elem().AssignableTo(typ) {
-			return nil, fmt.Errorf("forced type %s cannot be sent to chan %s", settings.forcedType, typ)
-		}
-		typ = settings.forcedType
-	}
-
-	err = b.withNode(typ.Elem(), func(n *node) {
-		n.sinks = append(n.sinks, refCh)
+func (b *bus) Subscribe(evtType interface{}, _ ...SubOption) (s <-chan interface{}, c CancelFunc, err error) {
+	err = b.withNode(evtType, func(n *node) {
+		out, i := n.sub(0)
+		s = out
 		c = func() {
 			n.lk.Lock()
-			for i := 0; i < len(n.sinks); i++ {
-				if n.sinks[i] == refCh {
-					n.sinks[i], n.sinks[len(n.sinks)-1] = n.sinks[len(n.sinks)-1], reflect.Value{}
-					n.sinks = n.sinks[:len(n.sinks)-1]
-					break
-				}
-			}
-			tryDrop := len(n.sinks) == 0 && atomic.LoadInt32(&n.nEmitters) == 0
+			n.sink.Delete(i)
+			close(out)
+			tryDrop := n.sinkLen() == 0 && n.nEmitters == 0
 			n.lk.Unlock()
 			if tryDrop {
-				b.tryDropNode(typ.Elem())
+				b.tryDropNode(evtType)
 			}
 		}
-	}, func(n *node) {
-		if n.keepLast {
-			lastVal, ok := n.last.Load().(reflect.Value)
-			if !ok {
-				return
-			}
-
-			refCh.Send(lastVal)
-		}
 	})
 	return
 }
 
-func (b *basicBus) Emitter(evtType interface{}, opts ...EmitterOption) (e EmitFunc, err error) {
-	var settings emitterSettings
-	for _, opt := range opts {
-		opt(&settings)
-	}
-
-	typ := reflect.TypeOf(evtType)
-	if typ.Kind() != reflect.Ptr {
-		return nil, errors.New("emitter called with non-pointer type")
-	}
-	typ = typ.Elem()
-
-	err = b.withNode(typ, func(n *node) {
+func (b *bus) Emitter(evtType interface{}, _ ...EmitterOption) (e EmitFunc, c CancelFunc, err error) {
+	err = b.withNode(evtType, func(n *node) {
 		atomic.AddInt32(&n.nEmitters, 1)
 		closed := false
-		n.keepLast = n.keepLast || settings.makeStateful
 
 		e = func(event interface{}) {
 			if closed {
 				panic("emitter is closed")
 			}
-			if event == closeEmit {
-				closed = true
-				if atomic.AddInt32(&n.nEmitters, -1) == 0 {
-					b.tryDropNode(typ)
-				}
-				return
-			}
 			n.emit(event)
 		}
-	}, func(_ *node) {})
+
+		c = func() {
+			closed = true
+			if atomic.AddInt32(&n.nEmitters, -1) == 0 {
+				b.tryDropNode(evtType)
+			}
+		}
+	})
 	return
 }
 
@@ -156,7 +111,10 @@ func (b *basicBus) Emitter(evtType interface{}, opts ...EmitterOption) (e EmitFu
 // NODE
 
 type node struct {
-	// Note: make sure to NEVER lock basicBus.lk when this lock is held
+	// not under lock
+	sink sync.Map
+
+	// Note: make sure to NEVER lock bus.lk when this lock is held
 	lk sync.RWMutex
 
 	typ reflect.Type
@@ -164,10 +122,8 @@ type node struct {
 	// emitter ref count
 	nEmitters int32
 
-	keepLast bool
-	last     atomic.Value
-
-	sinks []reflect.Value
+	// sink index counter
+	sinkC int
 }
 
 func newNode(typ reflect.Type) *node {
@@ -176,24 +132,40 @@ func newNode(typ reflect.Type) *node {
 	}
 }
 
+func (n *node) sinkLen() int {
+	ln := 0
+	n.sink.Range(func(_, _ interface{}) bool {
+		ln = ln + 1
+		return true
+	})
+	return ln
+}
+
+func (n *node) sub(buf int) (chan interface{}, int) {
+	out := make(chan interface{}, buf)
+	i := n.sinkC
+	n.sinkC++
+	n.sink.Store(i, out)
+	return out, i
+}
+
 func (n *node) emit(event interface{}) {
-	eval := reflect.ValueOf(event)
-	if eval.Type() != n.typ {
-		panic(fmt.Sprintf("Emit called with wrong type. expected: %s, got: %s", n.typ, eval.Type()))
+	etype := reflect.TypeOf(event)
+	if etype != n.typ {
+		panic(fmt.Sprintf("Emit called with wrong type. expected: %s, got: %s", n.typ, etype))
 	}
 
-	n.lk.RLock()
-	if n.keepLast {
-		n.last.Store(eval)
-	}
-
-	for _, ch := range n.sinks {
-		ch.Send(eval)
-	}
-	n.lk.RUnlock()
+	n.sink.Range(func(_, ch interface{}) bool {
+		ch.(chan interface{}) <- event
+		return true
+	})
 }
 
 ///////////////////////
 // UTILS
 
-var _ Bus = &basicBus{}
+func typePath(t reflect.Type) string {
+	return t.PkgPath() + "/" + t.String()
+}
+
+var _ Bus = &bus{}

basic_test.go

@@ -1,34 +1,18 @@
 package event
 
 import (
-	"fmt"
 	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
-
-	"github.com/jbenet/go-detect-race"
 )
 
 type EventA struct{}
 type EventB int
 
-func getN() int {
-	n := 50000
-	if detectrace.WithRace() {
-		n = 1000
-	}
-	return n
-}
-
-func (EventA) String() string {
-	return "Oh, Hello"
-}
-
 func TestEmit(t *testing.T) {
 	bus := NewBus()
-	events := make(chan EventA)
-	cancel, err := bus.Subscribe(events)
+	events, cancel, err := bus.Subscribe(new(EventA))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -38,19 +22,18 @@ func TestEmit(t *testing.T) {
 		<-events
 	}()
 
-	emit, err := bus.Emitter(new(EventA))
+	emit, cancel, err := bus.Emitter(new(EventA))
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer emit.Close()
+	defer cancel()
 
 	emit(EventA{})
 }
 
 func TestSub(t *testing.T) {
 	bus := NewBus()
-	events := make(chan EventB)
-	cancel, err := bus.Subscribe(events)
+	events, cancel, err := bus.Subscribe(new(EventB))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -62,15 +45,15 @@ func TestSub(t *testing.T) {
 
 	go func() {
 		defer cancel()
-		event = <-events
+		event = (<-events).(EventB)
 		wait.Done()
 	}()
 
-	emit, err := bus.Emitter(new(EventB))
+	emit, cancel, err := bus.Emitter(new(EventB))
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer emit.Close()
+	defer cancel()
 
 	emit(EventB(7))
 	wait.Wait()
@@ -83,11 +66,11 @@ func TestSub(t *testing.T) {
 func TestEmitNoSubNoBlock(t *testing.T) {
 	bus := NewBus()
 
-	emit, err := bus.Emitter(new(EventA))
+	emit, cancel, err := bus.Emitter(new(EventA))
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer emit.Close()
+	defer cancel()
 
 	emit(EventA{})
 }
@@ -95,11 +78,11 @@ func TestEmitNoSubNoBlock(t *testing.T) {
 func TestEmitOnClosed(t *testing.T) {
 	bus := NewBus()
 
-	emit, err := bus.Emitter(new(EventA))
+	emit, cancel, err := bus.Emitter(new(EventA))
 	if err != nil {
 		t.Fatal(err)
 	}
-	emit.Close()
+	cancel()
 
 	defer func() {
 		r := recover()
@@ -115,8 +98,8 @@ func TestEmitOnClosed(t *testing.T) {
 }
 
 func TestClosingRaces(t *testing.T) {
-	subs := getN()
-	emits := getN()
+	subs := 50000
+	emits := 50000
 
 	var wg sync.WaitGroup
 	var lk sync.RWMutex
@@ -131,7 +114,7 @@ func TestClosingRaces(t *testing.T) {
 			lk.RLock()
 			defer lk.RUnlock()
 
-			cancel, _ := b.Subscribe(make(chan EventA))
+			_, cancel, _ := b.Subscribe(new(EventA))
 			time.Sleep(10 * time.Millisecond)
 			cancel()
 
@@ -143,9 +126,9 @@ func TestClosingRaces(t *testing.T) {
 			lk.RLock()
 			defer lk.RUnlock()
 
-			emit, _ := b.Emitter(new(EventA))
+			_, cancel, _ := b.Emitter(new(EventA))
 			time.Sleep(10 * time.Millisecond)
-			emit.Close()
+			cancel()
 
 			wg.Done()
 		}()
@@ -156,7 +139,7 @@ func TestClosingRaces(t *testing.T) {
 
 	wg.Wait()
 
-	if len(b.nodes) != 0 {
+	if len(b.(*bus).nodes) != 0 {
 		t.Error("expected no nodes")
 	}
 }
@@ -166,7 +149,7 @@ func TestSubMany(t *testing.T) {
 
 	var r int32
 
-	n := getN()
+	n := 50000
 	var wait sync.WaitGroup
 	var ready sync.WaitGroup
 	wait.Add(n)
@@ -174,141 +157,35 @@ func TestSubMany(t *testing.T) {
 
 	for i := 0; i < n; i++ {
 		go func() {
-			events := make(chan EventB)
-			cancel, err := bus.Subscribe(events)
+			events, cancel, err := bus.Subscribe(new(EventB))
 			if err != nil {
 				panic(err)
 			}
 			defer cancel()
 
 			ready.Done()
-			atomic.AddInt32(&r, int32(<-events))
+			atomic.AddInt32(&r, int32((<-events).(EventB)))
 			wait.Done()
 		}()
 	}
 
-	emit, err := bus.Emitter(new(EventB))
+	emit, cancel, err := bus.Emitter(new(EventB))
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer emit.Close()
+	defer cancel()
 
 	ready.Wait()
 
 	emit(EventB(7))
 	wait.Wait()
 
-	if int(r) != 7*n {
+	if int(r) != 7 * n {
 		t.Error("got wrong result")
 	}
 }
 
-func TestSubType(t *testing.T) {
-	bus := NewBus()
-	events := make(chan fmt.Stringer)
-	cancel, err := bus.Subscribe(events, ForceSubType(new(EventA)))
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	var event fmt.Stringer
-
-	var wait sync.WaitGroup
-	wait.Add(1)
-
-	go func() {
-		defer cancel()
-		event = <-events
-		wait.Done()
-	}()
-
-	emit, err := bus.Emitter(new(EventA))
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer emit.Close()
-
-	emit(EventA{})
-	wait.Wait()
-
-	if event.String() != "Oh, Hello" {
-		t.Error("didn't get the correct message")
-	}
-}
-
-func TestNonStateful(t *testing.T) {
-	bus := NewBus()
-	emit, err := bus.Emitter(new(EventB))
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer emit.Close()
-
-	eventsA := make(chan EventB, 1)
-	cancelS, err := bus.Subscribe(eventsA)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer cancelS()
-
-	select {
-	case <-eventsA:
-		t.Fatal("didn't expect to get an event")
-	default:
-	}
-
-	emit(EventB(1))
-
-	select {
-	case e := <-eventsA:
-		if e != 1 {
-			t.Fatal("got wrong event")
-		}
-	default:
-		t.Fatal("expected to get an event")
-	}
-
-	eventsB := make(chan EventB, 1)
-	cancelS2, err := bus.Subscribe(eventsB)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer cancelS2()
-
-	select {
-	case <-eventsA:
-		t.Fatal("didn't expect to get an event")
-	default:
-	}
-}
-
-func TestStateful(t *testing.T) {
-	bus := NewBus()
-	emit, err := bus.Emitter(new(EventB), Stateful)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer emit.Close()
-
-	emit(EventB(2))
-
-	eventsA := make(chan EventB, 1)
-	cancelS, err := bus.Subscribe(eventsA)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer cancelS()
-
-	if <-eventsA != 2 {
-		t.Fatal("got wrong event")
-	}
-}
-
-func testMany(t testing.TB, subs, emits, msgs int, stateful bool) {
-	if detectrace.WithRace() && subs+emits > 5000 {
-		t.SkipNow()
-	}
-
+func testMany(t testing.TB, subs, emits, msgs int) {
 	bus := NewBus()
 
 	var r int64
@@ -320,16 +197,15 @@ func testMany(t testing.TB, subs, emits, msgs int, stateful bool) {
 
 	for i := 0; i < subs; i++ {
 		go func() {
-			events := make(chan EventB)
-			cancel, err := bus.Subscribe(events)
+			events, cancel, err := bus.Subscribe(new(EventB))
 			if err != nil {
 				panic(err)
 			}
 			defer cancel()
 
 			ready.Done()
-			for i := 0; i < emits*msgs; i++ {
-				atomic.AddInt64(&r, int64(<-events))
+			for i := 0; i < emits * msgs; i++ {
+				atomic.AddInt64(&r, int64((<-events).(EventB)))
 			}
 			wait.Done()
 		}()
@@ -337,13 +213,11 @@ func testMany(t testing.TB, subs, emits, msgs int, stateful bool) {
 
 	for i := 0; i < emits; i++ {
 		go func() {
-			emit, err := bus.Emitter(new(EventB), func(settings *emitterSettings) {
-				settings.makeStateful = stateful
-			})
+			emit, cancel, err := bus.Emitter(new(EventB))
 			if err != nil {
 				panic(err)
 			}
-			defer emit.Close()
+			defer cancel()
 
 			ready.Wait()
 
@@ -357,78 +231,60 @@ func testMany(t testing.TB, subs, emits, msgs int, stateful bool) {
 
 	wait.Wait()
 
-	if int(r) != 97*subs*emits*msgs {
+	if int(r) != 97 * subs * emits * msgs {
 		t.Fatal("got wrong result")
 	}
 }
 
 func TestBothMany(t *testing.T) {
-	testMany(t, 10000, 100, 10, false)
+	testMany(t, 10000, 100, 10)
 }
 
 func BenchmarkSubs(b *testing.B) {
 	b.ReportAllocs()
-	testMany(b, b.N, 100, 100, false)
+	testMany(b, b.N, 100, 100)
 }
 
 func BenchmarkEmits(b *testing.B) {
 	b.ReportAllocs()
-	testMany(b, 100, b.N, 100, false)
+	testMany(b, 100, b.N, 100)
 }
 
 func BenchmarkMsgs(b *testing.B) {
 	b.ReportAllocs()
-	testMany(b, 100, 100, b.N, false)
+	testMany(b, 100, 100, b.N)
 }
 
 func BenchmarkOneToMany(b *testing.B) {
 	b.ReportAllocs()
-	testMany(b, b.N, 1, 100, false)
+	testMany(b, b.N, 1, 100)
 }
 
 func BenchmarkManyToOne(b *testing.B) {
 	b.ReportAllocs()
-	testMany(b, 1, b.N, 100, false)
+	testMany(b, 1, b.N, 100)
 }
 
 func BenchmarkMs1e2m4(b *testing.B) {
 	b.N = 1000000
 	b.ReportAllocs()
-	testMany(b, 10, 100, 10000, false)
+	testMany(b, 10, 100, 10000)
 }
 
 func BenchmarkMs1e0m6(b *testing.B) {
-	b.N = 10000000
-	b.ReportAllocs()
-	testMany(b, 10, 1, 1000000, false)
-}
-
-func BenchmarkMs0e0m6(b *testing.B) {
 	b.N = 1000000
 	b.ReportAllocs()
-	testMany(b, 1, 1, 1000000, false)
-}
-
-func BenchmarkStatefulMs1e0m6(b *testing.B) {
-	b.N = 10000000
-	b.ReportAllocs()
-	testMany(b, 10, 1, 1000000, true)
-}
-
-func BenchmarkStatefulMs0e0m6(b *testing.B) {
-	b.N = 1000000
-	b.ReportAllocs()
-	testMany(b, 1, 1, 1000000, true)
+	testMany(b, 10, 1, 1000000)
 }
 
 func BenchmarkMs0e6m0(b *testing.B) {
 	b.N = 1000000
 	b.ReportAllocs()
-	testMany(b, 1, 1000000, 1, false)
+	testMany(b, 1, 1000000, 1)
 }
 
 func BenchmarkMs6e0m0(b *testing.B) {
 	b.N = 1000000
 	b.ReportAllocs()
-	testMany(b, 1000000, 1, 1, false)
+	testMany(b, 1000000, 1, 1)
 }

go.mod

@@ -1,5 +1,3 @@
 module github.com/libp2p/go-eventbus
 
 go 1.12
-
-require github.com/jbenet/go-detect-race v0.0.0-20150302022421-3463798d9574

go.sum

@@ -1,2 +0,0 @@
-github.com/jbenet/go-detect-race v0.0.0-20150302022421-3463798d9574 h1:Pxjl8Wn3cCU7nB/MCmPEUMbjMHxXFqODW6rce0jpxB4=
-github.com/jbenet/go-detect-race v0.0.0-20150302022421-3463798d9574/go.mod h1:gynVu6LUw+xMXD3XEvjHQcIbJkWEamnGjJDebRHqTd0=

interface.go

@@ -1,86 +1,27 @@
 package event
 
-import (
-	"errors"
-	"reflect"
-)
+type SubSettings struct {}
+type SubOption func(*SubSettings)
 
-var closeEmit struct{}
+type EmitterSettings struct {}
+type EmitterOption func(*EmitterSettings)
 
-type subSettings struct {
-	forcedType reflect.Type
-}
-
-type SubOption func(interface{}) error
-
-// ForceSubType is a Subscribe option which overrides the type to which
-// the subscription will be done. Note that the evtType must be assignable
-// to channel type.
-//
-// This also allows for subscribing to multiple eventbus channels with one
-// Go channel to get better ordering guarantees.
-//
-// Example:
-// type Event struct{}
-// func (Event) String() string {
-//    return "event"
-// }
-//
-// eventCh := make(chan fmt.Stringer) // interface { String() string }
-// cancel, err := eventbus.Subscribe(eventCh, event.ForceSubType(new(Event)))
-// [...]
-func ForceSubType(evtType interface{}) SubOption {
-	return func(settings interface{}) error {
-		s := settings.(*subSettings)
-		typ := reflect.TypeOf(evtType)
-		if typ.Kind() != reflect.Ptr {
-			return errors.New("ForceSubType called with non-pointer type")
-		}
-		s.forcedType = typ
-		return nil
-	}
-}
-
-type emitterSettings struct {
-	makeStateful bool
-}
-type EmitterOption func(interface{}) error
-
-// Stateful is an Emitter option which makes makes the eventbus channel
-// 'remember' last event sent, and when a new subscriber joins the
-// bus, the remembered event is immediately sent to the subscription
-// channel.
-//
-// This allows to provide state tracking for dynamic systems, and/or
-// allows new subscribers to verify that there are Emitters on the channel
-func Stateful(s *emitterSettings) {
-	s.makeStateful = true
-}
-
-// Bus is an interface to type-based event delivery system
 type Bus interface {
-	// Subscribe creates new subscription. Failing to drain the channel will cause
-	// publishers to get blocked. CancelFunc is guaranteed to return after last send
-	// to the channel
+	// Subscribe creates new subscription. Failing to drain the incoming channel
+	// will cause publishers to get blocked
 	//
-	// Example:
-	// ch := make(chan EventT, 10)
-	// defer close(ch)
-	// cancel, err := eventbus.Subscribe(ch)
-	// defer cancel()
-	Subscribe(typedChan interface{}, opts ...SubOption) (CancelFunc, error)
-
-	// Emitter creates new emitter
-	//
-	// eventType accepts typed nil pointers, and uses the type information to
+	// evtTypes only accepts typed nil pointers, and uses the type information to
 	// select output type
 	//
 	// Example:
-	// emit, err := eventbus.Emitter(new(EventT))
-	// defer emit.Close() // MUST call this after being done with the emitter
+	// sub, cancel, err := eventbus.Subscribe(new(os.Signal))
+	// defer cancel()
 	//
-	// emit(EventT{})
-	Emitter(eventType interface{}, opts ...EmitterOption) (EmitFunc, error)
+	// evt := (<-sub).(os.Signal) // guaranteed to be safe
+	Subscribe(eventType interface{}, opts ...SubOption) (<-chan interface{}, CancelFunc, error)
+
+
+	Emitter(eventType interface{}, opts ...EmitterOption) (EmitFunc, CancelFunc, error)
 }
 
 // EmitFunc emits events. If any channel subscribed to the topic is blocked,
@@ -89,8 +30,5 @@ type Bus interface {
 // Calling this function with wrong event type will cause a panic
 type EmitFunc func(event interface{})
 
-func (f EmitFunc) Close() {
-	f(closeEmit)
-}
-
 type CancelFunc func()
+