Merge pull request #47 from jolestar/fix_dead_lock

Fix dead lock

Author: jolestar

collections/queue_test.go

@@ -252,7 +252,7 @@ func (suit *LinkedBlockDequeTestSuite) TestInterrupt() {
 	for i := 0; i < 2; i++ {
 		_, e := suit.deque.TakeFirst(ctx)
 		_, ok := e.(*InterruptedErr)
-		suit.True(ok, "expect InterruptedErr bug get %v", reflect.TypeOf(e))
+		suit.True(ok, "expect InterruptedErr but get %v", reflect.TypeOf(e))
 		suit.NotNil(e.Error())
 	}
 	wait.Wait()
@@ -502,3 +502,21 @@ func (suit *LinkedBlockDequeTestSuite) TestHasTakeWaiters() {
 	suit.Equal(1, val)
 	suit.False(suit.deque.HasTakeWaiters())
 }
+
+// https://github.com/jolestar/go-commons-pool/issues/44
+func (suit *LinkedBlockDequeTestSuite) TestDeadLock() {
+	ctx := context.Background()
+	suit.deque = NewDeque(1)
+	suit.deque.PutFirst(ctx, 1)
+	count := 1000000
+	testWG := sync.WaitGroup{}
+	testWG.Add(count)
+	for i := 0; i < count; i++ {
+		o := suit.NoErrorWithResult(suit.deque.PollFirstWithContext(ctx))
+		go func() {
+			suit.deque.PutFirst(ctx, o)
+			testWG.Done()
+		}()
+	}
+	testWG.Wait()
+}

concurrent/cond.go

@@ -13,11 +13,12 @@ type TimeoutCond struct {
 	hasWaiters uint64
 	L          sync.Locker
 	signal     chan int
+	condL      sync.RWMutex
 }
 
 // NewTimeoutCond return a new TimeoutCond
 func NewTimeoutCond(l sync.Locker) *TimeoutCond {
-	cond := TimeoutCond{L: l, signal: make(chan int, 0)}
+	cond := TimeoutCond{L: l, signal: make(chan int, 1), condL: sync.RWMutex{}}
 	return &cond
 }
 
@@ -45,8 +46,11 @@ func (cond *TimeoutCond) HasWaiters() bool {
 // Wait waits for a signal, or for the context do be done. Returns true if signaled.
 func (cond *TimeoutCond) Wait(ctx context.Context) bool {
 	cond.addWaiter()
+
+	cond.condL.RLock()
 	//copy signal in lock, avoid data race with Interrupt
 	ch := cond.signal
+	cond.condL.RUnlock()
 	//wait should unlock mutex,  if not will cause deadlock
 	cond.L.Unlock()
 	defer cond.removeWaiter()
@@ -62,16 +66,18 @@ func (cond *TimeoutCond) Wait(ctx context.Context) bool {
 
 // Signal wakes one goroutine waiting on c, if there is any.
 func (cond *TimeoutCond) Signal() {
+	cond.condL.RLock()
 	select {
 	case cond.signal <- 1:
 	default:
 	}
+	cond.condL.RUnlock()
 }
 
 // Interrupt goroutine wait on this TimeoutCond
 func (cond *TimeoutCond) Interrupt() {
-	cond.L.Lock()
-	defer cond.L.Unlock()
+	cond.condL.Lock()
+	defer cond.condL.Unlock()
 	close(cond.signal)
 	cond.signal = make(chan int, 0)
 }

concurrent/cond_test.go

@@ -90,10 +90,10 @@ func TestTimeoutCondWaitTimeoutNotify(t *testing.T) {
 	wait := sync.WaitGroup{}
 	wait.Add(2)
 	ch := make(chan time.Duration, 1)
-	timeout := 2 * time.Second
+	timeout := 5 * time.Second
 	go func() {
 		begin := time.Now()
-		obj.lockAndWaitWithTimeout(time.Duration(timeout) * time.Millisecond)
+		obj.lockAndWaitWithTimeout(timeout * time.Millisecond)
 		elapsed := time.Since(begin)
 		ch <- elapsed
 		wait.Done()
@@ -196,7 +196,7 @@ func TestInterrupted(t *testing.T) {
 	wait.Wait()
 	for i := 0; i < count; i++ {
 		b := <-ch
-		assert.True(t, b, "expect %v interrupted bug get false", i)
+		assert.True(t, b, "expect %v interrupted but get false", i)
 	}
 }
 
@@ -222,7 +222,7 @@ func TestInterruptedWithTimeout(t *testing.T) {
 	wait.Wait()
 	for i := 0; i < count; i++ {
 		b := <-ch
-		assert.True(t, b, "expect %v interrupted bug get false", i)
+		assert.True(t, b, "expect %v interrupted but get false", i)
 	}
 }
 

pool_test.go

@@ -1317,7 +1317,6 @@ func (suit *PoolTestSuite) TestEvictionSoftMinIdle() {
 	suit.Equal(0, suit.pool.GetNumIdle(), "Idle count different than expected.")
 }
 
-
 func (suit *PoolTestSuite) TestEvictionNegativeIdleTime() {
 	suit.pool.Config.MaxIdle = 5
 	suit.pool.Config.MaxTotal = 5
@@ -2070,6 +2069,32 @@ func (suit *PoolTestSuite) TestValueFactory() {
 	})
 }
 
+// https://github.com/jolestar/go-commons-pool/issues/44
+func (suit *PoolTestSuite) TestDeadLock() {
+	ctx := context.Background()
+	suit.pool.Config.MinIdle = 1
+	suit.pool.Config.MaxIdle = 1
+	suit.pool.Config.MaxTotal = 1
+	count := 1000000
+	testWG := sync.WaitGroup{}
+	testWG.Add(count)
+
+	for i := 0; i < count; i++ {
+		obj, err := suit.pool.BorrowObject(ctx)
+		if err != nil {
+			panic(err)
+		}
+		go func(obj interface{}) {
+			err = suit.pool.ReturnObject(ctx, obj)
+			if err != nil {
+				panic(err)
+			}
+			testWG.Done()
+		}(obj)
+	}
+	testWG.Wait()
+}
+
 var perf bool
 
 func init() {