diff --git a/adapter/provider/adapter.go b/adapter/provider/adapter.go index c834223f..c8cc4d4e 100644 --- a/adapter/provider/adapter.go +++ b/adapter/provider/adapter.go @@ -28,6 +28,7 @@ type Adapter struct { logger log.ContextLogger providerType string providerTag string + outboundAccess sync.RWMutex outbounds []adapter.Outbound outboundsByTag map[string]adapter.Outbound ticker *time.Ticker @@ -85,7 +86,10 @@ func (a *Adapter) Start() error { if a.history == nil { a.history = urltest.NewHistoryStorage() } - go a.loopCheck() + if a.enabled { + a.ticker = time.NewTicker(a.interval) + go a.loopCheck() + } return nil } @@ -98,13 +102,14 @@ func (a *Adapter) Tag() string { } func (a *Adapter) Outbounds() []adapter.Outbound { + a.outboundAccess.RLock() + defer a.outboundAccess.RUnlock() return a.outbounds } func (a *Adapter) Outbound(tag string) (adapter.Outbound, bool) { - if a.outboundsByTag == nil { - return nil, false - } + a.outboundAccess.RLock() + defer a.outboundAccess.RUnlock() detour, ok := a.outboundsByTag[tag] return detour, ok } @@ -156,11 +161,13 @@ func (a *Adapter) UpdateOutbounds(oldOpts []option.Outbound, newOpts []option.Ou outbounds = append(outbounds, outbound) outboundsByTag[tag] = outbound } + a.outboundAccess.Lock() + a.outbounds = outbounds + a.outboundsByTag = outboundsByTag + a.outboundAccess.Unlock() if a.enabled && a.history != nil { go a.HealthCheck(a.ctx) } - a.outbounds = outbounds - a.outboundsByTag = outboundsByTag } func (a *Adapter) HealthCheck(ctx context.Context) (map[string]uint16, error) { @@ -196,8 +203,11 @@ func (a *Adapter) Close() error { if a.ticker != nil { a.ticker.Stop() } + a.outboundAccess.Lock() outbounds := a.outbounds a.outbounds = nil + a.outboundsByTag = nil + a.outboundAccess.Unlock() var err error for _, ob := range outbounds { if err2 := a.outbound.Remove(ob.Tag()); err2 != nil { @@ -210,10 +220,6 @@ func (a *Adapter) Close() error { } func (a *Adapter) loopCheck() { - if !a.enabled { - return - } - a.ticker = time.NewTicker(a.interval) a.healthcheck(a.ctx) for { select { @@ -234,7 +240,7 @@ func (a *Adapter) healthcheck(ctx context.Context) (map[string]uint16, error) { b, _ := batch.New(ctx, batch.WithConcurrencyNum[any](10)) var resultAccess sync.Mutex checked := make(map[string]bool) - for _, detour := range a.outbounds { + for _, detour := range a.Outbounds() { tag := detour.Tag() if checked[tag] { continue @@ -265,7 +271,8 @@ func (a *Adapter) healthcheck(ctx context.Context) (map[string]uint16, error) { } func (a *Adapter) removeUseless(newOpts []option.Outbound) { - if len(a.outbounds) == 0 { + outbounds := a.Outbounds() + if len(outbounds) == 0 { return } exists := make(map[string]bool) @@ -278,7 +285,7 @@ func (a *Adapter) removeUseless(newOpts []option.Outbound) { } exists[tag] = true } - for _, opt := range a.outbounds { + for _, opt := range outbounds { if !exists[opt.Tag()] { if err := a.outbound.Remove(opt.Tag()); err != nil { a.logger.Error(err, "close outbound [", opt.Tag(), "]") diff --git a/provider/local/provider.go b/provider/local/provider.go index 01e2491f..c989b55c 100644 --- a/provider/local/provider.go +++ b/provider/local/provider.go @@ -102,13 +102,17 @@ func (s *ProviderLocal) Start(stage adapter.StartStage, scope *adapter.Scope) er return err } s.UpdateGroups() + err = s.Adapter.Start() + if err != nil { + return err + } if s.watcher != nil { - err := s.watcher.Start() + err = s.watcher.Start() if err != nil { s.logger.Error(E.Cause(err, "watch provider file")) } } - return s.Adapter.Start() + return nil } func (s *ProviderLocal) UpdatedAt() time.Time { @@ -133,5 +137,5 @@ func (s *ProviderLocal) reloadFile(path string) error { } func (s *ProviderLocal) Close() error { - return common.Close(&s.Adapter, common.PtrOrNil(s.watcher)) + return common.Close(common.PtrOrNil(s.watcher), &s.Adapter) } diff --git a/provider/remote/provider.go b/provider/remote/provider.go index 3b9a3e07..4c2a5e0a 100644 --- a/provider/remote/provider.go +++ b/provider/remote/provider.go @@ -135,9 +135,13 @@ func (s *ProviderRemote) Start(stage adapter.StartStage, scope *adapter.Scope) e } else { s.dialer = s.outbound.Default() } - + err := s.Adapter.Start() + if err != nil { + return err + } + s.ticker = time.NewTicker(s.updateInterval) go s.loopUpdate() - return s.Adapter.Start() + return nil } func (s *ProviderRemote) References() []string { @@ -289,22 +293,17 @@ func (s *ProviderRemote) fetch(ctx context.Context) error { func (s *ProviderRemote) loopUpdate() { if time.Since(s.lastUpdated) < s.updateInterval { - select { - case <-s.ctx.Done(): - return - case <-time.After(time.Until(s.lastUpdated.Add(s.updateInterval))): - s.updateOnce() - } + s.ticker.Reset(time.Until(s.lastUpdated.Add(s.updateInterval))) } else { s.updateOnce() } - s.ticker = time.NewTicker(s.updateInterval) for { runtime.GC() select { case <-s.ctx.Done(): return case <-s.ticker.C: + s.ticker.Reset(s.updateInterval) s.updateOnce() } }