Merge branch 'release-5.0' into pr/2499

Signed-off-by: Neil Shen <overvenus@gmail.com>

pkg/workerpool/pool.go

@@ -38,11 +38,13 @@ type EventHandle interface {
 	// AddEvent adds an `event` object to the internal queue, so that the `f` used to register the handle can be called.
 	// Note: events are always processed in the order they are added.
 	// Unregistering the EventHandle MAY CAUSE EVENT LOSSES. But for an event lost, any event after it is guaranteed to be lost too.
+	// Cancelling `ctx` here will cancel the on-going or next execution of the event.
 	AddEvent(ctx context.Context, event interface{}) error
 
 	// SetTimer is used to provide a function that is periodic called, as long as the EventHandle has not been unregistered.
 	// The current implementation uses as the base clock source a ticker whose interval is the const workerPoolDefaultClockSourceInterval.
 	// DO NOT set an interval less than workerPoolDefaultClockSourceInterval.
+	// Cancelling `ctx` here will cancel the on-going or next execution of `f`.
 	SetTimer(ctx context.Context, interval time.Duration, f func(ctx context.Context) error) EventHandle
 
 	// Unregister removes the EventHandle from the WorkerPool.

pkg/workerpool/pool_impl.go

@@ -128,13 +128,7 @@ func (h *defaultEventHandle) AddEvent(ctx context.Context, event interface{}) er
 	task := task{
 		handle: h,
 		f: func(ctx1 context.Context) error {
-			// Here we merge the context passed down from WorkerPool.Run,
-			// with the context supplied by AddEvent,
-			// because we want operations to be cancellable by both contexts.
-			mContext, cancel := MergeContexts(ctx, ctx1)
-			// this cancels the merged context only.
-			defer cancel()
-			return h.f(mContext, event)
+			return h.f(ctx, event)
 		},
 	}
 
@@ -154,9 +148,7 @@ func (h *defaultEventHandle) SetTimer(ctx context.Context, interval time.Duratio
 
 	h.timerInterval = interval
 	h.timerHandler = func(ctx1 context.Context) error {
-		mContext, cancel := MergeContexts(ctx, ctx1)
-		defer cancel()
-		return f(mContext)
+		return f(ctx)
 	}
 	// mark the timer handler function as valid
 	atomic.StoreInt32(&h.hasTimer, 1)