basic_test.go 7.59 KB
Newer Older
1
package eventbus
Łukasz Magiera's avatar
Łukasz Magiera committed
2 3

import (
Łukasz Magiera's avatar
Łukasz Magiera committed
4
	"fmt"
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
5
	"sync"
Łukasz Magiera's avatar
Łukasz Magiera committed
6
	"sync/atomic"
Łukasz Magiera's avatar
Łukasz Magiera committed
7
	"testing"
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
8
	"time"
Łukasz Magiera's avatar
Łukasz Magiera committed
9

10
	"github.com/libp2p/go-libp2p-testing/race"
Łukasz Magiera's avatar
Łukasz Magiera committed
11 12 13
)

type EventA struct{}
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
14
type EventB int
Łukasz Magiera's avatar
Łukasz Magiera committed
15

Łukasz Magiera's avatar
Łukasz Magiera committed
16 17
func getN() int {
	n := 50000
18
	if race.WithRace() {
Łukasz Magiera's avatar
Łukasz Magiera committed
19 20 21 22 23
		n = 1000
	}
	return n
}

Łukasz Magiera's avatar
Łukasz Magiera committed
24 25 26 27
func (EventA) String() string {
	return "Oh, Hello"
}

Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
28
func TestEmit(t *testing.T) {
Łukasz Magiera's avatar
Łukasz Magiera committed
29
	bus := NewBus()
Łukasz Magiera's avatar
Łukasz Magiera committed
30
	sub, err := bus.Subscribe(new(EventA))
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
31 32 33
	if err != nil {
		t.Fatal(err)
	}
Łukasz Magiera's avatar
Łukasz Magiera committed
34

Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
35
	go func() {
Łukasz Magiera's avatar
Łukasz Magiera committed
36 37
		defer sub.Close()
		<-sub.Out()
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
38 39
	}()

Raúl Kripalani's avatar
Raúl Kripalani committed
40
	em, err := bus.Emitter(new(EventA))
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
41 42 43
	if err != nil {
		t.Fatal(err)
	}
Raúl Kripalani's avatar
Raúl Kripalani committed
44
	defer em.Close()
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
45

Raúl Kripalani's avatar
Raúl Kripalani committed
46
	em.Emit(EventA{})
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
47 48 49 50
}

func TestSub(t *testing.T) {
	bus := NewBus()
Łukasz Magiera's avatar
Łukasz Magiera committed
51
	sub, err := bus.Subscribe(new(EventB))
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
52 53 54 55 56 57 58 59 60 61
	if err != nil {
		t.Fatal(err)
	}

	var event EventB

	var wait sync.WaitGroup
	wait.Add(1)

	go func() {
Łukasz Magiera's avatar
Łukasz Magiera committed
62 63
		defer sub.Close()
		event = (<-sub.Out()).(EventB)
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
64 65 66
		wait.Done()
	}()

Raúl Kripalani's avatar
Raúl Kripalani committed
67
	em, err := bus.Emitter(new(EventB))
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
68 69 70
	if err != nil {
		t.Fatal(err)
	}
Raúl Kripalani's avatar
Raúl Kripalani committed
71
	defer em.Close()
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
72

Raúl Kripalani's avatar
Raúl Kripalani committed
73
	em.Emit(EventB(7))
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
74 75 76 77 78 79 80 81 82 83
	wait.Wait()

	if event != 7 {
		t.Error("got wrong event")
	}
}

func TestEmitNoSubNoBlock(t *testing.T) {
	bus := NewBus()

Raúl Kripalani's avatar
Raúl Kripalani committed
84
	em, err := bus.Emitter(new(EventA))
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
85 86 87
	if err != nil {
		t.Fatal(err)
	}
Raúl Kripalani's avatar
Raúl Kripalani committed
88
	defer em.Close()
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
89

Raúl Kripalani's avatar
Raúl Kripalani committed
90
	em.Emit(EventA{})
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
91 92 93 94 95
}

func TestEmitOnClosed(t *testing.T) {
	bus := NewBus()

Raúl Kripalani's avatar
Raúl Kripalani committed
96
	em, err := bus.Emitter(new(EventA))
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
97 98 99
	if err != nil {
		t.Fatal(err)
	}
Raúl Kripalani's avatar
Raúl Kripalani committed
100
	em.Close()
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
101 102 103 104 105 106 107 108 109 110 111

	defer func() {
		r := recover()
		if r == nil {
			t.Errorf("expected panic")
		}
		if r.(string) != "emitter is closed" {
			t.Error("unexpected message")
		}
	}()

Raúl Kripalani's avatar
Raúl Kripalani committed
112
	em.Emit(EventA{})
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
113 114 115
}

