diff --git a/.changeset/sync-raii-holders.md b/.changeset/sync-raii-holders.md new file mode 100644 index 000000000..aa2c994ec --- /dev/null +++ b/.changeset/sync-raii-holders.md @@ -0,0 +1,5 @@ +--- +"github.com/livekit/protocol": minor +--- + +Add Synchronized[T], a guarded-value mutex whose Holder handle makes releases exact and the current holder visible to stuck-lock diagnostics diff --git a/utils/synchronized.go b/utils/synchronized.go new file mode 100644 index 000000000..28d483d4e --- /dev/null +++ b/utils/synchronized.go @@ -0,0 +1,91 @@ +// 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 ( + "runtime" + "sync" + "unsafe" +) + +// Synchronized guards a value behind a mutex whose Lock returns the guarded +// value and a Holder handle; Unlock is a method on the handle, so a release +// always matches its acquisition, even when the lock is handed off between +// goroutines. The embedded lockTracker registers in the shared scan registry, +// so stuck-lock diagnostics (ScanTrackedLocks, PopulateHolderStacks) work +// like they do for Mutex. The zero value is ready to use and guards the zero +// value of T. +type Synchronized[T any] struct { + base syncBase + value T +} + +func NewSynchronized[T any](value T) *Synchronized[T] { + return &Synchronized[T]{value: value} +} + +// Lock acquires the mutex and returns the guarded value with the Holder that +// releases it. The value must not be used after Holder.Unlock. +func (s *Synchronized[T]) Lock() (*T, Holder) { + b := &s.base + enabled := lockTrackerEnabled + if enabled { + b.t.trackWait() + } + b.mu.Lock() + if enabled { + if !b.registered { + b.registered = true + b.t.stack = make([]uintptr, lockTrackerMaxStackDepth) + registerSyncBase(s, &b.t) + } + b.t.trackLock() + } + return &s.value, Holder{base: b} +} + +// Holder releases one acquisition of a Synchronized lock. Treat it as +// move-only: copies (e.g. sending it to another goroutine for handoff) share +// the acquisition, which must be unlocked exactly once across all copies. +type Holder struct { + base *syncBase +} + +func (h *Holder) Unlock() { + b := h.base + if b == nil { + panic("utils.Holder: unlocked twice") + } + h.base = nil + b.t.trackUnlock() + b.mu.Unlock() +} + +type syncBase struct { + mu sync.Mutex + registered bool // guarded by mu + t lockTracker +} + +// registerSyncBase adds the lock to the scan registry on first Lock. The +// finalizer goes on the owner — the tracker is embedded in its allocation — +// and captures only the slot index, since capturing the tracker would root +// the owner and leak it. +func registerSyncBase(owner any, t *lockTracker) { + ref := weakRefs.add(unsafe.Pointer(t)) + runtime.SetFinalizer(owner, func(any) { + weakRefs.remove(ref) + }) +} diff --git a/utils/synchronized_test.go b/utils/synchronized_test.go new file mode 100644 index 000000000..2c9c5f529 --- /dev/null +++ b/utils/synchronized_test.go @@ -0,0 +1,88 @@ +// 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_test + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/livekit/protocol/utils" +) + +func TestSynchronizedBasic(t *testing.T) { + t.Cleanup(cleanupTest) + + s := utils.NewSynchronized(map[string]int{}) + v, h := s.Lock() + (*v)["a"] = 1 + h.Unlock() + + v, h = s.Lock() + require.Equal(t, 1, (*v)["a"]) + h.Unlock() + require.Panics(t, func() { h.Unlock() }) +} + +func TestSynchronizedStuck(t *testing.T) { + t.Cleanup(cleanupTest) + require.Nil(t, utils.ScanTrackedLocks(time.Millisecond)) + + s := utils.NewSynchronized(0) + release := make(chan struct{}) + handoff := make(chan utils.Holder, 1) + done := make(chan struct{}) + + go func() { + _, h := s.Lock() + handoff <- h + parkHoldingLock(release) + }() + h := <-handoff + go func() { + v, h := s.Lock() + *v++ + h.Unlock() + close(done) + }() + + time.Sleep(100 * time.Millisecond) + locks := utils.ScanTrackedLocks(time.Millisecond) + require.NotNil(t, locks) + require.Len(t, locks[0].HolderGoroutineIDs(), 1) + require.Equal(t, 1, locks[0].NumGoroutineHeld()) + require.Equal(t, 1, locks[0].NumGoroutineWaiting()) + + utils.PopulateHolderStacks(locks) + require.Contains(t, locks[0].HolderStacks(), "parkHoldingLock(") + + // handle-based release: unlocking from a goroutine other than the locker + // is exact, and the waiter proceeds + h.Unlock() + close(release) + <-done + require.Nil(t, utils.ScanTrackedLocks(time.Millisecond)) +} + +func BenchmarkSynchronized(b *testing.B) { + s := utils.NewSynchronized(0) + b.ReportAllocs() + for i := 0; i < b.N; i++ { + v, h := s.Lock() + *v++ + h.Unlock() + } +}