check start waiters with unsafe semaphore inspection.

Author: rekby

internal/topic/topicwriterinternal/writer_reconnector_test.go

@@ -271,9 +271,11 @@ func TestWriterReconnector_Write_QueueLimit(t *testing.T) {
 		))
 		w.firstConnectionHandled.Store(true)
 
-		waitStartQueueWait := func() {
-			// TODO: fix with reflection and unsafe for count internal waiters
-			time.Sleep(time.Second / 10)
+		waitStartQueueWait := func(targetWaiters int) {
+			xtest.SpinWaitCondition(t, nil, func() bool {
+				res := getWaitersCount(w.semaphore) == targetWaiters
+				return res
+			})
 		}
 
 		err := w.Write(ctx, newTestMessages(1, 2))
@@ -282,7 +284,7 @@ func TestWriterReconnector_Write_QueueLimit(t *testing.T) {
 		ctxNoQueueSpace, ctxNoQueueSpaceCancel := context.WithCancel(ctx)
 
 		go func() {
-			waitStartQueueWait()
+			waitStartQueueWait(1)
 			ctxNoQueueSpaceCancel()
 		}()
 		err = w.Write(ctxNoQueueSpace, newTestMessages(3))
@@ -291,7 +293,7 @@ func TestWriterReconnector_Write_QueueLimit(t *testing.T) {
 		}
 
 		go func() {
-			waitStartQueueWait()
+			waitStartQueueWait(1)
 			ackErr := w.queue.AcksReceived([]rawtopicwriter.WriteAck{
 				{
 					SeqNo: 1,

internal/topic/topicwriterinternal/writer_reconnector_unsafe_test.go

@@ -0,0 +1,28 @@
+package topicwriterinternal
+
+import (
+	"container/list"
+	"reflect"
+	"runtime"
+	"sync"
+	"unsafe"
+
+	"golang.org/x/sync/semaphore"
+)
+
+func getWaitersCount(sem *semaphore.Weighted) int {
+	defer runtime.KeepAlive(sem)
+
+	semVal := reflect.ValueOf(sem).Elem()
+	mutexField := semVal.FieldByName("mu")
+
+	mutexAddr := unsafe.Pointer(mutexField.UnsafeAddr())
+	mutex := (*sync.Mutex)(mutexAddr)
+	mutex.Lock()
+	defer mutex.Unlock()
+
+	waitersField := semVal.FieldByName("waiters")
+	waitersPointer := unsafe.Pointer(waitersField.UnsafeAddr())
+	waiters := (*list.List)(waitersPointer)
+	return waiters.Len()
+}