func TestClosingRaces(t *testing.T) {
Łukasz Magiera's avatar
Łukasz Magiera committed
116 117
	subs := getN()
	emits := getN()
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
118 119 120 121 122 123 124

	var wg sync.WaitGroup
	var lk sync.RWMutex
	lk.Lock()

	wg.Add(subs + emits)

Łukasz Magiera's avatar
Łukasz Magiera committed
125
	b := NewBus()
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
126 127 128 129 130 131

	for i := 0; i < subs; i++ {
		go func() {
			lk.RLock()
			defer lk.RUnlock()

Łukasz Magiera's avatar
Łukasz Magiera committed
132
			sub, _ := b.Subscribe(new(EventA))
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
133
			time.Sleep(10 * time.Millisecond)
Łukasz Magiera's avatar
Łukasz Magiera committed
134
			sub.Close()
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
135 136 137 138 139 140 141 142 143

			wg.Done()
		}()
	}
	for i := 0; i < emits; i++ {
		go func() {
			lk.RLock()
			defer lk.RUnlock()

Łukasz Magiera's avatar
Łukasz Magiera committed
144
			emit, _ := b.Emitter(new(EventA))
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
145
			time.Sleep(10 * time.Millisecond)
Łukasz Magiera's avatar
Łukasz Magiera committed
146
			emit.Close()
Łukasz Magiera's avatar
MVP  
Łukasz Magiera committed
147 148 149 150 151 152 153 154 155

			wg.Done()
		}()
	}

	time.Sleep(10 * time.Millisecond)
	lk.Unlock() // start everything

	wg.Wait()
Łukasz Magiera's avatar
Łukasz Magiera committed
156

Raúl Kripalani's avatar
Raúl Kripalani committed
157
	if len(b.(*basicBus).nodes) != 0 {
Łukasz Magiera's avatar
Łukasz Magiera committed
158 159 160 161 162 163 164 165 166
		t.Error("expected no nodes")
	}
}

func TestSubMany(t *testing.T) {
	bus := NewBus()

	var r int32

Łukasz Magiera's avatar
Łukasz Magiera committed
167
	n := getN()
Łukasz Magiera's avatar
Łukasz Magiera committed
168 169 170 171 172 173 174
	var wait sync.WaitGroup
	var ready sync.WaitGroup
	wait.Add(n)
	ready.Add(n)

	for i := 0; i < n; i++ {
		go func() {
Łukasz Magiera's avatar
Łukasz Magiera committed
175
			sub, err := bus.Subscribe(new(EventB))
Łukasz Magiera's avatar
Łukasz Magiera committed
176 177 178
			if err != nil {
				panic(err)
			}
Łukasz Magiera's avatar
Łukasz Magiera committed
179
			defer sub.Close()
Łukasz Magiera's avatar
Łukasz Magiera committed
180 181

			ready.Done()
Łukasz Magiera's avatar
Łukasz Magiera committed
182
			atomic.AddInt32(&r, int32((<-sub.Out()).(EventB)))
Łukasz Magiera's avatar
Łukasz Magiera committed
183 184 185 186
			wait.Done()
		}()
	}

Raúl Kripalani's avatar
Raúl Kripalani committed
187
	em, err := bus.Emitter(new(EventB))
Łukasz Magiera's avatar
Łukasz Magiera committed
188 189 190
	if err != nil {
		t.Fatal(err)
	}
Raúl Kripalani's avatar
Raúl Kripalani committed
191
	defer em.Close()
Łukasz Magiera's avatar
Łukasz Magiera committed
192 193 194

	ready.Wait()

Raúl Kripalani's avatar
Raúl Kripalani committed
195
	em.Emit(EventB(7))
Łukasz Magiera's avatar
Łukasz Magiera committed
196 197
	wait.Wait()

Łukasz Magiera's avatar
Gofmt  
Łukasz Magiera committed
198
	if int(r) != 7*n {
Łukasz Magiera's avatar
Łukasz Magiera committed
199 200 201 202
		t.Error("got wrong result")
	}
}

