From ad0fc1953d1efd4c56810bb8bb18f77dbe7bb372 Mon Sep 17 00:00:00 2001 From: krauthaufen Date: Wed, 2 Sep 2026 14:51:16 +0200 Subject: [PATCH] Fix AddCallback deadlock under concurrent transactions (#120) MultiCallbackObject.check blocked on the callback-table lock while holding the object's monitor, the reverse of setMultiCallback's order. Release now uses Monitor.TryEnter and retries on the next Mark/remove when contended. --- src/FSharp.Data.Adaptive/Core/Callbacks.fs | 16 ++- .../FSharp.Data.Adaptive.Tests/Callbacks.fs | 101 ++++++++++++++++++ 2 files changed, 115 insertions(+), 2 deletions(-) diff --git a/src/FSharp.Data.Adaptive/Core/Callbacks.fs b/src/FSharp.Data.Adaptive/Core/Callbacks.fs index 102f61e7..8b39f0c5 100644 --- a/src/FSharp.Data.Adaptive/Core/Callbacks.fs +++ b/src/FSharp.Data.Adaptive/Core/Callbacks.fs @@ -67,7 +67,7 @@ type internal MultiCallbackObject(table : ConditionalWeakTable + let release () = #if !FABLE_COMPILER if cbs.Count = 0 then // since there are not more live callbacks we'd like to release this object #else @@ -85,7 +85,19 @@ type internal MultiCallbackObject(table : ConditionalWeakTable diff --git a/src/Test/FSharp.Data.Adaptive.Tests/Callbacks.fs b/src/Test/FSharp.Data.Adaptive.Tests/Callbacks.fs index f554cb31..cf48c1b9 100644 --- a/src/Test/FSharp.Data.Adaptive.Tests/Callbacks.fs +++ b/src/Test/FSharp.Data.Adaptive.Tests/Callbacks.fs @@ -160,3 +160,104 @@ let ``[AddWeakCallback] not surviving GC``() = + +// https://github.com/fsprojects/FSharp.Data.Adaptive/issues/120 +[] +let ``[AddCallback] concurrent add/dispose vs transact does not deadlock`` () = + let c = cval 1 + use __ = c.AddCallback(fun (_ : int) -> ()) + + let running = ref true + let writer = + Thread((fun () -> + while running.Value do + transact (fun () -> + Thread.Yield() |> ignore + c.Value <- c.Value + 1 + ) + ), IsBackground = true) + + let worker = + Thread((fun () -> + for _ in 1 .. 200000 do + let d = c.AddCallback(fun (_ : int) -> ()) + Thread.Yield() |> ignore + d.Dispose() + ), IsBackground = true) + + writer.Start() + worker.Start() + let finished = worker.Join(TimeSpan.FromSeconds 60.0) + running.Value <- false + finished |> should be True + writer.Join(TimeSpan.FromSeconds 10.0) |> should be True + +/// The private callback table (CallbackExtensions.callbackObjects) via reflection. +let private callbackTable () = + let asm = typeof.Assembly + let modType = asm.GetType("FSharp.Data.Adaptive.CallbackExtensions", true) + let flags = System.Reflection.BindingFlags.Static ||| System.Reflection.BindingFlags.NonPublic ||| System.Reflection.BindingFlags.Public + let isTable (t : Type) = t.IsGenericType && t.GetGenericTypeDefinition() = typedefof> + let fromProp = + modType.GetProperties(flags) |> Array.tryFind (fun p -> isTable p.PropertyType) |> Option.map (fun p -> p.GetValue(null)) + let fromField () = + asm.GetTypes() + |> Seq.collect (fun t -> t.GetFields(flags)) + |> Seq.tryFind (fun f -> isTable f.FieldType) + |> Option.map (fun f -> f.GetValue(null)) + match fromProp with + | Some t -> t + | None -> + match fromField() with + | Some t -> t + | None -> failwith "callback table not found" + +/// When the table lock is contended during Dispose, the MultiCallbackObject cannot release +/// itself and lingers as an (empty) output of the cval. That must be harmless: a new +/// subscription reuses it correctly and the next marking releases it. +[] +let ``[AddCallback] lingering callback object after contended dispose is harmless`` () = + let table = callbackTable () + let c = cval 1 + let outputs = (c :> IAdaptiveObject).Outputs + + let d = c.AddCallback(fun (_ : int) -> ()) + outputs.IsEmpty |> should be False + + // hold the table lock on another thread while disposing -> check's TryEnter fails + use locked = new ManualResetEventSlim(false) + use release = new ManualResetEventSlim(false) + let holder = Thread((fun () -> lock table (fun () -> locked.Set(); release.Wait())), IsBackground = true) + holder.Start() + locked.Wait() + d.Dispose() + release.Set() + holder.Join() + + // the empty callback object is still registered (linger) + outputs.IsEmpty |> should be False + + // a new subscription reuses the lingering object and works + let mutable fired = [] + let d2 = c.AddCallback(fun (v : int) -> fired <- v :: fired) + fired |> should equal [1] + transact (fun () -> c.Value <- 2) + fired |> should equal [2; 1] + outputs.IsEmpty |> should be False + d2.Dispose() + outputs.IsEmpty |> should be True + + // linger again, but this time nobody subscribes: the next marking releases it + let d3 = c.AddCallback(fun (_ : int) -> ()) + use locked2 = new ManualResetEventSlim(false) + use release2 = new ManualResetEventSlim(false) + let holder2 = Thread((fun () -> lock table (fun () -> locked2.Set(); release2.Wait())), IsBackground = true) + holder2.Start() + locked2.Wait() + d3.Dispose() + release2.Set() + holder2.Join() + outputs.IsEmpty |> should be False + transact (fun () -> c.Value <- 3) + outputs.IsEmpty |> should be True + fired |> should equal [2; 1]