解决inst.Lives 线程安全问题

This commit is contained in:
kira1928 committed 2026-02-19 17:11:00 +09:00
1 parent 4c0d2fa6b6
commit 507e7c286b
12 files changed
+301 -59

No files matched your search

+4
View File
@@ -1,6 +1,10 @@
// VS Code 调试配置模板
// 使用方法:复制此文件为 launch.json(即去掉 .example 后缀)
// 你可以在 launch.json 中自由添加或修改配置,不会影响仓库
//
// ⚠️ 调试器要求:Go 1.25+ 默认生成 DWARFv5 调试符号,需要更新 Delve:
// go install github.com/go-delve/delve/cmd/dlv@latest
// 如果调试时报错 "DWARFv5 or later",请运行上述命令更新 Delve。
{
"version": "0.1.0",
"configurations": [
-5
View File
@@ -41,11 +41,6 @@
"command": "make dev-incremental",
"isBackground": false,
"problemMatcher": "$go",
"options": {
"env": {
"GOEXPERIMENT": "nodwarf5"
}
},
"group": {
"kind": "build",
"isDefault": true
+2 -2
View File
@@ -111,7 +111,7 @@ openlist:
port: 5244
data_path: ""
update:
auto_check: true
auto_check: false
check_interval_hours: 6
auto_download: true
include_prerelease: false
include_prerelease: true
+5 -5
View File
@@ -222,6 +222,7 @@ func main() {
telemetry.Init(consts.AppVersion, true)
inst := new(instance.Instance)
inst.Lives = instance.NewLiveMap()
// TODO: Replace gcache with hashmap.
// LRU seems not necessary here.
inst.Cache = gcache.New(4096).LRU().Build()
@@ -467,7 +468,7 @@ func main() {
}
var statuses []iostats.RecorderStatus
for liveID, l := range inst.Lives {
for liveID, l := range inst.Lives.Snapshot() {
status, err := rm.GetRecorderStatus(context.Background(), liveID)
if err != nil {
continue
@@ -501,7 +502,6 @@ func main() {
// 初始化 live rooms
// 第一步:立即为所有配置的直播间创建 InitializingLive,让前端可以看到
inst.Lives = make(map[types.LiveID]live.Live)
cfg := configs.GetCurrentConfig()
// 确保所有平台都有最小访问限制(用于控制并行初始化时的请求速度)
@@ -538,11 +538,11 @@ func main() {
logger.WithField("url", room).Error(liveErr.Error())
continue
}
if _, ok := inst.Lives[l.GetLiveId()]; ok {
if inst.Lives.Has(l.GetLiveId()) {
logger.Errorf("%v is exist!", room)
continue
}
inst.Lives[l.GetLiveId()] = l
inst.Lives.Set(l.GetLiveId(), l)
configs.SetLiveRoomId(room.Url, l.GetLiveId())
// 从数据库加载缓存的直播间信息,用于在初始化完成前显示
@@ -648,7 +648,7 @@ func main() {
})
logger.Infof("Created %d live rooms (%d listening, %d not listening)",
len(inst.Lives), len(listeningRooms), len(nonListeningRooms))
inst.Lives.Len(), len(listeningRooms), len(nonListeningRooms))
c := make(chan os.Signal, 1)
// 使用 os.Interrupt 更跨平台,在 Windows 上 SIGHUP 可能不被支持
+1 -3
View File
@@ -4,14 +4,12 @@ import (
"sync"
"github.com/bililive-go/bililive-go/src/interfaces"
"github.com/bililive-go/bililive-go/src/live"
"github.com/bililive-go/bililive-go/src/types"
"github.com/bluele/gcache"
)
type Instance struct {
WaitGroup sync.WaitGroup
Lives map[types.LiveID]live.Live
Lives *LiveMap
Cache gcache.Cache
Server interfaces.Module
EventDispatcher interfaces.Module
+150
View File
@@ -0,0 +1,150 @@
package instance
import (
"sync"
"github.com/bililive-go/bililive-go/src/live"
"github.com/bililive-go/bililive-go/src/types"
)
// LiveMap 是一个并发安全的 Live 对象 map。
// 解决了 inst.Lives 在 HTTP handler goroutine 中迭代、
// 同时在事件处理器 goroutine 中写入导致的 concurrent map read/write panic。
//
// 所有公开方法对 nil receiver 安全:当 LiveMap 指针为 nil 时,
// 读方法返回零值(0/false/nil),写方法为 no-op。
// 这保证了即使 inst.Lives 尚未初始化(例如启动早期),
// HTTP handler 和各类 manager 调用时也不会触发 nil 指针 panic。
type LiveMap struct {
mu sync.RWMutex
m map[types.LiveID]live.Live
}
// NewLiveMap 创建一个新的 LiveMap。
func NewLiveMap() *LiveMap {
return &LiveMap{
m: make(map[types.LiveID]live.Live),
}
}
// Get 根据 LiveID 获取 Live 对象。
// nil receiver 安全:返回 (nil, false)。
func (lm *LiveMap) Get(id types.LiveID) (live.Live, bool) {
if lm == nil {
return nil, false
}
lm.mu.RLock()
defer lm.mu.RUnlock()
v, ok := lm.m[id]
return v, ok
}
// Set 设置一个 Live 对象。
// nil receiver 安全:no-op。
func (lm *LiveMap) Set(id types.LiveID, l live.Live) {
if lm == nil {
return
}
lm.mu.Lock()
defer lm.mu.Unlock()
lm.m[id] = l
}
// Delete 删除一个 Live 对象。
// nil receiver 安全:no-op。
func (lm *LiveMap) Delete(id types.LiveID) {
if lm == nil {
return
}
lm.mu.Lock()
defer lm.mu.Unlock()
delete(lm.m, id)
}
// Has 检查是否存在指定 LiveID。
// nil receiver 安全:返回 false。
func (lm *LiveMap) Has(id types.LiveID) bool {
if lm == nil {
return false
}
lm.mu.RLock()
defer lm.mu.RUnlock()
_, ok := lm.m[id]
return ok
}
// Len 返回 map 中元素的数量。
// nil receiver 安全:返回 0。
func (lm *LiveMap) Len() int {
if lm == nil {
return 0
}
lm.mu.RLock()
defer lm.mu.RUnlock()
return len(lm.m)
}
// Range 遍历 map 中的所有元素。
// 回调函数返回 false 时停止遍历。
// nil receiver 安全:直接返回,不调用回调。
//
// 警告:在回调函数执行期间持有读锁。回调中不得调用 Set、Delete、
// ReplaceKey 等写操作,否则会导致死锁(sync.RWMutex 不可重入)。
// 如果需要在遍历过程中修改 map,请使用 Snapshot() 获取快照后再操作。
func (lm *LiveMap) Range(fn func(id types.LiveID, l live.Live) bool) {
if lm == nil {
return
}
lm.mu.RLock()
defer lm.mu.RUnlock()
for id, l := range lm.m {
if !fn(id, l) {
break
}
}
}
// Snapshot 返回 map 的一个浅拷贝快照。
// 适用于需要长时间处理的场景(避免长时间持有读锁),
// 或者需要在遍历过程中修改 LiveMap 的场景。
// nil receiver 安全:返回空 map(非 nil)。
func (lm *LiveMap) Snapshot() map[types.LiveID]live.Live {
if lm == nil {
return make(map[types.LiveID]live.Live)
}
lm.mu.RLock()
defer lm.mu.RUnlock()
snapshot := make(map[types.LiveID]live.Live, len(lm.m))
for id, l := range lm.m {
snapshot[id] = l
}
return snapshot
}
// SetIfAbsent 如果 key 不存在则设置,返回是否设置成功。
// nil receiver 安全:返回 false。
func (lm *LiveMap) SetIfAbsent(id types.LiveID, l live.Live) bool {
if lm == nil {
return false
}
lm.mu.Lock()
defer lm.mu.Unlock()
if _, ok := lm.m[id]; ok {
return false
}
lm.m[id] = l
return true
}
// ReplaceKey 原子地删除旧 key 并设置新 key。
// 用于 InitializingLive 完成初始化后替换 LiveID 的场景。
// nil receiver 安全:no-op。
func (lm *LiveMap) ReplaceKey(oldID types.LiveID, newID types.LiveID, l live.Live) {
if lm == nil {
return
}
lm.mu.Lock()
defer lm.mu.Unlock()
delete(lm.m, oldID)
lm.m[newID] = l
}
+96
View File
@@ -0,0 +1,96 @@
package instance
import (
"testing"
"github.com/bililive-go/bililive-go/src/live"
"github.com/bililive-go/bililive-go/src/types"
)
// TestLiveMap_NilReceiver 验证所有方法在 nil receiver 上不会 panic。
// 这是为了防止 inst.Lives 尚未初始化时(启动早期),
// HTTP handler 和各类 manager 调用方法导致 nil 指针 panic。
func TestLiveMap_NilReceiver(t *testing.T) {
var lm *LiveMap
t.Run("Get", func(t *testing.T) {
v, ok := lm.Get("test")
if ok || v != nil {
t.Errorf("Get on nil LiveMap should return (nil, false), got (%v, %v)", v, ok)
}
})
t.Run("Has", func(t *testing.T) {
if lm.Has("test") {
t.Error("Has on nil LiveMap should return false")
}
})
t.Run("Len", func(t *testing.T) {
if lm.Len() != 0 {
t.Errorf("Len on nil LiveMap should return 0, got %d", lm.Len())
}
})
t.Run("Range", func(t *testing.T) {
called := false
lm.Range(func(id types.LiveID, l live.Live) bool {
called = true
return true
})
if called {
t.Error("Range on nil LiveMap should not call the callback")
}
})
t.Run("Snapshot", func(t *testing.T) {
snap := lm.Snapshot()
if snap == nil {
t.Error("Snapshot on nil LiveMap should return non-nil empty map")
}
if len(snap) != 0 {
t.Errorf("Snapshot on nil LiveMap should return empty map, got %d entries", len(snap))
}
})
t.Run("SetIfAbsent", func(t *testing.T) {
if lm.SetIfAbsent("test", nil) {
t.Error("SetIfAbsent on nil LiveMap should return false")
}
})
t.Run("Set", func(t *testing.T) {
lm.Set("test", nil) // 不应 panic
})
t.Run("Delete", func(t *testing.T) {
lm.Delete("test") // 不应 panic
})
t.Run("ReplaceKey", func(t *testing.T) {
lm.ReplaceKey("old", "new", nil) // 不应 panic
})
}
// TestLiveMap_BasicOperations 验证正常初始化后的基本操作。
func TestLiveMap_BasicOperations(t *testing.T) {
lm := NewLiveMap()
if lm.Len() != 0 {
t.Fatalf("new LiveMap should be empty, got Len=%d", lm.Len())
}
if lm.Has("id1") {
t.Error("Has should return false for non-existent key")
}
if _, ok := lm.Get("id1"); ok {
t.Error("Get should return false for non-existent key")
}
// Snapshot 应该返回空 map
snap := lm.Snapshot()
if len(snap) != 0 {
t.Errorf("Snapshot of empty LiveMap should be empty, got %d", len(snap))
}
}
+4 -8
View File
@@ -48,17 +48,13 @@ func (m *manager) registryListener(ctx context.Context, ed events.Dispatcher) {
inst := instance.GetInstance(ctx)
logger := originalLive.GetLogger()
// 删除旧的 InitializingLive
oldLiveId := initializingLive.GetLiveId()
delete(inst.Lives, oldLiveId)
// 将原始 Live 包装为 WrappedLive(使用全局缓存)
// 这样 parseInfo 可以通过缓存获取信息
// 传入 ctx 以便调度器可以被统一取消
oldLiveId := initializingLive.GetLiveId()
wrappedLive := live.NewWrappedLive(ctx, originalLive, inst.Cache)
// 添加新的 Live 对象到 Lives map
inst.Lives[wrappedLive.GetLiveId()] = wrappedLive
// 原子地替换 Lives map 中的条目(删除旧 InitializingLive,添加新 wrappedLive)
inst.Lives.ReplaceKey(oldLiveId, wrappedLive.GetLiveId(), wrappedLive)
// 直接将已有的 info 存入缓存,避免再次调用 GetInfo() 触发 API 请求
// info.Live 需要更新为新的 wrappedLive,确保后续 JSON 序列化正确
@@ -88,7 +84,7 @@ func (m *manager) registryListener(ctx context.Context, ed events.Dispatcher) {
func (m *manager) Start(ctx context.Context) error {
inst := instance.GetInstance(ctx)
if cfg := configs.GetCurrentConfig(); (cfg != nil && cfg.RPC.Enable) || len(inst.Lives) > 0 {
if cfg := configs.GetCurrentConfig(); (cfg != nil && cfg.RPC.Enable) || inst.Lives.Len() > 0 {
inst.WaitGroup.Add(1)
}
m.registryListener(ctx, inst.EventDispatcher.(events.Dispatcher))
+1 -1
View File
@@ -57,7 +57,7 @@ func bool2float64(b bool) float64 {
func (c collector) Collect(ch chan<- prometheus.Metric) {
wg := sync.WaitGroup{}
for id, l := range c.inst.Lives {
for id, l := range c.inst.Lives.Snapshot() {
wg.Add(1)
bilisentry.Go(func() {
defer wg.Done()
+1 -1
View File
@@ -110,7 +110,7 @@ func (m *manager) registryListener(ctx context.Context, ed events.Dispatcher) {
func (m *manager) Start(ctx context.Context) error {
inst := instance.GetInstance(ctx)
if cfg := configs.GetCurrentConfig(); (cfg != nil && cfg.RPC.Enable) || len(inst.Lives) > 0 {
if cfg := configs.GetCurrentConfig(); (cfg != nil && cfg.RPC.Enable) || inst.Lives.Len() > 0 {
inst.WaitGroup.Add(1)
}
m.registryListener(ctx, inst.EventDispatcher.(events.Dispatcher))
+29 -27
View File
@@ -93,9 +93,10 @@ func extractStreamAttributeCombinations(streams []*live.AvailableStreamInfo) []m
func getAllLives(writer http.ResponseWriter, r *http.Request) {
inst := instance.GetInstance(r.Context())
lives := liveSlice(make([]*live.Info, 0, 4))
for _, v := range inst.Lives {
inst.Lives.Range(func(_ types.LiveID, v live.Live) bool {
lives = append(lives, parseInfo(r.Context(), v))
}
return true
})
sort.Sort(lives)
writeJSON(writer, lives)
}
@@ -103,7 +104,7 @@ func getAllLives(writer http.ResponseWriter, r *http.Request) {
func getLive(writer http.ResponseWriter, r *http.Request) {
inst := instance.GetInstance(r.Context())
vars := mux.Vars(r)
liveObj, ok := inst.Lives[types.LiveID(vars["id"])]
liveObj, ok := inst.Lives.Get(types.LiveID(vars["id"]))
if !ok {
writeJsonWithStatusCode(writer, http.StatusNotFound, commonResp{
ErrNo: http.StatusNotFound,
@@ -347,7 +348,7 @@ func getLiveLogs(writer http.ResponseWriter, r *http.Request) {
liveID := types.LiveID(vars["id"])
// 查找直播间
liveInstance, ok := inst.Lives[liveID]
liveInstance, ok := inst.Lives.Get(liveID)
if !ok {
writeJsonWithStatusCode(writer, http.StatusNotFound, commonResp{
ErrNo: http.StatusNotFound,
@@ -388,7 +389,7 @@ func parseLiveAction(writer http.ResponseWriter, r *http.Request) {
inst := instance.GetInstance(r.Context())
vars := mux.Vars(r)
resp := commonResp{}
live, ok := inst.Lives[types.LiveID(vars["id"])]
live, ok := inst.Lives.Get(types.LiveID(vars["id"]))
if !ok {
resp.ErrNo = http.StatusNotFound
resp.ErrMsg = fmt.Sprintf("live id: %s can not find", vars["id"])
@@ -520,7 +521,7 @@ func switchStreamHandler(writer http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
resp := commonResp{}
live, ok := inst.Lives[types.LiveID(vars["id"])]
live, ok := inst.Lives.Get(types.LiveID(vars["id"]))
if !ok {
resp.ErrNo = http.StatusNotFound
resp.ErrMsg = fmt.Sprintf("直播间 ID: %s 未找到", vars["id"])
@@ -698,8 +699,7 @@ func addLiveImpl(ctx context.Context, urlStr string, isListen bool) (info *live.
}
// 记录 LiveId 到全局配置(并发安全)
configs.SetLiveRoomId(u.String(), newLive.GetLiveId())
if _, ok := inst.Lives[newLive.GetLiveId()]; !ok {
inst.Lives[newLive.GetLiveId()] = newLive
if inst.Lives.SetIfAbsent(newLive.GetLiveId(), newLive) {
if isListen {
inst.ListenerManager.(listeners.Manager).AddListener(ctx, newLive)
}
@@ -727,7 +727,7 @@ func addLiveImpl(ctx context.Context, urlStr string, isListen bool) (info *live.
func removeLive(writer http.ResponseWriter, r *http.Request) {
inst := instance.GetInstance(r.Context())
vars := mux.Vars(r)
live, ok := inst.Lives[types.LiveID(vars["id"])]
live, ok := inst.Lives.Get(types.LiveID(vars["id"]))
if !ok {
writeJsonWithStatusCode(writer, http.StatusNotFound, commonResp{
ErrNo: http.StatusNotFound,
@@ -756,7 +756,7 @@ func removeLiveImpl(ctx context.Context, live live.Live) error {
return err
}
}
delete(inst.Lives, liveId)
inst.Lives.Delete(liveId)
if _, err := configs.RemoveLiveRoomByUrl(live.GetRawUrl()); err != nil {
return err
}
@@ -863,7 +863,7 @@ func applyLiveRoomsByConfig(ctx context.Context, oldConfig *configs.Config, newC
return err
}
} else {
live, ok := inst.Lives[types.LiveID(room.LiveId)]
live, ok := inst.Lives.Get(types.LiveID(room.LiveId))
if !ok {
return fmt.Errorf("live id: %s can not find", room.LiveId)
}
@@ -887,7 +887,7 @@ func applyLiveRoomsByConfig(ctx context.Context, oldConfig *configs.Config, newC
for _, room := range loopRooms {
if _, ok := newUrlMap[room.Url]; !ok {
// remove live
live, ok := inst.Lives[types.LiveID(room.LiveId)]
live, ok := inst.Lives.Get(types.LiveID(room.LiveId))
if !ok {
return fmt.Errorf("live id: %s can not find", room.LiveId)
}
@@ -1037,7 +1037,7 @@ func getPlatformStats(writer http.ResponseWriter, r *http.Request) {
}
// 从缓存获取直播间信息(不触发网络请求)
if liveInstance, ok := inst.Lives[room.LiveId]; ok {
if liveInstance, ok := inst.Lives.Get(room.LiveId); ok {
if obj, err := inst.Cache.Get(liveInstance); err == nil {
if info, ok := obj.(*live.Info); ok && info != nil {
roomInfo["host_name"] = info.HostName
@@ -2196,10 +2196,10 @@ func getLiveHostCookie(writer http.ResponseWriter, r *http.Request) {
inst := instance.GetInstance(r.Context())
hostCookieMap := make(map[string]*live.InfoCookie)
keys := make([]string, 0)
for _, v := range inst.Lives {
inst.Lives.Range(func(_ types.LiveID, v live.Live) bool {
urltmp, _ := url.Parse(v.GetRawUrl())
if _, ok := hostCookieMap[urltmp.Host]; ok {
continue
return true
}
host := urltmp.Host
platformName := v.GetPlatformCNName()
@@ -2211,7 +2211,8 @@ func getLiveHostCookie(writer http.ResponseWriter, r *http.Request) {
hostCookieMap[host] = tmp
}
keys = append(keys, host)
}
return true
})
sort.Strings(keys)
result := make([]*live.InfoCookie, 0)
for _, v := range keys {
@@ -2261,22 +2262,23 @@ func putLiveHostCookie(writer http.ResponseWriter, r *http.Request) {
if tmpurl.Host != host {
continue
}
live := inst.Lives[v.LiveId]
if live == nil {
liveObj, _ := inst.Lives.Get(v.LiveId)
if liveObj == nil {
// fallback search by URL
for _, l := range inst.Lives {
inst.Lives.Range(func(_ types.LiveID, l live.Live) bool {
if l.GetRawUrl() == v.Url {
live = l
break
liveObj = l
return false
}
}
return true
})
}
if live == nil {
if liveObj == nil {
applog.GetLogger().Warn("can't find live by id or url: " + string(v.LiveId) + " " + v.Url)
continue
}
live.UpdateLiveOptionsbyConfig(ctx, &v)
liveObj.UpdateLiveOptionsbyConfig(ctx, &v)
}
if err := newCfg.Marshal(); err != nil {
applog.GetLogger().Error("failed to persistence config: " + err.Error())
@@ -2293,7 +2295,7 @@ func getLiveSessionHistory(writer http.ResponseWriter, r *http.Request) {
liveID := vars["id"]
// 检查直播间是否存在
if _, ok := inst.Lives[types.LiveID(liveID)]; !ok {
if !inst.Lives.Has(types.LiveID(liveID)) {
writeJsonWithStatusCode(writer, http.StatusNotFound, commonResp{
ErrNo: http.StatusNotFound,
ErrMsg: fmt.Sprintf("live id: %s can not find", liveID),
@@ -2336,7 +2338,7 @@ func getLiveNameHistory(writer http.ResponseWriter, r *http.Request) {
liveID := vars["id"]
// 检查直播间是否存在
if _, ok := inst.Lives[types.LiveID(liveID)]; !ok {
if !inst.Lives.Has(types.LiveID(liveID)) {
writeJsonWithStatusCode(writer, http.StatusNotFound, commonResp{
ErrNo: http.StatusNotFound,
ErrMsg: fmt.Sprintf("live id: %s can not find", liveID),
@@ -2387,7 +2389,7 @@ func getLiveHistory(writer http.ResponseWriter, r *http.Request) {
liveID := vars["id"]
// 检查直播间是否存在
if _, ok := inst.Lives[types.LiveID(liveID)]; !ok {
if !inst.Lives.Has(types.LiveID(liveID)) {
writeJsonWithStatusCode(writer, http.StatusNotFound, commonResp{
ErrNo: http.StatusNotFound,
ErrMsg: fmt.Sprintf("live id: %s can not find", liveID),
+8 -7
View File
@@ -210,10 +210,11 @@ func osrpGetTasks(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
inst := instance.GetInstance(ctx)
tasks := make([]OSRPTaskInfo, 0, len(inst.Lives))
for _, l := range inst.Lives {
tasks := make([]OSRPTaskInfo, 0, inst.Lives.Len())
inst.Lives.Range(func(_ types.LiveID, l live.Live) bool {
tasks = append(tasks, convertLiveToOSRPTask(ctx, l))
}
return true
})
osrpWriteSuccess(w, OSRPTaskListResponse{
Tasks: tasks,
@@ -228,7 +229,7 @@ func osrpGetTask(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
id := vars["id"]
l, ok := inst.Lives[types.LiveID(id)]
l, ok := inst.Lives.Get(types.LiveID(id))
if !ok {
osrpWriteError(w, http.StatusNotFound, "TASK_NOT_FOUND", "任务不存在")
return
@@ -267,7 +268,7 @@ func osrpAddTask(w http.ResponseWriter, r *http.Request) {
// 获取刚添加的 live
inst := instance.GetInstance(ctx)
l, ok := inst.Lives[info.Live.GetLiveId()]
l, ok := inst.Lives.Get(info.Live.GetLiveId())
if !ok {
osrpWriteError(w, http.StatusInternalServerError, "INTERNAL_ERROR", "添加成功但无法获取任务")
return
@@ -286,7 +287,7 @@ func osrpDeleteTask(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
id := vars["id"]
l, ok := inst.Lives[types.LiveID(id)]
l, ok := inst.Lives.Get(types.LiveID(id))
if !ok {
osrpWriteError(w, http.StatusNotFound, "TASK_NOT_FOUND", "任务不存在")
return
@@ -318,7 +319,7 @@ func osrpTaskAction(w http.ResponseWriter, r *http.Request) {
return
}
l, ok := inst.Lives[types.LiveID(id)]
l, ok := inst.Lives.Get(types.LiveID(id))
if !ok {
osrpWriteError(w, http.StatusNotFound, "TASK_NOT_FOUND", "任务不存在")
return