fix: if the pool is saturated, execute the job in the current goroutine

Author: Reasno

pool.go

@@ -120,9 +120,18 @@ func (p *Pool) Module() interface{} {
 // from http/grpc handler, and asyncContext is the context for async job
 // handling. The asyncContext contains all values from requestContext, but it's
 // cancellation has nothing to do with the request, but is determined the timeout
-// set in pool constructor.
+// set in pool constructor. If the pool has reached max concurrency, the job will
+// be executed in the current goroutine. In other word, the job will be executed
+// synchronously.
 func (p *Pool) Go(requestContext context.Context, function func(asyncContext context.Context)) {
-	p.ch <- job{ctx: requestContext, function: function}
+	select {
+	case p.ch <- job{ctx: requestContext, function: function}:
+	default:
+		cancelCtx, cancel := context.WithTimeout(context.Background(), p.timeout)
+		defer cancel()
+		newCtx := asyncContext{valueCtx: requestContext, cancelCtx: cancelCtx}
+		function(newCtx)
+	}
 }
 
 // Run starts the async worker pool and block until it finishes.

pool_test.go

@@ -41,6 +41,29 @@ func TestPool_Timeout(t *testing.T) {
 	p.Run(ctx)
 }
 
+func TestPool_FallbackToSyncMode(t *testing.T) {
+	t.Parallel()
+	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond)
+	defer cancel()
+
+	dispatcher := events.SyncDispatcher{}
+	p := newPool(WithTimeout(time.Second), WithConcurrency(1), WithShutdownEvents())(&dispatcher)
+	p.Run(ctx)
+
+	var executed = make(chan struct{})
+	go func() {
+		// saturate the pool
+		p.Go(ctx, func(asyncContext context.Context) {
+			time.Sleep(time.Second)
+		})
+		// fallback to sync mode
+		p.Go(ctx, func(asyncContext context.Context) {
+			close(executed)
+		})
+	}()
+	<-executed
+}
+
 func TestPool_contextValue(t *testing.T) {
 	t.Parallel()
 	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond)