daemon: Throttle group and outbound subscription pushes

- Limit SubscribeGroups and SubscribeOutbounds to one push per 250ms, coalescing bursts of URLTest updates
This commit is contained in:
世界
2026-08-30 17:41:45 +08:00
parent 87c4f84a89
commit 20b4b8e17d
+62
View File
@@ -36,6 +36,8 @@ import (
const APIVersion = 3 const APIVersion = 3
const urlTestPushMinInterval = 250 * time.Millisecond
var _ StartedServiceServer = (*StartedService)(nil) var _ StartedServiceServer = (*StartedService)(nil)
type StartedService struct { type StartedService struct {
@@ -488,6 +490,7 @@ func (s *StartedService) SubscribeGroups(empty *emptypb.Empty, server grpc.Serve
return err return err
} }
defer s.serviceStatusObserver.UnSubscribe(statusSubscription) defer s.serviceStatusObserver.UnSubscribe(statusSubscription)
var lastSendTime time.Time
for { for {
s.serviceAccess.RLock() s.serviceAccess.RLock()
var groups *Groups var groups *Groups
@@ -501,6 +504,7 @@ func (s *StartedService) SubscribeGroups(empty *emptypb.Empty, server grpc.Serve
if err != nil { if err != nil {
return err return err
} }
lastSendTime = time.Now()
select { select {
case <-subscription: case <-subscription:
case <-statusSubscription: case <-statusSubscription:
@@ -513,6 +517,34 @@ func (s *StartedService) SubscribeGroups(empty *emptypb.Empty, server grpc.Serve
case <-statusDone: case <-statusDone:
return nil return nil
} }
throttleDelay := urlTestPushMinInterval - time.Since(lastSendTime)
if throttleDelay <= 0 {
continue
}
throttleTimer := time.NewTimer(throttleDelay)
select {
case <-throttleTimer.C:
case <-s.ctx.Done():
throttleTimer.Stop()
return s.ctx.Err()
case <-server.Context().Done():
throttleTimer.Stop()
return server.Context().Err()
case <-done:
throttleTimer.Stop()
return nil
case <-statusDone:
throttleTimer.Stop()
return nil
}
select {
case <-subscription:
default:
}
select {
case <-statusSubscription:
default:
}
} }
} }
@@ -1122,6 +1154,7 @@ func (s *StartedService) SubscribeOutbounds(_ *emptypb.Empty, server grpc.Server
return err return err
} }
defer s.serviceStatusObserver.UnSubscribe(statusSubscription) defer s.serviceStatusObserver.UnSubscribe(statusSubscription)
var lastSendTime time.Time
for { for {
s.serviceAccess.RLock() s.serviceAccess.RLock()
boxService := s.instance boxService := s.instance
@@ -1157,6 +1190,7 @@ func (s *StartedService) SubscribeOutbounds(_ *emptypb.Empty, server grpc.Server
if err != nil { if err != nil {
return err return err
} }
lastSendTime = time.Now()
select { select {
case <-subscription: case <-subscription:
case <-statusSubscription: case <-statusSubscription:
@@ -1169,6 +1203,34 @@ func (s *StartedService) SubscribeOutbounds(_ *emptypb.Empty, server grpc.Server
case <-statusDone: case <-statusDone:
return nil return nil
} }
throttleDelay := urlTestPushMinInterval - time.Since(lastSendTime)
if throttleDelay <= 0 {
continue
}
throttleTimer := time.NewTimer(throttleDelay)
select {
case <-throttleTimer.C:
case <-s.ctx.Done():
throttleTimer.Stop()
return s.ctx.Err()
case <-server.Context().Done():
throttleTimer.Stop()
return server.Context().Err()
case <-done:
throttleTimer.Stop()
return nil
case <-statusDone:
throttleTimer.Stop()
return nil
}
select {
case <-subscription:
default:
}
select {
case <-statusSubscription:
default:
}
} }
} }