AlexStocks commented on code in PR #3482:
URL: https://github.com/apache/dubbo-go/pull/3482#discussion_r3578175443
##########
registry/polaris/registry.go:
##########
@@ -188,14 +219,73 @@ func (pr *polarisRegistry) LoadSubscribeInstances(url
*common.URL, notify regist
return perrors.New(fmt.Sprintf("could not query the instances
for serviceName=%s,namespace=%s,error=%v",
serviceName, pr.namespace, err))
}
+ initialSubscribeInstances := make([]model.Instance, 0,
len(resp.Instances))
for i := range resp.Instances {
if newUrl := generateUrl(resp.Instances[i]); newUrl != nil {
notify.Notify(®istry.ServiceEvent{Action:
remoting.EventTypeAdd, Service: newUrl})
+ initialSubscribeInstances =
append(initialSubscribeInstances, resp.Instances[i])
}
}
+ pr.storeInitialSubscribeInstances(key, initialSubscribeInstances)
return nil
}
+func newInitialSubscribeInstancesKey(serviceName string, notify
registry.NotifyListener) initialSubscribeInstancesKey {
+ value := reflect.ValueOf(notify)
+ if !value.IsValid() || value.Kind() != reflect.Pointer || value.IsNil()
{
Review Comment:
[P1] 非指针监听器会共享同一份初始基线
这里把所有非 `Pointer` 的 `NotifyListener` 都退化为仅包含 `serviceName` 的
key，但该接口允许合法的值类型实现。两个不同的可比较值监听器为同一 service 先后执行 `LoadSubscribeInstances`
时，第二次会覆盖第一次的基线；随后第一个监听器进入 `Subscribe` 会按错误快照对账，原 stale-provider
问题仍会在这类实现上出现。WSL overlay 用例已稳定复现实例 A 的基线被实例 B 覆盖。建议对动态类型可比较的监听器直接保留其身份（例如检查
`reflect.TypeOf(notify).Comparable()`），对不可比较实现返回明确错误或分配独立身份，并补两个值类型监听器的回归测试。
##########
registry/polaris/core.go:
##########
@@ -72,60 +88,172 @@ func (watcher *PolarisServiceWatcher) lazyRun() {
})
}
+func (watcher *PolarisServiceWatcher) addSubscriberWithInitialSnapshot(
+ initialSnapshot []model.Instance,
+ subscriber item,
+) {
+ state := &subscriberState{
+ notify: subscriber,
+ initialSnapshot: copyInstances(initialSnapshot),
+ }
+
+ func() {
+ watcher.lock.Lock()
+ defer watcher.lock.Unlock()
+
+ watcher.subscribers = append(watcher.subscribers, state)
+ if watcher.snapshotReady {
+ watcher.reconcileSubscriberLocked(state)
+ }
+ }()
+
+ // Start only after the subscriber is registered and any available
current
+ // snapshot has been replayed.
+ watcher.lazyRun()
+}
+
+// missingInitialInstances returns initial - current by model.InstanceKey while
+// preserving the order of the initial snapshot.
+func missingInitialInstances(initial []model.Instance, current
[]model.Instance) []model.Instance {
+ currentInstances := make(map[model.InstanceKey]struct{}, len(current))
+ for _, instance := range current {
+ currentInstances[instance.GetInstanceKey()] = struct{}{}
+ }
+
+ missing := make([]model.Instance, 0, len(initial))
+ for _, instance := range initial {
+ if _, ok := currentInstances[instance.GetInstanceKey()]; !ok {
+ missing = append(missing, instance)
+ }
+ }
+ return missing
+}
+
+// handleWatchSnapshot replaces the watcher's current state and reconciles each
+// subscriber's own synchronous-load baseline exactly once.
+func (watcher *PolarisServiceWatcher) handleWatchSnapshot(current
[]model.Instance) {
+ watcher.lock.Lock()
+ defer watcher.lock.Unlock()
+
+ watcher.currentInstances = copyInstances(current)
+ watcher.snapshotReady = true
+ for _, subscriber := range watcher.subscribers {
+ if subscriber.reconciled {
+ watcher.notifySubscriberLocked(subscriber,
remoting.EventTypeAdd, watcher.currentInstances)
+ continue
+ }
+ watcher.reconcileSubscriberLocked(subscriber)
+ }
+}
+
+func (watcher *PolarisServiceWatcher) reconcileSubscriberLocked(subscriber
*subscriberState) {
+ missing := missingInitialInstances(subscriber.initialSnapshot,
watcher.currentInstances)
+ if len(missing) > 0 {
+ watcher.notifySubscriberLocked(subscriber,
remoting.EventTypeDel, missing)
+ }
+ watcher.notifySubscriberLocked(subscriber, remoting.EventTypeAdd,
watcher.currentInstances)
+ subscriber.reconciled = true
+ subscriber.initialSnapshot = nil
+}
+
// startWatch start run work to watch target service by polaris
func (watcher *PolarisServiceWatcher) startWatch() {
for {
- resp, err :=
watcher.consumer.WatchService(watcher.subscribeParam)
- if err != nil {
+ if err := watcher.watchOnce(); err != nil {
time.Sleep(time.Duration(500 * time.Millisecond))
- continue
}
- watcher.notifyAllSubscriber(&config_center.ConfigChangeEvent{
- Value: resp.GetAllInstancesResp.Instances,
- ConfigType: remoting.EventTypeAdd,
- })
-
- for event := range resp.EventChannel {
- eType := event.GetSubScribeEventType()
- if eType == internalapi.EventInstance {
- insEvent := event.(*model.InstanceEvent)
-
- if insEvent.AddEvent != nil {
-
watcher.notifyAllSubscriber(&config_center.ConfigChangeEvent{
- Value:
insEvent.AddEvent.Instances,
- ConfigType:
remoting.EventTypeAdd,
- })
- }
- if insEvent.UpdateEvent != nil {
- instances := make([]model.Instance,
len(insEvent.UpdateEvent.UpdateList))
- for i := range
insEvent.UpdateEvent.UpdateList {
- instances[i] =
insEvent.UpdateEvent.UpdateList[i].After
- }
-
watcher.notifyAllSubscriber(&config_center.ConfigChangeEvent{
- Value: instances,
- ConfigType:
remoting.EventTypeUpdate,
- })
- }
- if insEvent.DeleteEvent != nil {
-
watcher.notifyAllSubscriber(&config_center.ConfigChangeEvent{
- Value:
insEvent.DeleteEvent.Instances,
- ConfigType:
remoting.EventTypeDel,
- })
- }
+ }
+}
+
+func (watcher *PolarisServiceWatcher) watchOnce() error {
+ resp, err := watcher.consumer.WatchService(watcher.subscribeParam)
+ if err != nil {
+ return err
+ }
+ watcher.handleWatchSnapshot(resp.GetAllInstancesResp.Instances)
+ for event := range resp.EventChannel {
+ if event.GetSubScribeEventType() == internalapi.EventInstance {
+
watcher.handleInstanceEvent(event.(*model.InstanceEvent))
+ }
+ }
+ return nil
+}
+
+func (watcher *PolarisServiceWatcher) handleInstanceEvent(event
*model.InstanceEvent) {
+ if event == nil {
+ return
+ }
+
+ watcher.lock.Lock()
+ defer watcher.lock.Unlock()
+
+ if event.AddEvent != nil {
+ instances := copyInstances(event.AddEvent.Instances)
+ for _, instance := range instances {
+ watcher.upsertCurrentInstanceLocked(instance)
+ }
+
watcher.notifyReconciledSubscribersLocked(remoting.EventTypeAdd, instances)
+ }
+ if event.UpdateEvent != nil {
+ instances := make([]model.Instance, 0,
len(event.UpdateEvent.UpdateList))
+ for _, update := range event.UpdateEvent.UpdateList {
+ if update.Before.GetInstanceKey() !=
update.After.GetInstanceKey() {
+
watcher.removeCurrentInstancesLocked([]model.Instance{update.Before})
}
+ watcher.upsertCurrentInstanceLocked(update.After)
+ instances = append(instances, update.After)
}
+
watcher.notifyReconciledSubscribersLocked(remoting.EventTypeUpdate, instances)
+ }
+ if event.DeleteEvent != nil {
+ instances := copyInstances(event.DeleteEvent.Instances)
+ watcher.removeCurrentInstancesLocked(instances)
+
watcher.notifyReconciledSubscribersLocked(remoting.EventTypeDel, instances)
+ }
+}
+func (watcher *PolarisServiceWatcher) upsertCurrentInstanceLocked(instance
model.Instance) {
+ key := instance.GetInstanceKey()
+ for i := range watcher.currentInstances {
+ if watcher.currentInstances[i].GetInstanceKey() == key {
+ watcher.currentInstances[i] = instance
+ return
+ }
}
+ watcher.currentInstances = append(watcher.currentInstances, instance)
}
-// notifyAllSubscriber notify config_center.ConfigChangeEvent to all subscriber
-func (watcher *PolarisServiceWatcher) notifyAllSubscriber(event
*config_center.ConfigChangeEvent) {
- watcher.lock.RLock()
- defer watcher.lock.RUnlock()
+func (watcher *PolarisServiceWatcher) removeCurrentInstancesLocked(instances
[]model.Instance) {
+ keys := make(map[model.InstanceKey]struct{}, len(instances))
+ for _, instance := range instances {
+ keys[instance.GetInstanceKey()] = struct{}{}
+ }
- for i := 0; i < len(watcher.subscribers); i++ {
- subscriber := watcher.subscribers[i]
- subscriber(event.ConfigType, event.Value.([]model.Instance))
+ current := watcher.currentInstances[:0]
+ for _, instance := range watcher.currentInstances {
+ if _, remove := keys[instance.GetInstanceKey()]; !remove {
+ current = append(current, instance)
+ }
}
+ watcher.currentInstances = current
Review Comment:
[P2] 删除后应清空切片尾部引用
这里用 `[:0]` 原地压缩后直接缩短长度，backing array 尾部仍保留已删除的 `model.Instance` 接口值及其
metadata 引用。watcher 生命周期很长且实例发生大规模下线后，这些对象会一直被底层数组持有，直到后续 append 覆盖或 watcher
释放。建议在赋值前对 `watcher.currentInstances[len(current):]` 执行 `clear`，并补一个大量
add/delete 后检查状态容量复用的测试。
--
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]
