Copilot commented on code in PR #3599:
URL: https://github.com/apache/dubbo-go/pull/3599#discussion_r3726192094
##########
registry/base_registry.go:
##########
@@ -340,23 +341,44 @@ func (r *BaseRegistry) Subscribe(url *common.URL,
notifyListener NotifyListener)
return err
}
Review Comment:
When DoSubscribe returns an error during shutdown, Subscribe currently
returns that subscribe error (line 341) instead of the stable unavailable
error. This makes shutdown exit behavior timing-dependent (Destroy between
DoSubscribe and waitRetryDelay changes the returned error) and contradicts the
intent of returning the existing unavailable error on destruction.
##########
cluster/cluster/failback/cluster_invoker.go:
##########
@@ -166,15 +199,129 @@ func (invoker *failbackClusterInvoker) Invoke(ctx
context.Context, invocation pr
return res
}
+func (invoker *failbackClusterInvoker) isStopped() bool {
+ invoker.lifecycleMu.Lock()
+ defer invoker.lifecycleMu.Unlock()
+ return invoker.stopped
+}
+
func (invoker *failbackClusterInvoker) Destroy() {
- invoker.BaseClusterInvoker.Destroy()
+ invoker.destroyOnce.Do(func() {
+ invoker.lifecycleMu.Lock()
+ invoker.stopped = true
+ if invoker.retryCancel != nil {
+ invoker.retryCancel()
+ }
+ taskList := invoker.taskList
+ processDone := invoker.processDone
+ retryDone := invoker.retryDone
+ if taskList != nil {
+ _ = taskList.Dispose()
+ }
+ invoker.lifecycleMu.Unlock()
+
+ invoker.waitForShutdown(processDone, retryDone)
+ invoker.BaseClusterInvoker.Destroy()
+ })
+}
- // stop ticker
- if invoker.ticker != nil {
- invoker.ticker.Stop()
+func (invoker *failbackClusterInvoker) enqueueInitialRetry(ctx
context.Context, retryTask *retryTimerTask) {
+ invoker.lifecycleMu.Lock()
+ defer invoker.lifecycleMu.Unlock()
+
+ if invoker.stopped || invoker.Destroyed.Load() {
+ return
+ }
+
+ if invoker.taskList == nil {
+ if ctx == nil {
+ ctx = context.Background()
+ }
+ invoker.retryCtx, invoker.retryCancel =
context.WithCancel(context.WithoutCancel(ctx))
+ invoker.taskList = queue.New(invoker.failbackTasks)
+ invoker.processDone = make(chan struct{})
+ go invoker.process(invoker.retryCtx, invoker.taskList,
invoker.processDone)
+ }
+
+ if invoker.taskList.Len() >= invoker.failbackTasks {
+ logger.Warnf("[Cluster][Failback] task list full, len=%d",
invoker.taskList.Len())
+ return
}
- _ = invoker.taskList.Dispose()
+ if err := invoker.taskList.Put(retryTask); err != nil {
+ logger.Warnf("[Cluster][Failback] put initial task failed,
err=%v", err)
+ }
+}
+
+func (invoker *failbackClusterInvoker) startRetry(ctx context.Context,
retryTask *retryTimerTask) {
+ invoker.lifecycleMu.Lock()
+ defer invoker.lifecycleMu.Unlock()
+
+ if invoker.stopped || ctx.Err() != nil {
+ return
+ }
+ if invoker.activeRetries == 0 {
+ invoker.retryDone = make(chan struct{})
+ }
+ invoker.activeRetries++
+ retryDone := invoker.retryDone
+ go func() {
+ defer invoker.finishRetry(retryDone)
+ invoker.tryTimerTaskProc(ctx, retryTask)
+ }()
+}
+
+func (invoker *failbackClusterInvoker) finishRetry(retryDone chan struct{}) {
+ invoker.lifecycleMu.Lock()
+ defer invoker.lifecycleMu.Unlock()
+
+ invoker.activeRetries--
+ if invoker.activeRetries == 0 && invoker.retryDone == retryDone {
+ close(retryDone)
+ }
+}
+
+func (invoker *failbackClusterInvoker) enqueueRetry(retryTask *retryTimerTask)
bool {
+ invoker.lifecycleMu.Lock()
+ defer invoker.lifecycleMu.Unlock()
+
+ if invoker.stopped || invoker.retryCtx == nil || invoker.retryCtx.Err()
!= nil || invoker.taskList == nil {
+ return false
+ }
+
+ retryTask.lastT = time.Now()
+ if err := invoker.taskList.Put(retryTask); err != nil {
+ logger.Warnf("[Cluster][Failback] put retry task failed,
err=%v", err)
+ return false
+ }
+ return true
+}
+
+func (invoker *failbackClusterInvoker) waitForShutdown(processDone, retryDone
<-chan struct{}) {
+ if processDone == nil && retryDone == nil {
+ return
+ }
+
+ timer := time.NewTimer(constant.DefaultShutdownConfigStepTimeout)
+ defer timer.Stop()
+
+ wait := func(done <-chan struct{}, name string) bool {
+ if done == nil {
+ return true
+ }
+ select {
+ case <-done:
+ return true
+ case <-timer.C:
+ logger.Warnf("[Cluster][Failback] timed out waiting for
%s shutdown", name)
+ return false
+ }
+ }
+
+ if !wait(processDone, "retry processor") {
+ return
+ }
+ _ = wait(retryDone, "retry tasks")
}
Review Comment:
waitForShutdown uses a single shared timer for both waits; if the first wait
consumes most of DefaultShutdownConfigStepTimeout, the second wait can time out
almost immediately. Since the constant name implies a per-step timeout, each
wait should have its own timer budget.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
