diff --git a/broadcast.go b/broadcast.go index c81f600..ef8dd3f 100644 --- a/broadcast.go +++ b/broadcast.go @@ -2,6 +2,7 @@ package events import ( "fmt" + "slices" "sync" "github.com/sirupsen/logrus" @@ -105,55 +106,46 @@ func (b *Broadcaster) Close() error { // Close is called, this goroutine will exit. func (b *Broadcaster) run() { defer close(b.closed) - remove := func(target Sink) { - for i, sink := range b.sinks { - if sink == target { - b.sinks = append(b.sinks[:i], b.sinks[i+1:]...) - break - } - } - } for { select { case event := <-b.events: - for _, sink := range b.sinks { + for i := 0; i < len(b.sinks); { + sink := b.sinks[i] if err := sink.Write(event); err != nil { if err == ErrSinkClosed { - // remove closed sinks - remove(sink) + b.sinks = slices.Delete(b.sinks, i, i+1) continue } - logrus.WithField("event", event).WithField("events.sink", sink).WithError(err). - Errorf("broadcaster: dropping event") + logrus.WithFields(logrus.Fields{ + "error": err, + "event": event, + "events.sink": sink, + }).Error("broadcaster: dropping event") } + i++ } case request := <-b.adds: // while we have to iterate for add/remove, common iteration for // send is faster against slice. - - var found bool - for _, sink := range b.sinks { - if request.sink == sink { - found = true - break - } - } - - if !found { + // b.sinks[request.sink] = struct{}{} + if !slices.Contains(b.sinks, request.sink) { b.sinks = append(b.sinks, request.sink) } - // b.sinks[request.sink] = struct{}{} request.response <- nil case request := <-b.removes: - remove(request.sink) + if i := slices.Index(b.sinks, request.sink); i >= 0 { + b.sinks = slices.Delete(b.sinks, i, i+1) + } request.response <- nil case <-b.shutdown: // close all the underlying sinks for _, sink := range b.sinks { if err := sink.Close(); err != nil && err != ErrSinkClosed { - logrus.WithField("events.sink", sink).WithError(err). - Errorf("broadcaster: closing sink failed") + logrus.WithFields(logrus.Fields{ + "error": err, + "events.sink": sink, + }).Error("broadcaster: closing sink failed") } } return @@ -164,8 +156,7 @@ func (b *Broadcaster) run() { func (b *Broadcaster) String() string { // Serialize copy of this broadcaster without the sync.Once, to avoid // a data race. - - b2 := map[string]interface{}{ + return fmt.Sprint(map[string]any{ "sinks": b.sinks, "events": b.events, "adds": b.adds, @@ -173,7 +164,5 @@ func (b *Broadcaster) String() string { "shutdown": b.shutdown, "closed": b.closed, - } - - return fmt.Sprint(b2) + }) } diff --git a/broadcast_test.go b/broadcast_test.go index 5108dd9..5afed3f 100644 --- a/broadcast_test.go +++ b/broadcast_test.go @@ -10,7 +10,7 @@ func TestBroadcaster(t *testing.T) { const nEvents = 1000 var sinks []Sink b := NewBroadcaster() - for i := 0; i < 10; i++ { + for i := range 10 { sinks = append(sinks, newTestSink(t, nEvents)) err := b.Add(sinks[i]) if err != nil { @@ -108,7 +108,7 @@ func benchmarkBroadcast(b *testing.B, nsinks int) { b.StopTimer() var sinks []Sink - for i := 0; i < nsinks; i++ { + for range nsinks { // counter.Inc(1) sinks = append(sinks, newTestSink(b, b.N)) // sinks = append(sinks, NewQueue(&testSink{t: b, expected: b.N})) diff --git a/channel.go b/channel.go index 802cf51..0eb7287 100644 --- a/channel.go +++ b/channel.go @@ -53,9 +53,8 @@ func (ch *Channel) Close() error { func (ch *Channel) String() string { // Serialize a copy of the Channel that doesn't contain the sync.Once, // to avoid a data race. - ch2 := map[string]interface{}{ + return fmt.Sprint(map[string]any{ "C": ch.C, "closed": ch.closed, - } - return fmt.Sprint(ch2) + }) } diff --git a/common_test.go b/common_test.go index 15e125e..e27053e 100644 --- a/common_test.go +++ b/common_test.go @@ -2,15 +2,15 @@ package events import ( "fmt" - "math/rand" + "math/rand/v2" "sync" "testing" "time" ) type tOrB interface { - Fatalf(format string, args ...interface{}) - Logf(format string, args ...interface{}) + Fatalf(format string, args ...any) + Logf(format string, args ...any) } type testSink struct { diff --git a/event.go b/event.go index f0f1d9e..8cd997a 100644 --- a/event.go +++ b/event.go @@ -1,7 +1,7 @@ package events // Event marks items that can be sent as events. -type Event interface{} +type Event any // Sink accepts and sends events. type Sink interface { diff --git a/filter_test.go b/filter_test.go index ccb46c1..18ccd8a 100644 --- a/filter_test.go +++ b/filter_test.go @@ -10,7 +10,7 @@ func TestFilter(t *testing.T) { return ok && i%2 == 0 })) - for i := 0; i < nevents; i++ { + for i := range nevents { if err := filter.Write(i); err != nil { t.Fatalf("unexpected error writing event: %v", err) } diff --git a/queue.go b/queue.go index 4bb770a..b286405 100644 --- a/queue.go +++ b/queue.go @@ -103,6 +103,7 @@ func (eq *Queue) next() Event { eq.cond.Wait() } + // Len is non-zero while holding eq.mu, so Front cannot be nil. front := eq.events.Front() block := front.Value.(Event) eq.events.Remove(front) diff --git a/retry.go b/retry.go index 9f75e23..70fad9c 100644 --- a/retry.go +++ b/retry.go @@ -2,7 +2,7 @@ package events import ( "fmt" - "math/rand" + "math/rand/v2" "sync" "sync/atomic" "time" @@ -26,13 +26,11 @@ type RetryingSink struct { // off on failure. Parameters threshold and backoff adjust the behavior of the // circuit breaker. func NewRetryingSink(sink Sink, strategy RetryStrategy) *RetryingSink { - rs := &RetryingSink{ + return &RetryingSink{ sink: sink, strategy: strategy, closed: make(chan struct{}), } - - return rs } // Write attempts to flush the events to the downstream sink until it succeeds @@ -66,14 +64,12 @@ retry: return err } - logger := logger.WithError(err) // shadow!! - if rs.strategy.Failure(event, err) { - logger.Errorf("retryingsink: dropped event") + logger.WithError(err).Error("retryingsink: dropped event") return nil } - logger.Errorf("retryingsink: error writing event, retrying") + logger.WithError(err).Error("retryingsink: error writing event, retrying") goto retry } @@ -93,7 +89,7 @@ func (rs *RetryingSink) Close() error { func (rs *RetryingSink) String() string { // Serialize a copy of the RetryingSink without the sync.Once, to avoid // a data race. - rs2 := map[string]interface{}{ + rs2 := map[string]any{ "sink": rs.sink, "strategy": rs.strategy, "closed": rs.closed, @@ -201,7 +197,7 @@ type ExponentialBackoffConfig struct { // ExponentialBackoff implements random backoff with exponentially increasing // bounds as the number consecutive failures increase. type ExponentialBackoff struct { - failures uint64 // consecutive failure counter (needs to be 64-bit aligned) + failures atomic.Uint64 // consecutive failure counter (needs to be 64-bit aligned) config ExponentialBackoffConfig } @@ -215,17 +211,17 @@ func NewExponentialBackoff(config ExponentialBackoffConfig) *ExponentialBackoff // Proceed returns the next randomly bound exponential backoff time. func (b *ExponentialBackoff) Proceed(event Event) time.Duration { - return b.backoff(atomic.LoadUint64(&b.failures)) + return b.backoff(b.failures.Load()) } // Success resets the failures counter. func (b *ExponentialBackoff) Success(event Event) { - atomic.StoreUint64(&b.failures, 0) + b.failures.Store(0) } // Failure increments the failure counter. func (b *ExponentialBackoff) Failure(event Event, err error) bool { - atomic.AddUint64(&b.failures, 1) + b.failures.Add(1) return false } @@ -242,17 +238,16 @@ func (b *ExponentialBackoff) backoff(failures uint64) time.Duration { factor = DefaultExponentialBackoffConfig.Factor } - backoff := b.config.Base + factor*time.Duration(1<<(failures-1)) - - max := b.config.Max - if max <= 0 { - max = DefaultExponentialBackoffConfig.Max + maxBackoff := b.config.Max + if maxBackoff <= 0 { + maxBackoff = DefaultExponentialBackoffConfig.Max } - if backoff > max || backoff < 0 { - backoff = max + backoff := b.config.Base + factor*time.Duration(1<<(failures-1)) + if backoff > maxBackoff || backoff < 0 { + backoff = maxBackoff } // Choose a uniformly distributed value from [0, backoff). - return time.Duration(rand.Int63n(int64(backoff))) + return rand.N(backoff) } diff --git a/retry_test.go b/retry_test.go index 1da4618..57ef092 100644 --- a/retry_test.go +++ b/retry_test.go @@ -84,7 +84,7 @@ func TestExponentialBackoff(t *testing.T) { t.Errorf("no facilities for dropping events in ExponentialBackoff") } - for j := 0; j < 1000; j++ { + for range 1000 { // sample this several thousand times. backoff := strategy.Proceed(nil) if backoff > expected { @@ -92,10 +92,7 @@ func TestExponentialBackoff(t *testing.T) { } } - expected = strategy.config.Base + strategy.config.Factor*time.Duration(1< strategy.config.Max { - expected = strategy.config.Max - } + expected = min(strategy.config.Base+strategy.config.Factor*time.Duration(1<