Łukasz Magiera's avatar
Łukasz Magiera committed
203
func TestSubType(t *testing.T) {
Łukasz Magiera's avatar
Łukasz Magiera committed
204
	bus := NewBus()
Łukasz Magiera's avatar
Łukasz Magiera committed
205
	sub, err := bus.Subscribe([]interface{}{new(EventA), new(EventB)})
Łukasz Magiera's avatar
Łukasz Magiera committed
206 207 208 209 210 211 212 213
	if err != nil {
		t.Fatal(err)
	}

	var event fmt.Stringer

	var wait sync.WaitGroup
	wait.Add(1)
Łukasz Magiera's avatar
Łukasz Magiera committed
214 215

	go func() {
Łukasz Magiera's avatar
Łukasz Magiera committed
216 217
		defer sub.Close()
		event = (<-sub.Out()).(EventA)
Łukasz Magiera's avatar
Łukasz Magiera committed
218
		wait.Done()
Łukasz Magiera's avatar
Łukasz Magiera committed
219 220
	}()

Raúl Kripalani's avatar
Raúl Kripalani committed
221
	em, err := bus.Emitter(new(EventA))
Łukasz Magiera's avatar
Łukasz Magiera committed
222
	if err != nil {
Łukasz Magiera's avatar
Łukasz Magiera committed
223
		t.Fatal(err)
Łukasz Magiera's avatar
Łukasz Magiera committed
224
	}
Raúl Kripalani's avatar
Raúl Kripalani committed
225
	defer em.Close()
Łukasz Magiera's avatar
Łukasz Magiera committed
226

Raúl Kripalani's avatar
Raúl Kripalani committed
227
	em.Emit(EventA{})
Łukasz Magiera's avatar
Łukasz Magiera committed
228
	wait.Wait()
Łukasz Magiera's avatar
Łukasz Magiera committed
229

Łukasz Magiera's avatar
Łukasz Magiera committed
230 231
	if event.String() != "Oh, Hello" {
		t.Error("didn't get the correct message")
Łukasz Magiera's avatar
Łukasz Magiera committed
232
	}
Łukasz Magiera's avatar
Łukasz Magiera committed
233
}
Łukasz Magiera's avatar
Łukasz Magiera committed
234

Łukasz Magiera's avatar
Łukasz Magiera committed
235 236
func TestNonStateful(t *testing.T) {
	bus := NewBus()
Raúl Kripalani's avatar
Raúl Kripalani committed
237
	em, err := bus.Emitter(new(EventB))
Łukasz Magiera's avatar
Łukasz Magiera committed
238 239 240
	if err != nil {
		t.Fatal(err)
	}
Raúl Kripalani's avatar
Raúl Kripalani committed
241
	defer em.Close()
Łukasz Magiera's avatar
Łukasz Magiera committed
242

Łukasz Magiera's avatar
Łukasz Magiera committed
243
	sub1, err := bus.Subscribe(new(EventB), BufSize(1))
Łukasz Magiera's avatar
Łukasz Magiera committed
244 245 246
	if err != nil {
		t.Fatal(err)
	}
Łukasz Magiera's avatar
Łukasz Magiera committed
247
	defer sub1.Close()
Łukasz Magiera's avatar
Łukasz Magiera committed
248 249

	select {
Łukasz Magiera's avatar
Łukasz Magiera committed
250
	case <-sub1.Out():
Łukasz Magiera's avatar
Łukasz Magiera committed
251 252 253 254
		t.Fatal("didn't expect to get an event")
	default:
	}

Raúl Kripalani's avatar
Raúl Kripalani committed
255
	em.Emit(EventB(1))
Łukasz Magiera's avatar
Łukasz Magiera committed
256 257

	select {
Łukasz Magiera's avatar
Łukasz Magiera committed
258 259
	case e := <-sub1.Out():
		if e.(EventB) != 1 {
Łukasz Magiera's avatar
Łukasz Magiera committed
260 261 262 263 264 265
			t.Fatal("got wrong event")
		}
	default:
		t.Fatal("expected to get an event")
	}

Łukasz Magiera's avatar
Łukasz Magiera committed
266
	sub2, err := bus.Subscribe(new(EventB), BufSize(1))
Łukasz Magiera's avatar
Łukasz Magiera committed
267 268 269
	if err != nil {
		t.Fatal(err)
	}
Łukasz Magiera's avatar
Łukasz Magiera committed
270
	defer sub2.Close()
Łukasz Magiera's avatar
Łukasz Magiera committed
271 272

	select {
Łukasz Magiera's avatar
Łukasz Magiera committed
273
	case <-sub2.Out():
Łukasz Magiera's avatar
Łukasz Magiera committed
274 275 276 277 278 279 280
		t.Fatal("didn't expect to get an event")
	default:
	}
}

