From 4935078aa4f25695df7024f70e1ad3ef08ed70f4 Mon Sep 17 00:00:00 2001 From: Andrew Thornton Date: Thu, 18 Nov 2021 16:57:24 +0000 Subject: [PATCH 1/2] Prevent deadlock in TestPersistableChannelQueue There is a potential deadlock in TestPersistableChannelQueue due to attempting to shutdown the test queue before it is ready. Signed-off-by: Andrew Thornton --- modules/queue/queue_disk_channel_test.go | 30 ++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/modules/queue/queue_disk_channel_test.go b/modules/queue/queue_disk_channel_test.go index d464a183a0f55..2b67587698595 100644 --- a/modules/queue/queue_disk_channel_test.go +++ b/modules/queue/queue_disk_channel_test.go @@ -42,13 +42,26 @@ func TestPersistableChannelQueue(t *testing.T) { }, &testData{}) assert.NoError(t, err) + readyForShutdown := make(chan struct{}) + readyForTerminate := make(chan struct{}) + go queue.Run(func(shutdown func()) { lock.Lock() defer lock.Unlock() + select { + case <-readyForShutdown: + default: + close(readyForShutdown) + } queueShutdown = append(queueShutdown, shutdown) }, func(terminate func()) { lock.Lock() defer lock.Unlock() + select { + case <-readyForTerminate: + default: + close(readyForTerminate) + } queueTerminate = append(queueTerminate, terminate) }) @@ -74,6 +87,7 @@ func TestPersistableChannelQueue(t *testing.T) { err = queue.Push(test1) assert.Error(t, err) + <-readyForShutdown // Now shutdown the queue lock.Lock() callbacks := make([]func(), len(queueShutdown)) @@ -97,6 +111,7 @@ func TestPersistableChannelQueue(t *testing.T) { } // terminate the queue + <-readyForTerminate lock.Lock() callbacks = make([]func(), len(queueTerminate)) copy(callbacks, queueTerminate) @@ -123,13 +138,26 @@ func TestPersistableChannelQueue(t *testing.T) { }, &testData{}) assert.NoError(t, err) + readyForShutdown = make(chan struct{}) + readyForTerminate = make(chan struct{}) + go queue.Run(func(shutdown func()) { lock.Lock() defer lock.Unlock() + select { + case <-readyForShutdown: + default: + close(readyForShutdown) + } queueShutdown = append(queueShutdown, shutdown) }, func(terminate func()) { lock.Lock() defer lock.Unlock() + select { + case <-readyForTerminate: + default: + close(readyForTerminate) + } queueTerminate = append(queueTerminate, terminate) }) @@ -141,6 +169,7 @@ func TestPersistableChannelQueue(t *testing.T) { assert.Equal(t, test2.TestString, result4.TestString) assert.Equal(t, test2.TestInt, result4.TestInt) + <-readyForShutdown lock.Lock() callbacks = make([]func(), len(queueShutdown)) copy(callbacks, queueShutdown) @@ -148,6 +177,7 @@ func TestPersistableChannelQueue(t *testing.T) { for _, callback := range callbacks { callback() } + <-readyForTerminate lock.Lock() callbacks = make([]func(), len(queueTerminate)) copy(callbacks, queueTerminate) From 624699383564f148c07aedcdf8949b7aae748029 Mon Sep 17 00:00:00 2001 From: Andrew Thornton Date: Thu, 18 Nov 2021 22:55:37 +0000 Subject: [PATCH 2/2] prevent npe Signed-off-by: Andrew Thornton --- modules/queue/queue_disk_channel_test.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/modules/queue/queue_disk_channel_test.go b/modules/queue/queue_disk_channel_test.go index 2b67587698595..c90d715a73cf2 100644 --- a/modules/queue/queue_disk_channel_test.go +++ b/modules/queue/queue_disk_channel_test.go @@ -18,6 +18,9 @@ func TestPersistableChannelQueue(t *testing.T) { handleChan := make(chan *testData) handle := func(data ...Data) { for _, datum := range data { + if datum == nil { + continue + } testDatum := datum.(*testData) handleChan <- testDatum }