Revert direct master commits

Revert "stop retry on unexpected call"

This reverts commit 88cebd8.

Revert "fix worker start after close"

This reverts commit 4a46a77.

Revert "fixed start func may lock while close wait stop backgrounds"

This reverts commit 5500363.

Revert "fix task loop"

This reverts commit 23061fb.

Author: rekby

internal/background/worker.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"runtime/pprof"
 	"sync"
+	"sync/atomic"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/empty"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xcontext"
@@ -23,19 +24,13 @@ type Worker struct {
 	workers  sync.WaitGroup
 	onceInit sync.Once
 
-	tasksCompleted empty.Chan
-
 	m xsync.Mutex
 
-	tasks chan backgroundTask
-
-	closed      bool
+	closed      uint32
 	stop        xcontext.CancelErrFunc
 	closeReason error
 }
 
-type CallbackFunc func(ctx context.Context)
-
 func NewWorker(parent context.Context) *Worker {
 	w := Worker{}
 	w.ctx, w.stop = xcontext.WithErrCancel(parent)
@@ -49,52 +44,54 @@ func (b *Worker) Context() context.Context {
 	return b.ctx
 }
 
-func (b *Worker) Start(name string, f CallbackFunc) {
+func (b *Worker) Start(name string, f func(ctx context.Context)) {
+	if atomic.LoadUint32(&b.closed) != 0 {
+		f(b.ctx)
+		return
+	}
+
 	b.init()
 
-	b.m.WithLock(func() {
-		if b.closed {
-			return
-		}
+	b.m.Lock()
+	defer b.m.Unlock()
 
-		b.tasks <- backgroundTask{
-			callback: f,
-			name:     name,
-		}
-	})
+	if b.ctx.Err() != nil {
+		return
+	}
+
+	b.workers.Add(1)
+	go func() {
+		defer b.workers.Done()
+
+		pprof.Do(b.ctx, pprof.Labels("background", name), f)
+	}()
 }
 
 func (b *Worker) Done() <-chan struct{} {
 	b.init()
 
+	b.m.Lock()
+	defer b.m.Unlock()
+
 	return b.ctx.Done()
 }
 
 func (b *Worker) Close(ctx context.Context, err error) error {
-	b.init()
-
-	var resErr error
-	b.m.WithLock(func() {
-		if b.closed {
-			resErr = xerrors.WithStackTrace(ErrAlreadyClosed)
-			return
-		}
+	if !atomic.CompareAndSwapUint32(&b.closed, 0, 1) {
+		return xerrors.WithStackTrace(ErrAlreadyClosed)
+	}
 
-		b.closed = true
+	b.init()
 
-		close(b.tasks)
-		b.closeReason = err
-		if b.closeReason == nil {
-			b.closeReason = errClosedWithNilReason
-		}
+	b.m.Lock()
+	defer b.m.Unlock()
 
-		b.stop(err)
-	})
-	if resErr != nil {
-		return resErr
+	b.closeReason = err
+	if b.closeReason == nil {
+		b.closeReason = errClosedWithNilReason
 	}
 
-	<-b.tasksCompleted
+	b.stop(err)
 
 	bgCompleted := make(empty.Chan)
 
@@ -123,27 +120,5 @@ func (b *Worker) init() {
 		if b.ctx == nil {
 			b.ctx, b.stop = xcontext.WithErrCancel(context.Background())
 		}
-		b.tasks = make(chan backgroundTask)
-		b.tasksCompleted = make(empty.Chan)
-		go b.starterLoop()
 	})
 }
-
-func (b *Worker) starterLoop() {
-	defer close(b.tasksCompleted)
-
-	for bgTask := range b.tasks {
-		b.workers.Add(1)
-
-		go func(task backgroundTask) {
-			defer b.workers.Done()
-
-			pprof.Do(b.ctx, pprof.Labels("background", task.name), task.callback)
-		}(bgTask)
-	}
-}
-
-type backgroundTask struct {
-	callback CallbackFunc
-	name     string
-}

internal/background/worker_test.go

@@ -222,7 +222,7 @@ func TestTopicReaderReconnectorConnectionLoop(t *testing.T) {
 			},
 			{
 				callback: func(ctx context.Context) (batchedStreamReader, error) {
-					t.Fatal()
+					t.Error()
 					return nil, errors.New("unexpected call")
 				},
 			},