func TestStateful(t *testing.T) {
	bus := NewBus()
Raúl Kripalani's avatar
Raúl Kripalani committed
281
	em, err := bus.Emitter(new(EventB), Stateful)
Łukasz Magiera's avatar
Łukasz Magiera committed
282 283 284
	if err != nil {
		t.Fatal(err)
	}
Raúl Kripalani's avatar
Raúl Kripalani committed
285
	defer em.Close()
Łukasz Magiera's avatar
Łukasz Magiera committed
286

Raúl Kripalani's avatar
Raúl Kripalani committed
287
	em.Emit(EventB(2))
Łukasz Magiera's avatar
Łukasz Magiera committed
288

Łukasz Magiera's avatar
Łukasz Magiera committed
289
	sub, err := bus.Subscribe(new(EventB), BufSize(1))
Łukasz Magiera's avatar
Łukasz Magiera committed
290 291 292
	if err != nil {
		t.Fatal(err)
	}
Łukasz Magiera's avatar
Łukasz Magiera committed
293
	defer sub.Close()
Łukasz Magiera's avatar
Łukasz Magiera committed
294

Łukasz Magiera's avatar
Łukasz Magiera committed
295
	if (<-sub.Out()).(EventB) != 2 {
Łukasz Magiera's avatar
Łukasz Magiera committed
296 297 298 299 300
		t.Fatal("got wrong event")
	}
}

func testMany(t testing.TB, subs, emits, msgs int, stateful bool) {
301
	if race.WithRace() && subs+emits > 5000 {
Łukasz Magiera's avatar
Łukasz Magiera committed
302 303 304
		t.SkipNow()
	}

Łukasz Magiera's avatar
Łukasz Magiera committed
305 306 307 308 309 310 311 312 313 314 315
	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() {
Łukasz Magiera's avatar
Łukasz Magiera committed
316
			sub, err := bus.Subscribe(new(EventB))
Łukasz Magiera's avatar
Łukasz Magiera committed
317 318 319
			if err != nil {
				panic(err)
			}
Łukasz Magiera's avatar
Łukasz Magiera committed
320
			defer sub.Close()
Łukasz Magiera's avatar
Łukasz Magiera committed
321 322

			ready.Done()
Łukasz Magiera's avatar
Gofmt  
Łukasz Magiera committed
323
			for i := 0; i < emits*msgs; i++ {
Łukasz Magiera's avatar
Łukasz Magiera committed
324 325 326 327 328
				e, ok := <-sub.Out()
				if !ok {
					panic("wat")
				}
				atomic.AddInt64(&r, int64(e.(EventB)))
Łukasz Magiera's avatar
Łukasz Magiera committed
329 330 331 332 333 334 335
			}
			wait.Done()
		}()
	}

	for i := 0; i < emits; i++ {
		go func() {
Raúl Kripalani's avatar
Raúl Kripalani committed
336
			em, err := bus.Emitter(new(EventB), func(settings interface{}) error {
Łukasz Magiera's avatar
Łukasz Magiera committed
337 338
				settings.(*emitterSettings).makeStateful = stateful
				return nil
Łukasz Magiera's avatar
Łukasz Magiera committed
339
			})
Łukasz Magiera's avatar
Łukasz Magiera committed
340 341 342
			if err != nil {
				panic(err)
			}
Raúl Kripalani's avatar
Raúl Kripalani committed
343
			defer em.Close()
Łukasz Magiera's avatar
Łukasz Magiera committed
344 345 346 347

			ready.Wait()

			for i := 0; i < msgs; i++ {
Raúl Kripalani's avatar
Raúl Kripalani committed
348
				em.Emit(EventB(97))
Łukasz Magiera's avatar
Łukasz Magiera committed
349 350 351 352 353 354 355 356
			}

			wait.Done()
		}()
	}

	wait.Wait()

Łukasz Magiera's avatar
Gofmt  
Łukasz Magiera committed
357
	if int(r) != 97*subs*emits*msgs {
Łukasz Magiera's avatar
Łukasz Magiera committed
358 359 360 361 362
		t.Fatal("got wrong result")
	}
}

func TestBothMany(t *testing.T) {
Łukasz Magiera's avatar
Łukasz Magiera committed
363
	testMany(t, 10000, 100, 10, false)
Łukasz Magiera's avatar
Łukasz Magiera committed
364 365
}

Jakub Sztandera's avatar
Jakub Sztandera committed
366 367 368 369
type benchCase struct {
	subs     int
	emits    int
	stateful bool
Łukasz Magiera's avatar
Łukasz Magiera committed
370 371
}

