fix potential race in listeners and add tests for that case

Author: AmrSaber

dequeue.go

@@ -14,20 +14,27 @@ type BlockingDequeue[T any] struct {
 
 	writeCond    *sync.Cond    // condition used to lock and notify about writing to the encapsulated list
 	capacityLock *sync.RWMutex // lock used to protect the capacity
+	onFullLock   *sync.RWMutex // lock used to protect the onFull callback
+	onEmptyLock  *sync.RWMutex // lock used to protect the onEmpty callback
 
 	capacity int
 
-	OnFull  func() // Optional callback function invoked when the dequeue is full
-	OnEmpty func() // Optional callback function invoked when the dequeue is empty
+	onFull  func() // Optional callback function invoked when the dequeue is full
+	onEmpty func() // Optional callback function invoked when the dequeue is empty
 }
 
 // Creates a new blocking dequeue with infinite capacity.
 // The dequeue MUST only be created using this method.
 func NewBlockingDequeue[T any]() *BlockingDequeue[T] {
 	d := new(BlockingDequeue[T])
 	d.list = list.New()
+
 	d.writeCond = sync.NewCond(&sync.Mutex{})
+
 	d.capacityLock = &sync.RWMutex{}
+	d.onFullLock = &sync.RWMutex{}
+	d.onEmptyLock = &sync.RWMutex{}
+
 	return d
 }
 
@@ -47,9 +54,12 @@ func (d *BlockingDequeue[T]) PushFront(item T) {
 	// Notify the consumer that an item has been added
 	defer d.writeCond.Broadcast()
 
+	d.onFullLock.RLock()
+	defer d.onFullLock.RUnlock()
+
 	// Call the OnFull callback if the dequeue is full
-	if d.isFull_unsafe() && d.OnFull != nil {
-		d.OnFull()
+	if d.isFull_unsafe() && d.onFull != nil {
+		d.onFull()
 	}
 }
 
@@ -68,8 +78,11 @@ func (d *BlockingDequeue[T]) PushBack(item T) {
 	defer d.writeCond.Broadcast()
 
 	// Call the OnFull callback if the dequeue is full
-	if d.isFull_unsafe() && d.OnFull != nil {
-		d.OnFull()
+	d.onFullLock.RLock()
+	defer d.onFullLock.RUnlock()
+
+	if d.isFull_unsafe() && d.onFull != nil {
+		d.onFull()
 	}
 }
 
@@ -88,8 +101,11 @@ func (d *BlockingDequeue[T]) PopFront() T {
 	defer d.writeCond.Broadcast()
 
 	// Call the OnEmpty callback if the dequeue is empty
-	if d.isEmpty_unsafe() && d.OnEmpty != nil {
-		d.OnEmpty()
+	d.onEmptyLock.RLock()
+	defer d.onEmptyLock.RUnlock()
+
+	if d.isEmpty_unsafe() && d.onEmpty != nil {
+		d.onEmpty()
 	}
 
 	return item
@@ -110,8 +126,11 @@ func (d *BlockingDequeue[T]) PopBack() T {
 	defer d.writeCond.Broadcast()
 
 	// Call the OnEmpty callback if the dequeue is empty
-	if d.isEmpty_unsafe() && d.OnEmpty != nil {
-		d.OnEmpty()
+	d.onEmptyLock.RLock()
+	defer d.onEmptyLock.RUnlock()
+
+	if d.isEmpty_unsafe() && d.onEmpty != nil {
+		d.onEmpty()
 	}
 
 	return item
@@ -143,6 +162,26 @@ func (d *BlockingDequeue[T]) PeekBack() T {
 	return element.Value.(T)
 }
 
+// ================================[Listeners related]================================
+
+// Set the callback function invoked when the dequeue is full.
+// Attempting to update the dequeue in the callback function will cause a deadlock.
+func (d *BlockingDequeue[T]) SetOnFull(onFull func()) {
+	d.onFullLock.Lock()
+	defer d.onFullLock.Unlock()
+
+	d.onFull = onFull
+}
+
+// Set the callback function invoked when the dequeue is empty.
+// Attempting to update the dequeue in the callback function will cause a deadlock.
+func (d *BlockingDequeue[T]) SetOnEmpty(onEmpty func()) {
+	d.onEmptyLock.Lock()
+	defer d.onEmptyLock.Unlock()
+
+	d.onEmpty = onEmpty
+}
+
 // ================================[Size/Capacity related]================================
 
 // Set dequeue capacity, if capacity is 0, dequeue is infinite.

dequeue_integration_test.go

@@ -235,3 +235,54 @@ func TestConcurrentCapacityChange(t *testing.T) {
 
 	// No checks need to be done, if the test is here without race conditions, it passed
 }
+
+func TestConcurrentUpdatesToOnEmptyListener(t *testing.T) {
+	dequeue := NewBlockingDequeue[int]()
+
+	getDummyCallback := func() func() { return func() {} }
+
+	// Update onEmpty callback while it's being called
+	go func() {
+		time.Sleep(10 * time.Millisecond)
+		for i := 0; i < 100; i++ {
+			dequeue.SetOnEmpty(getDummyCallback())
+		}
+	}()
+
+	// Update the dequeue to call the onEmpty callback
+	for i := 0; i < 100; i++ {
+		go func(val int) {
+			dequeue.PushBack(val)
+			dequeue.PopBack()
+
+			dequeue.PushFront(val)
+			dequeue.PopFront()
+		}(i)
+	}
+}
+
+func TestConcurrentUpdatesToOnFullListener(t *testing.T) {
+	dequeue := NewBlockingDequeue[int]()
+	dequeue.SetCapacity(1)
+
+	getDummyCallback := func() func() { return func() {} }
+
+	// Update onEmpty callback while it's being called
+	go func() {
+		time.Sleep(10 * time.Millisecond)
+		for i := 0; i < 100; i++ {
+			dequeue.SetOnFull(getDummyCallback())
+		}
+	}()
+
+	// Update the dequeue to call the onEmpty callback
+	for i := 0; i < 100; i++ {
+		go func(val int) {
+			dequeue.PushBack(val)
+			dequeue.PopBack()
+
+			dequeue.PushFront(val)
+			dequeue.PopFront()
+		}(i)
+	}
+}

dequeue_unit_test.go

@@ -386,9 +386,9 @@ func TestOnEmpty(t *testing.T) {
 	dequeue := NewBlockingDequeue[int]()
 
 	called := 0
-	dequeue.OnEmpty = func() {
+	dequeue.SetOnEmpty(func() {
 		called++
-	}
+	})
 
 	dequeue.PushBack(1)
 	dequeue.PopFront()
@@ -410,9 +410,9 @@ func TestOnFull(t *testing.T) {
 	dequeue.SetCapacity(3)
 
 	called := 0
-	dequeue.OnFull = func() {
+	dequeue.SetOnFull(func() {
 		called++
-	}
+	})
 
 	dequeue.PushBack(1)
 	dequeue.PushBack(2)