Copilot commented on code in PR #1455:
URL: https://github.com/apache/dubbo-admin/pull/1455#discussion_r3069425139
##########
pkg/core/events/component.go:
##########
@@ -55,52 +62,109 @@ func (b *eventBus) Order() int {
return math.MaxInt
}
-func (b *eventBus) Init(_ runtime.BuilderContext) error {
- b.subscriberDir = make(map[model.ResourceKind]Subscribers)
+func (b *eventBus) Init(ctx runtime.BuilderContext) error {
+ b.subscriberDir = make(map[model.ResourceKind][]*subscriberState)
+ if cfg := ctx.Config().EventBus; cfg != nil {
+ b.bufferSize = int(cfg.BufferSize)
+ } else {
+ b.bufferSize = int(eventbusconfig.Default().BufferSize)
+ }
return nil
}
func (b *eventBus) Start(_ runtime.Runtime, _ <-chan struct{}) error {
+ if !b.started.CompareAndSwap(false, true) {
+ return fmt.Errorf("eventBus already started")
+ }
+ b.rwMutex.RLock()
+ for _, states := range b.subscriberDir {
+ for _, st := range states {
+ if st.async {
+ b.launchDrainer(st)
+ }
+ }
+ }
+ b.rwMutex.RUnlock()
return nil
}
// Subscribe subscribes to a resource kind, ProcessEventFunc is synchronous
which is used to avoid event loss
func (b *eventBus) Subscribe(subscriber Subscriber) error {
b.rwMutex.Lock()
defer b.rwMutex.Unlock()
- subs, exists := b.subscriberDir[subscriber.ResourceKind()]
+ rk := subscriber.ResourceKind()
+ states, exists := b.subscriberDir[rk]
if !exists {
- subs = make(Subscribers, 0)
+ states = make([]*subscriberState, 0)
}
// check name if is unique
- for _, sub := range subs {
- if sub.Name() == subscriber.Name() {
+ for _, st := range states {
+ if st.subscriber.Name() == subscriber.Name() {
return fmt.Errorf("duplicated subscriber name %s,
skipped subscribing", subscriber.Name())
}
}
- b.subscriberDir[subscriber.ResourceKind()] = append(subs, subscriber)
+ isAsync := false
+ if b.bufferSize > 0 {
+ if as, ok := subscriber.(AsyncSubscriber); ok &&
as.AsyncEnabled() {
+ isAsync = true
+ }
+ }
+ state := &subscriberState{
+ subscriber: subscriber,
+ async: isAsync,
+ }
+ if isAsync {
+ state.ch = make(chan Event, b.bufferSize)
+ state.done = make(chan struct{})
+ }
+ b.subscriberDir[rk] = append(states, state)
+ if isAsync && b.started.Load() {
+ b.launchDrainer(state)
+ }
Review Comment:
`Subscribe()` can still launch a drainer goroutine after shutdown has begun.
If `WaitForDone()` is already running (and calling `wg.Wait()`), a concurrent
`launchDrainer()` will call `wg.Add(1)` and can trigger a `sync: WaitGroup
misuse: Add called concurrently with Wait` panic. Add a shutdown guard (e.g.,
return error / no-op when `b.stopped` is set) so no new drainers (and no new
`wg.Add`) can occur once `WaitForDone()` starts.
##########
pkg/core/runtime/runtime.go:
##########
@@ -136,6 +136,11 @@ func (rt *runtime) Start(stop <-chan struct{}) error {
logger.Info("Admin started successfully")
select {
case <-stop:
+ for _, com := range components {
+ if gc, ok := com.(GracefulComponent); ok {
+ gc.WaitForDone()
+ }
Review Comment:
`Start()` waits for graceful components by calling `WaitForDone()` with no
timeout/cancellation. If any component blocks indefinitely (e.g., stuck drainer
/ long I/O), `runtime.Start()` will never return and graceful shutdown will
hang until the process is force-killed. Consider adding a bounded wait
(timeout) around each `WaitForDone()` call (log and continue), or extending the
graceful contract to accept a context/deadline so shutdown can be interrupted.
##########
pkg/core/events/component.go:
##########
@@ -109,15 +173,62 @@ func (b *eventBus) Send(event Event) {
} else if event.OldObj() != nil {
rk = event.OldObj().ResourceKind()
}
- subs, exists := b.subscriberDir[rk]
+ states, exists := b.subscriberDir[rk]
if !exists {
logger.Infof("no subscriber for resource %s, skipped sending
event%v", rk, event)
return
}
- for _, sub := range subs {
- // TODO Do we need to support reprocess
- if err := sub.ProcessEvent(event); err != nil {
- logger.Errorf("failed to process event in %s, cause:
%s, event: %v", sub.Name(), err.Error(), event)
+ for _, st := range states {
+ if st.async && !st.closed.Load() {
+ select {
+ case st.ch <- event:
+ default:
+ logger.Warnf("async subscriber %s channel full
(cap=%d), event dropped: %v",
+ st.subscriber.Name(), cap(st.ch), event)
+ }
+ continue
+ }
+ if !st.async {
+ // TODO Do we need to support reprocess
+ if err := st.subscriber.ProcessEvent(event); err != nil
{
+ logger.Errorf("failed to process event in %s,
cause: %s, event: %v",
+ st.subscriber.Name(), err.Error(),
event)
+ }
+ }
+ }
+}
+
+func (b *eventBus) WaitForDone() {
+ b.stopped.Store(true)
+
+ b.rwMutex.Lock()
+ for _, states := range b.subscriberDir {
+ for _, st := range states {
+ if st.async && st.ch != nil && !st.closed.Load() {
+ st.closed.Store(true)
+ close(st.ch)
+ }
}
}
+ b.rwMutex.Unlock()
+
+ b.wg.Wait()
+}
+
+func (b *eventBus) launchDrainer(st *subscriberState) {
+ if !st.drainerStarted.CompareAndSwap(false, true) {
+ return
+ }
+
+ b.wg.Add(1)
Review Comment:
`launchDrainer()` unconditionally calls `b.wg.Add(1)` and can be invoked
concurrently with `WaitForDone()`'s `b.wg.Wait()`. Without an explicit guard,
this risks `WaitGroup` misuse panics during shutdown if a drainer is started
late (e.g., a subscriber registers while shutdown is in progress). Ensure
`launchDrainer()` refuses to start once shutdown begins (check `b.stopped`
before `wg.Add`), or otherwise synchronize drainer creation so `wg.Add` cannot
race with `wg.Wait`.
--
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]
