events/loop: Factory & reduce lock contention (#140)

Update the loop implementation to use an optimistic RLock for
enqueueing, falling back to a full Lock only when the queue segment is
full. This reduces lock contention.

Uses a generic typed Factory to allow for the loop to be sync.Pool
cached by the consumer, on demand. The loop does not cache itself as the
loop may continue to be used after being closed.

Signed-off-by: joshvanl <[email protected]>

Author: JoshVanL

events/loop/factory.go

@@ -0,0 +1,61 @@
+/*
+Copyright 2025 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package loop
+
+import "sync"
+
+type Factory[T any] struct {
+	size uint64
+
+	segPool  sync.Pool
+	loopPool sync.Pool
+}
+
+func New[T any](size uint64) *Factory[T] {
+	f := &Factory[T]{
+		size: size,
+		segPool: sync.Pool{
+			New: func() any {
+				return new(queueSegment[T])
+			},
+		},
+	}
+
+	f.loopPool = sync.Pool{
+		New: func() any {
+			return &loop[T]{
+				factory: f,
+			}
+		},
+	}
+
+	return f
+}
+
+func (f *Factory[T]) NewLoop(h Handler[T]) Interface[T] {
+	l := f.loopPool.Get().(*loop[T])
+
+	seg := l.getSegment()
+	l.head = seg
+	l.tail = seg
+	l.closeCh = make(chan struct{})
+	l.handler = h
+	l.closed.Store(false)
+
+	return l
+}
+
+func (f *Factory[T]) CacheLoop(l Interface[T]) {
+	f.loopPool.Put(l)
+}

events/loop/loop.go

@@ -16,6 +16,7 @@ package loop
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 )
 
 type Handler[T any] interface {
@@ -26,32 +27,21 @@ type Interface[T any] interface {
 	Run(ctx context.Context) error
 	Enqueue(t T)
 	Close(t T)
-	Reset(h Handler[T], size uint64) Interface[T]
 }
 
 type loop[T any] struct {
+	factory *Factory[T]
+
 	head *queueSegment[T]
 	tail *queueSegment[T]
 
-	segSize uint64
-
 	handler Handler[T]
 
-	closed  bool
-	closeCh chan struct{}
-
-	lock sync.Mutex
-
-	segPool sync.Pool
-}
+	closed atomic.Bool
 
-func New[T any](h Handler[T], size uint64) Interface[T] {
-	l := new(loop[T])
-	return l.Reset(h, size)
-}
+	closeCh chan struct{}
 
-func Empty[T any]() Interface[T] {
-	return new(loop[T])
+	lock sync.RWMutex
 }
 
 func (l *loop[T]) Run(ctx context.Context) error {
@@ -78,23 +68,37 @@ func (l *loop[T]) Run(ctx context.Context) error {
 }
 
 func (l *loop[T]) Enqueue(req T) {
-	l.lock.Lock()
-	defer l.lock.Unlock()
+	l.lock.RLock()
 
-	if l.closed {
+	if l.closed.Load() {
+		l.lock.RUnlock()
 		return
 	}
 
-	if l.tail == nil {
-		seg := l.getSegment()
-		l.head = seg
-		l.tail = seg
-	}
-
 	// First try to send to the current tail segment without blocking.
 	select {
 	case l.tail.ch <- req:
+		l.lock.RUnlock()
 		return
+	default:
+		l.lock.RUnlock()
+	}
+
+	// Tail is full; need to acquire write lock to roll over. If no longer full
+	// (lost race, another goroutine rolled over first), don't expand.
+
+	l.lock.Lock()
+	defer l.lock.Unlock()
+
+	if l.closed.Load() {
+		// Closed while we were waiting for the lock.
+		return
+	}
+
+	// Try again to send to the tail; if successful, another goroutine must
+	// have rolled over for us.
+	select {
+	case l.tail.ch <- req:
 	default:
 		// Tail is full: create a new segment, link it, close the old tail, and
 		// send into the new tail.
@@ -108,20 +112,13 @@ func (l *loop[T]) Enqueue(req T) {
 
 func (l *loop[T]) Close(req T) {
 	l.lock.Lock()
-	if l.closed {
+	if l.closed.Load() {
 		// Already closed; just unlock and wait for Run to finish.
 		l.lock.Unlock()
 		<-l.closeCh
 		return
 	}
-	l.closed = true
-
-	// Ensure at least one segment exists.
-	if l.tail == nil {
-		seg := l.getSegment()
-		l.head = seg
-		l.tail = seg
-	}
+	l.closed.Store(true)
 
 	// Enqueue the final request; if the tail is full, roll over as in Enqueue.
 	select {
@@ -141,28 +138,3 @@ func (l *loop[T]) Close(req T) {
 	// Wait for Run to finish draining everything.
 	<-l.closeCh
 }
-
-func (l *loop[T]) Reset(h Handler[T], size uint64) Interface[T] {
-	if l == nil {
-		return New[T](h, size)
-	}
-
-	l.lock.Lock()
-	defer l.lock.Unlock()
-
-	l.closed = false
-	l.closeCh = make(chan struct{})
-	l.handler = h
-	l.segSize = size
-
-	// Initialize pool for this instantiation of T.
-	l.segPool.New = func() any {
-		return new(queueSegment[T])
-	}
-
-	seg := l.getSegment()
-	l.head = seg
-	l.tail = seg
-
-	return l
-}

events/loop/loop_test.go

@@ -61,7 +61,8 @@ func TestLoop_EnqueueAndRunOrder_Unbounded(t *testing.T) {
 	h := &testHandler[int]{}
 	const segmentSize = 4
 
-	l := New[int](h, segmentSize)
+	f := New[int](segmentSize)
+	l := f.NewLoop(h)
 
 	var wg sync.WaitGroup
 	wg.Add(1)
@@ -103,7 +104,8 @@ func TestLoop_CloseTwiceIsSafe(t *testing.T) {
 	defer cancel()
 
 	h := &testHandler[int]{}
-	l := New[int](h, 2)
+	f := New[int](2)
+	l := f.NewLoop(h)
 
 	var wg sync.WaitGroup
 	wg.Add(1)
@@ -146,7 +148,8 @@ func TestLoop_Reset(t *testing.T) {
 	defer cancel()
 
 	h1 := &testHandler[int]{}
-	l := New[int](h1, 2)
+	f := New[int](2)
+	l := f.NewLoop(h1)
 
 	var wg1 sync.WaitGroup
 	wg1.Add(1)
@@ -167,7 +170,8 @@ func TestLoop_Reset(t *testing.T) {
 
 	// Reset to a new handler and buffer size.
 	h2 := &testHandler[int]{}
-	l = l.Reset(h2, 8)
+	f = New[int](8)
+	l = f.NewLoop(h2)
 
 	require.NotNil(t, l)
 
@@ -195,7 +199,8 @@ func TestLoop_EnqueueAfterCloseIsDropped(t *testing.T) {
 	defer cancel()
 
 	h := &testHandler[int]{}
-	l := New[int](h, 2)
+	f := New[int](2)
+	l := f.NewLoop(h)
 
 	var wg sync.WaitGroup
 	wg.Add(1)

events/loop/segment.go

@@ -25,16 +25,10 @@ type queueSegment[T any] struct {
 // getSegment gets a queueSegment from the pool or allocates a new one.
 // It always initializes a fresh channel of the configured size.
 func (l *loop[T]) getSegment() *queueSegment[T] {
-	if l.segPool.New == nil {
-		l.segPool.New = func() any {
-			return new(queueSegment[T])
-		}
-	}
-
-	seg := l.segPool.Get().(*queueSegment[T])
+	seg := l.factory.segPool.Get().(*queueSegment[T])
 	seg.next = nil
 
-	segSize := l.segSize
+	segSize := l.factory.size
 	if segSize == 0 {
 		segSize = 1
 	}
@@ -50,5 +44,5 @@ func (l *loop[T]) putSegment(seg *queueSegment[T]) {
 	}
 	seg.ch = nil
 	seg.next = nil
-	l.segPool.Put(seg)
+	l.factory.segPool.Put(seg)
 }