refactor(channelserver): replace global stagesLock with sync.Map-backed StageMap

The global stagesLock sync.RWMutex protected map[string]*Stage, causing
all stage operations to contend on a single lock even for unrelated
stages. Any stage creation or deletion blocked all reads server-wide.

Replace with a typed StageMap wrapper around sync.Map which provides
lock-free reads and allows concurrent writes to disjoint keys. Per-stage
sync.RWMutex remains unchanged for protecting individual stage state.

StageMap exposes Get, GetOrCreate, StoreIfAbsent, Store, Delete, and
Range methods. Updated ~50 call sites across 6 production files and
9 test files.
This commit is contained in:
Houmgaor
2026-02-22 15:47:21 +01:00
parent 2a5cd50e3f
commit ad4afb4d3b
15 changed files with 207 additions and 221 deletions

View File

@@ -56,15 +56,17 @@ func (r *LocalChannelRegistry) DisconnectUser(cids []uint32) {
func (r *LocalChannelRegistry) FindChannelForStage(stageSuffix string) string {
for _, channel := range r.channels {
channel.stagesLock.RLock()
for id := range channel.stages {
var gid string
channel.stages.Range(func(id string, _ *Stage) bool {
if strings.HasSuffix(id, stageSuffix) {
gid := channel.GlobalID
channel.stagesLock.RUnlock()
return gid
gid = channel.GlobalID
return false // stop iteration
}
return true
})
if gid != "" {
return gid
}
channel.stagesLock.RUnlock()
}
return ""
}
@@ -105,13 +107,14 @@ func (r *LocalChannelRegistry) SearchStages(stagePrefix string, max int) []Stage
if len(results) >= max {
break
}
c.stagesLock.RLock()
for _, stage := range c.stages {
cIP := net.ParseIP(c.IP).To4()
cPort := c.Port
c.stages.Range(func(_ string, stage *Stage) bool {
if len(results) >= max {
break
return false
}
if !strings.HasPrefix(stage.id, stagePrefix) {
continue
return true
}
stage.RLock()
bin0 := stage.rawBinaryData[stageBinaryKey{1, 0}]
@@ -125,8 +128,8 @@ func (r *LocalChannelRegistry) SearchStages(stagePrefix string, max int) []Stage
copy(bin3Copy, bin3)
results = append(results, StageSnapshot{
ServerIP: net.ParseIP(c.IP).To4(),
ServerPort: c.Port,
ServerIP: cIP,
ServerPort: cPort,
StageID: stage.id,
ClientCount: len(stage.clients) + len(stage.reservedClientSlots),
Reserved: len(stage.reservedClientSlots),
@@ -136,8 +139,8 @@ func (r *LocalChannelRegistry) SearchStages(stagePrefix string, max int) []Stage
RawBinData3: bin3Copy,
})
stage.RUnlock()
}
c.stagesLock.RUnlock()
return true
})
}
return results
}