fix: request-channel deadlock in some cases (#107)

* fix: request-channel deadlock in some cases

* fix: ut

Author: jjeffcaii

internal/socket/duplex.go

@@ -447,8 +447,6 @@ func (dc *DuplexConnection) RequestChannel(sending flux.Flux) (ret flux.Flux) {
 
 	toBeReleased := queue.NewLKQueue()
 
-	sendResult := make(chan error)
-
 	ret = receiving.
 		DoFinally(func(sig rx.SignalType) {
 			dc.unregister(sid)
@@ -460,20 +458,6 @@ func (dc *DuplexConnection) RequestChannel(sending flux.Flux) (ret flux.Flux) {
 				}
 				next.(common.Releasable).Release()
 			}
-			// process sending result
-			e, ok := <-sendResult
-			if ok {
-				dc.writeError(sid, e)
-			} else {
-				complete := framing.NewWriteablePayloadFrame(sid, nil, nil, core.FlagComplete)
-				done := make(chan struct{})
-				complete.HandleDone(func() {
-					close(done)
-				})
-				if dc.sendFrame(complete) {
-					<-done
-				}
-			}
 		}).
 		DoOnNext(func(next payload.Payload) error {
 			if nextRelease := toBeReleased.Dequeue(); nextRelease != nil {
@@ -486,7 +470,9 @@ func (dc *DuplexConnection) RequestChannel(sending flux.Flux) (ret flux.Flux) {
 		}).
 		DoOnRequest(func(initN int) {
 			n := ToUint32RequestN(initN)
-			if !rcvRequested.CAS(false, true) {
+			isFirstRequest := rcvRequested.CAS(false, true)
+			if !isFirstRequest {
+				// block send RequestN frame
 				frameN := framing.NewWriteableRequestNFrame(sid, n, 0)
 				done := make(chan struct{})
 				frameN.HandleDone(func() {
@@ -498,13 +484,11 @@ func (dc *DuplexConnection) RequestChannel(sending flux.Flux) (ret flux.Flux) {
 				return
 			}
 
-			sub := requestChannelSubscriber{
-				sid:          sid,
-				n:            n,
-				dc:           dc,
-				sndRequested: atomic.NewBool(false),
-				rcv:          receiving,
-				result:       sendResult,
+			sub := &requestChannelSubscriber{
+				sid: sid,
+				n:   n,
+				dc:  dc,
+				rcv: receiving,
 			}
 			sending.SubscribeOn(dc.reqSche).SubscribeWith(dc.ctx, sub)
 		})
@@ -645,13 +629,11 @@ func (dc *DuplexConnection) respondRequestChannel(req fragmentation.HeaderAndPay
 		return nil
 	}
 
-	receivingProcessor.Next(req)
-
 	// Ensure registering message success before func end.
 	subscribed := make(chan struct{})
 
 	// Create subscriber
-	sub := respondChannelSubscriber{
+	sub := &respondChannelSubscriber{
 		sid:        sid,
 		n:          initRequestN,
 		dc:         dc,
@@ -664,6 +646,8 @@ func (dc *DuplexConnection) respondRequestChannel(req fragmentation.HeaderAndPay
 		sending.SubscribeWith(dc.ctx, sub)
 	})
 
+	receivingProcessor.Next(req)
+
 	<-subscribed
 
 	return nil

internal/socket/subscriber_request_channel.go

@@ -13,26 +13,16 @@ import (
 	"go.uber.org/atomic"
 )
 
-type respondChannelSubscriber struct {
-	sid        uint32
-	n          uint32
-	dc         *DuplexConnection
-	rcv        flux.Processor
-	subscribed chan<- struct{}
-	calls      *atomic.Int32
-}
-
 type requestChannelSubscriber struct {
-	sid          uint32
-	n            uint32
-	dc           *DuplexConnection
-	sndRequested *atomic.Bool
-	rcv          flux.Processor
-	result       chan<- error
+	sid       uint32
+	n         uint32
+	dc        *DuplexConnection
+	requested atomic.Bool
+	rcv       flux.Processor
 }
 
-func (r requestChannelSubscriber) OnNext(item payload.Payload) {
-	if !r.sndRequested.CAS(false, true) {
+func (r *requestChannelSubscriber) OnNext(item payload.Payload) {
+	if !r.requested.CAS(false, true) {
 		r.dc.sendPayload(r.sid, item, core.FlagNext)
 		return
 	}
@@ -55,21 +45,22 @@ func (r requestChannelSubscriber) OnNext(item payload.Payload) {
 	})
 }
 
-func (r requestChannelSubscriber) OnError(err error) {
-	defer func() {
-		_ = recover()
-	}()
-	r.result <- err
+func (r *requestChannelSubscriber) OnError(err error) {
+	r.dc.writeError(r.sid, err)
 }
 
-func (r requestChannelSubscriber) OnComplete() {
-	defer func() {
-		_ = recover()
-	}()
-	close(r.result)
+func (r *requestChannelSubscriber) OnComplete() {
+	complete := framing.NewWriteablePayloadFrame(r.sid, nil, nil, core.FlagComplete)
+	done := make(chan struct{})
+	complete.HandleDone(func() {
+		close(done)
+	})
+	if r.dc.sendFrame(complete) {
+		<-done
+	}
 }
 
-func (r requestChannelSubscriber) OnSubscribe(ctx context.Context, s rx.Subscription) {
+func (r *requestChannelSubscriber) OnSubscribe(ctx context.Context, s rx.Subscription) {
 	select {
 	case <-ctx.Done():
 		r.OnError(reactor.ErrSubscribeCancelled)
@@ -83,18 +74,27 @@ func (r requestChannelSubscriber) OnSubscribe(ctx context.Context, s rx.Subscrip
 	}
 }
 
-func (r respondChannelSubscriber) OnNext(next payload.Payload) {
+type respondChannelSubscriber struct {
+	sid        uint32
+	n          uint32
+	dc         *DuplexConnection
+	rcv        flux.Processor
+	subscribed chan<- struct{}
+	calls      *atomic.Int32
+}
+
+func (r *respondChannelSubscriber) OnNext(next payload.Payload) {
 	r.dc.sendPayload(r.sid, next, core.FlagNext)
 }
 
-func (r respondChannelSubscriber) OnError(err error) {
+func (r *respondChannelSubscriber) OnError(err error) {
 	if r.calls.Inc() == 2 {
 		r.dc.unregister(r.sid)
 	}
 	r.dc.writeError(r.sid, err)
 }
 
-func (r respondChannelSubscriber) OnComplete() {
+func (r *respondChannelSubscriber) OnComplete() {
 	if r.calls.Inc() == 2 {
 		r.dc.unregister(r.sid)
 	}
@@ -108,7 +108,7 @@ func (r respondChannelSubscriber) OnComplete() {
 	}
 }
 
-func (r respondChannelSubscriber) OnSubscribe(ctx context.Context, s rx.Subscription) {
+func (r *respondChannelSubscriber) OnSubscribe(ctx context.Context, s rx.Subscription) {
 	select {
 	case <-ctx.Done():
 		r.OnError(reactor.ErrSubscribeCancelled)

rsocket_test.go

@@ -357,13 +357,13 @@ func testAll(t *testing.T, proto string, clientTp transport.ClientTransporter, s
 					RequestChannel(func(inputs flux.Flux) flux.Flux {
 						received := new(int32)
 						inputs.
-							DoFinally(func(s rx.SignalType) {
-								assert.Equal(t, channelElements, atomic.LoadInt32(received))
-							}).
 							DoOnNext(func(input payload.Payload) error {
 								atomic.AddInt32(received, 1)
 								return nil
 							}).
+							DoOnComplete(func() {
+								assert.Equal(t, channelElements, atomic.LoadInt32(received))
+							}).
 							Subscribe(context.Background())
 
 						return flux.Create(func(ctx context.Context, s flux.Sink) {