mirror of
https://github.com/fatedier/frp.git
synced 2026-03-13 13:29:14 +08:00
c7ac12ea0f
* server/group: refactor group package with shared abstractions and fix concurrency issues Extract common patterns into reusable components: - groupRegistry[G]: generic concurrent map for group lifecycle management - baseGroup: shared plumbing for listener-based groups (TCP, HTTPS, TCPMux) - Listener: unified virtual listener replacing 3 identical implementations Fix concurrency issues: - Stale-pointer race: isCurrent check + errGroupStale + controller retry loops - Worker generation safety: pass realLn and acceptCh as params instead of reading mutable fields - Connection leak: close conn on worker panic recovery path - ABBA deadlock in HTTP UnRegister: consistent lock ordering (group.mu -> registry.mu) - Round-robin overflow in HTTPGroup: use unsigned modulo Add unit tests (17 tests) for registry, listener, and baseGroup. Add TCPMux group load balancing e2e test. * server/group: replace tautological assertion with require.NotPanics * server/group: remove blank line between doc comment and type declaration
60 lines
1.2 KiB
Go
60 lines
1.2 KiB
Go
package group
|
|
|
|
import (
|
|
"sync"
|
|
)
|
|
|
|
// groupRegistry is a concurrent map of named groups with
|
|
// automatic creation on first access.
|
|
type groupRegistry[G any] struct {
|
|
groups map[string]G
|
|
mu sync.Mutex
|
|
}
|
|
|
|
func newGroupRegistry[G any]() groupRegistry[G] {
|
|
return groupRegistry[G]{
|
|
groups: make(map[string]G),
|
|
}
|
|
}
|
|
|
|
func (r *groupRegistry[G]) getOrCreate(key string, newFn func() G) G {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
g, ok := r.groups[key]
|
|
if !ok {
|
|
g = newFn()
|
|
r.groups[key] = g
|
|
}
|
|
return g
|
|
}
|
|
|
|
func (r *groupRegistry[G]) get(key string) (G, bool) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
g, ok := r.groups[key]
|
|
return g, ok
|
|
}
|
|
|
|
// isCurrent returns true if key exists in the registry and matchFn
|
|
// returns true for the stored value.
|
|
func (r *groupRegistry[G]) isCurrent(key string, matchFn func(G) bool) bool {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
g, ok := r.groups[key]
|
|
return ok && matchFn(g)
|
|
}
|
|
|
|
// removeIf atomically looks up the group for key, calls fn on it,
|
|
// and removes the entry if fn returns true.
|
|
func (r *groupRegistry[G]) removeIf(key string, fn func(G) bool) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
g, ok := r.groups[key]
|
|
if !ok {
|
|
return
|
|
}
|
|
if fn(g) {
|
|
delete(r.groups, key)
|
|
}
|
|
}
|