Fix AddCallback deadlock under concurrent transactions

src/FSharp.Data.Adaptive/Core/Callbacks.fs

@@ -67,7 +67,7 @@ type internal MultiCallbackObject(table : ConditionalWeakTable<IAdaptiveObject,
             // deciding to release this object before setMultiCallback can fetch this instance for a callback
             // the Count must be checked after the lock since we need to be sure no one is add a subscription
             // while we are in here.
-            lock table (fun _ ->                 
+            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<IAdaptiveObject,
                     false
                 else
                     true
-            )
+            #if FABLE_COMPILER
+            lock table release
+            #else
+            // We are called with the monitor of x held (Mark runs under Transaction's EnterWrite, remove
+            // under lock x) while setMultiCallback holds the table lock and then enters x in Subscribe.
+            // Blocking on the table here closes that cycle (issue #120), so only release when the table
+            // is uncontended; otherwise stay alive and let the next Mark/remove retry the release.
+            if Monitor.TryEnter table then
+                try release ()
+                finally Monitor.Exit table
+            else
+                true
+            #endif
 
     let remove (x : MultiCallbackObject) (id : int) =
         lock x (fun () ->

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
+[<Test>]
+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<IAdaptiveObject>.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<System.Runtime.CompilerServices.ConditionalWeakTable<_,_>>
+    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.
+[<Test>]
+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]