Merge pull request #1749 from DMwangnima/hotfix/grpc_client_deadlock

hotfix: task of cleaning up canceled streams in http2Client has a probability of deadlocking

Author: DMwangnima

pkg/remote/trans/nphttp2/grpc/http2_client.go

@@ -114,7 +114,8 @@ type http2Client struct {
 	// variable.
 	kpDormant bool
 	onGoAway  func(GoAwayReason)
-	onClose   func()
+	// Important: when onClose is invoked, the http2Client.mu is held
+	onClose func()
 
 	bufferPool *bufferPool
 }

pkg/remote/trans/nphttp2/grpc/http2_server.go

@@ -37,11 +37,12 @@ import (
 	"github.com/cloudwego/kitex/pkg/remote/codec/protobuf/encoding"
 	"github.com/cloudwego/kitex/pkg/remote/transmeta"
 
-	"github.com/cloudwego/netpoll"
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/hpack"
 	"google.golang.org/protobuf/proto"
 
+	"github.com/cloudwego/netpoll"
+
 	"github.com/cloudwego/kitex/pkg/gofunc"
 	"github.com/cloudwego/kitex/pkg/klog"
 	"github.com/cloudwego/kitex/pkg/remote/trans/nphttp2/codes"

pkg/remote/trans/nphttp2/grpc/transport_test.go

@@ -37,10 +37,11 @@ import (
 	"testing"
 	"time"
 
-	"github.com/cloudwego/netpoll"
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/hpack"
 
+	"github.com/cloudwego/netpoll"
+
 	"github.com/cloudwego/kitex/internal/test"
 	"github.com/cloudwego/kitex/pkg/remote/trans/nphttp2/codes"
 	"github.com/cloudwego/kitex/pkg/remote/trans/nphttp2/grpc/grpcframe"

pkg/utils/sharedticker.go

@@ -108,11 +108,17 @@ func (t *SharedTicker) Tick(interval time.Duration) {
 }
 
 func (t *SharedTicker) syncExec() {
+	var todoTasks []TickerTask
 	t.Lock()
+	todoTasks = make([]TickerTask, 0, len(t.tasks))
 	for task := range t.tasks {
-		task.Tick()
+		todoTasks = append(todoTasks, task)
 	}
 	t.Unlock()
+
+	for _, task := range todoTasks {
+		task.Tick()
+	}
 }
 
 func (t *SharedTicker) asyncExec() {

pkg/utils/sharedticker_test.go

@@ -88,3 +88,28 @@ func TestSharedTickerTick(t *testing.T) {
 		time.Sleep(150 * time.Millisecond)
 	}
 }
+
+type concurrentTask struct {
+	st       *SharedTicker
+	finishCh chan struct{}
+}
+
+func (ct *concurrentTask) Tick() {
+	// when Tick is invoked, SharedTicker.mu should not be hold
+	ct.st.Lock()
+	// prevents Tick from being executed multiple times due to goroutine scheduling.
+	select {
+	case ct.finishCh <- struct{}{}:
+	default:
+	}
+	ct.st.Unlock()
+}
+
+func TestSyncSharedTickerConcurrent(t *testing.T) {
+	duration := 100 * time.Millisecond
+	st := NewSyncSharedTicker(duration)
+	finishCh := make(chan struct{}, 1)
+	st.Add(&concurrentTask{st: st, finishCh: finishCh})
+	time.Sleep(150 * time.Millisecond)
+	<-finishCh
+}

version.go

@@ -19,5 +19,5 @@ package kitex
 // Name and Version info of this framework, used for statistics and debug
 const (
 	Name    = "Kitex"
-	Version = "v0.12.3"
+	Version = "v0.12.4"
 )