123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453 |
- // Copyright (C) 2014 The Syncthing Authors.
- //
- // This Source Code Form is subject to the terms of the Mozilla Public
- // License, v. 2.0. If a copy of the MPL was not distributed with this file,
- // You can obtain one at https://mozilla.org/MPL/2.0/.
- package events
- import (
- "context"
- "encoding/json"
- "fmt"
- "sync"
- "testing"
- "time"
- )
- const timeout = time.Second
- func init() {
- runningTests = true
- }
- func TestNewLogger(t *testing.T) {
- l := NewLogger()
- if l == nil {
- t.Fatal("Unexpected nil Logger")
- }
- }
- func setupLogger() (Logger, context.CancelFunc) {
- ctx, cancel := context.WithCancel(context.Background())
- l := NewLogger()
- go l.Serve(ctx)
- return l, cancel
- }
- func TestSubscriber(t *testing.T) {
- l, cancel := setupLogger()
- defer cancel()
- s := l.Subscribe(0)
- defer s.Unsubscribe()
- if s == nil {
- t.Fatal("Unexpected nil Subscription")
- }
- }
- func TestTimeout(t *testing.T) {
- l, cancel := setupLogger()
- defer cancel()
- s := l.Subscribe(0)
- defer s.Unsubscribe()
- _, err := s.Poll(timeout)
- if err != ErrTimeout {
- t.Fatal("Unexpected non-Timeout error:", err)
- }
- }
- func TestEventBeforeSubscribe(t *testing.T) {
- l, cancel := setupLogger()
- defer cancel()
- l.Log(DeviceConnected, "foo")
- s := l.Subscribe(0)
- defer s.Unsubscribe()
- _, err := s.Poll(timeout)
- if err != ErrTimeout {
- t.Fatal("Unexpected non-Timeout error:", err)
- }
- }
- func TestEventAfterSubscribe(t *testing.T) {
- l, cancel := setupLogger()
- defer cancel()
- s := l.Subscribe(AllEvents)
- defer s.Unsubscribe()
- l.Log(DeviceConnected, "foo")
- ev, err := s.Poll(timeout)
- if err != nil {
- t.Fatal("Unexpected error:", err)
- }
- if ev.Type != DeviceConnected {
- t.Error("Incorrect event type", ev.Type)
- }
- switch v := ev.Data.(type) {
- case string:
- if v != "foo" {
- t.Error("Incorrect Data string", v)
- }
- default:
- t.Errorf("Incorrect Data type %#v", v)
- }
- }
- func TestEventAfterSubscribeIgnoreMask(t *testing.T) {
- l, cancel := setupLogger()
- defer cancel()
- s := l.Subscribe(DeviceDisconnected)
- defer s.Unsubscribe()
- l.Log(DeviceConnected, "foo")
- _, err := s.Poll(timeout)
- if err != ErrTimeout {
- t.Fatal("Unexpected non-Timeout error:", err)
- }
- }
- func TestBufferOverflow(t *testing.T) {
- l, cancel := setupLogger()
- defer cancel()
- s := l.Subscribe(AllEvents)
- defer s.Unsubscribe()
- // The first BufferSize events will be logged pretty much
- // instantaneously. The next BufferSize events will each block for up to
- // 15ms, plus overhead from race detector and thread scheduling latency
- // etc. This latency can sometimes be significant and is incurred for
- // each call. We just verify that the whole test completes in a
- // reasonable time, taking no more than 15 seconds in total.
- t0 := time.Now()
- const nEvents = BufferSize * 2
- for i := 0; i < nEvents; i++ {
- l.Log(DeviceConnected, "foo")
- }
- if d := time.Since(t0); d > 15*time.Second {
- t.Fatal("Logging took too long,", d, "avg", d/nEvents, "expected <", eventLogTimeout)
- }
- }
- func TestUnsubscribe(t *testing.T) {
- l, cancel := setupLogger()
- defer cancel()
- s := l.Subscribe(AllEvents)
- l.Log(DeviceConnected, "foo")
- _, err := s.Poll(timeout)
- if err != nil {
- t.Fatal("Unexpected error:", err)
- }
- s.Unsubscribe()
- l.Log(DeviceConnected, "foo")
- _, err = s.Poll(timeout)
- if err != ErrClosed {
- t.Fatal("Unexpected non-Closed error:", err)
- }
- }
- func TestGlobalIDs(t *testing.T) {
- l, cancel := setupLogger()
- defer cancel()
- s := l.Subscribe(AllEvents)
- defer s.Unsubscribe()
- l.Log(DeviceConnected, "foo")
- l.Subscribe(AllEvents)
- l.Log(DeviceConnected, "bar")
- ev, err := s.Poll(timeout)
- if err != nil {
- t.Fatal("Unexpected error:", err)
- }
- if ev.Data.(string) != "foo" {
- t.Fatal("Incorrect event:", ev)
- }
- id := ev.GlobalID
- ev, err = s.Poll(timeout)
- if err != nil {
- t.Fatal("Unexpected error:", err)
- }
- if ev.Data.(string) != "bar" {
- t.Fatal("Incorrect event:", ev)
- }
- if ev.GlobalID != id+1 {
- t.Fatalf("ID not incremented (%d != %d)", ev.GlobalID, id+1)
- }
- }
- func TestSubscriptionIDs(t *testing.T) {
- l, cancel := setupLogger()
- defer cancel()
- s := l.Subscribe(DeviceConnected)
- defer s.Unsubscribe()
- l.Log(DeviceDisconnected, "a")
- l.Log(DeviceConnected, "b")
- l.Log(DeviceConnected, "c")
- l.Log(DeviceDisconnected, "d")
- ev, err := s.Poll(timeout)
- if err != nil {
- t.Fatal("Unexpected error:", err)
- }
- if ev.GlobalID != 2 {
- t.Fatal("Incorrect GlobalID:", ev.GlobalID)
- }
- if ev.SubscriptionID != 1 {
- t.Fatal("Incorrect SubscriptionID:", ev.SubscriptionID)
- }
- ev, err = s.Poll(timeout)
- if err != nil {
- t.Fatal("Unexpected error:", err)
- }
- if ev.GlobalID != 3 {
- t.Fatal("Incorrect GlobalID:", ev.GlobalID)
- }
- if ev.SubscriptionID != 2 {
- t.Fatal("Incorrect SubscriptionID:", ev.SubscriptionID)
- }
- ev, err = s.Poll(timeout)
- if err != ErrTimeout {
- t.Fatal("Unexpected error:", err)
- }
- }
- func TestBufferedSub(t *testing.T) {
- l, cancel := setupLogger()
- defer cancel()
- s := l.Subscribe(AllEvents)
- defer s.Unsubscribe()
- bs := NewBufferedSubscription(s, 10*BufferSize)
- go func() {
- for i := 0; i < 10*BufferSize; i++ {
- l.Log(DeviceConnected, fmt.Sprintf("event-%d", i))
- if i%30 == 0 {
- // Give the buffer routine time to pick up the events
- time.Sleep(20 * time.Millisecond)
- }
- }
- }()
- recv := 0
- for recv < 10*BufferSize {
- evs := bs.Since(recv, nil, time.Minute)
- for _, ev := range evs {
- if ev.GlobalID != recv+1 {
- t.Fatalf("Incorrect ID; %d != %d", ev.GlobalID, recv+1)
- }
- recv = ev.GlobalID
- }
- }
- }
- func BenchmarkBufferedSub(b *testing.B) {
- l, cancel := setupLogger()
- defer cancel()
- s := l.Subscribe(AllEvents)
- defer s.Unsubscribe()
- bufferSize := BufferSize
- bs := NewBufferedSubscription(s, bufferSize)
- // The coord channel paces the sender according to the receiver,
- // ensuring that no events are dropped. The benchmark measures sending +
- // receiving + synchronization overhead.
- coord := make(chan struct{}, bufferSize)
- for i := 0; i < bufferSize-1; i++ {
- coord <- struct{}{}
- }
- // Receive the events
- done := make(chan error)
- go func() {
- recv := 0
- var evs []Event
- for i := 0; i < b.N; {
- evs = bs.Since(recv, evs[:0], time.Minute)
- for _, ev := range evs {
- if ev.GlobalID != recv+1 {
- done <- fmt.Errorf("skipped event %v %v", ev.GlobalID, recv)
- return
- }
- recv = ev.GlobalID
- coord <- struct{}{}
- }
- i += len(evs)
- }
- done <- nil
- }()
- // Send the events
- eventData := map[string]string{
- "foo": "bar",
- "other": "data",
- "and": "something else",
- }
- for i := 0; i < b.N; i++ {
- l.Log(DeviceConnected, eventData)
- <-coord
- }
- if err := <-done; err != nil {
- b.Error(err)
- }
- b.ReportAllocs()
- }
- func TestSinceUsesSubscriptionId(t *testing.T) {
- l, cancel := setupLogger()
- defer cancel()
- s := l.Subscribe(DeviceConnected)
- defer s.Unsubscribe()
- bs := NewBufferedSubscription(s, 10*BufferSize)
- l.Log(DeviceConnected, "a") // SubscriptionID = 1
- l.Log(DeviceDisconnected, "b")
- l.Log(DeviceDisconnected, "c")
- l.Log(DeviceConnected, "d") // SubscriptionID = 2
- // We need to loop for the events, as they may not all have been
- // delivered to the buffered subscription when we get here.
- t0 := time.Now()
- for time.Since(t0) < time.Second {
- events := bs.Since(0, nil, time.Minute)
- if len(events) == 2 {
- break
- }
- if len(events) > 2 {
- t.Fatal("Incorrect number of events:", len(events))
- }
- }
- events := bs.Since(1, nil, time.Minute)
- if len(events) != 1 {
- t.Fatal("Incorrect number of events:", len(events))
- }
- }
- func TestUnmarshalEvent(t *testing.T) {
- var event Event
- s := `
- {
- "id": 1,
- "globalID": 1,
- "time": "2006-01-02T15:04:05.999999999Z",
- "type": "Starting",
- "data": {}
- }`
- if err := json.Unmarshal([]byte(s), &event); err != nil {
- t.Fatal("Failed to unmarshal event:", err)
- }
- }
- func TestUnsubscribeContention(t *testing.T) {
- // Check that we can unsubscribe without blocking the whole system.
- const (
- listeners = 50
- senders = 1000
- )
- l, cancel := setupLogger()
- defer cancel()
- // Start listeners. These will poll until the stop channel is closed,
- // then exit and unsubscribe.
- stopListeners := make(chan struct{})
- var listenerWg sync.WaitGroup
- listenerWg.Add(listeners)
- for i := 0; i < listeners; i++ {
- go func() {
- defer listenerWg.Done()
- s := l.Subscribe(AllEvents)
- defer s.Unsubscribe()
- for {
- select {
- case <-s.C():
- case <-stopListeners:
- return
- }
- }
- }()
- }
- // Start senders. These send pointless events until the stop channel is
- // closed.
- stopSenders := make(chan struct{})
- defer close(stopSenders)
- var senderWg sync.WaitGroup
- senderWg.Add(senders)
- for i := 0; i < senders; i++ {
- go func() {
- defer senderWg.Done()
- t := time.NewTicker(time.Millisecond)
- for {
- select {
- case <-t.C:
- l.Log(StateChanged, nil)
- case <-stopSenders:
- return
- }
- }
- }()
- }
- // Give everything time to start up.
- time.Sleep(time.Second)
- // Stop the listeners and wait for them to exit. This should happen in a
- // reasonable time frame.
- t0 := time.Now()
- close(stopListeners)
- listenerWg.Wait()
- if d := time.Since(t0); d > time.Minute {
- t.Error("It should not take", d, "to unsubscribe from an event stream")
- }
- }
- func BenchmarkLogEvent(b *testing.B) {
- l, cancel := setupLogger()
- defer cancel()
- s := l.Subscribe(AllEvents)
- defer s.Unsubscribe()
- NewBufferedSubscription(s, 1) // runs in the background
- for i := 0; i < b.N; i++ {
- l.Log(StateChanged, nil)
- }
- }
|