diff --git a/.changeset/tough-locks-talk.md b/.changeset/tough-locks-talk.md new file mode 100644 index 000000000..8ebacbba0 --- /dev/null +++ b/.changeset/tough-locks-talk.md @@ -0,0 +1,5 @@ +--- +"github.com/livekit/protocol": patch +--- + +Track holder goroutines of stuck locks (all RWMutex readers up to 8) and resolve their current stacks diff --git a/go.mod b/go.mod index 356244140..3d26c67aa 100644 --- a/go.mod +++ b/go.mod @@ -20,6 +20,7 @@ require ( github.com/mackerelio/go-osstat v0.2.8 github.com/maxbrunsfeld/counterfeiter/v6 v6.12.2 github.com/nyaruka/phonenumbers v1.8.1 + github.com/petermattis/goid v0.0.0-20260725062400-500c67a39b75 github.com/pion/logging v0.2.4 github.com/pion/sdp/v3 v3.0.19 github.com/pion/webrtc/v4 v4.2.17 diff --git a/go.sum b/go.sum index ee809f649..1597a5af8 100644 --- a/go.sum +++ b/go.sum @@ -121,6 +121,8 @@ github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJw github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M= github.com/ory/dockertest/v4 v4.0.0 h1:i19aFsO/VXE0VrMk4ifnKW4G/KIJ93PCjLOslxXoPME= github.com/ory/dockertest/v4 v4.0.0/go.mod h1:b5Ofu8VIxWNhXFvQcLu17pRNQdoUBKtXBW74G4Ygzx8= +github.com/petermattis/goid v0.0.0-20260725062400-500c67a39b75 h1:VmZ6mKVkxavKEhEy4ZYyV7BwBYBFBP0TwIqmLk84fpU= +github.com/petermattis/goid v0.0.0-20260725062400-500c67a39b75/go.mod h1:pxMtw7cyUw6B2bRH0ZBANSPg+AoSud1I1iyJHI69jH4= github.com/pion/datachannel v1.6.2 h1:7EXQ8TH3vTouBUdRWYbcX2edSx9Yj6k5zl5P+qyxEPc= github.com/pion/datachannel v1.6.2/go.mod h1:pzbdAZvyGtXbcHM1hBbsFaOTf40lZizU/dNlvVOak6E= github.com/pion/dtls/v3 v3.1.5 h1:9xJtVsHwMYeSjPp5Hh1FTis4DchnQWtnOa5o+6ygqfc= diff --git a/infra/link.pb.go b/infra/link.pb.go index ee44b5b08..97745edcd 100644 --- a/infra/link.pb.go +++ b/infra/link.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: infra/link.proto diff --git a/livekit/agent/livekit_agent_dev.pb.go b/livekit/agent/livekit_agent_dev.pb.go index b673894a8..9b3f92317 100644 --- a/livekit/agent/livekit_agent_dev.pb.go +++ b/livekit/agent/livekit_agent_dev.pb.go @@ -1,6 +1,6 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: agent/livekit_agent_dev.proto diff --git a/livekit/agent/livekit_agent_inference.pb.go b/livekit/agent/livekit_agent_inference.pb.go index 24154c774..b14c54fd8 100644 --- a/livekit/agent/livekit_agent_inference.pb.go +++ b/livekit/agent/livekit_agent_inference.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: agent/livekit_agent_inference.proto diff --git a/livekit/agent/livekit_agent_session.pb.go b/livekit/agent/livekit_agent_session.pb.go index 2268c2e1f..9b0eac540 100644 --- a/livekit/agent/livekit_agent_session.pb.go +++ b/livekit/agent/livekit_agent_session.pb.go @@ -1,6 +1,6 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: agent/livekit_agent_session.proto diff --git a/livekit/cloud_replay.pb.go b/livekit/cloud_replay.pb.go index d8c64fa71..12e72de7a 100644 --- a/livekit/cloud_replay.pb.go +++ b/livekit/cloud_replay.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: cloud_replay.proto diff --git a/livekit/livekit_agent.pb.go b/livekit/livekit_agent.pb.go index 1100ee05c..a1052dee1 100644 --- a/livekit/livekit_agent.pb.go +++ b/livekit/livekit_agent.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_agent.proto diff --git a/livekit/livekit_agent_dispatch.pb.go b/livekit/livekit_agent_dispatch.pb.go index 7b5e1b6be..0f0702e4c 100644 --- a/livekit/livekit_agent_dispatch.pb.go +++ b/livekit/livekit_agent_dispatch.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_agent_dispatch.proto diff --git a/livekit/livekit_agent_simulation.pb.go b/livekit/livekit_agent_simulation.pb.go index 82dff485d..98dd41601 100644 --- a/livekit/livekit_agent_simulation.pb.go +++ b/livekit/livekit_agent_simulation.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_agent_simulation.proto diff --git a/livekit/livekit_agent_worker.pb.go b/livekit/livekit_agent_worker.pb.go index 223829a5a..97b9ddc2e 100644 --- a/livekit/livekit_agent_worker.pb.go +++ b/livekit/livekit_agent_worker.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_agent_worker.proto diff --git a/livekit/livekit_analytics.pb.go b/livekit/livekit_analytics.pb.go index f767f0ab7..23f8572bc 100644 --- a/livekit/livekit_analytics.pb.go +++ b/livekit/livekit_analytics.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_analytics.proto diff --git a/livekit/livekit_cloud_agent.pb.go b/livekit/livekit_cloud_agent.pb.go index 58432e641..2f049e67e 100644 --- a/livekit/livekit_cloud_agent.pb.go +++ b/livekit/livekit_cloud_agent.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_cloud_agent.proto diff --git a/livekit/livekit_connector.pb.go b/livekit/livekit_connector.pb.go index 2be9c64aa..d44083615 100644 --- a/livekit/livekit_connector.pb.go +++ b/livekit/livekit_connector.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_connector.proto diff --git a/livekit/livekit_connector_twilio.pb.go b/livekit/livekit_connector_twilio.pb.go index 733bcf059..b0cb87c83 100644 --- a/livekit/livekit_connector_twilio.pb.go +++ b/livekit/livekit_connector_twilio.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_connector_twilio.proto diff --git a/livekit/livekit_connector_whatsapp.pb.go b/livekit/livekit_connector_whatsapp.pb.go index 502cf5cc9..f99fd95bf 100644 --- a/livekit/livekit_connector_whatsapp.pb.go +++ b/livekit/livekit_connector_whatsapp.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_connector_whatsapp.proto diff --git a/livekit/livekit_egress.pb.go b/livekit/livekit_egress.pb.go index 17413136c..f71ca8cac 100644 --- a/livekit/livekit_egress.pb.go +++ b/livekit/livekit_egress.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_egress.proto diff --git a/livekit/livekit_ingress.pb.go b/livekit/livekit_ingress.pb.go index 667b15c06..b05701fcf 100644 --- a/livekit/livekit_ingress.pb.go +++ b/livekit/livekit_ingress.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_ingress.proto diff --git a/livekit/livekit_internal.pb.go b/livekit/livekit_internal.pb.go index b231df286..ef6e48a30 100644 --- a/livekit/livekit_internal.pb.go +++ b/livekit/livekit_internal.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_internal.proto diff --git a/livekit/livekit_metrics.pb.go b/livekit/livekit_metrics.pb.go index 5b29b86fb..b57b0d139 100644 --- a/livekit/livekit_metrics.pb.go +++ b/livekit/livekit_metrics.pb.go @@ -1,6 +1,6 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_metrics.proto diff --git a/livekit/livekit_models.pb.go b/livekit/livekit_models.pb.go index deb149afd..19b2f14e2 100644 --- a/livekit/livekit_models.pb.go +++ b/livekit/livekit_models.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_models.proto diff --git a/livekit/livekit_phone_number.pb.go b/livekit/livekit_phone_number.pb.go index ab7d99ffd..0b94ce34d 100644 --- a/livekit/livekit_phone_number.pb.go +++ b/livekit/livekit_phone_number.pb.go @@ -1,6 +1,6 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_phone_number.proto diff --git a/livekit/livekit_room.pb.go b/livekit/livekit_room.pb.go index ed6d6a08e..b228265cb 100644 --- a/livekit/livekit_room.pb.go +++ b/livekit/livekit_room.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_room.proto diff --git a/livekit/livekit_rtc.pb.go b/livekit/livekit_rtc.pb.go index f1cbc4556..9fc24411a 100644 --- a/livekit/livekit_rtc.pb.go +++ b/livekit/livekit_rtc.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_rtc.proto diff --git a/livekit/livekit_sip.pb.go b/livekit/livekit_sip.pb.go index 50e990b66..ee6678022 100644 --- a/livekit/livekit_sip.pb.go +++ b/livekit/livekit_sip.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_sip.proto diff --git a/livekit/livekit_token_source.pb.go b/livekit/livekit_token_source.pb.go index c2e782841..156d50d3e 100644 --- a/livekit/livekit_token_source.pb.go +++ b/livekit/livekit_token_source.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_token_source.proto diff --git a/livekit/livekit_webhook.pb.go b/livekit/livekit_webhook.pb.go index 79631ce8b..f466a3809 100644 --- a/livekit/livekit_webhook.pb.go +++ b/livekit/livekit_webhook.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: livekit_webhook.proto diff --git a/livekit/logger/options.pb.go b/livekit/logger/options.pb.go index b84bfab65..768c4bcdb 100644 --- a/livekit/logger/options.pb.go +++ b/livekit/logger/options.pb.go @@ -1,6 +1,6 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: logger/options.proto diff --git a/livekit/roomrpc/siprpc/roomrpc_sip_v1.pb.go b/livekit/roomrpc/siprpc/roomrpc_sip_v1.pb.go index 81c848c26..4f2a3f39d 100644 --- a/livekit/roomrpc/siprpc/roomrpc_sip_v1.pb.go +++ b/livekit/roomrpc/siprpc/roomrpc_sip_v1.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: roomrpc/siprpc/roomrpc_sip_v1.proto diff --git a/rpc/agent.pb.go b/rpc/agent.pb.go index 98a49ca06..2d66e26f2 100644 --- a/rpc/agent.pb.go +++ b/rpc/agent.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: rpc/agent.proto diff --git a/rpc/agent_dispatch.pb.go b/rpc/agent_dispatch.pb.go index 3825da610..d61e54299 100644 --- a/rpc/agent_dispatch.pb.go +++ b/rpc/agent_dispatch.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: rpc/agent_dispatch.proto diff --git a/rpc/analytics.pb.go b/rpc/analytics.pb.go index d1c072f2c..45885e577 100644 --- a/rpc/analytics.pb.go +++ b/rpc/analytics.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: rpc/analytics.proto diff --git a/rpc/egress.pb.go b/rpc/egress.pb.go index ac36d2500..9a83890c4 100644 --- a/rpc/egress.pb.go +++ b/rpc/egress.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: rpc/egress.proto diff --git a/rpc/ingress.pb.go b/rpc/ingress.pb.go index d6402467a..119bb3166 100644 --- a/rpc/ingress.pb.go +++ b/rpc/ingress.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: rpc/ingress.proto diff --git a/rpc/io.pb.go b/rpc/io.pb.go index 863b70df6..73b8be366 100644 --- a/rpc/io.pb.go +++ b/rpc/io.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: rpc/io.proto diff --git a/rpc/keepalive.pb.go b/rpc/keepalive.pb.go index e31ee779f..36c1d58ab 100644 --- a/rpc/keepalive.pb.go +++ b/rpc/keepalive.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: rpc/keepalive.proto diff --git a/rpc/participant.pb.go b/rpc/participant.pb.go index d7fe40fd4..84da23d12 100644 --- a/rpc/participant.pb.go +++ b/rpc/participant.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: rpc/participant.proto diff --git a/rpc/room.pb.go b/rpc/room.pb.go index 97c630aa0..28a8cd92f 100644 --- a/rpc/room.pb.go +++ b/rpc/room.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: rpc/room.proto diff --git a/rpc/roommanager.pb.go b/rpc/roommanager.pb.go index 84e5f200e..bbdd62d08 100644 --- a/rpc/roommanager.pb.go +++ b/rpc/roommanager.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: rpc/roommanager.proto diff --git a/rpc/signal.pb.go b/rpc/signal.pb.go index f9efdd445..6acb08680 100644 --- a/rpc/signal.pb.go +++ b/rpc/signal.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: rpc/signal.proto diff --git a/rpc/sip.pb.go b/rpc/sip.pb.go index adfdf6663..f1b774c4d 100644 --- a/rpc/sip.pb.go +++ b/rpc/sip.pb.go @@ -14,7 +14,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: rpc/sip.proto diff --git a/rpc/whip_signal.pb.go b/rpc/whip_signal.pb.go index 6229f3b1b..bd6d57cc9 100644 --- a/rpc/whip_signal.pb.go +++ b/rpc/whip_signal.pb.go @@ -1,6 +1,6 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc v7.35.1 // source: rpc/whip_signal.proto diff --git a/utils/lock_tracker.go b/utils/lock_tracker.go index 61b6b3c30..f28702966 100644 --- a/utils/lock_tracker.go +++ b/utils/lock_tracker.go @@ -24,6 +24,7 @@ import ( "time" "unsafe" + "github.com/petermattis/goid" "golang.org/x/exp/slices" ) @@ -58,14 +59,59 @@ func updateLowResTime() { } } -var weakRefs []uintptr -var weakRefFree []int +// weakRefList is a registry of tracker pointers held as uintptrs so they +// don't keep their owners alive; finalizers clear the slots. +type weakRefList struct { + refs []uintptr + free []int + next int +} + var weakRefLock sync.Mutex +var weakRefs weakRefList + +func (l *weakRefList) add(p unsafe.Pointer) int { + weakRefLock.Lock() + defer weakRefLock.Unlock() + if fi := len(l.free) - 1; fi >= 0 { + ref := l.free[fi] + l.refs[ref] = uintptr(p) + l.free = l.free[:fi] + return ref + } + l.refs = append(l.refs, uintptr(p)) + return len(l.refs) - 1 +} + +func (l *weakRefList) remove(ref int) { + weakRefLock.Lock() + defer weakRefLock.Unlock() + l.refs[ref] = 0 + l.free = append(l.free, ref) +} + +func (l *weakRefList) count() int { + return len(l.refs) - len(l.free) +} + +// window returns the next n-element window of refs, wrapping to the start +// when the end is reached. +func (l *weakRefList) window(n int) []uintptr { + min := l.next + max := min + n + if len(l.refs) <= max { + max = len(l.refs) + l.next = 0 + } else { + l.next = max + } + return l.refs[min:max] +} func NumMutexes() int { weakRefLock.Lock() defer weakRefLock.Unlock() - return len(weakRefs) - len(weakRefFree) + return weakRefs.count() } // ScanTrackedLocks check all lock trackers @@ -74,11 +120,9 @@ func ScanTrackedLocks(threshold time.Duration) []*StuckLock { weakRefLock.Lock() defer weakRefLock.Unlock() - return scanTrackedLocks(weakRefs, minTS) + return scanTrackedLocks(weakRefs.refs, minTS) } -var nextScanMin int - // ScanTrackedLocksI check lock trackers incrementally n at a time func ScanTrackedLocksI(threshold time.Duration, n int) []*StuckLock { minTS := uint32(time.Now().Add(-threshold).Unix()) @@ -89,16 +133,7 @@ func ScanTrackedLocksI(threshold time.Duration, n int) []*StuckLock { weakRefLock.Lock() defer weakRefLock.Unlock() - min := nextScanMin - max := nextScanMin + n - if rl := len(weakRefs); rl <= max { - max = rl - nextScanMin = 0 - } else { - nextScanMin = max - } - - return scanTrackedLocks(weakRefs[min:max], minTS) + return scanTrackedLocks(weakRefs.window(n), minTS) } //go:norace @@ -111,23 +146,63 @@ func scanTrackedLocks(refs []uintptr, minTS uint32) []*StuckLock { ts := atomic.LoadUint32(&t.ts) waiting := atomic.LoadInt32(&t.waiting) if ts <= minTS && waiting > 0 { - stuck = append(stuck, &StuckLock{ - stack: slices.Clone(t.stack), - ts: ts, - waiting: waiting, - held: atomic.LoadInt32(&t.held), - }) + stuck = append(stuck, t.toStuckLock(ts, waiting)) } } } return stuck } +//go:norace +func (t *lockTracker) toStuckLock(ts uint32, waiting int32) *StuckLock { + d := &StuckLock{ + stack: slices.Clone(t.stack), + ts: ts, + waiting: waiting, + } + if gid := t.gid; gid != 0 { + d.gids = append(d.gids, gid) + d.held = 1 + } + if t.rw { + r := (*rwLockTracker)(unsafe.Pointer(t)) + d.held += r.rheld.Load() + if len(d.gids) == 0 && d.held > 0 { + d.holderStrength = HolderShared + } + for i := range r.rgids { + // dedupe: a recursive RLock occupies multiple slots with one gid + if gid := r.rgids[i].Load(); gid != 0 && !slices.Contains(d.gids, gid) { + d.gids = append(d.gids, gid) + } + } + } + return d +} + +// HolderStrength describes how a stuck lock is held. +type HolderStrength int32 + +const ( + HolderExclusive HolderStrength = iota // Mutex, RWMutex write lock + HolderShared // RWMutex read locks +) + +func (s HolderStrength) String() string { + if s == HolderShared { + return "shared" + } + return "exclusive" +} + type StuckLock struct { - stack []uintptr - ts uint32 - waiting int32 - held int32 + stack []uintptr + ts uint32 + waiting int32 + held int32 + holderStrength HolderStrength + gids []int64 + holderStacks []string } func (d *StuckLock) FirstLockedAtStack() string { @@ -165,103 +240,149 @@ func (d *StuckLock) NumGoroutineWaiting() int { return int(d.waiting) } +// HolderGoroutineIDs returns the ids of the goroutines holding the lock: the +// single holder for a Mutex, the writer or up to 8 concurrent readers for an +// RWMutex (compare with NumGoroutineHeld to detect overflow). +func (d *StuckLock) HolderGoroutineIDs() []int64 { + return d.gids +} + +// HolderStrength reports whether the lock is held exclusively or shared by +// readers. +func (d *StuckLock) HolderStrength() HolderStrength { + return d.holderStrength +} + +// HolderStacks returns the holder goroutines' stacks as resolved by +// PopulateHolderStacks, or "" if not resolved. +func (d *StuckLock) HolderStacks() string { + return strings.Join(d.holderStacks, "\n\n") +} + +// PopulateHolderStacks resolves the current stack of each stuck lock's holder +// goroutines from a single snapshot of all goroutine stacks. The snapshot +// stops the world; call it once per detection, not per lock. +func PopulateHolderStacks(stuck []*StuckLock) { + byGID := make(map[int64][]*StuckLock, len(stuck)) + for _, d := range stuck { + for _, gid := range d.gids { + byGID[gid] = append(byGID[gid], d) + } + } + if len(byGID) == 0 { + return + } + + // size for the live goroutine count up front: each doubling pass repeats + // the stop-the-world snapshot, worst exactly when diagnosing a + // goroutine-heavy stuck process + buf := make([]byte, min(max(1<<20, runtime.NumGoroutine()*4096), 1<<26)) + for { + n := runtime.Stack(buf, true) + if n < len(buf) || len(buf) >= 1<<26 { + buf = buf[:n] + break + } + buf = make([]byte, len(buf)*2) + } + + for _, g := range strings.Split(string(buf), "\n\n") { + if gid, ok := parseGoroutineHeader(g); ok { + for _, d := range byGID[gid] { + d.holderStacks = append(d.holderStacks, g) + } + } + } +} + +// parseGoroutineHeader extracts the goroutine id from a stack block formatted +// like "goroutine 123 [chan receive]:\n..." +func parseGoroutineHeader(g string) (int64, bool) { + const prefix = "goroutine " + if !strings.HasPrefix(g, prefix) { + return 0, false + } + rest := g[len(prefix):] + sp := strings.IndexByte(rest, ' ') + if sp <= 0 { + return 0, false + } + gid, err := strconv.ParseInt(rest[:sp], 10, 64) + return gid, err == nil +} + +// lockTracker tracks an exclusive lock. trackLock and trackUnlock run while +// the lock is held, so the single holder's bookkeeping needs no atomics; +// only the waiter count, maintained outside the lock, does. The scanner +// reads holder state without the lock and tolerates racy reads (go:norace). +// rw marks trackers that are really rwLockTrackers, which extend this with +// read-holder slots. type lockTracker struct { stack []uintptr ts uint32 waiting int32 - held int32 - ref int + gid int64 + rw bool } func (t *lockTracker) trackWait() { - if t != nil { - atomic.AddInt32(&t.waiting, 1) - } + atomic.AddInt32(&t.waiting, 1) } func (t *lockTracker) trackLock() { - if t != nil { - atomic.AddInt32(&t.waiting, -1) - if atomic.AddInt32(&t.held, 1) == 1 { - atomic.StoreUint32(&t.ts, atomic.LoadUint32(&lowResTime)) + atomic.AddInt32(&t.waiting, -1) + t.gid = goid.Get() + atomic.StoreUint32(&t.ts, atomic.LoadUint32(&lowResTime)) - if atomic.LoadUint32(&enableLockTrackerStackTrace) == 1 { - n := runtime.Callers(2, t.stack[:lockTrackerMaxStackDepth]) - t.stack = t.stack[:n] - } - } + if atomic.LoadUint32(&enableLockTrackerStackTrace) == 1 { + n := runtime.Callers(2, t.stack[:lockTrackerMaxStackDepth]) + t.stack = t.stack[:n] } } func (t *lockTracker) trackUnlock() { - if t != nil { - if atomic.AddInt32(&t.held, -1) == 0 { - atomic.StoreUint32(&t.ts, math.MaxUint32) - } - } + t.gid = 0 + atomic.StoreUint32(&t.ts, math.MaxUint32) } +// newLockTracker allocates a tracker without side effects; lazyInitTracker +// registers whichever allocation wins publication. func newLockTracker() *lockTracker { - t := &lockTracker{ + return &lockTracker{ stack: make([]uintptr, lockTrackerMaxStackDepth), ts: math.MaxUint32, } - - runtime.SetFinalizer(t, finalizeLockTracker) - - weakRefLock.Lock() - defer weakRefLock.Unlock() - - if fi := len(weakRefFree) - 1; fi >= 0 { - t.ref = weakRefFree[fi] - weakRefs[t.ref] = uintptr((unsafe.Pointer)(t)) - weakRefFree = weakRefFree[:fi] - } else { - t.ref = len(weakRefs) - weakRefs = append(weakRefs, uintptr((unsafe.Pointer)(t))) - } - - return t } -func finalizeLockTracker(t *lockTracker) { - weakRefLock.Lock() - defer weakRefLock.Unlock() - - weakRefs[t.ref] = 0 - weakRefFree = append(weakRefFree, t.ref) +// loadTracker returns the tracker, or nil if there is none. Unlock paths use +// it instead of lazyInitTracker: a lock being unlocked was necessarily +// locked first, so the tracker either exists or tracking was disabled at +// Lock time. +func loadTracker[T any](p **T) *T { + return (*T)(atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(p)))) } -var waiting lockTracker - -//go:linkname sync_runtime_canSpin sync.runtime_canSpin -func sync_runtime_canSpin(int) bool - -//go:linkname sync_runtime_doSpin sync.runtime_doSpin -func sync_runtime_doSpin() - -func lazyInitLockTracker(p **lockTracker) *lockTracker { +// lazyInitTracker returns the field's tracker, constructing and registering +// it on first use. construct must be free of side effects: racing +// initializers may each allocate, but only the publishing CAS winner +// registers its tracker; losers' allocations are left to the GC. +func lazyInitTracker[T any](p **T, construct func() *T) *T { if !lockTrackerEnabled { return nil } up := (*unsafe.Pointer)(unsafe.Pointer(p)) - iter := 0 - for { - if t := atomic.LoadPointer(up); t == nil { - if atomic.CompareAndSwapPointer(up, nil, (unsafe.Pointer)(&waiting)) { - atomic.StorePointer(up, (unsafe.Pointer)(newLockTracker())) - } - } else if t == (unsafe.Pointer)(&waiting) { - if sync_runtime_canSpin(iter) { - sync_runtime_doSpin() - iter++ - } else { - runtime.Gosched() - } - } else { - return (*lockTracker)(t) - } + if t := atomic.LoadPointer(up); t != nil { + return (*T)(t) + } + t := construct() + if !atomic.CompareAndSwapPointer(up, nil, unsafe.Pointer(t)) { + return (*T)(atomic.LoadPointer(up)) } + ref := weakRefs.add(unsafe.Pointer(t)) + runtime.SetFinalizer(t, func(*T) { + weakRefs.remove(ref) + }) + return t } type Mutex struct { @@ -270,45 +391,17 @@ type Mutex struct { } func (m *Mutex) Lock() { - t := lazyInitLockTracker(&m.t) - t.trackWait() + t := lazyInitTracker(&m.t, newLockTracker) + if t != nil { + t.trackWait() + defer t.trackLock() + } m.Mutex.Lock() - t.trackLock() } func (m *Mutex) Unlock() { - t := lazyInitLockTracker(&m.t) - t.trackUnlock() + if t := loadTracker(&m.t); t != nil { + t.trackUnlock() + } m.Mutex.Unlock() } - -type RWMutex struct { - sync.RWMutex - t *lockTracker -} - -func (m *RWMutex) Lock() { - t := lazyInitLockTracker(&m.t) - t.trackWait() - m.RWMutex.Lock() - t.trackLock() -} - -func (m *RWMutex) Unlock() { - t := lazyInitLockTracker(&m.t) - t.trackUnlock() - m.RWMutex.Unlock() -} - -func (m *RWMutex) RLock() { - t := lazyInitLockTracker(&m.t) - t.trackWait() - m.RWMutex.RLock() - t.trackLock() -} - -func (m *RWMutex) RUnlock() { - t := lazyInitLockTracker(&m.t) - t.trackUnlock() - m.RWMutex.RUnlock() -} diff --git a/utils/lock_tracker_rw.go b/utils/lock_tracker_rw.go new file mode 100644 index 000000000..bf8d39c96 --- /dev/null +++ b/utils/lock_tracker_rw.go @@ -0,0 +1,146 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package utils + +import ( + "math" + "runtime" + "sync" + "sync/atomic" + + "github.com/petermattis/goid" +) + +const lockTrackerMaxReadHolders = 8 + +// rwLockTracker extends lockTracker with read-holder tracking. The write +// holder uses the embedded single-holder bookkeeping (writes are exclusive); +// readers are concurrent, so their gids live in atomic slots. +// +// Read-holder accounting: trackRLock records the reader's gid in a free +// slot, or increments roverflow when all slots are taken. trackRUnlock +// removes exactly one entry: the caller's own gid if present, else an +// overflow credit, else an arbitrary slot — Go permits unlocking from a +// goroutine other than the locker, so like go-deadlock we keep the count +// consistent at the price of a possibly misattributed entry in that rare +// case. Invariant: occupied slots plus roverflow equals the reader count. +// Typed atomics guarantee 64-bit alignment regardless of field layout. +type rwLockTracker struct { + lockTracker + rheld atomic.Int32 + roverflow atomic.Int32 + rgids [lockTrackerMaxReadHolders]atomic.Int64 +} + +func (t *rwLockTracker) trackRLock() { + atomic.AddInt32(&t.waiting, -1) + + gid := goid.Get() + claimed := false + for i := range t.rgids { + if t.rgids[i].CompareAndSwap(0, gid) { + claimed = true + break + } + } + if !claimed { + t.roverflow.Add(1) + } + + if t.rheld.Add(1) == 1 { + atomic.StoreUint32(&t.ts, atomic.LoadUint32(&lowResTime)) + + if atomic.LoadUint32(&enableLockTrackerStackTrace) == 1 { + n := runtime.Callers(2, t.stack[:lockTrackerMaxStackDepth]) + t.stack = t.stack[:n] + } + } +} + +func (t *rwLockTracker) trackRUnlock() { + gid := goid.Get() + removed := false + for i := range t.rgids { + if t.rgids[i].CompareAndSwap(gid, 0) { + removed = true + break + } + } + for !removed { + if o := t.roverflow.Load(); o > 0 { + removed = t.roverflow.CompareAndSwap(o, o-1) + continue + } + removed = true + for i := range t.rgids { + if g := t.rgids[i].Load(); g != 0 && t.rgids[i].CompareAndSwap(g, 0) { + break + } + } + } + + if t.rheld.Add(-1) == 0 { + atomic.StoreUint32(&t.ts, math.MaxUint32) + } +} + +// newRWLockTracker allocates a tracker without side effects; lazyInitTracker +// registers whichever allocation wins publication. +func newRWLockTracker() *rwLockTracker { + return &rwLockTracker{ + lockTracker: lockTracker{ + stack: make([]uintptr, lockTrackerMaxStackDepth), + ts: math.MaxUint32, + rw: true, + }, + } +} + +type RWMutex struct { + sync.RWMutex + t *rwLockTracker +} + +func (m *RWMutex) Lock() { + t := lazyInitTracker(&m.t, newRWLockTracker) + if t != nil { + t.trackWait() + defer t.trackLock() + } + m.RWMutex.Lock() +} + +func (m *RWMutex) Unlock() { + if t := loadTracker(&m.t); t != nil { + t.trackUnlock() + } + m.RWMutex.Unlock() +} + +func (m *RWMutex) RLock() { + t := lazyInitTracker(&m.t, newRWLockTracker) + if t != nil { + t.trackWait() + defer t.trackRLock() + } + m.RWMutex.RLock() +} + +func (m *RWMutex) RUnlock() { + if t := loadTracker(&m.t); t != nil { + t.trackRUnlock() + } + m.RWMutex.RUnlock() +} diff --git a/utils/lock_tracker_test.go b/utils/lock_tracker_test.go index 383c41edc..b3deb3877 100644 --- a/utils/lock_tracker_test.go +++ b/utils/lock_tracker_test.go @@ -17,6 +17,7 @@ package utils_test import ( "fmt" "runtime" + "strings" "sync" "testing" "time" @@ -92,6 +93,202 @@ func TestFirstLockStackTrace(t *testing.T) { m.Unlock() } +func parkHoldingLock(release chan struct{}) { + <-release +} + +func TestHolderStacks(t *testing.T) { + t.Cleanup(cleanupTest) + require.Nil(t, utils.ScanTrackedLocks(time.Millisecond)) + + m := &utils.RWMutex{} + release := make(chan struct{}) + done := make(chan struct{}) + + var held sync.WaitGroup + held.Add(2) + for range 2 { + go func() { + m.RLock() + held.Done() + parkHoldingLock(release) + m.RUnlock() + }() + } + + held.Wait() + go func() { + m.Lock() + noop() + m.Unlock() + close(done) + }() + + time.Sleep(100 * time.Millisecond) + locks := utils.ScanTrackedLocks(time.Millisecond) + require.NotNil(t, locks) + require.Len(t, locks[0].HolderGoroutineIDs(), 2) + require.Equal(t, utils.HolderShared, locks[0].HolderStrength()) + require.Equal(t, "", locks[0].HolderStacks()) + + utils.PopulateHolderStacks(locks) + for _, gid := range locks[0].HolderGoroutineIDs() { + require.Contains(t, locks[0].HolderStacks(), fmt.Sprintf("goroutine %d ", gid)) + } + require.Equal(t, 2, strings.Count(locks[0].HolderStacks(), "parkHoldingLock(")) + + close(release) + <-done +} + +func TestHolderCrossGoroutineUnlock(t *testing.T) { + t.Cleanup(cleanupTest) + require.Nil(t, utils.ScanTrackedLocks(time.Millisecond)) + + m := &utils.Mutex{} + locked := make(chan struct{}) + go func() { + m.Lock() + close(locked) + }() + <-locked + m.Unlock() // legal: different goroutine than the locker + + release := make(chan struct{}) + held := make(chan struct{}) + done := make(chan struct{}) + go func() { + m.Lock() + close(held) + parkHoldingLock(release) + m.Unlock() + }() + <-held + go func() { + m.Lock() + noop() + m.Unlock() + close(done) + }() + + time.Sleep(100 * time.Millisecond) + locks := utils.ScanTrackedLocks(time.Millisecond) + require.NotNil(t, locks) + // the cross-goroutine unlock must not leak the original locker's slot + require.Len(t, locks[0].HolderGoroutineIDs(), 1) + require.Equal(t, utils.HolderExclusive, locks[0].HolderStrength()) + utils.PopulateHolderStacks(locks) + require.Contains(t, locks[0].HolderStacks(), "parkHoldingLock(") + + close(release) + <-done +} + +func TestHolderOverflow(t *testing.T) { + t.Cleanup(cleanupTest) + require.Nil(t, utils.ScanTrackedLocks(time.Millisecond)) + + const readers = 10 // more than the 8 holder slots + + m := &utils.RWMutex{} + release := make(chan struct{}) + done := make(chan struct{}) + + var held, finished sync.WaitGroup + held.Add(readers) + finished.Add(readers) + for range readers { + go func() { + m.RLock() + held.Done() + parkHoldingLock(release) + m.RUnlock() + finished.Done() + }() + } + + held.Wait() + go func() { + m.Lock() + noop() + m.Unlock() + close(done) + }() + + time.Sleep(100 * time.Millisecond) + locks := utils.ScanTrackedLocks(time.Millisecond) + require.NotNil(t, locks) + require.Len(t, locks[0].HolderGoroutineIDs(), 8) + require.Equal(t, readers, locks[0].NumGoroutineHeld()) + + close(release) + finished.Wait() + <-done + + // slot and overflow accounting must drain fully for the next episode + release2 := make(chan struct{}) + held2 := make(chan struct{}) + done2 := make(chan struct{}) + go func() { + m.RLock() + close(held2) + parkHoldingLock(release2) + m.RUnlock() + }() + <-held2 + go func() { + m.Lock() + noop() + m.Unlock() + close(done2) + }() + time.Sleep(100 * time.Millisecond) + locks = utils.ScanTrackedLocks(time.Millisecond) + require.NotNil(t, locks) + require.Len(t, locks[0].HolderGoroutineIDs(), 1) + + close(release2) + <-done2 +} + +func TestRecursiveRLockDedupe(t *testing.T) { + t.Cleanup(cleanupTest) + require.Nil(t, utils.ScanTrackedLocks(time.Millisecond)) + + m := &utils.RWMutex{} + release := make(chan struct{}) + held := make(chan struct{}) + done := make(chan struct{}) + go func() { + m.RLock() + m.RLock() + close(held) + parkHoldingLock(release) + m.RUnlock() + m.RUnlock() + }() + <-held + go func() { + m.Lock() + noop() + m.Unlock() + close(done) + }() + + time.Sleep(100 * time.Millisecond) + locks := utils.ScanTrackedLocks(time.Millisecond) + require.NotNil(t, locks) + // recursive RLock occupies two slots but is one holder + require.Len(t, locks[0].HolderGoroutineIDs(), 1) + require.Equal(t, 2, locks[0].NumGoroutineHeld()) + + utils.PopulateHolderStacks(locks) + require.Equal(t, 1, strings.Count(locks[0].HolderStacks(), "parkHoldingLock(")) + + close(release) + <-done +} + func TestMutexFinalizer(t *testing.T) { cleanupTest() require.Equal(t, 0, utils.NumMutexes()) @@ -301,6 +498,29 @@ func BenchmarkLockTracker(b *testing.B) { }) } +func BenchmarkLockTrackerParallel(b *testing.B) { + b.Run("wrapped rwmutex rlock", func(b *testing.B) { + var m utils.RWMutex + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + m.RLock() + noop() + m.RUnlock() + } + }) + }) + b.Run("native rwmutex rlock", func(b *testing.B) { + var m sync.RWMutex + b.RunParallel(func(pb *testing.PB) { + for pb.Next() { + m.RLock() + noop() + m.RUnlock() + } + }) + }) +} + func BenchmarkGetBlocked(b *testing.B) { for n := 100; n <= 1000000; n *= 100 { n := n