Jakub Sztandera's avatar
Jakub Sztandera committed
372
func (bc benchCase) name() string {
Jakub Sztandera's avatar
Jakub Sztandera committed
373
	return fmt.Sprintf("subs-%03d/emits-%03d/stateful-%t", bc.subs, bc.emits, bc.stateful)
Łukasz Magiera's avatar
Łukasz Magiera committed
374 375
}

Jakub Sztandera's avatar
Jakub Sztandera committed
376 377 378
func genTestCases() []benchCase {
	ret := make([]benchCase, 0, 200)
	for stateful := 0; stateful < 2; stateful++ {
Jakub Sztandera's avatar
Jakub Sztandera committed
379 380
		for subs := uint(0); subs <= 8; subs = subs + 4 {
			for emits := uint(0); emits <= 8; emits = emits + 4 {
Jakub Sztandera's avatar
Jakub Sztandera committed
381 382 383 384 385
				ret = append(ret, benchCase{1 << subs, 1 << emits, stateful == 1})
			}
		}
	}
	return ret
Łukasz Magiera's avatar
Łukasz Magiera committed
386 387
}

Jakub Sztandera's avatar
Jakub Sztandera committed
388 389 390 391
func BenchmarkEvents(b *testing.B) {
	for _, bc := range genTestCases() {
		b.Run(bc.name(), benchMany(bc))
	}
Łukasz Magiera's avatar
Łukasz Magiera committed
392 393
}

Jakub Sztandera's avatar
Jakub Sztandera committed
394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424
func benchMany(bc benchCase) func(*testing.B) {
	return func(b *testing.B) {
		b.ReportAllocs()
		subs := bc.subs
		emits := bc.emits
		stateful := bc.stateful
		bus := NewBus()
		var wait sync.WaitGroup
		var ready sync.WaitGroup
		wait.Add(subs + emits)
		ready.Add(subs + emits)

		for i := 0; i < subs; i++ {
			go func() {
				sub, err := bus.Subscribe(new(EventB))
				if err != nil {
					panic(err)
				}
				defer sub.Close()

				ready.Done()
				ready.Wait()
				for i := 0; i < (b.N/emits)*emits; i++ {
					_, ok := <-sub.Out()
					if !ok {
						panic("wat")
					}
				}
				wait.Done()
			}()
		}
Łukasz Magiera's avatar
Łukasz Magiera committed
425

Jakub Sztandera's avatar
Jakub Sztandera committed
426 427 428 429 430 431 432 433 434 435
		for i := 0; i < emits; i++ {
			go func() {
				em, err := bus.Emitter(new(EventB), func(settings interface{}) error {
					settings.(*emitterSettings).makeStateful = stateful
					return nil
				})
				if err != nil {
					panic(err)
				}
				defer em.Close()
Łukasz Magiera's avatar
Łukasz Magiera committed
436

Jakub Sztandera's avatar
Jakub Sztandera committed
437 438
				ready.Done()
				ready.Wait()
Łukasz Magiera's avatar
Łukasz Magiera committed
439

Jakub Sztandera's avatar
Jakub Sztandera committed
440 441 442
				for i := 0; i < b.N/emits; i++ {
					em.Emit(EventB(97))
				}
Łukasz Magiera's avatar
Łukasz Magiera committed
443

Jakub Sztandera's avatar
Jakub Sztandera committed
444 445 446 447 448 449 450
				wait.Done()
			}()
		}
		ready.Wait()
		b.ResetTimer()
		wait.Wait()
	}
Łukasz Magiera's avatar
Łukasz Magiera committed
451
}
452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484

var div = 100

func BenchmarkSubscribe(b *testing.B) {
	b.ReportAllocs()
	for i := 0; i < b.N/div; i++ {
		bus := NewBus()
		for j := 0; j < div; j++ {
			bus.Subscribe(new(EventA))
		}
	}
}

func BenchmarkEmitter(b *testing.B) {
	b.ReportAllocs()
	for i := 0; i < b.N/div; i++ {
		bus := NewBus()
		for j := 0; j < div; j++ {
			bus.Emitter(new(EventA))
		}
	}
}

func BenchmarkSubscribeAndEmitter(b *testing.B) {
	b.ReportAllocs()
	for i := 0; i < b.N/div; i++ {
		bus := NewBus()
		for j := 0; j < div; j++ {
			bus.Subscribe(new(EventA))
			bus.Emitter(new(EventA))
		}
	}
}