Merge pull request #114 from rueian/avoid-background-ping-stuck

fix: stucked background ping when its ring is full

pipe.go

@@ -450,9 +450,15 @@ func (p *pipe) _backgroundPing(stop <-chan struct{}) (err error) {
 			if atomic.LoadInt32(&p.blcksig) != 0 {
 				continue
 			}
-			ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
-			err = p.Do(ctx, cmds.PingCmd).NonRedisError()
-			cancel()
+			ch := make(chan error, 1)
+			tm := time.NewTimer(p.timeout)
+			go func() { ch <- p.Do(context.Background(), cmds.PingCmd).NonRedisError() }()
+			select {
+			case <-tm.C:
+				err = context.DeadlineExceeded
+			case err = <-ch:
+				tm.Stop()
+			}
 			if err != nil && atomic.LoadInt32(&p.blcksig) != 0 {
 				err = nil
 			}

pipe_test.go

@@ -2472,6 +2472,32 @@ func TestExitOnRingFullAndConnError(t *testing.T) {
 	}
 }
 
+func TestExitOnRingFullAndPingTimout(t *testing.T) {
+	p, mock, _, _ := setup(t, ClientOption{
+		RingScaleEachConn: 1,
+		ConnWriteTimeout:  500 * time.Millisecond,
+		Dialer:            net.Dialer{KeepAlive: 500 * time.Millisecond},
+	})
+	p.background()
+
+	// fill the ring
+	for i := 0; i < len(p.queue.(*ring).store); i++ {
+		go func() {
+			if err := p.Do(context.Background(), cmds.NewCompleted([]string{"GET", "a"})).Error(); err != context.DeadlineExceeded {
+				t.Errorf("unexpected result, expected context.DeadlineExceeded, got %v", err)
+			}
+		}()
+	}
+	// let writer loop over the ring
+	for i := 0; i < len(p.queue.(*ring).store); i++ {
+		mock.Expect("GET", "a")
+	}
+
+	if err := p.Do(context.Background(), cmds.NewCompleted([]string{"GET", "a"})).Error(); err != context.DeadlineExceeded {
+		t.Errorf("unexpected result, expected context.DeadlineExceeded, got %v", err)
+	}
+}
+
 func TestExitAllGoroutineOnWriteError(t *testing.T) {
 	conn, mock, _, closeConn := setup(t, ClientOption{})