@@ -57,6 +57,8 @@ import Data.Proxy
5757import Data.These
5858import Data.Traversable
5959import Data.Word
60+import Data.FastMutableIntMap (FastMutableIntMap, PatchIntMap (..))
61+import qualified Data.FastMutableIntMap as FastMutableIntMap
6062import GHC.Exts
6163import GHC.IORef (IORef (..))
6264import GHC.Stack
@@ -246,10 +248,10 @@ pushCheap :: HasSpiderTimeline x => (a -> ComputeM x (Maybe b)) -> Event x a ->
246248pushCheap !f e = Event $ \sub -> do
247249 (subscription, occ) <- subscribeAndRead e $ sub
248250 { subscriberPropagate = \a -> do
249− mb <- runComputeM $ f a
251+ mb <- f a
250252 mapM_ (subscriberPropagate sub) mb
251253 }
252− occ' <- join <$> mapM (runComputeM . f) occ
254+ occ' <- join <$> mapM f occ
253255 return (subscription, occ')
254256
255257data CacheSubscribed x a
@@ -762,6 +764,7 @@ data EventEnv x
762764 , eventEnvMergeUpdates :: !(IORef [SomeMergeUpdate x])
763765 , eventEnvMergeInits :: !(IORef [SomeMergeInit x]) -- Needed for Subscribe
764766 , eventEnvClears :: !(IORef [SomeClear]) -- Needed for Subscribe
767+ , eventEnvIntClears :: !(IORef [SomeIntClear])
765768 , eventEnvRootClears :: !(IORef [SomeRootClear])
766769 , eventEnvCurrentHeight :: !(IORef Height) -- Needed for Subscribe
767770 , eventEnvResetCoincidences :: !(IORef [SomeResetCoincidence x]) -- Needed for Subscribe
@@ -846,6 +849,14 @@ instance HasSpiderTimeline x => Defer SomeClear (EventM x) where
846849scheduleClear :: Defer SomeClear m => IORef (Maybe a) -> m ()
847850scheduleClear r = defer $ SomeClear r
848851
852+instance HasSpiderTimeline x => Defer SomeIntClear (EventM x) where
853+ {-# INLINE getDeferralQueue #-}
854+ getDeferralQueue = asksEventEnv eventEnvIntClears
855+
856+{-# INLINE scheduleIntClear #-}
857+scheduleIntClear :: Defer SomeIntClear m => IORef (IntMap a) -> m ()
858+scheduleIntClear r = defer $ SomeIntClear r
859+
849860instance HasSpiderTimeline x => Defer SomeRootClear (EventM x) where
850861 {-# INLINE getDeferralQueue #-}
851862 getDeferralQueue = asksEventEnv eventEnvRootClears
@@ -981,21 +992,6 @@ newtype SomeMergeInit x = SomeMergeInit { unSomeMergeInit :: EventM x () }
981992-- EventM can do everything BehaviorM can, plus create holds
982993newtype EventM x a = EventM { unEventM :: IO a } deriving (Functor, Applicative, Monad, MonadIO, MonadFix, MonadException, MonadAsyncException) -- The environment should be Nothing if we are not in a frame, and Just if we are - in which case it is a list of assignments to be done after the frame is over
983994
984−newtype ComputeM x a = ComputeM (ReaderT (IORef [SomeHoldInit x], IORef [SomeDynInit x]) IO a) deriving (Functor, Applicative, Monad, MonadIO, MonadFix)
985−
986−instance Defer (SomeHoldInit x) (ComputeM x) where
987− getDeferralQueue = ComputeM $ asks fst
988−
989−instance Defer (SomeDynInit x) (ComputeM x) where
990− getDeferralQueue = ComputeM $ asks snd
991−
992−{-# INLINABLE runComputeM #-}
993−runComputeM :: (Defer (SomeHoldInit x) m, Defer (SomeDynInit x) m) => ComputeM x a -> m a
994−runComputeM (ComputeM a) = do
995− holds <- getDeferralQueue
996− dyns <- getDeferralQueue
997− liftIO $ runReaderT a (holds, dyns)
998−
999995newtype MergeSubscribedParent x a = MergeSubscribedParent { unMergeSubscribedParent :: EventSubscription x }
1000996
1001997data MergeSubscribedParentWithMove x k a = MergeSubscribedParentWithMove
@@ -1239,9 +1235,11 @@ scheduleMerge' initialHeight heightRef a = scheduleMerge initialHeight $ do
12391235 GT -> scheduleMerge' height heightRef a -- The height has been increased (by a coincidence event; TODO: is this the only way?)
12401236 EQ -> a
12411237
1242−data SomeClear = forall a. SomeClear (IORef (Maybe a))
1238+data SomeClear = forall a. SomeClear {-# UNPACK #-} !(IORef (Maybe a))
1239+
1240+data SomeIntClear = forall a. SomeIntClear {-# UNPACK #-} !(IORef (IntMap a))
12431241
1244−data SomeRootClear = forall k. SomeRootClear (IORef (DMap k Identity))
1242+data SomeRootClear = forall k. SomeRootClear {-# UNPACK #-} !(IORef (DMap k Identity))
12451243
12461244data SomeAssignment x = forall a. SomeAssignment {-# UNPACK #-} !(IORef a) {-# UNPACK #-} !(IORef [Weak (Invalidator x)]) a
12471245
@@ -1377,6 +1375,7 @@ type CanSubscribe x m =
13771375 , HasCurrentHeight x m
13781376 , Defer (SomeHoldInit x) m, Defer (SomeMergeInit x) m -- The things we need to lazily initialize state
13791377 , Defer SomeClear m -- Wiping out event firings
1378+ , Defer SomeIntClear m
13801379 , Defer (SomeAssignment x) m -- Updating state
13811380 , Defer (SomeResetCoincidence x) m
13821381 , m ~ EventM x
@@ -1437,6 +1436,68 @@ cleanupRootSubscribed self@RootSubscribed { rootSubscribedKey = k, rootSubscribe
14371436subscribeRootSubscribed :: RootSubscribed x a -> Subscriber x a -> IO WeakBagTicket
14381437subscribeRootSubscribed subscribed sub = WeakBag.insert sub (rootSubscribedSubscribers subscribed) (rootSubscribedWeakSelf subscribed) cleanupRootSubscribed
14391438
1439+newtype EventSelectorInt x a = EventSelectorInt { selectInt :: Int -> Event x a }
1440+
1441+data FanInt x a = FanInt
1442+ { _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
1443+ , _fanInt_subscriptionRef :: {-# UNPACK #-} !(IORef (EventSubscription x)) -- This should have a valid subscription iff subscribers is non-empty
1444+ , _fanInt_occRef :: {-# UNPACK #-} !(IORef (IntMap a))
1445+ }
1446+
1447+newFanInt :: IO (FanInt x a)
1448+newFanInt = do
1449+ 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
1450+ subscriptionRef <- newIORef $ error "fanInt: no subscription"
1451+ occRef <- newIORef $ error "fanInt: no occurrence"
1452+ return $ FanInt
1453+ { _fanInt_subscribers = subscribers
1454+ , _fanInt_subscriptionRef = subscriptionRef
1455+ , _fanInt_occRef = occRef
1456+ }
1457+
1458+{-# NOINLINE unsafeNewFanInt #-}
1459+unsafeNewFanInt :: b -> FanInt x a
1460+unsafeNewFanInt _ = unsafePerformIO newFanInt
1461+
1462+{-# NOINLINE fanInt #-}
1463+fanInt :: HasSpiderTimeline x => Event x (IntMap a) -> EventSelectorInt x a
1464+fanInt p =
1465+ let self = unsafeNewFanInt p
1466+ in EventSelectorInt $ \k -> Event $ \sub -> do
1467+ isEmpty <- liftIO $ FastMutableIntMap.isEmpty (_fanInt_subscribers self)
1468+ when isEmpty $ do -- This is the first subscriber, so we need to subscribe to our input
1469+ (subscription, parentOcc) <- subscribeAndRead p $ Subscriber
1470+ { subscriberPropagate = \m -> do
1471+ liftIO $ writeIORef (_fanInt_occRef self) m
1472+ scheduleIntClear $ _fanInt_occRef self
1473+ FastMutableIntMap.forIntersectionWithImmutable_ (_fanInt_subscribers self) m $ \b v -> do
1474+ FastWeakBag.traverse b $ \s -> do
1475+ subscriberPropagate s v
1476+ , subscriberInvalidateHeight = \old -> do
1477+ FastMutableIntMap.for_ (_fanInt_subscribers self) $ \b -> do
1478+ FastWeakBag.traverse b $ \s -> do
1479+ subscriberInvalidateHeight s old
1480+ , subscriberRecalculateHeight = \new -> do
1481+ FastMutableIntMap.for_ (_fanInt_subscribers self) $ \b -> do
1482+ FastWeakBag.traverse b $ \s -> do
1483+ subscriberRecalculateHeight s new
1484+ }
1485+ liftIO $ do
1486+ writeIORef (_fanInt_subscriptionRef self) subscription
1487+ writeIORef (_fanInt_occRef self) $ fromMaybe IntMap.empty parentOcc
1488+ scheduleIntClear $ _fanInt_occRef self
1489+ liftIO $ do
1490+ b <- FastMutableIntMap.lookup (_fanInt_subscribers self) k >>= \case
1491+ Nothing -> do
1492+ b <- FastWeakBag.empty
1493+ FastMutableIntMap.insert (_fanInt_subscribers self) k b
1494+ return b
1495+ Just b -> return b
1496+ t <- liftIO $ FastWeakBag.insert sub b
1497+ currentOcc <- readIORef (_fanInt_occRef self)
1498+ (EventSubscription _ (EventSubscribed !heightRef _)) <- readIORef (_fanInt_subscriptionRef self)
1499+ return (EventSubscription (FastWeakBag.remove t) $! EventSubscribed heightRef $! toAny $! _fanInt_subscriptionRef self, IntMap.lookup k currentOcc)
1500+
14401501{-# INLINABLE getFanSubscribed #-}
14411502getFanSubscribed :: (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 ~ EventM x, GCompare k) => k a -> Fan x k -> Subscriber x a -> m (WeakBagTicket, FanSubscribed x k, Maybe a)
14421503getFanSubscribed k f sub = do
@@ -1616,7 +1677,7 @@ mergeWithMove d = cacheEvent (mergeCheapWithMove d)
16161677
16171678{-# INLINE [1] mergeCheap #-}
16181679mergeCheap :: forall k x. (HasSpiderTimeline x, GCompare k) => Dynamic x (PatchDMap k (Event x)) -> Event x (DMap k Identity)
1619−mergeCheap = mergeCheap' getInitialSubscribers updateMe destroy DMap.size
1680+mergeCheap = mergeCheap' getInitialSubscribers updateMe destroy
16201681 where
16211682 updateMe :: MergeUpdateFunc k x (PatchDMap k (Event x)) (MergeSubscribedParent x)
16221683 updateMe subscriber heightBagRef oldParents (PatchDMap p) = do
@@ -1651,7 +1712,7 @@ mergeCheap = mergeCheap' getInitialSubscribers updateMe destroy DMap.size
16511712
16521713{-# INLINE [1] mergeCheapWithMove #-}
16531714mergeCheapWithMove :: forall k x. (HasSpiderTimeline x, GCompare k) => Dynamic x (PatchDMapWithMove k (Event x)) -> Event x (DMap k Identity)
1654−mergeCheapWithMove = mergeCheap' getInitialSubscribers updateMe destroy DMap.size
1715+mergeCheapWithMove = mergeCheap' getInitialSubscribers updateMe destroy
16551716 where
16561717 updateMe :: MergeUpdateFunc k x (PatchDMapWithMove k (Event x)) (MergeSubscribedParentWithMove x k)
16571718 updateMe subscriber heightBagRef oldParents p = do
@@ -1709,78 +1770,108 @@ type MergeDestroyFunc k s
17091770 = DMap k s
17101771 -> IO ()
17111772
1773+data Merge x k s = Merge
1774+ { _merge_parentsRef :: {-# UNPACK #-} !(IORef (DMap k s))
1775+ , _merge_heightBagRef :: {-# UNPACK #-} !(IORef HeightBag)
1776+ , _merge_heightRef :: {-# UNPACK #-} !(IORef Height)
1777+ , _merge_sub :: {-# UNPACK #-} !(Subscriber x (DMap k Identity))
1778+ , _merge_accumRef :: {-# UNPACK #-} !(IORef (DMap k Identity))
1779+ }
1780+
1781+invalidateMergeHeight :: Merge x k s -> IO ()
1782+invalidateMergeHeight m = invalidateMergeHeight' (_merge_heightRef m) (_merge_sub m)
1783+
1784+invalidateMergeHeight' :: IORef Height -> Subscriber x a -> IO ()
1785+invalidateMergeHeight' heightRef sub = do
1786+ oldHeight <- readIORef heightRef
1787+ 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
1788+ writeIORef heightRef $! invalidHeight
1789+ subscriberInvalidateHeight sub oldHeight
1790+
1791+
1792+revalidateMergeHeight :: Merge x k s -> IO ()
1793+revalidateMergeHeight m = do
1794+ currentHeight <- readIORef $ _merge_heightRef m
1795+ when (currentHeight == invalidHeight) $ do -- revalidateMergeHeight may be called multiple times; perhaps the's a way to finesse it to avoid this check
1796+ heights <- readIORef $ _merge_heightBagRef m
1797+ parents <- readIORef $ _merge_parentsRef m
1798+ -- When the number of heights in the bag reaches the number of parents, we should have a valid height
1799+ case heightBagSize heights `compare` DMap.size parents of
1800+ LT -> return ()
1801+ EQ -> do
1802+ let height = succHeight $ heightBagMax heights
1803+ when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: height: " <> show height
1804+ writeIORef (_merge_heightRef m) $! height
1805+ subscriberRecalculateHeight (_merge_sub m) height
1806+ GT -> error $ "revalidateMergeHeight: more heights (" <> show (heightBagSize heights) <> ") than parents (" <> show (DMap.size parents) <> ") for Merge"
1807+
1808+scheduleMergeSelf :: HasSpiderTimeline x => Merge x k s -> Height -> EventM x ()
1809+scheduleMergeSelf m height = scheduleMerge' height (_merge_heightRef m) $ do
1810+ vals <- liftIO $ readIORef $ _merge_accumRef m
1811+ liftIO $ writeIORef (_merge_accumRef m) $! DMap.empty -- Once we're done with this, we can clear it immediately, because if there's a cacheEvent in front of us, it'll handle subsequent subscribers, and if not, we won't get subsequent subscribers
1812+ --TODO: Assert that m is not empty
1813+ subscriberPropagate (_merge_sub m) vals
1814+
1815+mergeSubscriber :: forall x k s a. (HasSpiderTimeline x, GCompare k) => Merge x k s -> EventM x (k a) -> Subscriber x a
1816+mergeSubscriber m getKey = Subscriber
1817+ { subscriberPropagate = \a -> do
1818+ oldM <- liftIO $ readIORef $ _merge_accumRef m
1819+ k <- getKey
1820+ let newM = DMap.insertWith (error $ "Same key fired multiple times for Merge") k (Identity a) oldM
1821+ tracePropagate (Proxy :: Proxy x) $ " DMap.size oldM = " <> show (DMap.size oldM) <> "; DMap.size newM = " <> show (DMap.size newM)
1822+ liftIO $ writeIORef (_merge_accumRef m) $! newM
1823+ when (DMap.null oldM) $ do -- Only schedule the firing once
1824+ height <- liftIO $ readIORef $ _merge_heightRef m
1825+ --TODO: assertions about height
1826+ currentHeight <- getCurrentHeight
1827+ when (height <= currentHeight) $ do
1828+ if height /= invalidHeight
1829+ then do
1830+ myStack <- liftIO $ whoCreatedIORef undefined --TODO
1831+ error $ "Height (" ++ show height ++ ") is not greater than current height (" ++ show currentHeight ++ ")\n" ++ unlines (reverse myStack)
1832+ else liftIO $ do
1833+#ifdef DEBUG_CYCLES
1834+ nodesInvolvedInCycle <- walkInvalidHeightParents $ eventSubscribedMerge subscribed
1835+ stacks <- forM nodesInvolvedInCycle $ \(Some.This es) -> whoCreatedEventSubscribed es
1836+ let cycleInfo = ":\n" <> drawForest (listsToForest stacks)
1837+#else
1838+ let cycleInfo = ""
1839+#endif
1840+ error $ "Causality loop found" <> cycleInfo
1841+ scheduleMergeSelf m height
1842+ , subscriberInvalidateHeight = \old -> do --TODO: When removing a parent doesn't actually change the height, maybe we can avoid invalidating
1843+ modifyIORef' (_merge_heightBagRef m) $ heightBagRemove old
1844+ invalidateMergeHeight m
1845+ , subscriberRecalculateHeight = \new -> do
1846+ modifyIORef' (_merge_heightBagRef m) $ heightBagAdd new
1847+ revalidateMergeHeight m
1848+ }
1849+
1850+--TODO: Be able to run as much of this as possible promptly
1851+updateMerge :: (HasSpiderTimeline x, GCompare k) => Merge x k s -> MergeUpdateFunc k x p s -> p -> SomeMergeUpdate x
1852+updateMerge m updateFunc p = SomeMergeUpdate updateMe (invalidateMergeHeight m) (revalidateMergeHeight m)
1853+ where updateMe = do
1854+ oldParents <- liftIO $ readIORef $ _merge_parentsRef m
1855+ (subscriptionsToKill, newParents) <- updateFunc (mergeSubscriber m) (_merge_heightBagRef m) oldParents p
1856+ liftIO $ writeIORef (_merge_parentsRef m) $! newParents
1857+ return subscriptionsToKill
1858+
17121859{-# INLINE mergeCheap' #-}
1713−mergeCheap' :: forall k x p s. (HasSpiderTimeline x, GCompare k, PatchTarget p ~ DMap k (Event x)) => MergeInitFunc k x s -> MergeUpdateFunc k x p s -> MergeDestroyFunc k s -> (DMap k s -> Int) -> Dynamic x p -> Event x (DMap k Identity)
1714−mergeCheap' getInitialSubscribers updateFunc destroy numParents d = Event $ \sub -> do
1860+mergeCheap' :: forall k x p s. (HasSpiderTimeline x, GCompare k, PatchTarget p ~ DMap k (Event x)) => MergeInitFunc k x s -> MergeUpdateFunc k x p s -> MergeDestroyFunc k s -> Dynamic x p -> Event x (DMap k Identity)
1861+mergeCheap' getInitialSubscribers updateFunc destroy d = Event $ \sub -> do
17151862 initialParents <- readBehaviorUntracked $ dynamicCurrent d
17161863 accumRef <- liftIO $ newIORef $ error "merge: accumRef not yet initialized"
17171864 heightRef <- liftIO $ newIORef $ error "merge: heightRef not yet initialized"
17181865 heightBagRef <- liftIO $ newIORef $ error "merge: heightBagRef not yet initialized"
17191866 parentsRef :: IORef (DMap k s) <- liftIO $ newIORef $ error "merge: parentsRef not yet initialized"
1720− let scheduleSelf height = scheduleMerge' height heightRef $ do
1721− m <- liftIO $ readIORef accumRef
1722− liftIO $ writeIORef accumRef $! DMap.empty
1723− --TODO: Assert that m is not empty
1724− subscriberPropagate sub m
1725− invalidateMyHeight = do
1726− oldHeight <- readIORef heightRef
1727− 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
1728− writeIORef heightRef $! invalidHeight
1729− subscriberInvalidateHeight sub oldHeight
1730− revalidateMyHeight = do
1731− currentHeight <- readIORef heightRef
1732− when (currentHeight == invalidHeight) $ do -- revalidateMergeHeight may be called multiple times; perhaps the's a way to finesse it to avoid this check
1733− heights <- readIORef heightBagRef
1734− parents <- readIORef parentsRef
1735− -- When the number of heights in the bag reaches the number of parents, we should have a valid height
1736− case heightBagSize heights `compare` numParents parents of
1737− LT -> return ()
1738− EQ -> do
1739− let height = succHeight $ heightBagMax heights
1740− when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: height: " <> show height
1741− writeIORef heightRef $! height
1742− subscriberRecalculateHeight sub height
1743− GT -> error $ "revalidateMergeHeight: more heights (" <> show (heightBagSize heights) <> ") than parents (" <> show (numParents parents) <> ") for Merge"
1744− subscriber :: forall a. EventM x (k a) -> Subscriber x a
1745− subscriber getKey = Subscriber
1746− { subscriberPropagate = \a -> do
1747− oldM <- liftIO $ readIORef accumRef
1748− k <- getKey
1749− let newM = DMap.insertWith (error $ "Same key fired multiple times for Merge") k (Identity a) oldM
1750− tracePropagate (Proxy :: Proxy x) $ " DMap.size oldM = " <> show (DMap.size oldM) <> "; DMap.size newM = " <> show (DMap.size newM)
1751− liftIO $ writeIORef accumRef $! newM
1752− when (DMap.null oldM) $ do -- Only schedule the firing once
1753− height <- liftIO $ readIORef heightRef
1754− --TODO: assertions about height
1755− currentHeight <- getCurrentHeight
1756− when (height <= currentHeight) $ do
1757− if height /= invalidHeight
1758− then do
1759− myStack <- liftIO $ whoCreatedIORef undefined --TODO
1760− error $ "Height (" ++ show height ++ ") is not greater than current height (" ++ show currentHeight ++ ")\n" ++ unlines (reverse myStack)
1761− else liftIO $ do
1762−#ifdef DEBUG_CYCLES
1763− nodesInvolvedInCycle <- walkInvalidHeightParents $ eventSubscribedMerge subscribed
1764− stacks <- forM nodesInvolvedInCycle $ \(Some.This es) -> whoCreatedEventSubscribed es
1765− let cycleInfo = ":\n" <> drawForest (listsToForest stacks)
1766−#else
1767− let cycleInfo = ""
1768−#endif
1769− error $ "Causality loop found" <> cycleInfo
1770− scheduleSelf height
1771− , subscriberInvalidateHeight = \old -> do --TODO: When removing a parent doesn't actually change the height, maybe we can avoid invalidating
1772− modifyIORef' heightBagRef $ heightBagRemove old
1773− invalidateMyHeight
1774− , subscriberRecalculateHeight = \new -> do
1775− modifyIORef' heightBagRef $ heightBagAdd new
1776− revalidateMyHeight
1867+ let m = Merge
1868+ { _merge_parentsRef = parentsRef
1869+ , _merge_heightBagRef = heightBagRef
1870+ , _merge_heightRef = heightRef
1871+ , _merge_sub = sub
1872+ , _merge_accumRef = accumRef
17771873 }
1778− let updateMe x = do
1779− oldParents <- liftIO $ readIORef parentsRef
1780− (subscriptionsToKill, newParents) <- updateFunc subscriber heightBagRef oldParents x
1781− liftIO $ writeIORef parentsRef $! newParents
1782− return subscriptionsToKill
1783− (dm, heights, initialParentState) <- getInitialSubscribers initialParents subscriber
1874+ (dm, heights, initialParentState) <- getInitialSubscribers initialParents $ mergeSubscriber m
17841875 let myHeightBag = heightBagFromList $ filter (/= invalidHeight) heights
17851876 myHeight = if invalidHeight `elem` heights
17861877 then invalidHeight
@@ -1790,9 +1881,7 @@ mergeCheap' getInitialSubscribers updateFunc destroy numParents d = Event $ \sub
17901881 then (if DMap.null dm then Nothing else Just dm, DMap.empty)
17911882 else (Nothing, dm)
17921883 unless (DMap.null accum) $ do
1793− scheduleSelf myHeight
1794− occRef <- liftIO $ newIORef occ
1795− when (isJust occ) $ scheduleClear occRef
1884+ scheduleMergeSelf m myHeight
17961885 liftIO $ writeIORef accumRef $! accum
17971886 liftIO $ writeIORef heightRef $! myHeight
17981887 liftIO $ writeIORef heightBagRef $! myHeightBag
@@ -1802,12 +1891,12 @@ mergeCheap' getInitialSubscribers updateFunc destroy numParents d = Event $ \sub
18021891 let s = Subscriber
18031892 { subscriberPropagate = \a -> {-# SCC "traverseMergeChange" #-} do
18041893 tracePropagate (Proxy :: Proxy x) $ "SubscriberMerge/Change"
1805− defer $ SomeMergeUpdate (updateMe a) invalidateMyHeight revalidateMyHeight
1894+ defer $ updateMerge m updateFunc a
18061895 , subscriberInvalidateHeight = \_ -> return ()
18071896 , subscriberRecalculateHeight = \_ -> return ()
18081897 }
18091898 (changeSubscription, change) <- subscribeAndRead (dynamicUpdated d) s
1810− forM_ change $ \c -> defer $ SomeMergeUpdate (updateMe c) invalidateMyHeight revalidateMyHeight
1899+ forM_ change $ \c -> defer $ updateMerge m updateFunc c
18111900 -- We explicitly hold on to the unsubscribe function from subscribing to the update event.
18121901 -- If we don't do this, there are certain cases where mergeCheap will fail to properly retain
18131902 -- its subscription.
@@ -1817,6 +1906,106 @@ mergeCheap' getInitialSubscribers updateFunc destroy numParents d = Event $ \sub
18171906 , occ
18181907 )
18191908
1909+mergeInt :: forall x a. (HasSpiderTimeline x) => Dynamic x (PatchIntMap (Event x a)) -> Event x (IntMap a)
1910+mergeInt = cacheEvent . mergeIntCheap
1911+
1912+{-# INLINABLE mergeIntCheap #-}
1913+mergeIntCheap :: forall x a. (HasSpiderTimeline x) => Dynamic x (PatchIntMap (Event x a)) -> Event x (IntMap a)
1914+mergeIntCheap d = Event $ \sub -> do
1915+ initialParents <- readBehaviorUntracked $ dynamicCurrent d
1916+ accum <- liftIO $ FastMutableIntMap.newEmpty
1917+ heightRef <- liftIO $ newIORef zeroHeight
1918+ heightBagRef <- liftIO $ newIORef heightBagEmpty
1919+ parents <- liftIO $ FastMutableIntMap.newEmpty
1920+ let scheduleSelf = do
1921+ height <- liftIO $ readIORef $ heightRef
1922+ scheduleMerge' height heightRef $ do
1923+ vals <- liftIO $ FastMutableIntMap.getFrozenAndClear accum
1924+ subscriberPropagate sub vals
1925+ invalidateMyHeight = do
1926+ invalidateMergeHeight' heightRef sub
1927+ recalculateMyHeight = do
1928+ currentHeight <- readIORef heightRef
1929+ 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?
1930+ heights <- readIORef heightBagRef
1931+ numParents <- FastMutableIntMap.size parents
1932+ case heightBagSize heights `compare` numParents of
1933+ LT -> return ()
1934+ EQ -> do
1935+ let height = succHeight $ heightBagMax heights
1936+ when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: height: " <> show height
1937+ writeIORef heightRef $! height
1938+ subscriberRecalculateHeight sub height
1939+ GT -> error $ "revalidateMergeHeight: more heights (" <> show (heightBagSize heights) <> ") than parents (" <> show numParents <> ") for Merge"
1940+ mySubscriber k = Subscriber
1941+ { subscriberPropagate = \a -> do
1942+ wasEmpty <- liftIO $ FastMutableIntMap.isEmpty accum
1943+ liftIO $ FastMutableIntMap.insert accum k a
1944+ when wasEmpty scheduleSelf
1945+ , subscriberInvalidateHeight = \old -> do
1946+ modifyIORef' heightBagRef $ heightBagRemove old
1947+ invalidateMyHeight
1948+ , subscriberRecalculateHeight = \new -> do
1949+ modifyIORef' heightBagRef $ heightBagAdd new
1950+ recalculateMyHeight
1951+ }
1952+ forM_ (IntMap.toList initialParents) $ \(k, p) -> do
1953+ (subscription@(EventSubscription _ parentSubd), parentOcc) <- subscribeAndRead p $ mySubscriber k
1954+ liftIO $ do
1955+ forM_ parentOcc $ FastMutableIntMap.insert accum k
1956+ FastMutableIntMap.insert parents k subscription
1957+ height <- getEventSubscribedHeight parentSubd
1958+ if height == invalidHeight
1959+ then writeIORef heightRef invalidHeight
1960+ else do
1961+ modifyIORef' heightBagRef $ heightBagAdd height
1962+ modifyIORef' heightRef $ \oldHeight ->
1963+ if oldHeight == invalidHeight
1964+ then invalidHeight
1965+ else max (succHeight height) oldHeight
1966+ myHeight <- liftIO $ readIORef heightRef
1967+ currentHeight <- getCurrentHeight
1968+ isEmpty <- liftIO $ FastMutableIntMap.isEmpty accum
1969+ occ <- if currentHeight >= myHeight -- If we should have fired by now
1970+ then if isEmpty
1971+ then return Nothing
1972+ else liftIO $ Just <$> FastMutableIntMap.getFrozenAndClear accum
1973+ else do when (not isEmpty) scheduleSelf -- We have things accumulated, but we shouldn't have fired them yet
1974+ return Nothing
1975+ changeSubdRef <- liftIO $ newIORef $ error "getMergeSubscribed: changeSubdRef not yet initialized"
1976+ defer $ SomeMergeInit $ do
1977+ let updateMe a = SomeMergeUpdate u invalidateMyHeight recalculateMyHeight
1978+ where
1979+ u = do
1980+ let f k newParent = do
1981+ subscription@(EventSubscription _ subd) <- subscribe newParent $ mySubscriber k
1982+ newParentHeight <- liftIO $ getEventSubscribedHeight subd
1983+ liftIO $ modifyIORef' heightBagRef $ heightBagAdd newParentHeight
1984+ return subscription
1985+ newSubscriptions <- FastMutableIntMap.traverseIntMapPatchWithKey f a
1986+ oldParents <- liftIO $ FastMutableIntMap.applyPatch parents newSubscriptions
1987+ liftIO $ for_ oldParents $ \oldParent -> do
1988+ oldParentHeight <- getEventSubscribedHeight $ _eventSubscription_subscribed oldParent
1989+ modifyIORef' heightBagRef $ heightBagRemove oldParentHeight
1990+ return $ IntMap.elems oldParents
1991+ let s = Subscriber
1992+ { subscriberPropagate = \a -> {-# SCC "traverseMergeChange" #-} do
1993+ tracePropagate (Proxy :: Proxy x) $ "SubscriberMergeInt/Change"
1994+ defer $ updateMe a
1995+ , subscriberInvalidateHeight = \_ -> return ()
1996+ , subscriberRecalculateHeight = \_ -> return ()
1997+ }
1998+ (changeSubscription, change) <- subscribeAndRead (dynamicUpdated d) s
1999+ forM_ change $ \c -> defer $ updateMe c
2000+ -- We explicitly hold on to the unsubscribe function from subscribing to the update event.
2001+ -- If we don't do this, there are certain cases where mergeCheap will fail to properly retain
2002+ -- its subscription.
2003+ liftIO $ writeIORef changeSubdRef (s, changeSubscription)
2004+ let unsubscribeAll = traverse_ unsubscribe =<< FastMutableIntMap.getFrozenAndClear parents
2005+ return ( EventSubscription unsubscribeAll $ EventSubscribed heightRef $ toAny (parents, changeSubdRef)
2006+ , occ
2007+ )
2008+
18202009newtype EventSelector x k = EventSelector { select :: forall a. k a -> Event x a }
18212010
18222011fan :: (HasSpiderTimeline x, GCompare k) => Event x (DMap k Identity) -> EventSelector x k
@@ -1856,13 +2045,14 @@ newEventEnv = do
18562045 mergeInitRef <- newIORef []
18572046 heightRef <- newIORef zeroHeight
18582047 toClearRef <- newIORef []
2048+ toClearIntRef <- newIORef []
18592049 toClearRootRef <- newIORef []
18602050 coincidenceInfosRef <- newIORef []
18612051 delayedRef <- newIORef IntMap.empty
1862− return $ EventEnv toAssignRef holdInitRef dynInitRef mergeUpdateRef mergeInitRef toClearRef toClearRootRef heightRef coincidenceInfosRef delayedRef
2052+ return $ EventEnv toAssignRef holdInitRef dynInitRef mergeUpdateRef mergeInitRef toClearRef toClearIntRef toClearRootRef heightRef coincidenceInfosRef delayedRef
18632053
18642054clearEventEnv :: EventEnv x -> IO ()
1865−clearEventEnv (EventEnv toAssignRef holdInitRef dynInitRef mergeUpdateRef mergeInitRef toClearRef toClearRootRef heightRef coincidenceInfosRef delayedRef) = do
2055+clearEventEnv (EventEnv toAssignRef holdInitRef dynInitRef mergeUpdateRef mergeInitRef toClearRef toClearIntRef toClearRootRef heightRef coincidenceInfosRef delayedRef) = do
18662056 writeIORef toAssignRef []
18672057 writeIORef holdInitRef []
18682058 writeIORef dynInitRef []
@@ -1870,6 +2060,7 @@ clearEventEnv (EventEnv toAssignRef holdInitRef dynInitRef mergeUpdateRef mergeI
18702060 writeIORef mergeInitRef []
18712061 writeIORef heightRef zeroHeight
18722062 writeIORef toClearRef []
2063+ writeIORef toClearIntRef []
18732064 writeIORef toClearRootRef []
18742065 writeIORef coincidenceInfosRef []
18752066 writeIORef delayedRef IntMap.empty
@@ -1885,6 +2076,8 @@ runFrame a = SpiderHost $ ask >>= \_ -> lift $ do
18852076 result <- runEventM go
18862077 toClear <- readIORef $ eventEnvClears env
18872078 forM_ toClear $ \(SomeClear ref) -> {-# SCC "clear" #-} writeIORef ref Nothing
2079+ toClearInt <- readIORef $ eventEnvIntClears env
2080+ forM_ toClearInt $ \(SomeIntClear ref) -> {-# SCC "intClear" #-} writeIORef ref $! IntMap.empty
18882081 toClearRoot <- readIORef $ eventEnvRootClears env
18892082 forM_ toClearRoot $ \(SomeRootClear ref) -> {-# SCC "rootClear" #-} writeIORef ref $! DMap.empty
18902083 toAssign <- readIORef $ eventEnvAssignments env
@@ -2079,6 +2272,8 @@ withSpiderTimeline k = do
20792272
20802273newtype SpiderPullM x a = SpiderPullM (BehaviorM x a) deriving (Functor, Applicative, Monad, MonadIO, MonadFix)
20812274
2275+type ComputeM = EventM
2276+
20822277newtype SpiderPushM x a = SpiderPushM (ComputeM x a) deriving (Functor, Applicative, Monad, MonadIO, MonadFix)
20832278
20842279#ifndef SPECIALIZE_TO_SPIDERTIMELINE_GLOBAL
@@ -2131,6 +2326,8 @@ instance HasSpiderTimeline x => R.Reflex (SpiderTimeline x) where
21312326 eventCoercion Coercion = Coercion
21322327 behaviorCoercion Coercion = Coercion
21332328 dynamicCoercion = unsafeCoerce --TODO: How can we avoid this unsafeCoerce? This is safe only because we know how Identity works as a Patch instance
2329+ mergeIntIncremental = SpiderEvent . mergeInt . (unsafeCoerce :: Dynamic x (PatchIntMap (R.Event (SpiderTimeline x) a)) -> Dynamic x (PatchIntMap (Event x a))) . unSpiderIncremental
2330+ fanInt e = R.EventSelectorInt $ SpiderEvent . selectInt (fanInt (unSpiderEvent e))
21342331#endif
21352332
21362333data RootTrigger x a = forall k. GCompare k => RootTrigger (WeakBag (Subscriber x a), IORef (DMap k Identity), k a)