@@ -20,6 +20,7 @@
2020{-# LANGUAGE DeriveFunctor #-}
2121{-# LANGUAGE StandaloneDeriving #-}
2222{-# LANGUAGE InstanceSigs #-}
23+{-# LANGUAGE RecursiveDo #-}
2324
2425#ifdef USE_REFLEX_OPTIMIZER
2526{-# OPTIONS_GHC -fplugin=Reflex.Optimizer #-}
@@ -59,7 +60,7 @@ import qualified Data.IntMap.Strict as IntMap
5960import Data.IORef
6061import Data.Kind (Type)
6162import Data.Maybe hiding (mapMaybe)
62−import Data.Monoid ((<>))
63+import Data.Monoid (mempty, (<>))
6364import Data.Proxy
6465import Data.These
6566import Data.Traversable
@@ -81,10 +82,11 @@ import Data.These.Combinators (justThese)
8182#ifdef DEBUG_CYCLES
8283import Control.Monad.State hiding (forM, forM_, mapM, mapM_, sequence)
8384import Data.List.NonEmpty (NonEmpty (..), nonEmpty)
85+#endif
86+
8487import qualified Data.List.NonEmpty as NonEmpty
85−import Data.Monoid (mempty)
8688import Data.Tree (Forest, Tree (..), drawForest)
87−#endif
89+import Data.List (isPrefixOf)
8890
8991import Data.FastWeakBag (FastWeakBag, FastWeakBagTicket)
9092import qualified Data.FastWeakBag as FastWeakBag
@@ -111,7 +113,7 @@ withStackOneLine :: (BS8.ByteString -> a) -> a
111113withStackOneLine expr = unsafePerformIO $ do
112114 stack <- currentCallStack
113115 return (expr . BS8.pack . unwords . dropInternal . reverse $ stack)
114− where dropInternal = filter (not . isPrefixOf "Reflex.Spider.Internal")
116+ where dropInternal = filterStack "Reflex.Spider.Internal"
115117
116118#endif
117119
@@ -121,6 +123,12 @@ withStackOneLine expr = unsafePerformIO $ do
121123class HasNodeId a where
122124 getNodeId :: a -> Int
123125
126+instance HasNodeId (CacheSubscribed x a) where
127+ getNodeId = _cacheSubscribed_nodeId
128+
129+instance HasNodeId (FanInt x a) where
130+ getNodeId = _fanInt_nodeId
131+
124132instance HasNodeId (Hold x p) where
125133 getNodeId = holdNodeId
126134
@@ -141,7 +149,10 @@ instance HasNodeId (Pull x a) where
141149
142150{-# INLINE showNodeId #-}
143151showNodeId :: HasNodeId a => a -> String
144−showNodeId = ("#"<>) . show . getNodeId
152+showNodeId = showNodeId' . getNodeId
153+
154+showNodeId' :: Int -> String
155+showNodeId' = ("#"<>) . show
145156
146157
147158#else
@@ -196,13 +207,13 @@ subscribeAndRead = unEvent
196207-- caching; if the computation function is very cheap, this is (much) more
197208-- efficient than 'push'
198209{-# INLINE [1] pushCheap #-}
199−pushCheap :: (a -> ComputeM x (Maybe b)) -> Event x a -> Event x b
210+pushCheap :: HasSpiderTimeline x => (a -> ComputeM x (Maybe b)) -> Event x a -> Event x b
200211pushCheap !f e = Event $ \sub -> do
201− (subscription, occ) <- subscribeAndRead e $ sub
212+ (subscription, occ) <- subscribeAndRead e $ debugSubscriber' "push" (sub
202213 { subscriberPropagate = \a -> do
203214 mb <- f a
204215 mapM_ (subscriberPropagate sub) mb
205− }
216+ })
206217 occ' <- join <$> mapM f occ
207218 return (subscription, occ')
208219
@@ -217,10 +228,10 @@ terminalSubscriber p = Subscriber
217228
218229-- | Subscribe to an Event only for the duration of one occurrence
219230{-# INLINE subscribeAndReadHead #-}
220−subscribeAndReadHead :: Event x a -> Subscriber x a -> EventM x (EventSubscription x, Maybe a)
231+subscribeAndReadHead :: HasSpiderTimeline x => Event x a -> Subscriber x a -> EventM x (EventSubscription x, Maybe a)
221232subscribeAndReadHead e sub = do
222233 subscriptionRef <- liftIO $ newIORef $ error "subscribeAndReadHead: not initialized"
223− (subscription, occ) <- subscribeAndRead e $ sub
234+ (subscription, occ) <- subscribeAndRead e $ debugSubscriber' "head" $ sub
224235 { subscriberPropagate = \a -> do
225236 liftIO $ unsubscribe =<< readIORef subscriptionRef
226237 subscriberPropagate sub a
@@ -231,14 +242,14 @@ subscribeAndReadHead e sub = do
231242 return (subscription, occ)
232243
233244--TODO: Make this lazy in its input event
234−headE :: (MonadIO m, Defer (SomeMergeInit x) m) => Event x a -> m (Event x a)
245+headE :: (MonadIO m, Defer (SomeMergeInit x) m, HasSpiderTimeline x) => Event x a -> m (Event x a)
235246headE originalE = do
236247 parent <- liftIO $ newIORef $ Just originalE
237248 defer $ SomeMergeInit $ do --TODO: Rename SomeMergeInit appropriately
238249 let clearParent = liftIO $ writeIORef parent Nothing
239− (_, occ) <- subscribeAndReadHead originalE $ terminalSubscriber $ \_ -> clearParent
250+ (_, occ) <- subscribeAndReadHead originalE $ terminalSubscriber $ const clearParent
240251 when (isJust occ) clearParent
241− return $ Event $ \sub -> do
252+ return $ Event $ \sub ->
242253 liftIO (readIORef parent) >>= \case
243254 Nothing -> subscribeAndReadNever
244255 Just e -> subscribeAndReadHead e sub
@@ -247,8 +258,13 @@ data CacheSubscribed x a
247258 = CacheSubscribed { _cacheSubscribed_subscribers :: {-# UNPACK #-} !(FastWeakBag (Subscriber x a))
248259 , _cacheSubscribed_parent :: {-# UNPACK #-} !(EventSubscription x)
249260 , _cacheSubscribed_occurrence :: {-# UNPACK #-} !(IORef (Maybe a))
261+#ifdef DEBUG_NODEIDS
262+ , _cacheSubscribed_nodeId :: {-# UNPACK #-} !Int
263+#endif
250264 }
251265
266+
267+
252268-- | Construct an 'Event' whose value is guaranteed not to be recomputed
253269-- repeatedly
254270--
@@ -273,16 +289,20 @@ cacheEvent e =
273289 subscribedTicket <- liftIO (readIORef mSubscribedRef >>= getFastWeakTicket) >>= \case
274290 Just subscribedTicket -> return subscribedTicket
275291 Nothing -> do
292+#ifdef DEBUG_NODEIDS
293+ nodeId <- liftIO newNodeId
294+#endif
276295 subscribers <- liftIO FastWeakBag.empty
277296 occRef <- liftIO $ newIORef Nothing -- This should never be read prior to being set below
278− (parentSub, occ) <- subscribeAndRead e $ Subscriber
279− { subscriberPropagate = \a -> do
280− liftIO $ writeIORef occRef (Just a)
281− scheduleClear occRef
282− propagateFast a subscribers
283− , subscriberInvalidateHeight = FastWeakBag.traverse subscribers . invalidateSubscriberHeight
284− , subscriberRecalculateHeight = FastWeakBag.traverse subscribers . recalculateSubscriberHeight
285− }
297+ (parentSub, occ) <- subscribeAndRead e $ debugSubscriber' ("cacheEvent" <> showNodeId' nodeId) $ Subscriber
298+ { subscriberPropagate = \a -> do
299+ -- checkCycle (_eventSubscription_subscribed parentSub)
300+ liftIO $ writeIORef occRef (Just a)
301+ scheduleClear occRef
302+ propagateFast a subscribers
303+ , subscriberInvalidateHeight = FastWeakBag.traverse subscribers . invalidateSubscriberHeight
304+ , subscriberRecalculateHeight = FastWeakBag.traverse subscribers . recalculateSubscriberHeight
305+ }
286306 when (isJust occ) $ do
287307 liftIO $ writeIORef occRef occ -- Set the initial value of occRef; we don't need to do this if occ is Nothing
288308 scheduleClear occRef
@@ -290,6 +310,9 @@ cacheEvent e =
290310 { _cacheSubscribed_subscribers = subscribers
291311 , _cacheSubscribed_parent = parentSub
292312 , _cacheSubscribed_occurrence = occRef
313+#ifdef DEBUG_NODEIDS
314+ , _cacheSubscribed_nodeId = nodeId
315+#endif
293316 }
294317 subscribedTicket <- liftIO $ mkFastWeakTicket subscribed
295318 liftIO $ writeIORef mSubscribedRef =<< getFastWeakTicketWeak subscribedTicket
@@ -321,8 +344,8 @@ cacheSubscription sub mSubscribedRef subscribedTicket = do
321344 , eventSubscribedGetParents = return [_eventSubscription_subscribed parentSub]
322345 , eventSubscribedHasOwnHeightRef = False
323346 , eventSubscribedWhoCreated = whoCreatedIORef mSubscribedRef
324− }
325347#endif
348+ }
326349 }
327350 return (es, occ)
328351
@@ -446,7 +469,7 @@ newSubscriberCoincidenceOuter subscribed = debugSubscriber ("SubscriberCoinciden
446469 liftIO $ writeIORef (coincidenceSubscribedInnerParent subscribed) $ Just innerSubd
447470 scheduleClear $ coincidenceSubscribedInnerParent subscribed
448471 case occ of
449− Nothing -> do
472+ Nothing ->
450473 when (innerHeight > outerHeight) $ liftIO $ do -- If the event fires, it will fire at a later height
451474 writeIORef (coincidenceSubscribedHeight subscribed) $! innerHeight
452475 WeakBag.traverse (coincidenceSubscribedSubscribers subscribed) $ invalidateSubscriberHeight outerHeight
@@ -477,18 +500,18 @@ invalidateSubscriberHeight :: Height -> Subscriber x a -> IO ()
477500invalidateSubscriberHeight = flip subscriberInvalidateHeight
478501
479502recalculateSubscriberHeight :: Height -> Subscriber x a -> IO ()
480−recalculateSubscriberHeight = flip subscriberRecalculateHeight
503+recalculateSubscriberHeight = flip subscriberRecalculateHeight
481504
482505-- | Propagate everything at the current height
483−propagate :: a -> WeakBag (Subscriber x a) -> EventM x ()
484−propagate a subscribers = withIncreasedDepth (Proxy::Proxy x) $ do
506+propagate :: forall x a. HasSpiderTimeline x => a -> WeakBag (Subscriber x a) -> EventM x ()
507+propagate a subscribers = withIncreasedDepth (Proxy::Proxy x) $
485508 -- 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
486509 --TODO: Should we check if nodes already have their values before propagating? Maybe we're re-doing work
487510 WeakBag.traverse subscribers $ \s -> subscriberPropagate s a
488511
489512-- | Propagate everything at the current height
490−propagateFast :: a -> FastWeakBag (Subscriber x a) -> EventM x ()
491−propagateFast a subscribers = withIncreasedDepth (Proxy::Proxy x) $ do
513+propagateFast :: forall x a. HasSpiderTimeline x => a -> FastWeakBag (Subscriber x a) -> EventM x ()
514+propagateFast a subscribers = withIncreasedDepth (Proxy::Proxy x) $
492515 -- 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
493516 --TODO: Should we check if nodes already have their values before propagating? Maybe we're re-doing work
494517 FastWeakBag.traverse subscribers $ \s -> subscriberPropagate s a
@@ -999,7 +1022,8 @@ heightBagFromList :: [Height] -> HeightBag
9991022heightBagFromList heights = heightBagVerify $ foldl' (flip heightBagAdd) heightBagEmpty heights
10001023
10011024heightBagAdd :: Height -> HeightBag -> HeightBag
1002−heightBagAdd (Height h) (HeightBag s c) = heightBagVerify $ HeightBag (succ s) $ IntMap.insertWithKey (\_ _ old -> succ old) h 0 c
1025+heightBagAdd (Height h) (HeightBag s c) = heightBagVerify $ HeightBag (succ s) $
1026+ IntMap.insertWithKey (\_ _ old -> succ old) h 0 c
10031027
10041028heightBagRemove :: Height -> HeightBag -> HeightBag
10051029heightBagRemove (Height h) b@(HeightBag s c) = heightBagVerify $ case IntMap.lookup h c of
@@ -1008,6 +1032,13 @@ heightBagRemove (Height h) b@(HeightBag s c) = heightBagVerify $ case IntMap.loo
10081032 0 -> IntMap.delete h c
10091033 _ -> IntMap.insert h (pred old) c
10101034
1035+heightBagRemoveMaybe :: Height -> HeightBag -> Maybe HeightBag
1036+heightBagRemoveMaybe (Height h) b@(HeightBag s c) = heightBagVerify . removed <$> IntMap.lookup h c where
1037+ removed old = HeightBag (pred s) $ case old of
1038+ 0 -> IntMap.delete h c
1039+ _ -> IntMap.insert h (pred old) c
1040+
1041+
10111042heightBagMax :: HeightBag -> Height
10121043heightBagMax (HeightBag _ c) = case IntMap.maxViewWithKey c of
10131044 Just ((h, _), _) -> Height h
@@ -1265,17 +1296,21 @@ type CanTrace x m = (HasSpiderTimeline x, MonadIO m)
12651296#ifdef DEBUG
12661297
12671298debugSubscriber :: forall x a. HasSpiderTimeline x => String -> Subscriber x a -> IO (Subscriber x a)
1268−debugSubscriber description subscribed = return $ Subscriber
1299+debugSubscriber description = return . debugSubscriber' description
1300+
1301+
1302+debugSubscriber' :: forall x a. HasSpiderTimeline x => String -> Subscriber x a -> Subscriber x a
1303+debugSubscriber' description subscribed = Subscriber
12691304 {
12701305 subscriberPropagate = \m -> do
12711306 tracePropagate (Proxy :: Proxy x) ("subscriberPropagate: " <> description)
12721307 subscriberPropagate subscribed m
12731308 , subscriberInvalidateHeight = \old -> do
1274− putStrLn $ "invalidateSubscriberHeight: " <> description
1309+ putStrLn $ "invalidateSubscriberHeight: " <> description <> ", old = " <> show (unHeight old)
12751310 subscriberInvalidateHeight subscribed old
12761311 putStrLn $ "invalidateSubscriberHeight: " <> description <> ", done"
12771312 , subscriberRecalculateHeight = \new -> do
1278− putStrLn $ "subscriberRecalculateHeight: " <> description
1313+ putStrLn $ "subscriberRecalculateHeight: " <> description <> ", new = " <> show (unHeight new)
12791314 subscriberRecalculateHeight subscribed new
12801315 putStrLn $ "subscriberRecalculateHeight: " <> description <> ", done"
12811316 }
@@ -1336,18 +1371,9 @@ traceM _ _ = return ()
13361371
13371372#endif
13381373
1339−
13401374whoCreatedIORef :: IORef a -> IO [String]
13411375whoCreatedIORef (IORef a) = whoCreated $! a
13421376
1343−#ifdef DEBUG_CYCLES
1344−
1345−data EventLoopException = EventLoopException (Forest String)
1346−instance Exception EventLoopException
1347−
1348−instance Show EventLoopException where
1349− show (EventLoopException stacks) = "Causality loop detected:\n" <> drawForest stacks
1350−
13511377groupByHead :: Eq a => [NonEmpty a] -> [(a, NonEmpty [a])]
13521378groupByHead = \case
13531379 [] -> []
@@ -1358,7 +1384,25 @@ groupByHead = \case
13581384 | otherwise -> (x, xs :| []) : l
13591385
13601386listsToForest :: Eq a => [[a]] -> Forest a
1361−listsToForest l = fmap (\(a, l') -> Node a $ listsToForest $ toList l') $ groupByHead $ catMaybes $ fmap nonEmpty l
1387+listsToForest lists = buildForest <$> groupByHead (mapMaybe nonEmpty lists)
1388+ where buildForest (a, lists') = Node a $ listsToForest $ toList lists'
1389+
1390+showStacks :: [[String]] -> String
1391+showStacks = drawForest . listsToForest . fmap (filterStack "Reflex.Spider.Internal")
1392+
1393+filterStack :: String -> [String] -> [String]
1394+filterStack prefix = filter (not . (prefix `isPrefixOf`))
1395+
1396+
1397+#ifdef DEBUG_CYCLES
1398+
1399+data EventLoopException = EventLoopException [[String]]
1400+instance Exception EventLoopException
1401+
1402+instance Show EventLoopException where
1403+ show (EventLoopException stacks) = "causality loop detected:\n" <> if null stacks
1404+ then "no location information, compile with profiling enabled for stack tree"
1405+ else showStacks stacks
13621406
13631407#else
13641408
@@ -1366,8 +1410,8 @@ data EventLoopException = EventLoopException
13661410instance Exception EventLoopException
13671411
13681412instance Show EventLoopException where
1369− show EventLoopException = "Causality loop detected: \n" <>
1370− "no location information, compile with --enable-profiling and reflex with flag 'debug-cycles'"
1413+ show EventLoopException = "causality loop detected: \n" <>
1414+ "compile reflex with flag 'debug-cycles' and compile with profiling enabled for stack tree"
13711415
13721416#endif
13731417
@@ -1504,6 +1548,9 @@ data FanInt x a = FanInt
15041548 { _fanInt_subscribers :: {-# UNPACK #-} !(FastMutableIntMap (FastWeakBag (Subscriber x a))) --TODO: Clean up the keys in here when their child weak bags get empty --TODO: Remove our own subscription when the subscribers list is completely empty
15051549 , _fanInt_subscriptionRef :: {-# UNPACK #-} !(IORef (EventSubscription x)) -- This should have a valid subscription iff subscribers is non-empty
15061550 , _fanInt_occRef :: {-# UNPACK #-} !(IORef (IntMap a))
1551+#ifdef DEBUG_NODEIDS
1552+ , _fanInt_nodeId :: {-# UNPACK #-} !Int
1553+#endif
15071554 }
15081555
15091556newFanInt :: IO (FanInt x a)
@@ -1511,10 +1558,16 @@ newFanInt = do
15111558 subscribers <- FastMutableIntMap.newEmpty --TODO: Clean up the keys in here when their child weak bags get empty --TODO: Remove our own subscription when the subscribers list is completely empty
15121559 subscriptionRef <- newIORef $ error "fanInt: no subscription"
15131560 occRef <- newIORef $ error "fanInt: no occurrence"
1561+#ifdef DEBUG_NODEIDS
1562+ nodeId <- newNodeId
1563+#endif
15141564 return $ FanInt
15151565 { _fanInt_subscribers = subscribers
15161566 , _fanInt_subscriptionRef = subscriptionRef
15171567 , _fanInt_occRef = occRef
1568+#ifdef DEBUG_NODEIDS
1569+ , _fanInt_nodeId = nodeId
1570+#endif
15181571 }
15191572
15201573fanInt :: HasSpiderTimeline x => Event x (IntMap a) -> EventSelectorInt x a
@@ -1523,7 +1576,8 @@ fanInt p = unsafePerformIO $ do
15231576 pure $ EventSelectorInt $ \k -> Event $ \sub -> do
15241577 isEmpty <- liftIO $ FastMutableIntMap.isEmpty (_fanInt_subscribers self)
15251578 when isEmpty $ do -- This is the first subscriber, so we need to subscribe to our input
1526− (subscription, parentOcc) <- subscribeAndRead p $ Subscriber
1579+ let desc = "fanInt" <> showNodeId self <> ", k = " <> show k
1580+ (subscription, parentOcc) <- subscribeAndRead p $ debugSubscriber' desc $ Subscriber
15271581 { subscriberPropagate = \m -> do
15281582 liftIO $ writeIORef (_fanInt_occRef self) m
15291583 scheduleIntClear $ _fanInt_occRef self
@@ -1919,26 +1973,29 @@ scheduleMergeSelf m height = scheduleMerge' height (_merge_heightRef m) $ do
19191973 --TODO: Assert that m is not empty
19201974 subscriberPropagate (_merge_sub m) vals
19211975
1922−checkCycle :: HasSpiderTimeline x => EventSubscribed x -> Height -> EventM x ()
1923−checkCycle subscribed height = do
1924− currentHeight <- getCurrentHeight
1925− when (height <= currentHeight) $ if height /= invalidHeight
1926− then do
1927− myStack <- liftIO $ whoCreatedIORef undefined --TODO
1928− error $ "Height (" ++ show height ++ ") is not greater than current height (" ++ show currentHeight ++ ")\n" ++ unlines (reverse myStack)
1929− else liftIO $
1976+checkCycle :: HasSpiderTimeline x => EventSubscribed x -> EventM x ()
1977+checkCycle subscribed = liftIO $ do
1978+ height <- readIORef (eventSubscribedHeightRef subscribed)
1979+
1980+ -- currentHeight <- getCurrentHeight
1981+ -- when (height <= currentHeight) $ if height /= invalidHeight
1982+ -- then do
1983+ -- myStack <- liftIO $ whoCreatedIORef undefined --TODO
1984+ -- error $ "Height (" ++ show height ++ ") is not greater than current height (" ++ show currentHeight ++ ")\n" ++ unlines (reverse myStack)
1985+ -- else liftIO $
1986+ when (height == invalidHeight) $
19301987#ifdef DEBUG_CYCLES
1931− do
1932− nodesInvolvedInCycle <- walkInvalidHeightParents subscribed
1933− stacks <- forM nodesInvolvedInCycle whoCreatedEventSubscribed
1934− throwIO (EventLoopException (listsToForest stacks))
1988+ do
1989+ nodesInvolvedInCycle <- walkInvalidHeightParents subscribed
1990+ stacks <- forM nodesInvolvedInCycle whoCreatedEventSubscribed
1991+ throwIO (EventLoopException stacks)
19351992#else
1936− throwIO EventLoopException
1993+ throwIO EventLoopException
19371994#endif
1938−
19391995
1940−mergeSubscriber :: forall x k v s a. (HasSpiderTimeline x, GCompare k) => Merge x k v s -> EventM x (k a) -> Subscriber x (v a)
1941−mergeSubscriber m getKey = Subscriber
1996+
1997+mergeSubscriber :: forall x k v s a. (HasSpiderTimeline x, GCompare k) => EventSubscribed x -> Merge x k v s -> EventM x (k a) -> Subscriber x (v a)
1998+mergeSubscriber subscribed m getKey = Subscriber
19421999 { subscriberPropagate = \a -> do
19432000 oldM <- liftIO $ readIORef $ _merge_accumRef m
19442001 k <- getKey
@@ -1947,7 +2004,7 @@ mergeSubscriber m getKey = Subscriber
19472004 liftIO $ writeIORef (_merge_accumRef m) $! newM
19482005 when (DMap.null oldM) $ do -- Only schedule the firing once
19492006 height <- liftIO $ readIORef $ _merge_heightRef m
1950− -- checkCycle (_merge_sub m) height --TODO: assertions about height
2007+ checkCycle subscribed
19512008
19522009 scheduleMergeSelf m height
19532010 , subscriberInvalidateHeight = \old -> do --TODO: When removing a parent doesn't actually change the height, maybe we can avoid invalidating
@@ -1959,11 +2016,11 @@ mergeSubscriber m getKey = Subscriber
19592016 }
19602017
19612018--TODO: Be able to run as much of this as possible promptly
1962−updateMerge :: (HasSpiderTimeline x, GCompare k) => Merge x k v s -> MergeUpdateFunc k v x p s -> p -> SomeMergeUpdate x
1963−updateMerge m updateFunc p = SomeMergeUpdate updateMe (invalidateMergeHeight m) (revalidateMergeHeight m)
2019+updateMerge :: (HasSpiderTimeline x, GCompare k) => EventSubscribed x -> Merge x k v s -> MergeUpdateFunc k v x p s -> p -> SomeMergeUpdate x
2020+updateMerge subscribed m updateFunc p = SomeMergeUpdate updateMe (invalidateMergeHeight m) (revalidateMergeHeight m)
19642021 where updateMe = do
19652022 oldParents <- liftIO $ readIORef $ _merge_parentsRef m
1966− (subscriptionsToKill, newParents) <- updateFunc (mergeSubscriber m) (_merge_heightBagRef m) oldParents p
2023+ (subscriptionsToKill, newParents) <- updateFunc (mergeSubscriber subscribed m) (_merge_heightBagRef m) oldParents p
19672024 liftIO $ writeIORef (_merge_parentsRef m) $! newParents
19682025 return subscriptionsToKill
19692026
@@ -1976,14 +2033,28 @@ mergeGCheap' getParent getInitialSubscribers updateFunc destroy d = Event $ \sub
19762033 heightRef <- liftIO $ newIORef $ error "merge: heightRef not yet initialized"
19772034 heightBagRef <- liftIO $ newIORef $ error "merge: heightBagRef not yet initialized"
19782035 parentsRef :: IORef (DMap k s) <- liftIO $ newIORef $ error "merge: parentsRef not yet initialized"
1979− let m = Merge
2036+ changeSubdRef <- liftIO $ newIORef $ error "getMergeSubscribed: changeSubdRef not yet initialized"
2037+
2038+ let subscribed = EventSubscribed
2039+ { eventSubscribedHeightRef = heightRef
2040+ , eventSubscribedRetained = toAny (parentsRef, changeSubdRef)
2041+#ifdef DEBUG_CYCLES
2042+ , eventSubscribedGetParents = do
2043+ let getParent' (_ :=> v) = _eventSubscription_subscribed (getParent v)
2044+ fmap getParent' . DMap.toList <$> readIORef parentsRef
2045+ , eventSubscribedHasOwnHeightRef = False
2046+ , eventSubscribedWhoCreated = whoCreatedIORef heightRef
2047+#endif
2048+ }
2049+
2050+ m = Merge
19802051 { _merge_parentsRef = parentsRef
19812052 , _merge_heightBagRef = heightBagRef
19822053 , _merge_heightRef = heightRef
19832054 , _merge_sub = sub
19842055 , _merge_accumRef = accumRef
19852056 }
1986− (dm, heights, initialParentState) <- getInitialSubscribers initialParents $ mergeSubscriber m
2057+ (dm, heights, initialParentState) <- getInitialSubscribers initialParents $ mergeSubscriber subscribed m
19872058 let myHeightBag = heightBagFromList $ filter (/= invalidHeight) heights
19882059 myHeight = if invalidHeight `elem` heights
19892060 then invalidHeight
@@ -1992,38 +2063,27 @@ mergeGCheap' getParent getInitialSubscribers updateFunc destroy d = Event $ \sub
19922063 let (occ, accum) = if currentHeight >= myHeight -- If we should have fired by now
19932064 then (if DMap.null dm then Nothing else Just dm, DMap.empty)
19942065 else (Nothing, dm)
1995− unless (DMap.null accum) $ do
1996− scheduleMergeSelf m myHeight
2066+ unless (DMap.null accum) $ scheduleMergeSelf m myHeight
19972067 liftIO $ writeIORef accumRef $! accum
19982068 liftIO $ writeIORef heightRef $! myHeight
19992069 liftIO $ writeIORef heightBagRef $! myHeightBag
2000− changeSubdRef <- liftIO $ newIORef $ error "getMergeSubscribed: changeSubdRef not yet initialized"
20012070 liftIO $ writeIORef parentsRef $! initialParentState
20022071 defer $ SomeMergeInit $ do
2003− let s = Subscriber
2072+ let changeSubscriber = Subscriber
20042073 { subscriberPropagate = \a -> {-# SCC "traverseMergeChange" #-} do
2005− tracePropagate (Proxy :: Proxy x) $ "SubscriberMerge/Change"
2006− defer $ updateMerge m updateFunc a
2074+ tracePropagate (Proxy :: Proxy x) "SubscriberMerge/Change"
2075+ defer $ updateMerge subscribed m updateFunc a
20072076 , subscriberInvalidateHeight = \_ -> return ()
20082077 , subscriberRecalculateHeight = \_ -> return ()
20092078 }
2010− (changeSubscription, change) <- subscribeAndRead (dynamicUpdated d) s
2011− forM_ change $ \c -> defer $ updateMerge m updateFunc c
2079+ (changeSubscription, change) <- subscribeAndRead (dynamicUpdated d) changeSubscriber
2080+ forM_ change $ \c -> defer $ updateMerge subscribed m updateFunc c
20122081 -- We explicitly hold on to the unsubscribe function from subscribing to the update event.
20132082 -- If we don't do this, there are certain cases where mergeCheap will fail to properly retain
20142083 -- its subscription.
2015− liftIO $ writeIORef changeSubdRef (s, changeSubscription)
2084+ liftIO $ writeIORef changeSubdRef (changeSubscriber, changeSubscription)
20162085 let unsubscribeAll = destroy =<< readIORef parentsRef
2017− getParent' = \(_ :=> v) -> _eventSubscription_subscribed (getParent v)
2018− subscribed = EventSubscribed
2019− { eventSubscribedHeightRef = heightRef
2020− , eventSubscribedRetained = toAny (parentsRef, changeSubdRef)
2021−#ifdef DEBUG_CYCLES
2022− , eventSubscribedGetParents = fmap getParent' . DMap.toList <$> readIORef parentsRef
2023− , eventSubscribedHasOwnHeightRef = False
2024− , eventSubscribedWhoCreated = whoCreatedIORef heightRef
2025− }
2026−#endif
2086+
20272087 return (EventSubscription unsubscribeAll subscribed, occ)
20282088
20292089
@@ -2038,13 +2098,22 @@ mergeIntCheap d = Event $ \sub -> do
20382098 heightRef <- liftIO $ newIORef zeroHeight
20392099 heightBagRef <- liftIO $ newIORef heightBagEmpty
20402100 parents <- liftIO $ FastMutableIntMap.newEmpty
2101+ changeSubdRef <- liftIO $ newIORef $ error "getMergeSubscribed: changeSubdRef not yet initialized"
2102+ let subscribed = EventSubscribed
2103+ { eventSubscribedHeightRef = heightRef
2104+ , eventSubscribedRetained = toAny (parents, changeSubdRef)
2105+#ifdef DEBUG_CYCLES
2106+ , eventSubscribedGetParents = fmap (_eventSubscription_subscribed . snd) <$> FastMutableIntMap.toList parents
2107+ , eventSubscribedHasOwnHeightRef = False
2108+ , eventSubscribedWhoCreated = whoCreatedIORef heightRef
2109+#endif
2110+ }
20412111 let scheduleSelf = do
20422112 height <- liftIO $ readIORef $ heightRef
20432113 scheduleMerge' height heightRef $ do
20442114 vals <- liftIO $ FastMutableIntMap.getFrozenAndClear accum
20452115 subscriberPropagate sub vals
2046− invalidateMyHeight = do
2047− invalidateMergeHeight' heightRef sub
2116+ invalidateMyHeight = invalidateMergeHeight' heightRef sub
20482117 recalculateMyHeight = do
20492118 currentHeight <- readIORef heightRef
20502119 when (currentHeight == invalidHeight) $ do --TODO: This will almost always be true; can we get rid of this check and just proceed to the next one always?
@@ -2060,8 +2129,8 @@ mergeIntCheap d = Event $ \sub -> do
20602129 GT -> error $ "revalidateMergeHeight: more heights (" <> show (heightBagSize heights) <> ") than parents (" <> show numParents <> ") for Merge"
20612130 mySubscriber k = Subscriber
20622131 { subscriberPropagate = \a -> do
2063− -- liftIO (readIORef heightRef) >>= checkCycle
2064−
2132+ checkCycle subscribed
2133+
20652134 wasEmpty <- liftIO $ FastMutableIntMap.isEmpty accum
20662135 liftIO $ FastMutableIntMap.insert accum k a
20672136 when wasEmpty scheduleSelf
@@ -2095,7 +2164,6 @@ mergeIntCheap d = Event $ \sub -> do
20952164 else liftIO $ Just <$> FastMutableIntMap.getFrozenAndClear accum
20962165 else do when (not isEmpty) scheduleSelf -- We have things accumulated, but we shouldn't have fired them yet
20972166 return Nothing
2098− changeSubdRef <- liftIO $ newIORef $ error "getMergeSubscribed: changeSubdRef not yet initialized"
20992167 defer $ SomeMergeInit $ do
21002168 let updateMe a = SomeMergeUpdate u invalidateMyHeight recalculateMyHeight
21012169 where
@@ -2109,31 +2177,25 @@ mergeIntCheap d = Event $ \sub -> do
21092177 oldParents <- liftIO $ FastMutableIntMap.applyPatch parents newSubscriptions
21102178 liftIO $ for_ oldParents $ \oldParent -> do
21112179 oldParentHeight <- getEventSubscribedHeight $ _eventSubscription_subscribed oldParent
2180+
2181+ print ("updateMe", oldParentHeight)
21122182 modifyIORef' heightBagRef $ heightBagRemove oldParentHeight
21132183 return $ IntMap.elems oldParents
2114− let s = Subscriber
2184+ let changeSubscriber = Subscriber
21152185 { subscriberPropagate = \a -> {-# SCC "traverseMergeChange" #-} do
21162186 tracePropagate (Proxy :: Proxy x) $ "SubscriberMergeInt/Change"
21172187 defer $ updateMe a
21182188 , subscriberInvalidateHeight = \_ -> return ()
21192189 , subscriberRecalculateHeight = \_ -> return ()
21202190 }
2121− (changeSubscription, change) <- subscribeAndRead (dynamicUpdated d) s
2191+ (changeSubscription, change) <- subscribeAndRead (dynamicUpdated d) changeSubscriber
21222192 forM_ change $ \c -> defer $ updateMe c
21232193 -- We explicitly hold on to the unsubscribe function from subscribing to the update event.
21242194 -- If we don't do this, there are certain cases where mergeCheap will fail to properly retain
21252195 -- its subscription.
2126− liftIO $ writeIORef changeSubdRef (s, changeSubscription)
2196+ liftIO $ writeIORef changeSubdRef (changeSubscriber, changeSubscription)
21272197 let unsubscribeAll = traverse_ unsubscribe =<< FastMutableIntMap.getFrozenAndClear parents
2128− subscribed = EventSubscribed
2129− { eventSubscribedHeightRef = heightRef
2130− , eventSubscribedRetained = toAny (parents, changeSubdRef)
2131−#ifdef DEBUG_CYCLES
2132− , eventSubscribedGetParents = fmap (_eventSubscription_subscribed . snd) <$> FastMutableIntMap.toList parents
2133− , eventSubscribedHasOwnHeightRef = False
2134− , eventSubscribedWhoCreated = whoCreatedIORef heightRef
2135− }
2136−#endif
2198+
21372199
21382200 return (EventSubscription unsubscribeAll subscribed, occ)
21392201