fix: deadlock when using switch_if_error (#43)

Author: jjeffcaii

hc/hc_test.go

@@ -6,12 +6,13 @@ import (
 	"net/http"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	"github.com/jjeffcaii/reactor-go"
 	"github.com/jjeffcaii/reactor-go/hc"
 	"github.com/jjeffcaii/reactor-go/mono"
 	"github.com/jjeffcaii/reactor-go/scheduler"
 	"github.com/jjeffcaii/reactor-go/tuple"
-	"github.com/stretchr/testify/assert"
 )
 
 var httpBinUrl = "https://httpbin.org/anything"
@@ -89,7 +90,7 @@ func TestClient_Do(t *testing.T) {
 }
 
 func TestDo_Failed(t *testing.T) {
-	req, _ := http.NewRequest(http.MethodGet, "http://127.0.0.1/not-exists-path", nil)
+	req, _ := http.NewRequest(http.MethodGet, "http://127.0.0.1:8080/not-exists-path", nil)
 	_, err := hc.Do(req).Block(context.Background())
 	assert.Error(t, err)
 }

mono/create.go

@@ -2,38 +2,14 @@ package mono
 
 import (
 	"context"
-	"math"
-	"sync"
 	"sync/atomic"
 
+	"github.com/pkg/errors"
+
 	"github.com/jjeffcaii/reactor-go"
 	"github.com/jjeffcaii/reactor-go/hooks"
-	"github.com/pkg/errors"
 )
 
-var globalSinkPool sinkPool
-
-type sinkPool struct {
-	inner sync.Pool
-}
-
-func (p *sinkPool) get() *sink {
-	if exist, _ := p.inner.Get().(*sink); exist != nil {
-		atomic.StoreInt32(&exist.stat, 0)
-		return exist
-	}
-	return &sink{}
-}
-
-func (p *sinkPool) put(s *sink) {
-	if s == nil {
-		return
-	}
-	atomic.StoreInt32(&s.stat, math.MinInt32)
-	s.actual = nil
-	p.inner.Put(s)
-}
-
 type Sink interface {
 	Success(Any)
 	Error(error)
@@ -84,23 +60,21 @@ func (s *sink) Success(v Any) {
 	s.Complete()
 }
 
-func (s *sink) Request(n int) {
+func (s *sink) Request(_ int) {
 	// ignore
 }
 
 func (s *sink) Cancel() {
 	if !atomic.CompareAndSwapInt32(&s.stat, 0, statCancel) {
 		return
 	}
-	defer globalSinkPool.put(s)
 	s.actual.OnError(reactor.ErrSubscribeCancelled)
 }
 
 func (s *sink) Complete() {
 	if !atomic.CompareAndSwapInt32(&s.stat, 0, statComplete) {
 		return
 	}
-	defer globalSinkPool.put(s)
 	s.actual.OnComplete()
 }
 
@@ -109,7 +83,6 @@ func (s *sink) Error(err error) {
 		hooks.Global().OnErrorDrop(err)
 		return
 	}
-	defer globalSinkPool.put(s)
 	s.actual.OnError(err)
 }
 
@@ -129,8 +102,8 @@ func (s *sink) Next(v Any) {
 }
 
 func (m monoCreate) SubscribeWith(ctx context.Context, s reactor.Subscriber) {
-	sk := globalSinkPool.get()
+	var sk sink
 	sk.actual = s
-	s.OnSubscribe(ctx, sk)
-	m.sinker(ctx, sk)
+	s.OnSubscribe(ctx, &sk)
+	m.sinker(ctx, &sk)
 }

mono/create_test.go

@@ -6,8 +6,8 @@ import (
 	"testing"
 
 	"github.com/golang/mock/gomock"
+
 	"github.com/jjeffcaii/reactor-go"
-	"github.com/stretchr/testify/assert"
 )
 
 func TestMonoCreate_SubscribeWith(t *testing.T) {
@@ -97,9 +97,3 @@ func TestMonoCreate_Cancel(t *testing.T) {
 		s.Success(1)
 	}).SubscribeWith(context.Background(), s)
 }
-
-func TestSinkPool_PutWithNilValue(t *testing.T) {
-	assert.NotPanics(t, func() {
-		globalSinkPool.put(nil)
-	})
-}

mono/peek.go

@@ -4,9 +4,10 @@ import (
 	"context"
 	"sync/atomic"
 
+	"github.com/pkg/errors"
+
 	"github.com/jjeffcaii/reactor-go"
 	"github.com/jjeffcaii/reactor-go/internal"
-	"github.com/pkg/errors"
 )
 
 type monoPeek struct {
@@ -20,10 +21,11 @@ type monoPeek struct {
 }
 
 type peekSubscriber struct {
-	actual reactor.Subscriber
-	parent *monoPeek
-	s      reactor.Subscription
-	stat   int32
+	actual    reactor.Subscriber
+	parent    *monoPeek
+	s         reactor.Subscription
+	stat      int32
+	cancelled int32
 }
 
 func newMonoPeek(source reactor.RawPublisher, first monoPeekOption, others ...monoPeekOption) *monoPeek {
@@ -52,7 +54,7 @@ func (p *peekSubscriber) Request(n int) {
 }
 
 func (p *peekSubscriber) Cancel() {
-	if !atomic.CompareAndSwapInt32(&p.stat, 0, statCancel) {
+	if !atomic.CompareAndSwapInt32(&p.cancelled, 0, 1) {
 		return
 	}
 	if call := p.parent.onCancelCall; call != nil {
@@ -73,7 +75,9 @@ func (p *peekSubscriber) OnComplete() {
 
 func (p *peekSubscriber) OnError(err error) {
 	if !atomic.CompareAndSwapInt32(&p.stat, 0, statError) {
-		return
+		if isCancelled := atomic.LoadInt32(&p.stat) == statCancel && err == reactor.ErrSubscribeCancelled; !isCancelled {
+			return
+		}
 	}
 
 	defer func() {