Use sync map

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

.travis.yml

@@ -1,30 +0,0 @@
-os:
-  - linux
-
-language: go
-
-go:
-  - 1.12.x
-
-env:
-  global:
-    - GOTFLAGS="-race"
-  matrix:
-    - BUILD_DEPTYPE=gomod
-
-
-# disable travis install
-install:
-  - true
-
-script:
-  - bash <(curl -s https://raw.githubusercontent.com/ipfs/ci-helpers/master/travis-ci/run-standard-tests.sh)
-
-
-cache:
-  directories:
-    - $GOPATH/pkg/mod
-    - $HOME/.cache/go-build
-
-notifications:
-  email: false

README.md

@@ -0,0 +1,25 @@
+# go-eventbus
+
+[![](https://img.shields.io/badge/made%20by-Protocol%20Labs-blue.svg?style=flat-square)](https://protocol.ai)
+[![](https://img.shields.io/badge/project-libp2p-yellow.svg?style=flat-square)](https://libp2p.io/)
+[![](https://img.shields.io/badge/freenode-%23libp2p-yellow.svg?style=flat-square)](http://webchat.freenode.net/?channels=%23libp2p)
+[![GoDoc](https://godoc.org/github.com/libp2p/go-eventbus?status.svg)](https://godoc.org/github.com/libp2p/go-eventbus)
+[![Coverage Status](https://coveralls.io/repos/github/libp2p/go-eventbus/badge.svg?branch=master)](https://coveralls.io/github/libp2p/go-eventbus?branch=master)
+[![Build Status](https://travis-ci.org/libp2p/go-eventbus.svg?branch=master)](https://travis-ci.org/libp2p/go-eventbus)
+[![Discourse posts](https://img.shields.io/discourse/https/discuss.libp2p.io/posts.svg)](https://discuss.libp2p.io)
+
+> Simple and fast Go event bus
+
+## Install
+
+```sh
+go get github.com/libp2p/go-eventbus
+```
+
+## Usage
+
+Check out the [GoDocs](https://godoc.org/github.com/libp2p/go-eventbus).
+
+## License
+
+Dual-licensed under MIT and ASLv2, by way of the [Permissive License Stack](https://protocol.ai/blog/announcing-the-permissive-license-stack/).

basic.go

@@ -0,0 +1,171 @@
+package event
+
+import (
+	"errors"
+	"fmt"
+	"reflect"
+	"sync"
+	"sync/atomic"
+)
+
+///////////////////////
+// BUS
+
+type bus struct {
+	lk    sync.Mutex
+	nodes map[string]*node
+}
+
+func NewBus() Bus {
+	return &bus{
+		nodes: map[string]*node{},
+	}
+}
+
+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[path]
+	if !ok {
+		n = newNode(typ)
+		b.nodes[path] = n
+	}
+
+	n.lk.Lock()
+	b.lk.Unlock()
+	defer n.lk.Unlock()
+	cb(n)
+	return nil
+}
+
+func (b *bus) tryDropNode(evtType interface{}) {
+	path := typePath(reflect.TypeOf(evtType).Elem())
+
+	b.lk.Lock()
+	n, ok := b.nodes[path]
+	if !ok { // already dropped
+		b.lk.Unlock()
+		return
+	}
+
+	n.lk.Lock()
+	if n.nEmitters > 0 || n.sinkLen() > 0 {
+		n.lk.Unlock()
+		b.lk.Unlock()
+		return // still in use
+	}
+	n.lk.Unlock()
+
+	delete(b.nodes, path)
+	b.lk.Unlock()
+}
+
+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()
+			n.sink.Delete(i)
+			close(out)
+			tryDrop := n.sinkLen() == 0 && n.nEmitters == 0
+			n.lk.Unlock()
+			if tryDrop {
+				b.tryDropNode(evtType)
+			}
+		}
+	})
+	return
+}
+
+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
+
+		e = func(event interface{}) {
+			if closed {
+				panic("emitter is closed")
+			}
+			n.emit(event)
+		}
+
+		c = func() {
+			closed = true
+			if atomic.AddInt32(&n.nEmitters, -1) == 0 {
+				b.tryDropNode(evtType)
+			}
+		}
+	})
+	return
+}
+
+///////////////////////
+// NODE
+
+type node struct {
+	// 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
+
+	// emitter ref count
+	nEmitters int32
+
+	// sink index counter
+	sinkC int
+}
+
+func newNode(typ reflect.Type) *node {
+	return &node{
+		typ: typ,
+	}
+}
+
+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{}) {
+	etype := reflect.TypeOf(event)
+	if etype != n.typ {
+		panic(fmt.Sprintf("Emit called with wrong type. expected: %s, got: %s", n.typ, etype))
+	}
+
+	n.sink.Range(func(_, ch interface{}) bool {
+		ch.(chan interface{}) <- event
+		return true
+	})
+}
+
+///////////////////////
+// UTILS
+
+func typePath(t reflect.Type) string {
+	return t.PkgPath() + "/" + t.String()
+}
+
+var _ Bus = &bus{}

basic_test.go

@@ -0,0 +1,290 @@
+package event
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+type EventA struct{}
+type EventB int
+
+func TestEmit(t *testing.T) {
+	bus := NewBus()
+	events, cancel, err := bus.Subscribe(new(EventA))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	go func() {
+		defer cancel()
+		<-events
+	}()
+
+	emit, cancel, err := bus.Emitter(new(EventA))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cancel()
+
+	emit(EventA{})
+}
+
+func TestSub(t *testing.T) {
+	bus := NewBus()
+	events, cancel, err := bus.Subscribe(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var event EventB
+
+	var wait sync.WaitGroup
+	wait.Add(1)
+
+	go func() {
+		defer cancel()
+		event = (<-events).(EventB)
+		wait.Done()
+	}()
+
+	emit, cancel, err := bus.Emitter(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cancel()
+
+	emit(EventB(7))
+	wait.Wait()
+
+	if event != 7 {
+		t.Error("got wrong event")
+	}
+}
+
+func TestEmitNoSubNoBlock(t *testing.T) {
+	bus := NewBus()
+
+	emit, cancel, err := bus.Emitter(new(EventA))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cancel()
+
+	emit(EventA{})
+}
+
+func TestEmitOnClosed(t *testing.T) {
+	bus := NewBus()
+
+	emit, cancel, err := bus.Emitter(new(EventA))
+	if err != nil {
+		t.Fatal(err)
+	}
+	cancel()
+
+	defer func() {
+		r := recover()
+		if r == nil {
+			t.Errorf("expected panic")
+		}
+		if r.(string) != "emitter is closed" {
+			t.Error("unexpected message")
+		}
+	}()
+
+	emit(EventA{})
+}
+
+func TestClosingRaces(t *testing.T) {
+	subs := 50000
+	emits := 50000
+
+	var wg sync.WaitGroup
+	var lk sync.RWMutex
+	lk.Lock()
+
+	wg.Add(subs + emits)
+
+	b := NewBus()
+
+	for i := 0; i < subs; i++ {
+		go func() {
+			lk.RLock()
+			defer lk.RUnlock()
+
+			_, cancel, _ := b.Subscribe(new(EventA))
+			time.Sleep(10 * time.Millisecond)
+			cancel()
+
+			wg.Done()
+		}()
+	}
+	for i := 0; i < emits; i++ {
+		go func() {
+			lk.RLock()
+			defer lk.RUnlock()
+
+			_, cancel, _ := b.Emitter(new(EventA))
+			time.Sleep(10 * time.Millisecond)
+			cancel()
+
+			wg.Done()
+		}()
+	}
+
+	time.Sleep(10 * time.Millisecond)
+	lk.Unlock() // start everything
+
+	wg.Wait()
+
+	if len(b.(*bus).nodes) != 0 {
+		t.Error("expected no nodes")
+	}
+}
+
+func TestSubMany(t *testing.T) {
+	bus := NewBus()
+
+	var r int32
+
+	n := 50000
+	var wait sync.WaitGroup
+	var ready sync.WaitGroup
+	wait.Add(n)
+	ready.Add(n)
+
+	for i := 0; i < n; i++ {
+		go func() {
+			events, cancel, err := bus.Subscribe(new(EventB))
+			if err != nil {
+				panic(err)
+			}
+			defer cancel()
+
+			ready.Done()
+			atomic.AddInt32(&r, int32((<-events).(EventB)))
+			wait.Done()
+		}()
+	}
+
+	emit, cancel, err := bus.Emitter(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cancel()
+
+	ready.Wait()
+
+	emit(EventB(7))
+	wait.Wait()
+
+	if int(r) != 7 * n {
+		t.Error("got wrong result")
+	}
+}
+
+func testMany(t testing.TB, subs, emits, msgs int) {
+	bus := NewBus()
+
+	var r int64
+
+	var wait sync.WaitGroup
+	var ready sync.WaitGroup
+	wait.Add(subs + emits)
+	ready.Add(subs)
+
+	for i := 0; i < subs; i++ {
+		go func() {
+			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).(EventB)))
+			}
+			wait.Done()
+		}()
+	}
+
+	for i := 0; i < emits; i++ {
+		go func() {
+			emit, cancel, err := bus.Emitter(new(EventB))
+			if err != nil {
+				panic(err)
+			}
+			defer cancel()
+
+			ready.Wait()
+
+			for i := 0; i < msgs; i++ {
+				emit(EventB(97))
+			}
+
+			wait.Done()
+		}()
+	}
+
+	wait.Wait()
+
+	if int(r) != 97 * subs * emits * msgs {
+		t.Fatal("got wrong result")
+	}
+}
+
+func TestBothMany(t *testing.T) {
+	testMany(t, 10000, 100, 10)
+}
+
+func BenchmarkSubs(b *testing.B) {
+	b.ReportAllocs()
+	testMany(b, b.N, 100, 100)
+}
+
+func BenchmarkEmits(b *testing.B) {
+	b.ReportAllocs()
+	testMany(b, 100, b.N, 100)
+}
+
+func BenchmarkMsgs(b *testing.B) {
+	b.ReportAllocs()
+	testMany(b, 100, 100, b.N)
+}
+
+func BenchmarkOneToMany(b *testing.B) {
+	b.ReportAllocs()
+	testMany(b, b.N, 1, 100)
+}
+
+func BenchmarkManyToOne(b *testing.B) {
+	b.ReportAllocs()
+	testMany(b, 1, b.N, 100)
+}
+
+func BenchmarkMs1e2m4(b *testing.B) {
+	b.N = 1000000
+	b.ReportAllocs()
+	testMany(b, 10, 100, 10000)
+}
+
+func BenchmarkMs1e0m6(b *testing.B) {
+	b.N = 1000000
+	b.ReportAllocs()
+	testMany(b, 10, 1, 1000000)
+}
+
+func BenchmarkMs0e6m0(b *testing.B) {
+	b.N = 1000000
+	b.ReportAllocs()
+	testMany(b, 1, 1000000, 1)
+}
+
+func BenchmarkMs6e0m0(b *testing.B) {
+	b.N = 1000000
+	b.ReportAllocs()
+	testMany(b, 1000000, 1, 1)
+}

codecov.yml

@@ -1,3 +0,0 @@
-coverage:
-  range: "50...100"
-comment: off

interface.go

@@ -0,0 +1,34 @@
+package event
+
+type SubSettings struct {}
+type SubOption func(*SubSettings)
+
+type EmitterSettings struct {}
+type EmitterOption func(*EmitterSettings)
+
+type Bus interface {
+	// Subscribe creates new subscription. Failing to drain the incoming channel
+	// will cause publishers to get blocked
+	//
+	// evtTypes only accepts typed nil pointers, and uses the type information to
+	// select output type
+	//
+	// Example:
+	// sub, cancel, err := eventbus.Subscribe(new(os.Signal))
+	// defer cancel()
+	//
+	// 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,
+// calls to EmitFunc will block
+//
+// Calling this function with wrong event type will cause a panic
+type EmitFunc func(event interface{})
+
+type CancelFunc func()
+