@@ -102,9 +102,6 @@ instance HasNodeId (PushSubscribed x a b) where
102102instance HasNodeId (SwitchSubscribed x a) where
103103 getNodeId = switchSubscribedNodeId
104104
105−instance HasNodeId (MergeSubscribed x a) where
106− getNodeId = mergeSubscribedNodeId
107−
108105instance HasNodeId (FanSubscribed x a) where
109106 getNodeId = fanSubscribedNodeId
110107
@@ -225,7 +222,7 @@ cacheEvent e = Event $ \sub -> do
225222 { eventSubscribedHeightRef = eventSubscribedHeightRef $ _eventSubscription_subscribed $ _cacheSubscribed_parent subscribed
226223 , eventSubscribedRetained = toAny subscribed
227224 }
228− return (EventSubscription (Just ticket) es, occ)
225+ return (EventSubscription (WeakBag.remove ticket >> touch ticket) es, occ)
229226 Nothing -> do
230227 weakSelf <- liftIO $ newIORef $ error "cacheEvent: weakSelf not yet intialized"
231228 (subscribers, ticket) <- liftIO $ WeakBag.singleton sub weakSelf cleanupCacheSubscribed
@@ -251,7 +248,7 @@ cacheEvent e = Event $ \sub -> do
251248 }
252249 liftIO $ writeIORef mSubscribedRef $ Just subscribed
253250 liftIO $ writeIORef weakSelf =<< evaluate =<< mkWeakPtrWithDebug subscribed "PushSubscribed"
254− return (EventSubscription (Just ticket) (EventSubscribed (eventSubscribedHeightRef $ _eventSubscription_subscribed parentSub) (toAny subscribed)), occ)
251+ return (EventSubscription (WeakBag.remove ticket >> touch ticket) (EventSubscribed (eventSubscribedHeightRef $ _eventSubscription_subscribed parentSub) (toAny subscribed)), occ)
255252
256253cleanupCacheSubscribed :: CacheSubscribed x a -> IO ()
257254cleanupCacheSubscribed self = do
@@ -266,16 +263,13 @@ wrap :: MonadIO m => (t -> EventSubscribed x) -> (Subscriber x a -> m (WeakBagTi
266263wrap tag getSpecificSubscribed sub = do
267264 (sln, subd, occ) <- getSpecificSubscribed sub
268265 let es = tag subd
269− return (EventSubscription (Just sln) es, occ)
266+ return (EventSubscription (WeakBag.remove sln >> touch sln) es, occ)
270267
271268eventRoot :: GCompare k => k a -> Root x k -> Event x a
272269eventRoot !k !r = Event $ wrap eventSubscribedRoot $ liftIO . getRootSubscribed k r
273270
274271eventNever :: Event x a
275−eventNever = Event $ \_ -> return (EventSubscription Nothing eventSubscribedNever, Nothing)
276−
277−eventMerge :: GCompare k => Merge x k -> Event x (DMap k Identity)
278−eventMerge !m = Event $ wrap eventSubscribedMerge $ getMergeSubscribed m
272+eventNever = Event $ \_ -> return (EventSubscription (return ()) eventSubscribedNever, Nothing)
279273
280274eventFan :: GCompare k => k a -> Fan x k -> Event x a
281275eventFan !k !f = Event $ wrap eventSubscribedFan $ getFanSubscribed k f
@@ -329,54 +323,6 @@ newSubscriberHold h = return $ Subscriber
329323 , subscriberRecalculateHeight = \_ -> return ()
330324 }
331325
332−newSubscriberMerge :: GCompare k => k a -> MergeSubscribed x k -> IO (Subscriber x a)
333−newSubscriberMerge k subscribed = return $ Subscriber
334− { subscriberPropagate = \a -> {-# SCC "traverseMerge" #-} do
335− tracePropagate $ "SubscriberMerge" <> showNodeId subscribed
336− oldM <- liftIO $ readIORef $ mergeSubscribedAccum subscribed
337− let newM = DMap.insertWith (error $ "Same key fired multiple times for Merge" <> showNodeId subscribed) k (Identity a) oldM
338− tracePropagate $ " DMap.size oldM = " <> show (DMap.size oldM) <> "; DMap.size newM = " <> show (DMap.size newM)
339− liftIO $ writeIORef (mergeSubscribedAccum subscribed) $! newM
340− when (DMap.null oldM) $ do -- Only schedule the firing once
341− height <- liftIO $ readIORef $ mergeSubscribedHeight subscribed
342− --TODO: assertions about height
343− currentHeight <- getCurrentHeight
344− when (height <= currentHeight) $ do
345− if height /= invalidHeight
346− then do
347− myStack <- liftIO $ whoCreatedIORef $ mergeSubscribedCachedSubscribed subscribed
348− error $ "Height (" ++ show height ++ ") is not greater than current height (" ++ show currentHeight ++ ")\n" ++ unlines (reverse myStack)
349− else liftIO $ do
350−#ifdef DEBUG_CYCLES
351− nodesInvolvedInCycle <- walkInvalidHeightParents $ eventSubscribedMerge subscribed
352− stacks <- forM nodesInvolvedInCycle $ \(Some.This es) -> whoCreatedEventSubscribed es
353− let cycleInfo = ":\n" <> drawForest (listsToForest stacks)
354−#else
355− let cycleInfo = ""
356−#endif
357− error $ "Causality loop found" <> cycleInfo
358− scheduleMerge height subscribed
359− , subscriberInvalidateHeight = \old -> do
360− when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberMerge" <> showNodeId subscribed
361− removeMergeHeight old subscribed
362− invalidateMergeHeight subscribed
363− when debugInvalidateHeight $ putStrLn $ "invalidateSubscriberHeight: SubscriberMerge" <> showNodeId subscribed <> " done"
364− , subscriberRecalculateHeight = \new -> do
365− when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberMerge" <> showNodeId subscribed
366− addMergeHeight new subscribed
367− revalidateMergeHeight subscribed
368− when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: SubscriberMerge" <> showNodeId subscribed <> " done"
369− }
370−
371−newSubscriberMergeChange :: GCompare k => MergeSubscribed x k -> IO (Subscriber x (R.PatchDMap k (Event x)))
372−newSubscriberMergeChange subscribed = return $ Subscriber
373− { subscriberPropagate = \a -> {-# SCC "traverseMergeChange" #-} do
374− tracePropagate $ "SubscriberMerge" <> showNodeId subscribed
375− defer $ SomeMergeUpdate (mkUpdateMerge a subscribed) (invalidateMergeHeight subscribed) (revalidateMergeHeight subscribed)
376− , subscriberInvalidateHeight = \_ -> return ()
377− , subscriberRecalculateHeight = \_ -> return ()
378− }
379−
380326newSubscriberFan :: GCompare k => FanSubscribed x k -> IO (Subscriber x (DMap k Identity))
381327newSubscriberFan subscribed = return $ Subscriber
382328 { subscriberPropagate = \a -> {-# SCC "traverseFan" #-} do
@@ -525,19 +471,6 @@ eventSubscribedNever = EventSubscribed
525471#endif
526472 }
527473
528−eventSubscribedMerge :: MergeSubscribed x k -> EventSubscribed x
529−eventSubscribedMerge !subscribed = EventSubscribed
530− { eventSubscribedHeightRef = mergeSubscribedHeight subscribed
531− , eventSubscribedRetained = toAny subscribed
532−#ifdef DEBUG_CYCLES
533− , eventSubscribedGetParents = do
534− ps <- readIORef $ mergeSubscribedParents subscribed
535− return $ R.ffor (DMap.toList ps) $ \(_ :=> v) -> Some.This $ _eventSubscription_subscribed $ mergeSubscribedParentSubscription v
536− , eventSubscribedHasOwnHeightRef = True
537− , eventSubscribedWhoCreated = whoCreatedIORef $ mergeSubscribedCachedSubscribed subscribed
538−#endif
539− }
540−
541474eventSubscribedFan :: FanSubscribed x k -> EventSubscribed x
542475eventSubscribedFan !subscribed = EventSubscribed
543476 { eventSubscribedHeightRef = eventSubscribedHeightRef $ _eventSubscription_subscribed $ fanSubscribedParent subscribed
@@ -605,12 +538,12 @@ subscribeHoldEvent = subscribeAndRead . holdEvent
605538--------------------------------------------------------------------------------
606539
607540data EventSubscription x = EventSubscription
608− { _eventSubscription_ticket :: !(Maybe WeakBagTicket)
541+ { _eventSubscription_unsubscribe :: !(IO ())
609542 , _eventSubscription_subscribed :: {-# UNPACK #-} !(EventSubscribed x)
610543 }
611544
612545unsubscribe :: EventSubscription x -> IO ()
613−unsubscribe (EventSubscription msln _) = forM_ msln $ \sln -> WeakBag.remove sln
546+unsubscribe (EventSubscription u _) = u
614547
615548--------------------------------------------------------------------------------
616549-- Behavior
@@ -733,7 +666,7 @@ data EventEnv x
733666 , eventEnvRootClears :: !(IORef [SomeRootClear])
734667 , eventEnvCurrentHeight :: !(IORef Height) -- Needed for Subscribe
735668 , eventEnvResetCoincidences :: !(IORef [SomeResetCoincidence x]) -- Needed for Subscribe
736− , eventEnvDelayedMerges :: !(IORef (IntMap [DelayedMerge x]))
669+ , eventEnvDelayedMerges :: !(IORef (IntMap [EventM x ()]))
737670 }
738671
739672{-# INLINE runEventM #-}
@@ -771,7 +704,7 @@ instance Defer (SomeMergeInit x) (EventM x) where
771704
772705class HasCurrentHeight x m | m -> x where
773706 getCurrentHeight :: m Height
774− scheduleMerge :: Height -> MergeSubscribed x a -> m ()
707+ scheduleMerge :: Height -> EventM x () -> m ()
775708
776709instance HasCurrentHeight x (EventM x) where
777710 {-# INLINE getCurrentHeight #-}
@@ -781,7 +714,7 @@ instance HasCurrentHeight x (EventM x) where
781714 {-# INLINE scheduleMerge #-}
782715 scheduleMerge height subscribed = EventM $ do
783716 delayedRef <- asks eventEnvDelayedMerges
784− liftIO $ modifyIORef' delayedRef $ IntMap.insertWith (++) (unHeight height) [DelayedMerge subscribed]
717+ liftIO $ modifyIORef' delayedRef $ IntMap.insertWith (++) (unHeight height) [subscribed]
785718
786719class HasSpiderTimeline x m | m -> x where
787720 -- | Retrieve the current SpiderTimeline
@@ -951,7 +884,7 @@ instance Defer (SomeHoldInit x) (ComputeM x) where
951884runComputeM :: Defer (SomeHoldInit x) m => ComputeM x a -> m a
952885runComputeM (ComputeM a) = liftIO . runReaderT a =<< getDeferralQueue
953886
954−data MergeSubscribedParent (x :: *) k a = MergeSubscribedParent !(EventSubscription x) !(Subscriber x a)
887+data MergeSubscribedParent (x :: *) (k :: * -> *) a = MergeSubscribedParent !(EventSubscription x) !(Subscriber x a)
955888
956889mergeSubscribedParentSubscription :: MergeSubscribedParent x k a -> EventSubscription x
957890mergeSubscribedParentSubscription (MergeSubscribedParent es _) = es
@@ -998,27 +931,6 @@ heightBagVerify b@(HeightBag s c) = if
998931heightBagVerify = id
999932#endif
1000933
1001−data MergeSubscribed x k
1002− = MergeSubscribed { mergeSubscribedCachedSubscribed :: !(IORef (Maybe (MergeSubscribed x k))) -- From the original merge
1003− , mergeSubscribedOccurrence :: !(IORef (Maybe (DMap k Identity)))
1004− , mergeSubscribedAccum :: !(IORef (DMap k Identity)) -- This will accumulate occurrences until our height is reached, at which point it will be transferred to mergeSubscribedOccurrence
1005− , mergeSubscribedHeight :: !(IORef Height)
1006− , mergeSubscribedHeightBag :: !(IORef HeightBag)
1007− , mergeSubscribedSubscribers :: !(WeakBag (Subscriber x (DMap k Identity)))
1008− , mergeSubscribedChange :: !(IORef (Subscriber x (R.PatchDMap k (Event x)), EventSubscribed x))
1009− , mergeSubscribedParents :: !(IORef (DMap k (MergeSubscribedParent x k)))
1010− , mergeSubscribedWeakSelf :: !(IORef (Weak (MergeSubscribed x k)))
1011−#ifdef DEBUG_NODEIDS
1012− , mergeSubscribedNodeId :: Int
1013−#endif
1014− }
1015−
1016−--TODO: DMap sucks; we should really write something better (with a functor for the value as well as the key)
1017−data Merge x k
1018− = Merge { mergeParents :: !(Dynamic x (R.PatchDMap k (Event x)))
1019− , mergeSubscribed :: !(IORef (Maybe (MergeSubscribed x k)))
1020− }
1021−
1022934data FanSubscribedChildren (x :: *) k a = FanSubscribedChildren
1023935 { _fanSubscribedChildren_list :: !(WeakBag (Subscriber x a))
1024936 , _fanSubscribedChildren_self :: {-# NOUNPACK #-} !(k a, FanSubscribed x k)
@@ -1204,19 +1116,7 @@ run roots after = do
12041116 tracePropagate $ "Running height " ++ show currentHeight
12051117 putCurrentHeight $ Height currentHeight
12061118 liftIO $ writeIORef delayedRef $! future
1207− forM_ cur $ \d -> case d of
1208− DelayedMerge subscribed -> do
1209− height <- liftIO $ readIORef $ mergeSubscribedHeight subscribed
1210− case height `compare` Height currentHeight of
1211− LT -> error "Somehow a merge's height has been decreased after it was scheduled"
1212− GT -> scheduleMerge height subscribed -- The height has been increased (by a coincidence event; TODO: is this the only way?)
1213− EQ -> do
1214− m <- liftIO $ readIORef $ mergeSubscribedAccum subscribed
1215− liftIO $ writeIORef (mergeSubscribedAccum subscribed) $! DMap.empty
1216− --TODO: Assert that m is not empty
1217− liftIO $ writeIORef (mergeSubscribedOccurrence subscribed) $ Just m
1218− scheduleClear $ mergeSubscribedOccurrence subscribed
1219− propagate m $ mergeSubscribedSubscribers subscribed
1119+ sequence_ cur
12201120 go
12211121 go
12221122 putCurrentHeight maxBound
@@ -1224,14 +1124,21 @@ run roots after = do
12241124 tracePropagate "Done running an event frame"
12251125 return result
12261126
1127+scheduleMerge' :: Height -> IORef Height -> EventM x () -> EventM x ()
1128+scheduleMerge' initialHeight heightRef a = scheduleMerge initialHeight $ do
1129+ height <- liftIO $ readIORef heightRef
1130+ currentHeight <- getCurrentHeight
1131+ case height `compare` currentHeight of
1132+ LT -> error "Somehow a merge's height has been decreased after it was scheduled"
1133+ GT -> scheduleMerge' height heightRef a -- The height has been increased (by a coincidence event; TODO: is this the only way?)
1134+ EQ -> a
1135+
12271136data SomeClear = forall a. SomeClear (IORef (Maybe a))
12281137
12291138data SomeRootClear = forall k. SomeRootClear (IORef (DMap k Identity))
12301139
12311140data SomeAssignment x = forall a. SomeAssignment {-# UNPACK #-} !(IORef a) {-# UNPACK #-} !(IORef [Weak (Invalidator x)]) a
12321141
1233−data DelayedMerge x = forall k. DelayedMerge (MergeSubscribed x k)
1234−
12351142debugFinalize :: Bool
12361143debugFinalize = False
12371144
@@ -1412,76 +1319,6 @@ cleanupRootSubscribed self@(RootSubscribed { rootSubscribedKey = k, rootSubscrib
14121319subscribeRootSubscribed :: RootSubscribed x a -> Subscriber x a -> IO WeakBagTicket
14131320subscribeRootSubscribed subscribed sub = WeakBag.insert sub (rootSubscribedSubscribers subscribed) (rootSubscribedWeakSelf subscribed) cleanupRootSubscribed
14141321
1415−{-# SPECIALIZE getMergeSubscribed :: GCompare k => Merge x k -> Subscriber x (DMap k Identity) -> EventM x (WeakBagTicket, MergeSubscribed x k, Maybe (DMap k Identity)) #-}
1416−getMergeSubscribed :: forall x k m. (CanSubscribe x m, GCompare k) => Merge x k -> Subscriber x (DMap k Identity) -> m (WeakBagTicket, MergeSubscribed x k, Maybe (DMap k Identity))
1417−getMergeSubscribed m sub = do
1418− mSubscribed <- liftIO $ readIORef $ mergeSubscribed m
1419− case mSubscribed of
1420− Just subscribed -> {-# SCC "hitMerge" #-} liftIO $ do
1421− sln <- subscribeMergeSubscribed subscribed sub
1422− occ <- readIORef $ mergeSubscribedOccurrence subscribed
1423− return (sln, subscribed, occ)
1424− Nothing -> {-# SCC "missMerge" #-} do
1425− subscribedRef <- liftIO $ newIORef $ error "getMergeSubscribed: subscribedRef not yet initialized"
1426− subscribedUnsafe <- liftIO $ unsafeInterleaveIO $ readIORef subscribedRef
1427− initialParents <- readBehaviorUntracked $ dynamicCurrent $ mergeParents m
1428− subscribers :: [(Maybe (DSum k Identity), Height, DSum k (MergeSubscribedParent x k))] <- forM (DMap.toList initialParents) $ \(k :=> e) -> do
1429− s <- liftIO $ newSubscriberMerge k subscribedUnsafe
1430− (subscription@(EventSubscription _ parentSubd), parentOcc) <- subscribeAndRead e s
1431− height <- liftIO $ getEventSubscribedHeight parentSubd
1432− return $ (fmap (\x -> k :=> Identity x) parentOcc, height, k :=> MergeSubscribedParent subscription s)
1433− let dm = DMap.fromDistinctAscList $ mapMaybe (\(x, _, _) -> x) subscribers
1434− heights = fmap (\(_, h, _) -> h) subscribers --TODO: Assert that there's no invalidHeight in here
1435− myHeightBag = heightBagFromList $ filter (/= invalidHeight) heights
1436− myHeight = if invalidHeight `elem` heights
1437− then invalidHeight
1438− else succHeight $ heightBagMax myHeightBag
1439− currentHeight <- getCurrentHeight
1440− let (occ, accum) = if currentHeight >= myHeight -- If we should have fired by now
1441− then (if DMap.null dm then Nothing else Just dm, DMap.empty)
1442− else (Nothing, dm)
1443− unless (DMap.null accum) $ do
1444− scheduleMerge myHeight subscribedUnsafe
1445− occRef <- liftIO $ newIORef occ
1446− when (isJust occ) $ scheduleClear occRef
1447− accumRef <- liftIO $ newIORef accum
1448− heightRef <- liftIO $ newIORef myHeight
1449− heightMapRef <- liftIO $ newIORef myHeightBag
1450− weakSelf <- liftIO $ newIORef $ error "getMergeSubscribed: weakSelf not yet initialized"
1451− (subs, sln) <- liftIO $ WeakBag.singleton sub weakSelf cleanupMergeSubscribed
1452− changeSubdRef <- liftIO $ newIORef $ error "getMergeSubscribed: changeSubdRef not yet initialized"
1453− parentsRef <- liftIO $ newIORef $ DMap.fromDistinctAscList $ map (\(_, _, x) -> x) subscribers
1454− let !subscribed = MergeSubscribed
1455− { mergeSubscribedCachedSubscribed = mergeSubscribed m
1456− , mergeSubscribedOccurrence = occRef
1457− , mergeSubscribedAccum = accumRef
1458− , mergeSubscribedHeight = heightRef
1459− , mergeSubscribedHeightBag = heightMapRef
1460− , mergeSubscribedSubscribers = subs
1461− , mergeSubscribedChange = changeSubdRef
1462− , mergeSubscribedParents = parentsRef
1463− , mergeSubscribedWeakSelf = weakSelf
1464−#ifdef DEBUG_NODEIDS
1465− , mergeSubscribedNodeId = unsafeNodeId m
1466−#endif
1467− }
1468− defer $ SomeMergeInit $ mkInitMerge subscribed $ dynamicUpdated $ mergeParents m
1469− liftIO $ writeIORef subscribedRef $! subscribed
1470− liftIO $ writeIORef weakSelf =<< evaluate =<< mkWeakPtrWithDebug subscribed "MergeSubscribed"
1471− liftIO $ writeIORef (mergeSubscribed m) $ Just subscribed
1472− return (sln, subscribed, occ)
1473−
1474−cleanupMergeSubscribed :: MergeSubscribed x k -> IO ()
1475−cleanupMergeSubscribed subscribed = do
1476− parents <- readIORef $ mergeSubscribedParents subscribed
1477− forM_ (DMap.toList parents) $ \(_ :=> msp) -> unsubscribe $ mergeSubscribedParentSubscription msp
1478− -- Not necessary, because this whole MergeSubscribed is dead: writeIORef (mergeSubscribedParents subscribed) DMap.empty
1479− writeIORef (mergeSubscribedCachedSubscribed subscribed) $ Nothing -- Get rid of the cached subscribed
1480−
1481−{-# INLINE subscribeMergeSubscribed #-}
1482−subscribeMergeSubscribed :: MergeSubscribed x k -> Subscriber x (DMap k Identity) -> IO WeakBagTicket
1483−subscribeMergeSubscribed subscribed sub = WeakBag.insert sub (mergeSubscribedSubscribers subscribed) (mergeSubscribedWeakSelf subscribed) cleanupMergeSubscribed
1484−
14851322{-# SPECIALIZE getFanSubscribed :: GCompare k => k a -> Fan x k -> Subscriber x a -> EventM x (WeakBagTicket, FanSubscribed x k, Maybe a) #-}
14861323getFanSubscribed :: (Defer (SomeAssignment x) m, Defer (SomeHoldInit x) m, Defer (SomeMergeInit x) m, Defer SomeClear m, HasCurrentHeight x m, Defer (SomeResetCoincidence x) m, HasSpiderTimeline x m, m ~ EventM x, GCompare k) => k a -> Fan x k -> Subscriber x a -> m (WeakBagTicket, FanSubscribed x k, Maybe a)
14871324getFanSubscribed k f sub = do
@@ -1652,11 +1489,130 @@ subscribeCoincidenceSubscribed :: CoincidenceSubscribed x a -> Subscriber x a ->
16521489subscribeCoincidenceSubscribed subscribed sub = WeakBag.insert sub (coincidenceSubscribedSubscribers subscribed) (coincidenceSubscribedWeakSelf subscribed) cleanupCoincidenceSubscribed
16531490
16541491{-# INLINE merge #-}
1655−merge :: GCompare k => Dynamic x (R.PatchDMap k (Event x)) -> Event x (DMap k Identity)
1656−merge m = eventMerge $ Merge
1657− { mergeParents = m
1658− , mergeSubscribed = unsafeNewIORef m Nothing
1659− }
1492+merge :: forall k x. GCompare k => Dynamic x (R.PatchDMap k (Event x)) -> Event x (DMap k Identity)
1493+merge d = cacheEvent $ Event $ \sub -> do
1494+ initialParents <- readBehaviorUntracked $ dynamicCurrent d
1495+ accumRef <- liftIO $ newIORef $ error "merge: accumRef not yet initialized"
1496+ heightRef <- liftIO $ newIORef $ error "merge: heightRef not yet initialized"
1497+ heightBagRef <- liftIO $ newIORef $ error "merge: heightBagRef not yet initialized"
1498+ parentsRef <- liftIO $ newIORef $ error "merge: parentsRef not yet initialized"
1499+ let scheduleSelf height = scheduleMerge' height heightRef $ do
1500+ m <- liftIO $ readIORef accumRef
1501+ liftIO $ writeIORef accumRef $! DMap.empty
1502+ --TODO: Assert that m is not empty
1503+ subscriberPropagate sub m
1504+ invalidateMyHeight = do
1505+ oldHeight <- readIORef heightRef
1506+ when (oldHeight /= invalidHeight) $ do -- If the height used to be valid, it must be invalid now; we should never have *more* heights than we have parents
1507+ writeIORef heightRef $! invalidHeight
1508+ subscriberInvalidateHeight sub oldHeight
1509+ revalidateMyHeight = do
1510+ currentHeight <- readIORef heightRef
1511+ when (currentHeight == invalidHeight) $ do -- revalidateMergeHeight may be called multiple times; perhaps the's a way to finesse it to avoid this check
1512+ heights <- readIORef heightBagRef
1513+ parents <- readIORef parentsRef
1514+ -- When the number of heights in the bag reaches the number of parents, we should have a valid height
1515+ case heightBagSize heights `compare` DMap.size parents of
1516+ LT -> return ()
1517+ EQ -> do
1518+ let height = succHeight $ heightBagMax heights
1519+ when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: height: " <> show height
1520+ writeIORef heightRef $! height
1521+ subscriberRecalculateHeight sub height
1522+ GT -> error $ "revalidateMergeHeight: more heights than parents for Merge"
1523+ subscriber :: forall a. k a -> Subscriber x a
1524+ subscriber k = Subscriber
1525+ { subscriberPropagate = \a -> do
1526+ oldM <- liftIO $ readIORef accumRef
1527+ let newM = DMap.insertWith (error $ "Same key fired multiple times for Merge") k (Identity a) oldM
1528+ tracePropagate $ " DMap.size oldM = " <> show (DMap.size oldM) <> "; DMap.size newM = " <> show (DMap.size newM)
1529+ liftIO $ writeIORef accumRef $! newM
1530+ when (DMap.null oldM) $ do -- Only schedule the firing once
1531+ height <- liftIO $ readIORef heightRef
1532+ --TODO: assertions about height
1533+ currentHeight <- getCurrentHeight
1534+ when (height <= currentHeight) $ do
1535+ if height /= invalidHeight
1536+ then do
1537+ myStack <- liftIO $ whoCreatedIORef undefined --TODO
1538+ error $ "Height (" ++ show height ++ ") is not greater than current height (" ++ show currentHeight ++ ")\n" ++ unlines (reverse myStack)
1539+ else liftIO $ do
1540+#ifdef DEBUG_CYCLES
1541+ nodesInvolvedInCycle <- walkInvalidHeightParents $ eventSubscribedMerge subscribed
1542+ stacks <- forM nodesInvolvedInCycle $ \(Some.This es) -> whoCreatedEventSubscribed es
1543+ let cycleInfo = ":\n" <> drawForest (listsToForest stacks)
1544+#else
1545+ let cycleInfo = ""
1546+#endif
1547+ error $ "Causality loop found" <> cycleInfo
1548+ scheduleSelf height
1549+ , subscriberInvalidateHeight = \old -> do --TODO: When removing a parent doesn't actually change the height, maybe we can avoid invalidating
1550+ modifyIORef' heightBagRef $ heightBagRemove old
1551+ invalidateMyHeight
1552+ , subscriberRecalculateHeight = \new -> do
1553+ modifyIORef' heightBagRef $ heightBagAdd new
1554+ revalidateMyHeight
1555+ }
1556+ subscribers :: [(Maybe (DSum k Identity), Height, DSum k (MergeSubscribedParent x k))] <- forM (DMap.toList initialParents) $ \(k :=> e) -> do
1557+ let s = subscriber k
1558+ (subscription@(EventSubscription _ parentSubd), parentOcc) <- subscribeAndRead e s
1559+ height <- liftIO $ getEventSubscribedHeight parentSubd
1560+ return $ (fmap (\x -> k :=> Identity x) parentOcc, height, k :=> MergeSubscribedParent subscription s)
1561+ let dm = DMap.fromDistinctAscList $ mapMaybe (\(x, _, _) -> x) subscribers
1562+ heights = fmap (\(_, h, _) -> h) subscribers --TODO: Assert that there's no invalidHeight in here
1563+ myHeightBag = heightBagFromList $ filter (/= invalidHeight) heights
1564+ myHeight = if invalidHeight `elem` heights
1565+ then invalidHeight
1566+ else succHeight $ heightBagMax myHeightBag
1567+ currentHeight <- getCurrentHeight
1568+ let (occ, accum) = if currentHeight >= myHeight -- If we should have fired by now
1569+ then (if DMap.null dm then Nothing else Just dm, DMap.empty)
1570+ else (Nothing, dm)
1571+ unless (DMap.null accum) $ do
1572+ scheduleSelf myHeight
1573+ occRef <- liftIO $ newIORef occ
1574+ when (isJust occ) $ scheduleClear occRef
1575+ liftIO $ writeIORef accumRef $! accum
1576+ liftIO $ writeIORef heightRef $! myHeight
1577+ liftIO $ writeIORef heightBagRef $! myHeightBag
1578+ changeSubdRef <- liftIO $ newIORef $ error "getMergeSubscribed: changeSubdRef not yet initialized"
1579+ liftIO $ writeIORef parentsRef $! DMap.fromDistinctAscList $ map (\(_, _, x) -> x) subscribers
1580+ let updateMe (R.PatchDMap p) = do
1581+ oldParents <- liftIO $ readIORef parentsRef
1582+ let f (subscriptionsToKill, ps) (k :=> R.ComposeMaybe me) = do
1583+ (mOldSubd, newPs) <- case me of
1584+ Nothing -> return $ DMap.updateLookupWithKey (\_ _ -> Nothing) k ps
1585+ Just e -> do
1586+ let s = subscriber k
1587+ subscription@(EventSubscription _ subd) <- subscribe e s
1588+ newParentHeight <- liftIO $ getEventSubscribedHeight subd
1589+ let newParent = MergeSubscribedParent subscription s
1590+ liftIO $ modifyIORef' heightBagRef $ heightBagAdd newParentHeight
1591+ return $ DMap.insertLookupWithKey' (\_ new _ -> new) k newParent ps
1592+ forM_ mOldSubd $ \oldSubd -> do
1593+ oldHeight <- liftIO $ getEventSubscribedHeight $ _eventSubscription_subscribed $ mergeSubscribedParentSubscription oldSubd
1594+ liftIO $ modifyIORef heightBagRef $ heightBagRemove oldHeight
1595+ return (maybeToList (mergeSubscribedParentSubscription <$> mOldSubd) ++ subscriptionsToKill, newPs)
1596+ (subscriptionsToKill, newParents) <- foldM f ([], oldParents) $ DMap.toList p --TODO: I think this runEventM is OK, since no events are firing at this time, but it might not be
1597+ liftIO $ writeIORef parentsRef $! newParents
1598+ return subscriptionsToKill
1599+ defer $ SomeMergeInit $ do
1600+ let s = Subscriber
1601+ { subscriberPropagate = \a -> {-# SCC "traverseMergeChange" #-} do
1602+ tracePropagate $ "SubscriberMerge/Change"
1603+ defer $ SomeMergeUpdate (updateMe a) invalidateMyHeight revalidateMyHeight
1604+ , subscriberInvalidateHeight = \_ -> return ()
1605+ , subscriberRecalculateHeight = \_ -> return ()
1606+ }
1607+ (EventSubscription _ changeSubd, change) <- subscribeAndRead (dynamicUpdated d) s
1608+ forM_ change $ \c -> defer $ SomeMergeUpdate (updateMe c) invalidateMyHeight revalidateMyHeight
1609+ liftIO $ writeIORef changeSubdRef (s, changeSubd)
1610+ let unsubscribeAll = do
1611+ parents <- readIORef parentsRef
1612+ forM_ (DMap.toList parents) $ \(_ :=> MergeSubscribedParent s _) -> unsubscribe s
1613+ return ( EventSubscription unsubscribeAll $ EventSubscribed heightRef $ toAny (parentsRef, changeSubdRef)
1614+ , occ
1615+ )
16601616
16611617newtype EventSelector x k = EventSelector { select :: forall a. k a -> Event x a }
16621618
@@ -1682,34 +1638,6 @@ runHoldInits holdInitRef mergeInitRef = do
16821638initHold :: SomeHoldInit x -> EventM x ()
16831639initHold (SomeHoldInit h) = void $ getHoldEventSubscription h
16841640
1685−mkInitMerge :: GCompare k => MergeSubscribed x k -> Event x (R.PatchDMap k (Event x)) -> EventM x ()
1686−mkInitMerge subscribed changed = do
1687− changeSub <- liftIO $ newSubscriberMergeChange subscribed
1688− (EventSubscription _ changeSubd, change) <- subscribeAndRead changed changeSub
1689− forM_ change $ \c -> defer $ SomeMergeUpdate (mkUpdateMerge c subscribed) (invalidateMergeHeight subscribed) (revalidateMergeHeight subscribed)
1690− liftIO $ writeIORef (mergeSubscribedChange subscribed) (changeSub, changeSubd)
1691−
1692−mkUpdateMerge :: forall k x. GCompare k => R.PatchDMap k (Event x) -> MergeSubscribed x k -> EventM x [EventSubscription x]
1693−mkUpdateMerge (R.PatchDMap p) m = do
1694− oldParents <- liftIO $ readIORef $ mergeSubscribedParents m
1695− let f (subscriptionsToKill, ps) (k :=> R.ComposeMaybe me) = do
1696− (mOldSubd, newPs) <- case me of
1697− Nothing -> return $ DMap.updateLookupWithKey (\_ _ -> Nothing) k ps
1698− Just e -> do
1699− s <- liftIO $ newSubscriberMerge k m
1700− subscription@(EventSubscription _ subd) <- subscribe e s
1701− newParentHeight <- liftIO $ getEventSubscribedHeight subd
1702− let newParent = MergeSubscribedParent subscription s
1703− liftIO $ addMergeHeight newParentHeight m
1704− return $ DMap.insertLookupWithKey' (\_ new _ -> new) k newParent ps
1705− forM_ mOldSubd $ \oldSubd -> do
1706− oldHeight <- liftIO $ getEventSubscribedHeight $ _eventSubscription_subscribed $ mergeSubscribedParentSubscription oldSubd
1707− liftIO $ removeMergeHeight oldHeight m
1708− return (maybeToList (mergeSubscribedParentSubscription <$> mOldSubd) ++ subscriptionsToKill, newPs)
1709− (subscriptionsToKill, newParents) <- foldM f ([], oldParents) $ DMap.toList p --TODO: I think this runEventM is OK, since no events are firing at this time, but it might not be
1710− liftIO $ writeIORef (mergeSubscribedParents m) $! newParents
1711− return subscriptionsToKill
1712−
17131641-- | Run an event action outside of a frame
17141642runFrame :: forall x a. EventM x a -> SpiderHost x a --TODO: This function also needs to hold the mutex
17151643runFrame a = SpiderHost $ ask >>= \spiderTimeline -> lift $ do
@@ -1816,44 +1744,6 @@ succHeight h@(Height a) =
18161744 then invalidHeight
18171745 else Height $ succ a
18181746
1819−removeMergeHeight :: Height -> MergeSubscribed x k -> IO ()
1820−removeMergeHeight oldParentHeight subscribed = do
1821− when debugInvalidateHeight $ putStrLn $ "removeMergeHeight: " <> show oldParentHeight <> " Merge" <> showNodeId subscribed
1822− oldHeights <- readIORef $ mergeSubscribedHeightBag subscribed
1823− let newHeights = heightBagRemove oldParentHeight oldHeights
1824− writeIORef (mergeSubscribedHeightBag subscribed) $! newHeights
1825−
1826−invalidateMergeHeight :: MergeSubscribed x k -> IO ()
1827−invalidateMergeHeight subscribed = do
1828− oldHeight <- readIORef $ mergeSubscribedHeight subscribed
1829− -- If the height used to be valid, it must be invalid now; we should never have *more* heights than we have parents
1830− when (oldHeight /= invalidHeight) $ do
1831− writeIORef (mergeSubscribedHeight subscribed) $! invalidHeight
1832− WeakBag.traverse (mergeSubscribedSubscribers subscribed) $ invalidateSubscriberHeight oldHeight
1833−
1834−addMergeHeight :: Height -> MergeSubscribed x k -> IO ()
1835−addMergeHeight newParentHeight subscribed = do
1836− when debugInvalidateHeight $ putStrLn $ "addMergeHeight: " <> show newParentHeight <> " Merge" <> showNodeId subscribed
1837− oldHeights <- readIORef $ mergeSubscribedHeightBag subscribed
1838− let newHeights = heightBagAdd newParentHeight oldHeights
1839− writeIORef (mergeSubscribedHeightBag subscribed) $! newHeights
1840−
1841−revalidateMergeHeight :: MergeSubscribed x k -> IO ()
1842−revalidateMergeHeight subscribed = do
1843− currentHeight <- readIORef $ mergeSubscribedHeight subscribed
1844− when (currentHeight == invalidHeight) $ do -- revalidateMergeHeight may be called multiple times; perhaps the's a way to finesse it to avoid this check
1845− heights <- readIORef $ mergeSubscribedHeightBag subscribed
1846− parents <- readIORef $ mergeSubscribedParents subscribed
1847− -- When the number of heights in the bag reaches the number of parents, we should have a valid height
1848− case heightBagSize heights `compare` DMap.size parents of
1849− LT -> return ()
1850− EQ -> do
1851− let height = succHeight $ heightBagMax heights
1852− when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: height: " <> show height
1853− writeIORef (mergeSubscribedHeight subscribed) $! height
1854− WeakBag.traverse (mergeSubscribedSubscribers subscribed) $ recalculateSubscriberHeight height
1855− GT -> error $ "revalidateMergeHeight: more heights than parents for Merge" <> showNodeId subscribed
1856−
18571747invalidateCoincidenceHeight :: CoincidenceSubscribed x a -> IO ()
18581748invalidateCoincidenceHeight subscribed = do
18591749 oldHeight <- readIORef $ coincidenceSubscribedHeight subscribed