Use slice instead of map

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

basic.go

@@ -11,61 +11,59 @@ 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(typ reflect.Type, cb func(*node)) error {
+	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(typ reflect.Type) {
+	path := typePath(typ)
+
 	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 || len(n.sinks) > 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
+func (b *bus) 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
@@ -89,66 +87,54 @@ func (b *basicBus) Subscribe(typedChan interface{}, opts ...SubOption) (c Cancel
 	}
 
 	err = b.withNode(typ.Elem(), func(n *node) {
-		n.sinks = append(n.sinks, refCh)
+		// when all subs are waiting on this channel, setting this to 1 doesn't
+		// really affect benchmarks
+		n.sub(refCh)
+
 		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[i] = n.sinks[len(n.sinks)-1]
 					n.sinks = n.sinks[:len(n.sinks)-1]
 					break
 				}
 			}
-			tryDrop := len(n.sinks) == 0 && atomic.LoadInt32(&n.nEmitters) == 0
+			tryDrop := len(n.sinks) == 0 && n.nEmitters == 0
 			n.lk.Unlock()
 			if tryDrop {
 				b.tryDropNode(typ.Elem())
 			}
 		}
-	}, 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)
-	}
-
+func (b *bus) Emitter(evtType interface{}, _ ...EmitterOption) (e EmitFunc, c CancelFunc, err error) {
 	typ := reflect.TypeOf(evtType)
 	if typ.Kind() != reflect.Ptr {
-		return nil, errors.New("emitter called with non-pointer type")
+		return nil, nil, errors.New("emitter called with non-pointer type")
 	}
 	typ = typ.Elem()
 
 	err = b.withNode(typ, 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(typ)
+			}
+		}
+	})
 	return
 }
 
@@ -156,7 +142,7 @@ 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
+	// Note: make sure to NEVER lock bus.lk when this lock is held
 	lk sync.RWMutex
 
 	typ reflect.Type
@@ -165,7 +151,7 @@ type node struct {
 	nEmitters int32
 
 	keepLast bool
-	last     atomic.Value
+	last     reflect.Value
 
 	sinks []reflect.Value
 }
@@ -176,6 +162,10 @@ func newNode(typ reflect.Type) *node {
 	}
 }
 
+func (n *node) sub(outChan reflect.Value) {
+	n.sinks = append(n.sinks, outChan)
+}
+
 func (n *node) emit(event interface{}) {
 	eval := reflect.ValueOf(event)
 	if eval.Type() != n.typ {
@@ -183,10 +173,7 @@ func (n *node) emit(event interface{}) {
 	}
 
 	n.lk.RLock()
-	if n.keepLast {
-		n.last.Store(eval)
-	}
-
+	// TODO: try using reflect.Select
 	for _, ch := range n.sinks {
 		ch.Send(eval)
 	}
@@ -196,4 +183,8 @@ func (n *node) emit(event interface{}) {
 ///////////////////////
 // UTILS
 
-var _ Bus = &basicBus{}
+func typePath(t reflect.Type) string {
+	return t.PkgPath() + "/" + t.String()
+}
+
+var _ Bus = &bus{}

basic_test.go

@@ -6,21 +6,11 @@ import (
 	"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"
 }
@@ -38,11 +28,11 @@ 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{})
 }
@@ -66,11 +56,11 @@ func TestSub(t *testing.T) {
 		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 +73,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 +85,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 +105,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
@@ -143,9 +133,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 +146,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 +156,7 @@ func TestSubMany(t *testing.T) {
 
 	var r int32
 
-	n := getN()
+	n := 50000
 	var wait sync.WaitGroup
 	var ready sync.WaitGroup
 	wait.Add(n)
@@ -187,11 +177,11 @@ func TestSubMany(t *testing.T) {
 		}()
 	}
 
-	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()
 
@@ -222,11 +212,11 @@ func TestSubType(t *testing.T) {
 		wait.Done()
 	}()
 
-	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{})
 	wait.Wait()
@@ -236,79 +226,7 @@ func TestSubType(t *testing.T) {
 	}
 }
 
-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
@@ -337,13 +255,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()
 
@@ -363,72 +279,72 @@ func testMany(t testing.TB, subs, emits, msgs int, stateful bool) {
 }
 
 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)
+	testMany(b, 10, 1, 1000000)
 }
 
 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, 1, 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)
+}
+
+func t() {
+	bus := NewBus()
+
+	events := make(chan fmt.Stringer)
+	cancel, err := bus.Subscribe(events, Stateful)
+	if err != nil {
+		//
+	}
+	defer cancel()
+
 }

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

@@ -5,33 +5,13 @@ import (
 	"reflect"
 )
 
-var closeEmit struct{}
-
-type subSettings struct {
+type SubSettings struct {
 	forcedType reflect.Type
 }
+type SubOption func(*SubSettings) error
 
-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)
+	return func(s *SubSettings) error {
 		typ := reflect.TypeOf(evtType)
 		if typ.Kind() != reflect.Ptr {
 			return errors.New("ForceSubType called with non-pointer type")
@@ -41,33 +21,12 @@ func ForceSubType(evtType interface{}) SubOption {
 	}
 }
 
-type emitterSettings struct {
-	makeStateful bool
-}
-type EmitterOption func(interface{}) error
+type EmitterSettings struct{}
+type EmitterOption func(*EmitterSettings)
 
-// 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
-	//
-	// Example:
-	// ch := make(chan EventT, 10)
-	// defer close(ch)
-	// cancel, err := eventbus.Subscribe(ch)
-	// defer cancel()
+	// publishers to get blocked
 	Subscribe(typedChan interface{}, opts ...SubOption) (CancelFunc, error)
 
 	// Emitter creates new emitter
@@ -76,11 +35,9 @@ type Bus interface {
 	// select output type
 	//
 	// Example:
-	// emit, err := eventbus.Emitter(new(EventT))
-	// defer emit.Close() // MUST call this after being done with the emitter
-	//
-	// emit(EventT{})
-	Emitter(eventType interface{}, opts ...EmitterOption) (EmitFunc, error)
+	// sub, cancel, err := eventbus.Subscribe(new(os.Signal))
+	// defer cancel()
+	Emitter(eventType interface{}, opts ...EmitterOption) (EmitFunc, CancelFunc, error)
 }
 
 // EmitFunc emits events. If any channel subscribed to the topic is blocked,
@@ -89,8 +46,4 @@ 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()