@@ -238,6 +238,225 @@ subscribeCoincidenceInner inner outerHeight subscribedUnsafe = do
238238 defer $ SomeResetCoincidence subscription $ if height > outerHeight then Just subscribedUnsafe else Nothing
239239 return (innerOcc, height, innerSubd)
240240
241+--------------------------------------------------------------------------------
242+-- Subscriber
243+--------------------------------------------------------------------------------
244+
245+data Subscriber x a
246+ = forall b. SubscriberPush (a -> ComputeM x (Maybe b)) (PushSubscribed x a b) --TODO: Why does this have the pushCompute in it?
247+ | forall k. GCompare k => SubscriberMerge !(k a) (MergeSubscribed x k) --TODO: Can we inline the GCompare?
248+ | forall k. (a ~ R.PatchDMap (DMap k (Event x)), GCompare k) => SubscriberMergeChange (MergeSubscribed x k)
249+ | forall k. (GCompare k, a ~ DMap k Identity) => SubscriberFan (FanSubscribed x k)
250+ | forall p b. (a ~ p b, R.Patch p) => SubscriberHold !(Hold x p b)
251+ | forall b. (a ~ Identity b) => SubscriberHoldIdentity !(Hold x Identity b)
252+ | SubscriberSwitch (SwitchSubscribed x a)
253+ | forall b. a ~ Event x b => SubscriberCoincidenceOuter (CoincidenceSubscribed x b)
254+ | SubscriberCoincidenceInner (CoincidenceSubscribed x a)
255+ | SubscriberHandle
256+
257+-- These function are constructor functions that are marked NOINLINE so they are
258+-- opaque to GHC. If we do not do this, then GHC will sometimes fuse the constructor away
259+-- so any weak references that are attached to the constructors will have their
260+-- finalizer run. Using the opaque constructor, does not see the
261+-- constructor application, so it behaves like an IORef and cannot be fused away.
262+--
263+-- The result is also evaluated to WHNF, since forcing a thunk invalidates
264+-- the weak pointer to it in some cases.
265+
266+--TODO: I think all of these noinlines are now unnecessary
267+{-# NOINLINE newSubscriberPush #-}
268+newSubscriberPush :: (a -> ComputeM x (Maybe b)) -> PushSubscribed x a b -> IO (Subscriber x a)
269+newSubscriberPush compute subd = return $! SubscriberPush compute subd
270+
271+{-# RULES
272+ "newSubscriberHold/Identity" forall h. newSubscriberHold h = newSubscriberHoldIdentity h
273+ #-}
274+{-# NOINLINE newSubscriberHold #-}
275+newSubscriberHold :: R.Patch p => Hold x p a -> IO (Subscriber x (p a))
276+newSubscriberHold h = return $! SubscriberHold h
277+
278+{-# NOINLINE newSubscriberHoldIdentity #-}
279+newSubscriberHoldIdentity :: Hold x Identity a -> IO (Subscriber x (Identity a))
280+newSubscriberHoldIdentity h = return $! SubscriberHoldIdentity h
281+
282+{-# NOINLINE newSubscriberMerge #-}
283+newSubscriberMerge :: GCompare k => k a -> MergeSubscribed x k -> IO (Subscriber x a)
284+newSubscriberMerge k subd = return $! SubscriberMerge k subd
285+
286+{-# NOINLINE newSubscriberMergeChange #-}
287+newSubscriberMergeChange :: GCompare k => MergeSubscribed x k -> IO (Subscriber x (R.PatchDMap (DMap k (Event x))))
288+newSubscriberMergeChange subd = return $! SubscriberMergeChange subd
289+
290+{-# NOINLINE newSubscriberFan #-}
291+newSubscriberFan :: GCompare k => FanSubscribed x k -> IO (Subscriber x (DMap k Identity))
292+newSubscriberFan subd = return $! SubscriberFan subd
293+
294+{-# NOINLINE newSubscriberSwitch #-}
295+newSubscriberSwitch :: SwitchSubscribed x a -> IO (Subscriber x a)
296+newSubscriberSwitch subd = return $! SubscriberSwitch subd
297+
298+{-# NOINLINE newSubscriberCoincidenceOuter #-}
299+newSubscriberCoincidenceOuter :: CoincidenceSubscribed x b -> IO (Subscriber x (Event x b))
300+newSubscriberCoincidenceOuter subd = return $! SubscriberCoincidenceOuter subd
301+
302+{-# NOINLINE newSubscriberCoincidenceInner #-}
303+newSubscriberCoincidenceInner :: CoincidenceSubscribed x a -> IO (Subscriber x a)
304+newSubscriberCoincidenceInner subd = return $! SubscriberCoincidenceInner subd
305+
306+invalidateSubscriberHeight :: Height -> Subscriber x a -> IO ()
307+invalidateSubscriberHeight old s = case s of
308+ SubscriberPush _ subscribed -> do
309+ when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberPush" <> showNodeId subscribed
310+ WeakBag.traverse (pushSubscribedSubscribers subscribed) $ invalidateSubscriberHeight old
311+ when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberPush" <> showNodeId subscribed <> " done"
312+ SubscriberMerge _ subscribed -> do
313+ when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberMerge" <> showNodeId subscribed
314+ removeMergeHeight old subscribed
315+ invalidateMergeHeight subscribed
316+ when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberMerge" <> showNodeId subscribed <> " done"
317+ SubscriberMergeChange _ -> return () -- Doesn't affect the height of the merge, because its changes are delayed
318+ SubscriberFan subscribed -> do
319+ when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberFan" <> showNodeId subscribed
320+ subscribers <- readIORef $ fanSubscribedSubscribers subscribed
321+ forM_ (DMap.toList subscribers) $ \(_ :=> v) -> WeakBag.traverse (_fanSubscribedChildren_list v) $ invalidateSubscriberHeight old
322+ when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberFan" <> showNodeId subscribed <> " done"
323+ SubscriberHold _ -> return ()
324+ SubscriberHoldIdentity _ -> return ()
325+ SubscriberSwitch subscribed -> do
326+ when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberSwitch" <> showNodeId subscribed
327+ oldHeight <- readIORef $ switchSubscribedHeight subscribed
328+ when (oldHeight /= invalidHeight) $ do
329+ writeIORef (switchSubscribedHeight subscribed) $! invalidHeight
330+ WeakBag.traverse (switchSubscribedSubscribers subscribed) $ invalidateSubscriberHeight oldHeight
331+ when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberSwitch" <> showNodeId subscribed <> " done"
332+ SubscriberCoincidenceOuter subscribed -> do
333+ when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberCoincidenceOuter" <> showNodeId subscribed
334+ invalidateCoincidenceHeight subscribed
335+ when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberCoincidenceOuter" <> showNodeId subscribed <> " done"
336+ SubscriberCoincidenceInner subscribed -> do
337+ when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberCoincidenceInner" <> showNodeId subscribed
338+ invalidateCoincidenceHeight subscribed
339+ when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberCoincidenceInner" <> showNodeId subscribed <> " done"
340+ SubscriberHandle -> return ()
341+
342+recalculateSubscriberHeight :: Height -> Subscriber x a -> IO ()
343+recalculateSubscriberHeight new s = case s of
344+ SubscriberPush _ subscribed -> do
345+ when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberPush" <> showNodeId subscribed
346+ WeakBag.traverse (pushSubscribedSubscribers subscribed) $ recalculateSubscriberHeight new
347+ when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberPush" <> showNodeId subscribed <> " done"
348+ SubscriberMerge _ subscribed -> do
349+ when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberMerge" <> showNodeId subscribed
350+ addMergeHeight new subscribed
351+ revalidateMergeHeight subscribed
352+ when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberMerge" <> showNodeId subscribed <> " done"
353+ SubscriberMergeChange _ -> return ()
354+ SubscriberFan subscribed -> do
355+ when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberFan" <> showNodeId subscribed
356+ subscribers <- readIORef $ fanSubscribedSubscribers subscribed
357+ forM_ (DMap.toList subscribers) $ \(_ :=> v) -> WeakBag.traverse (_fanSubscribedChildren_list v) $ recalculateSubscriberHeight new
358+ when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberFan" <> showNodeId subscribed <> " done"
359+ SubscriberHold _ -> return ()
360+ SubscriberHoldIdentity _ -> return ()
361+ SubscriberSwitch subscribed -> do
362+ when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberSwitch" <> showNodeId subscribed
363+ updateSwitchHeight new subscribed
364+ when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberSwitch" <> showNodeId subscribed <> " done"
365+ SubscriberCoincidenceOuter subscribed -> do
366+ when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberCoincidenceOuter" <> showNodeId subscribed
367+ recalculateCoincidenceHeight subscribed
368+ when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberCoincidenceOuter" <> showNodeId subscribed <> " done"
369+ SubscriberCoincidenceInner subscribed -> do
370+ when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberCoincidenceInner" <> showNodeId subscribed
371+ recalculateCoincidenceHeight subscribed
372+ when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberCoincidenceInner" <> showNodeId subscribed <> " done"
373+ SubscriberHandle -> return ()
374+
375+-- | Propagate everything at the current height
376+propagate :: a -> SubscriberList x a -> EventM x ()
377+propagate a subscribers = withIncreasedDepth $ do
378+ -- Note: in the following traversal, we do not visit nodes that are added to the list during our traversal; they are new events, which will necessarily have full information already, so there is no need to traverse them
379+ --TODO: Should we check if nodes already have their values before propagating? Maybe we're re-doing work
380+ WeakBag.traverse subscribers $ \s -> case s of
381+ SubscriberPush compute subscribed -> {-# SCC "traversePush" #-} do
382+ tracePropagate $ "SubscriberPush" <> showNodeId subscribed
383+ occ <- runComputeM $ compute a
384+ case occ of
385+ Nothing -> return () -- No need to write a Nothing back into the Ref
386+ Just o -> do
387+ liftIO $ writeIORef (pushSubscribedOccurrence subscribed) occ
388+ scheduleClear $ pushSubscribedOccurrence subscribed
389+ propagate o $ pushSubscribedSubscribers subscribed
390+ SubscriberMerge k subscribed -> {-# SCC "traverseMerge" #-} do
391+ tracePropagate $ "SubscriberMerge" <> showNodeId subscribed
392+ oldM <- liftIO $ readIORef $ mergeSubscribedAccum subscribed
393+ let newM = DMap.insertWith (error $ "Same key fired multiple times for Merge" <> showNodeId subscribed) k (Identity a) oldM
394+ tracePropagate $ " DMap.size oldM = " <> show (DMap.size oldM) <> "; DMap.size newM = " <> show (DMap.size newM)
395+ liftIO $ writeIORef (mergeSubscribedAccum subscribed) $! newM
396+ when (DMap.null oldM) $ do -- Only schedule the firing once
397+ height <- liftIO $ readIORef $ mergeSubscribedHeight subscribed
398+ --TODO: assertions about height
399+ currentHeight <- getCurrentHeight
400+ when (height <= currentHeight) $ do
401+ myStack <- liftIO $ whoCreatedIORef $ mergeSubscribedCachedSubscribed subscribed
402+ if height /= invalidHeight
403+ then error $ "Height (" ++ show height ++ ") is not greater than current height (" ++ show currentHeight ++ ")\n" ++ unlines (reverse myStack)
404+ else liftIO $ do
405+ nodesInvolvedInCycle <- walkInvalidHeightParents $ EventSubscribedMerge subscribed
406+ stacks <- forM nodesInvolvedInCycle $ \(Some.This es) -> whoCreatedEventSubscribed es
407+ error $ "Causality loop found:\n" <> drawForest (listsToForest stacks)
408+ scheduleMerge height subscribed
409+ SubscriberMergeChange subscribed -> {-# SCC "traverseMergeChange" #-} do
410+ tracePropagate $ "SubscriberMerge" <> showNodeId subscribed
411+ defer $ SomeMergeUpdate a subscribed
412+ SubscriberFan subscribed -> {-# SCC "traverseFan" #-} do
413+ subs <- liftIO $ readIORef $ fanSubscribedSubscribers subscribed
414+ tracePropagate $ "SubscriberFan" <> showNodeId subscribed <> ": " ++ show (DMap.size subs) ++ " keys subscribed, " ++ show (DMap.size a) ++ " keys firing"
415+ --TODO: We need a better DMap intersection; here, we are assuming that the number of firing keys is small and the number of subscribers is large
416+ forM_ (DMap.toList a) $ \(k :=> Identity v) -> {-# SCC "SubscriberFan" #-} case DMap.lookup k subs of
417+ Nothing -> do
418+ tracePropagate "No subscriber for key"
419+ return ()
420+ Just subsubs -> do
421+ propagate v $ _fanSubscribedChildren_list subsubs
422+ return ()
423+ SubscriberHold h -> {-# SCC "traverseHold" #-} propagateSubscriberHold h a
424+ SubscriberHoldIdentity h -> {-# SCC "traverseHoldIdentity" #-} propagateSubscriberHold h a
425+ SubscriberSwitch subscribed -> {-# SCC "traverseSwitch" #-} do
426+ tracePropagate $ "SubscriberSwitch" <> showNodeId subscribed
427+ liftIO $ writeIORef (switchSubscribedOccurrence subscribed) $ Just a
428+ scheduleClear $ switchSubscribedOccurrence subscribed
429+ propagate a $ switchSubscribedSubscribers subscribed
430+ SubscriberCoincidenceOuter subscribed -> {-# SCC "traverseCoincidenceOuter" #-} do
431+ tracePropagate $ "SubscriberCoincidenceOuter" <> showNodeId subscribed
432+ outerHeight <- liftIO $ readIORef $ coincidenceSubscribedHeight subscribed
433+ tracePropagate $ " outerHeight = " <> show outerHeight
434+ (occ, innerHeight, innerSubd) <- subscribeCoincidenceInner a outerHeight subscribed
435+ tracePropagate $ " isJust occ = " <> show (isJust occ)
436+ tracePropagate $ " innerHeight = " <> show innerHeight
437+ liftIO $ writeIORef (coincidenceSubscribedInnerParent subscribed) $ Just innerSubd
438+ scheduleClear $ coincidenceSubscribedInnerParent subscribed
439+ case occ of
440+ Nothing -> do
441+ when (innerHeight > outerHeight) $ liftIO $ do -- If the event fires, it will fire at a later height
442+ writeIORef (coincidenceSubscribedHeight subscribed) $! innerHeight
443+ WeakBag.traverse (coincidenceSubscribedSubscribers subscribed) $ invalidateSubscriberHeight outerHeight
444+ WeakBag.traverse (coincidenceSubscribedSubscribers subscribed) $ recalculateSubscriberHeight innerHeight
445+ Just o -> do -- Since it's already firing, no need to adjust height
446+ liftIO $ writeIORef (coincidenceSubscribedOccurrence subscribed) occ
447+ scheduleClear $ coincidenceSubscribedOccurrence subscribed
448+ propagate o $ coincidenceSubscribedSubscribers subscribed
449+ SubscriberCoincidenceInner subscribed -> {-# SCC "traverseCoincidenceInner" #-} do
450+ tracePropagate $ "SubscriberCoincidenceInner" <> showNodeId subscribed
451+ occ <- liftIO $ readIORef $ coincidenceSubscribedOccurrence subscribed
452+ case occ of
453+ Just _ -> return () -- SubscriberCoincidenceOuter must have already propagated this event
454+ Nothing -> do
455+ liftIO $ writeIORef (coincidenceSubscribedOccurrence subscribed) $ Just a
456+ scheduleClear $ coincidenceSubscribedOccurrence subscribed
457+ propagate a $ coincidenceSubscribedSubscribers subscribed
458+ SubscriberHandle -> {-# SCC "traverseHandle" #-} tracePropagate "SubscriberHandle"
459+
241460--------------------------------------------------------------------------------
242461-- EventSubscribed
243462--------------------------------------------------------------------------------
@@ -351,6 +570,13 @@ walkInvalidHeightParents s0 = do
351570 forM_ subscribers $ \(Some.This s) -> writeIORef (eventSubscribedHeightRef s) $! invalidHeight
352571 return subscribers
353572
573+{-# INLINE subscribeHoldEvent #-}
574+subscribeHoldEvent :: (CanSubscribe x m, R.Patch p) => Hold x p a -> Subscriber x (p a) -> m (EventSubscription x (p a))
575+subscribeHoldEvent h sub = do
576+ EventSubscription _ subd <- getHoldEventSubscription h
577+ sln <- liftIO $ subscribeEventSubscribed subd sub
578+ return $ EventSubscription sln subd
579+
354580--------------------------------------------------------------------------------
355581-- EventSubscription
356582--------------------------------------------------------------------------------
@@ -753,67 +979,6 @@ data Coincidence x a
753979 , coincidenceSubscribed :: !(IORef (Maybe (CoincidenceSubscribed x a)))
754980 }
755981
756−data Subscriber x a
757− = forall b. SubscriberPush (a -> ComputeM x (Maybe b)) (PushSubscribed x a b) --TODO: Why does this have the pushCompute in it?
758− | forall k. GCompare k => SubscriberMerge !(k a) (MergeSubscribed x k) --TODO: Can we inline the GCompare?
759− | forall k. (a ~ R.PatchDMap (DMap k (Event x)), GCompare k) => SubscriberMergeChange (MergeSubscribed x k)
760− | forall k. (GCompare k, a ~ DMap k Identity) => SubscriberFan (FanSubscribed x k)
761− | forall p b. (a ~ p b, R.Patch p) => SubscriberHold !(Hold x p b)
762− | forall b. (a ~ Identity b) => SubscriberHoldIdentity !(Hold x Identity b)
763− | SubscriberSwitch (SwitchSubscribed x a)
764− | forall b. a ~ Event x b => SubscriberCoincidenceOuter (CoincidenceSubscribed x b)
765− | SubscriberCoincidenceInner (CoincidenceSubscribed x a)
766− | SubscriberHandle
767−
768−-- These function are constructor functions that are marked NOINLINE so they are
769−-- opaque to GHC. If we do not do this, then GHC will sometimes fuse the constructor away
770−-- so any weak references that are attached to the constructors will have their
771−-- finalizer run. Using the opaque constructor, does not see the
772−-- constructor application, so it behaves like an IORef and cannot be fused away.
773−--
774−-- The result is also evaluated to WHNF, since forcing a thunk invalidates
775−-- the weak pointer to it in some cases.
776−
777−--TODO: I think all of these noinlines are now unnecessary
778−{-# NOINLINE newSubscriberPush #-}
779−newSubscriberPush :: (a -> ComputeM x (Maybe b)) -> PushSubscribed x a b -> IO (Subscriber x a)
780−newSubscriberPush compute subd = return $! SubscriberPush compute subd
781−
782−{-# RULES
783− "newSubscriberHold/Identity" forall h. newSubscriberHold h = newSubscriberHoldIdentity h
784− #-}
785−{-# NOINLINE newSubscriberHold #-}
786−newSubscriberHold :: R.Patch p => Hold x p a -> IO (Subscriber x (p a))
787−newSubscriberHold h = return $! SubscriberHold h
788−
789−{-# NOINLINE newSubscriberHoldIdentity #-}
790−newSubscriberHoldIdentity :: Hold x Identity a -> IO (Subscriber x (Identity a))
791−newSubscriberHoldIdentity h = return $! SubscriberHoldIdentity h
792−
793−{-# NOINLINE newSubscriberMerge #-}
794−newSubscriberMerge :: GCompare k => k a -> MergeSubscribed x k -> IO (Subscriber x a)
795−newSubscriberMerge k subd = return $! SubscriberMerge k subd
796−
797−{-# NOINLINE newSubscriberMergeChange #-}
798−newSubscriberMergeChange :: GCompare k => MergeSubscribed x k -> IO (Subscriber x (R.PatchDMap (DMap k (Event x))))
799−newSubscriberMergeChange subd = return $! SubscriberMergeChange subd
800−
801−{-# NOINLINE newSubscriberFan #-}
802−newSubscriberFan :: GCompare k => FanSubscribed x k -> IO (Subscriber x (DMap k Identity))
803−newSubscriberFan subd = return $! SubscriberFan subd
804−
805−{-# NOINLINE newSubscriberSwitch #-}
806−newSubscriberSwitch :: SwitchSubscribed x a -> IO (Subscriber x a)
807−newSubscriberSwitch subd = return $! SubscriberSwitch subd
808−
809−{-# NOINLINE newSubscriberCoincidenceOuter #-}
810−newSubscriberCoincidenceOuter :: CoincidenceSubscribed x b -> IO (Subscriber x (Event x b))
811−newSubscriberCoincidenceOuter subd = return $! SubscriberCoincidenceOuter subd
812−
813−{-# NOINLINE newSubscriberCoincidenceInner #-}
814−newSubscriberCoincidenceInner :: CoincidenceSubscribed x a -> IO (Subscriber x a)
815−newSubscriberCoincidenceInner subd = return $! SubscriberCoincidenceInner subd
816−
817982{-# NOINLINE newInvalidatorSwitch #-}
818983newInvalidatorSwitch :: SwitchSubscribed x a -> IO (Invalidator x)
819984newInvalidatorSwitch subd = return $! InvalidatorSwitch subd
@@ -962,10 +1127,6 @@ coincidence a = eventCoincidence $ Coincidence
9621127 , coincidenceSubscribed = unsafeNewIORef a Nothing
9631128 }
9641129
965−propagateAndUpdateSubscribersRef :: SubscriberList x a -> a -> EventM x ()
966−propagateAndUpdateSubscribersRef subscribersRef a = do
967− propagate a subscribersRef
968−
9691130-- Propagate the given event occurrence; before cleaning up, run the given action, which may read the state of events and behaviors
9701131run :: [DSum (RootTrigger x) Identity] -> ResultM x b -> SpiderHost x b
9711132run roots after = do
@@ -982,7 +1143,7 @@ run roots after = do
9821143 return $ Just r
9831144 else return Nothing
9841145 forM_ (catMaybes rootsToPropagate) $ \(RootTrigger (subscribersRef, _, _) :=> Identity a) -> do
985− propagateAndUpdateSubscribersRef subscribersRef a
1146+ propagate a subscribersRef
9861147 delayedRef <- EventM $ asks eventEnvDelayedMerges
9871148 let go = do
9881149 delayed <- liftIO $ readIORef delayedRef
@@ -1004,7 +1165,7 @@ run roots after = do
10041165 --TODO: Assert that m is not empty
10051166 liftIO $ writeIORef (mergeSubscribedOccurrence subscribed) $ Just m
10061167 scheduleClear $ mergeSubscribedOccurrence subscribed
1007− propagateAndUpdateSubscribersRef (mergeSubscribedSubscribers subscribed) m
1168+ propagate m $ mergeSubscribedSubscribers subscribed
10081169 go
10091170 go
10101171 putCurrentHeight maxBound
@@ -1092,91 +1253,6 @@ groupByHead = \case
10921253listsToForest :: Eq a => [[a]] -> Forest a
10931254listsToForest l = fmap (\(a, l') -> Node a $ listsToForest $ toList l') $ groupByHead $ catMaybes $ fmap nonEmpty l
10941255
1095−-- | Propagate everything at the current height
1096−propagate :: a -> SubscriberList x a -> EventM x ()
1097−propagate a subscribers = withIncreasedDepth $ do
1098− -- Note: in the following traversal, we do not visit nodes that are added to the list during our traversal; they are new events, which will necessarily have full information already, so there is no need to traverse them
1099− --TODO: Should we check if nodes already have their values before propagating? Maybe we're re-doing work
1100− WeakBag.traverse subscribers $ \s -> case s of
1101− SubscriberPush compute subscribed -> {-# SCC "traversePush" #-} do
1102− tracePropagate $ "SubscriberPush" <> showNodeId subscribed
1103− occ <- runComputeM $ compute a
1104− case occ of
1105− Nothing -> return () -- No need to write a Nothing back into the Ref
1106− Just o -> do
1107− liftIO $ writeIORef (pushSubscribedOccurrence subscribed) occ
1108− scheduleClear $ pushSubscribedOccurrence subscribed
1109− propagate o $ pushSubscribedSubscribers subscribed
1110− SubscriberMerge k subscribed -> {-# SCC "traverseMerge" #-} do
1111− tracePropagate $ "SubscriberMerge" <> showNodeId subscribed
1112− oldM <- liftIO $ readIORef $ mergeSubscribedAccum subscribed
1113− let newM = DMap.insertWith (error $ "Same key fired multiple times for Merge" <> showNodeId subscribed) k (Identity a) oldM
1114− tracePropagate $ " DMap.size oldM = " <> show (DMap.size oldM) <> "; DMap.size newM = " <> show (DMap.size newM)
1115− liftIO $ writeIORef (mergeSubscribedAccum subscribed) $! newM
1116− when (DMap.null oldM) $ do -- Only schedule the firing once
1117− height <- liftIO $ readIORef $ mergeSubscribedHeight subscribed
1118− --TODO: assertions about height
1119− currentHeight <- getCurrentHeight
1120− when (height <= currentHeight) $ do
1121− myStack <- liftIO $ whoCreatedIORef $ mergeSubscribedCachedSubscribed subscribed
1122− if height /= invalidHeight
1123− then error $ "Height (" ++ show height ++ ") is not greater than current height (" ++ show currentHeight ++ ")\n" ++ unlines (reverse myStack)
1124− else liftIO $ do
1125− nodesInvolvedInCycle <- walkInvalidHeightParents $ EventSubscribedMerge subscribed
1126− stacks <- forM nodesInvolvedInCycle $ \(Some.This es) -> whoCreatedEventSubscribed es
1127− error $ "Causality loop found:\n" <> drawForest (listsToForest stacks)
1128− scheduleMerge height subscribed
1129− SubscriberMergeChange subscribed -> {-# SCC "traverseMergeChange" #-} do
1130− tracePropagate $ "SubscriberMerge" <> showNodeId subscribed
1131− defer $ SomeMergeUpdate a subscribed
1132− SubscriberFan subscribed -> {-# SCC "traverseFan" #-} do
1133− subs <- liftIO $ readIORef $ fanSubscribedSubscribers subscribed
1134− tracePropagate $ "SubscriberFan" <> showNodeId subscribed <> ": " ++ show (DMap.size subs) ++ " keys subscribed, " ++ show (DMap.size a) ++ " keys firing"
1135− --TODO: We need a better DMap intersection; here, we are assuming that the number of firing keys is small and the number of subscribers is large
1136− forM_ (DMap.toList a) $ \(k :=> Identity v) -> {-# SCC "SubscriberFan" #-} case DMap.lookup k subs of
1137− Nothing -> do
1138− tracePropagate "No subscriber for key"
1139− return ()
1140− Just subsubs -> do
1141− propagate v $ _fanSubscribedChildren_list subsubs
1142− return ()
1143− SubscriberHold h -> {-# SCC "traverseHold" #-} propagateSubscriberHold h a
1144− SubscriberHoldIdentity h -> {-# SCC "traverseHoldIdentity" #-} propagateSubscriberHold h a
1145− SubscriberSwitch subscribed -> {-# SCC "traverseSwitch" #-} do
1146− tracePropagate $ "SubscriberSwitch" <> showNodeId subscribed
1147− liftIO $ writeIORef (switchSubscribedOccurrence subscribed) $ Just a
1148− scheduleClear $ switchSubscribedOccurrence subscribed
1149− propagate a $ switchSubscribedSubscribers subscribed
1150− SubscriberCoincidenceOuter subscribed -> {-# SCC "traverseCoincidenceOuter" #-} do
1151− tracePropagate $ "SubscriberCoincidenceOuter" <> showNodeId subscribed
1152− outerHeight <- liftIO $ readIORef $ coincidenceSubscribedHeight subscribed
1153− tracePropagate $ " outerHeight = " <> show outerHeight
1154− (occ, innerHeight, innerSubd) <- subscribeCoincidenceInner a outerHeight subscribed
1155− tracePropagate $ " isJust occ = " <> show (isJust occ)
1156− tracePropagate $ " innerHeight = " <> show innerHeight
1157− liftIO $ writeIORef (coincidenceSubscribedInnerParent subscribed) $ Just innerSubd
1158− scheduleClear $ coincidenceSubscribedInnerParent subscribed
1159− case occ of
1160− Nothing -> do
1161− when (innerHeight > outerHeight) $ liftIO $ do -- If the event fires, it will fire at a later height
1162− writeIORef (coincidenceSubscribedHeight subscribed) $! innerHeight
1163− WeakBag.traverse (coincidenceSubscribedSubscribers subscribed) $ invalidateSubscriberHeight outerHeight
1164− WeakBag.traverse (coincidenceSubscribedSubscribers subscribed) $ recalculateSubscriberHeight innerHeight
1165− Just o -> do -- Since it's already firing, no need to adjust height
1166− liftIO $ writeIORef (coincidenceSubscribedOccurrence subscribed) occ
1167− scheduleClear $ coincidenceSubscribedOccurrence subscribed
1168− propagate o $ coincidenceSubscribedSubscribers subscribed
1169− SubscriberCoincidenceInner subscribed -> {-# SCC "traverseCoincidenceInner" #-} do
1170− tracePropagate $ "SubscriberCoincidenceInner" <> showNodeId subscribed
1171− occ <- liftIO $ readIORef $ coincidenceSubscribedOccurrence subscribed
1172− case occ of
1173− Just _ -> return () -- SubscriberCoincidenceOuter must have already propagated this event
1174− Nothing -> do
1175− liftIO $ writeIORef (coincidenceSubscribedOccurrence subscribed) $ Just a
1176− scheduleClear $ coincidenceSubscribedOccurrence subscribed
1177− propagate a $ coincidenceSubscribedSubscribers subscribed
1178− SubscriberHandle -> {-# SCC "traverseHandle" #-} tracePropagate "SubscriberHandle"
1179−
11801256{-# INLINE propagateSubscriberHold #-}
11811257propagateSubscriberHold :: R.Patch p => Hold x p a -> p a -> EventM x ()
11821258propagateSubscriberHold h a = do
@@ -1315,18 +1391,10 @@ readCoincidenceSubscribed subscribed = do
13151391zeroRef :: IORef Height
13161392zeroRef = unsafePerformIO $ newIORef zeroHeight
13171393
1318−{-# INLINE subscribeHoldEvent #-}
1319−subscribeHoldEvent :: (CanSubscribe x m, R.Patch p) => Hold x p a -> Subscriber x (p a) -> m (EventSubscription x (p a))
1320−subscribeHoldEvent h sub = do
1321− EventSubscription _ subd <- getHoldEventSubscription h
1322− sln <- liftIO $ subscribeEventSubscribed subd sub
1323− return $ EventSubscription sln subd
1324−
13251394{-# INLINE debugSubscribe #-}
13261395debugSubscribe :: Bool
13271396debugSubscribe = False
13281397
1329−
13301398type CanSubscribe x m =
13311399 ( HasSpiderEnv x m
13321400 , HasCurrentHeight x m
@@ -1823,42 +1891,6 @@ succHeight h@(Height a) =
18231891 then invalidHeight
18241892 else Height $ succ a
18251893
1826−invalidateSubscriberHeight :: Height -> Subscriber x a -> IO ()
1827−invalidateSubscriberHeight old s = case s of
1828− SubscriberPush _ subscribed -> do
1829− when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberPush" <> showNodeId subscribed
1830− WeakBag.traverse (pushSubscribedSubscribers subscribed) $ invalidateSubscriberHeight old
1831− when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberPush" <> showNodeId subscribed <> " done"
1832− SubscriberMerge _ subscribed -> do
1833− when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberMerge" <> showNodeId subscribed
1834− removeMergeHeight old subscribed
1835− invalidateMergeHeight subscribed
1836− when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberMerge" <> showNodeId subscribed <> " done"
1837− SubscriberMergeChange _ -> return () -- Doesn't affect the height of the merge, because its changes are delayed
1838− SubscriberFan subscribed -> do
1839− when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberFan" <> showNodeId subscribed
1840− subscribers <- readIORef $ fanSubscribedSubscribers subscribed
1841− forM_ (DMap.toList subscribers) $ \(_ :=> v) -> WeakBag.traverse (_fanSubscribedChildren_list v) $ invalidateSubscriberHeight old
1842− when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberFan" <> showNodeId subscribed <> " done"
1843− SubscriberHold _ -> return ()
1844− SubscriberHoldIdentity _ -> return ()
1845− SubscriberSwitch subscribed -> do
1846− when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberSwitch" <> showNodeId subscribed
1847− oldHeight <- readIORef $ switchSubscribedHeight subscribed
1848− when (oldHeight /= invalidHeight) $ do
1849− writeIORef (switchSubscribedHeight subscribed) $! invalidHeight
1850− WeakBag.traverse (switchSubscribedSubscribers subscribed) $ invalidateSubscriberHeight oldHeight
1851− when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberSwitch" <> showNodeId subscribed <> " done"
1852− SubscriberCoincidenceOuter subscribed -> do
1853− when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberCoincidenceOuter" <> showNodeId subscribed
1854− invalidateCoincidenceHeight subscribed
1855− when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberCoincidenceOuter" <> showNodeId subscribed <> " done"
1856− SubscriberCoincidenceInner subscribed -> do
1857− when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberCoincidenceInner" <> showNodeId subscribed
1858− invalidateCoincidenceHeight subscribed
1859− when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberCoincidenceInner" <> showNodeId subscribed <> " done"
1860− SubscriberHandle -> return ()
1861−
18621894removeMergeHeight :: Height -> MergeSubscribed x k -> IO ()
18631895removeMergeHeight oldParentHeight subscribed = do
18641896 when debugInvalidateHeight $ putStrLn $ "removeMergeHeight: " <> show oldParentHeight <> " Merge" <> showNodeId subscribed
@@ -1904,39 +1936,6 @@ invalidateCoincidenceHeight subscribed = do
19041936 writeIORef (coincidenceSubscribedHeight subscribed) $! invalidHeight
19051937 WeakBag.traverse (coincidenceSubscribedSubscribers subscribed) $ invalidateSubscriberHeight oldHeight
19061938
1907−recalculateSubscriberHeight :: Height -> Subscriber x a -> IO ()
1908−recalculateSubscriberHeight new s = case s of
1909− SubscriberPush _ subscribed -> do
1910− when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberPush" <> showNodeId subscribed
1911− WeakBag.traverse (pushSubscribedSubscribers subscribed) $ recalculateSubscriberHeight new
1912− when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberPush" <> showNodeId subscribed <> " done"
1913− SubscriberMerge _ subscribed -> do
1914− when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberMerge" <> showNodeId subscribed
1915− addMergeHeight new subscribed
1916− revalidateMergeHeight subscribed
1917− when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberMerge" <> showNodeId subscribed <> " done"
1918− SubscriberMergeChange _ -> return ()
1919− SubscriberFan subscribed -> do
1920− when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberFan" <> showNodeId subscribed
1921− subscribers <- readIORef $ fanSubscribedSubscribers subscribed
1922− forM_ (DMap.toList subscribers) $ \(_ :=> v) -> WeakBag.traverse (_fanSubscribedChildren_list v) $ recalculateSubscriberHeight new
1923− when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberFan" <> showNodeId subscribed <> " done"
1924− SubscriberHold _ -> return ()
1925− SubscriberHoldIdentity _ -> return ()
1926− SubscriberSwitch subscribed -> do
1927− when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberSwitch" <> showNodeId subscribed
1928− updateSwitchHeight new subscribed
1929− when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberSwitch" <> showNodeId subscribed <> " done"
1930− SubscriberCoincidenceOuter subscribed -> do
1931− when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberCoincidenceOuter" <> showNodeId subscribed
1932− recalculateCoincidenceHeight subscribed
1933− when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberCoincidenceOuter" <> showNodeId subscribed <> " done"
1934− SubscriberCoincidenceInner subscribed -> do
1935− when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberCoincidenceInner" <> showNodeId subscribed
1936− recalculateCoincidenceHeight subscribed
1937− when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberCoincidenceInner" <> showNodeId subscribed <> " done"
1938− SubscriberHandle -> return ()
1939−
19401939updateSwitchHeight :: Height -> SwitchSubscribed x a -> IO ()
19411940updateSwitchHeight new subscribed = do
19421941 oldHeight <- readIORef $ switchSubscribedHeight subscribed