@@ -8,9 +8,7 @@
8{-# LANGUAGE FunctionalDependencies #-}
8{-# LANGUAGE FunctionalDependencies #-}
9{-# LANGUAGE GADTs #-}
9{-# LANGUAGE GADTs #-}
10{-# LANGUAGE GeneralizedNewtypeDeriving #-}
10{-# LANGUAGE GeneralizedNewtypeDeriving #-}
11−#if defined(DEBUG) || defined(DEBUG_CYCLES)
12{-# LANGUAGE LambdaCase #-}
11{-# LANGUAGE LambdaCase #-}
14{-# LANGUAGE MagicHash #-}
12{-# LANGUAGE MagicHash #-}
15{-# LANGUAGE MultiParamTypeClasses #-}
13{-# LANGUAGE MultiParamTypeClasses #-}
16{-# LANGUAGE MultiWayIf #-}
14{-# LANGUAGE MultiWayIf #-}
@@ -59,9 +57,12 @@ import Data.Proxy
59import Data.These
57import Data.These
60import Data.Traversable
58import Data.Traversable
61import Data.Word
59import Data.Word
60+import Data.FastMutableIntMap (FastMutableIntMap, PatchIntMap (..))
61+import qualified Data.FastMutableIntMap as FastMutableIntMap
62import GHC.Exts
62import GHC.Exts
63import GHC.IORef (IORef (..))
63import GHC.IORef (IORef (..))
64import GHC.Stack
64import GHC.Stack
65+import Reflex.FastWeak
65import Reflex.FunctorMaybe
66import Reflex.FunctorMaybe
66import System.IO.Unsafe
67import System.IO.Unsafe
67import System.Mem.Weak
68import System.Mem.Weak
@@ -75,6 +76,8 @@ import Data.Monoid (mempty)
75import Data.Tree (Forest, Tree (..), drawForest)
76import Data.Tree (Forest, Tree (..), drawForest)
79+import Data.FastWeakBag (FastWeakBag)
80+import qualified Data.FastWeakBag as FastWeakBag
78import Data.WeakBag (WeakBag, WeakBagTicket, _weakBag_children)
81import Data.WeakBag (WeakBag, WeakBagTicket, _weakBag_children)
79import qualified Data.WeakBag as WeakBag
82import qualified Data.WeakBag as WeakBag
80#ifndef SPECIALIZE_TO_SPIDERTIMELINE_GLOBAL
83#ifndef SPECIALIZE_TO_SPIDERTIMELINE_GLOBAL
@@ -240,22 +243,57 @@ subscribeAndRead = unEvent
240-- | Construct an 'Event' equivalent to that constructed by 'push', but with no
243-- | Construct an 'Event' equivalent to that constructed by 'push', but with no
241-- caching; if the computation function is very cheap, this is (much) more
244-- caching; if the computation function is very cheap, this is (much) more
242-- efficient than 'push'
245-- efficient than 'push'
243−{-# NOINLINE pushCheap #-}
246+{-# INLINE [1] pushCheap #-}
244pushCheap :: HasSpiderTimeline x => (a -> ComputeM x (Maybe b)) -> Event x a -> Event x b
247pushCheap :: HasSpiderTimeline x => (a -> ComputeM x (Maybe b)) -> Event x a -> Event x b
245−pushCheap f e = Event $ \sub -> do
248+pushCheap !f e = Event $ \sub -> do
246 (subscription, occ) <- subscribeAndRead e $ sub
249 (subscription, occ) <- subscribeAndRead e $ sub
247 { subscriberPropagate = \a -> do
250 { subscriberPropagate = \a -> do
248− mb <- runComputeM $ f a
251+ mb <- f a
249 mapM_ (subscriberPropagate sub) mb
252 mapM_ (subscriberPropagate sub) mb
251− occ' <- join <$> mapM (runComputeM . f) occ
254+ occ' <- join <$> mapM f occ
252 return (subscription, occ')
255 return (subscription, occ')
257+-- | A subscriber that never triggers other 'Event's
258+{-# INLINE terminalSubscriber #-}
259+terminalSubscriber :: (a -> EventM x ()) -> Subscriber x a
260+terminalSubscriber p = Subscriber
261+ { subscriberPropagate = p
262+ , subscriberInvalidateHeight = \_ -> return ()
263+ , subscriberRecalculateHeight = \_ -> return ()
266+-- | Subscribe to an Event only for the duration of one occurrence
267+{-# INLINE subscribeAndReadHead #-}
268+subscribeAndReadHead :: Event x a -> Subscriber x a -> EventM x (EventSubscription x, Maybe a)
269+subscribeAndReadHead e sub = do
270+ subscriptionRef <- liftIO $ newIORef $ error "subscribeAndReadHead: not initialized"
271+ (subscription, occ) <- subscribeAndRead e $ sub
272+ { subscriberPropagate = \a -> do
273+ liftIO $ unsubscribe =<< readIORef subscriptionRef
274+ subscriberPropagate sub a
276+ liftIO $ case occ of
277+ Nothing -> writeIORef subscriptionRef $! subscription
278+ Just _ -> unsubscribe subscription
279+ return (subscription, occ)
281+--TODO: Make this lazy in its input event
282+headE :: (MonadIO m, Defer (SomeMergeInit x) m) => Event x a -> m (Event x a)
284+ parent <- liftIO $ newIORef $ Just originalE
285+ defer $ SomeMergeInit $ do --TODO: Rename SomeMergeInit appropriately
286+ let clearParent = liftIO $ writeIORef parent Nothing
287+ (_, occ) <- subscribeAndReadHead originalE $ terminalSubscriber $ \_ -> clearParent
288+ when (isJust occ) clearParent
289+ return $ Event $ \sub -> do
290+ liftIO (readIORef parent) >>= \case
291+ Nothing -> return (EventSubscription (return ()) $ EventSubscribed zeroRef $ toAny (), Nothing)
292+ Just e -> subscribeAndReadHead e sub
254data CacheSubscribed x a
294data CacheSubscribed x a
255− = CacheSubscribed { _cacheSubscribed_subscribers :: {-# UNPACK #-} !(WeakBag (Subscriber x a))
295+ = CacheSubscribed { _cacheSubscribed_subscribers :: {-# UNPACK #-} !(FastWeakBag (Subscriber x a))
256− , _cacheSubscribed_weakSelf :: {-# UNPACK #-} !(IORef (Weak (CacheSubscribed x a)))
257 , _cacheSubscribed_parent :: {-# UNPACK #-} !(EventSubscription x)
296 , _cacheSubscribed_parent :: {-# UNPACK #-} !(EventSubscription x)
258− , _cacheSubscribed_cached :: {-# UNPACK #-} !(IORef (Maybe (CacheSubscribed x a)))
259 , _cacheSubscribed_occurrence :: {-# UNPACK #-} !(IORef (Maybe a))
297 , _cacheSubscribed_occurrence :: {-# UNPACK #-} !(IORef (Maybe a))
@@ -263,62 +301,67 @@ data CacheSubscribed x a
263-- repeatedly
301-- repeatedly
265--TODO: Try a caching strategy where we subscribe directly to the parent when
303--TODO: Try a caching strategy where we subscribe directly to the parent when
266−--there's only one subscriber, and then build our own WeakBag only when a second
304+--there's only one subscriber, and then build our own FastWeakBag only when a second
267--subscriber joins
305--subscriber joins
268−{-# NOINLINE cacheEvent #-}
306+{-# NOINLINE [0] cacheEvent #-}
269−cacheEvent :: HasSpiderTimeline x => Event x a -> Event x a
307+cacheEvent :: forall x a. HasSpiderTimeline x => Event x a -> Event x a
270cacheEvent e =
308cacheEvent e =
271#ifdef DEBUG_TRACE_EVENTS
309#ifdef DEBUG_TRACE_EVENTS
272 withStackOneLine $ \callSite -> Event $
310 withStackOneLine $ \callSite -> Event $
314+ let mSubscribedRef :: IORef (FastWeak (CacheSubscribed x a))
276− let !mSubscribedRef = unsafeNewIORef e Nothing
315+ !mSubscribedRef = unsafeNewIORef e emptyFastWeak
277 in \sub -> {-# SCC "cacheEvent" #-} do
316 in \sub -> {-# SCC "cacheEvent" #-} do
278#ifdef DEBUG_TRACE_EVENTS
317#ifdef DEBUG_TRACE_EVENTS
279 unless (BS8.null callSite) $ liftIO $ BS8.hPutStrLn stderr callSite
318 unless (BS8.null callSite) $ liftIO $ BS8.hPutStrLn stderr callSite
281− mSubscribed <- liftIO $ readIORef mSubscribedRef
320+ subscribedTicket <- liftIO (readIORef mSubscribedRef >>= getFastWeakTicket) >>= \case
282− case mSubscribed of
321+ Just subscribedTicket -> return subscribedTicket
283− Just subscribed -> liftIO $ do
284− ticket <- WeakBag.insert sub (_cacheSubscribed_subscribers subscribed) (_cacheSubscribed_weakSelf subscribed) cleanupCacheSubscribed
285− occ <- readIORef $ _cacheSubscribed_occurrence subscribed
286− let es = EventSubscribed
287− { eventSubscribedHeightRef = eventSubscribedHeightRef $ _eventSubscription_subscribed $ _cacheSubscribed_parent subscribed
288− , eventSubscribedRetained = toAny subscribed
290− return (EventSubscription (WeakBag.remove ticket >> touch ticket) es, occ)
291 Nothing -> do
322 Nothing -> do
292− weakSelf <- liftIO $ newIORef $ error "cacheEvent: weakSelf not yet intialized"
323+ subscribers <- liftIO FastWeakBag.empty
293− (subscribers, ticket) <- liftIO $ WeakBag.singleton sub weakSelf cleanupCacheSubscribed
294− occRef <- liftIO $ newIORef undefined
324+ occRef <- liftIO $ newIORef Nothing -- This should never be read prior to being set below
295 (parentSub, occ) <- subscribeAndRead e $ Subscriber
325 (parentSub, occ) <- subscribeAndRead e $ Subscriber
296 { subscriberPropagate = \a -> do
326 { subscriberPropagate = \a -> do
297 liftIO $ writeIORef occRef $ Just a
327 liftIO $ writeIORef occRef $ Just a
298 scheduleClear occRef
328 scheduleClear occRef
299− propagate a subscribers
329+ propagateFast a subscribers
300 , subscriberInvalidateHeight = \old -> do
330 , subscriberInvalidateHeight = \old -> do
301− WeakBag.traverse subscribers $ invalidateSubscriberHeight old
331+ FastWeakBag.traverse subscribers $ invalidateSubscriberHeight old
302 , subscriberRecalculateHeight = \new -> do
332 , subscriberRecalculateHeight = \new -> do
303− WeakBag.traverse subscribers $ recalculateSubscriberHeight new
333+ FastWeakBag.traverse subscribers $ recalculateSubscriberHeight new
305− liftIO $ writeIORef occRef occ
335+ when (isJust occ) $ do
336+ liftIO $ writeIORef occRef occ -- Set the initial value of occRef; we don't need to do this if occ is Nothing
306− when (isJust occ) $ scheduleClear occRef
337+ scheduleClear occRef
307 let !subscribed = CacheSubscribed
338 let !subscribed = CacheSubscribed
308 { _cacheSubscribed_subscribers = subscribers
339 { _cacheSubscribed_subscribers = subscribers
309− , _cacheSubscribed_weakSelf = weakSelf
310 , _cacheSubscribed_parent = parentSub
340 , _cacheSubscribed_parent = parentSub
311− , _cacheSubscribed_cached = mSubscribedRef
312 , _cacheSubscribed_occurrence = occRef
341 , _cacheSubscribed_occurrence = occRef
343+ subscribedTicket <- liftIO $ mkFastWeakTicket subscribed
314− liftIO $ writeIORef mSubscribedRef $ Just subscribed
344+ liftIO $ writeIORef mSubscribedRef =<< getFastWeakTicketWeak subscribedTicket
315− liftIO $ writeIORef weakSelf =<< evaluate =<< mkWeakPtrWithDebug subscribed "PushSubscribed"
345+ return subscribedTicket
347+ subscribed <- getFastWeakTicketValue subscribedTicket
348+ ticket <- FastWeakBag.insert sub $ _cacheSubscribed_subscribers subscribed
349+ occ <- readIORef $ _cacheSubscribed_occurrence subscribed
350+ let es = EventSubscription
351+ { _eventSubscription_unsubscribe = do
352+ FastWeakBag.remove ticket
353+ isEmpty <- FastWeakBag.isEmpty $ _cacheSubscribed_subscribers subscribed
355+ writeIORef mSubscribedRef emptyFastWeak
356+ unsubscribe $ _cacheSubscribed_parent subscribed
358+ touch subscribedTicket
359+ , _eventSubscription_subscribed = EventSubscribed
316− return (EventSubscription (WeakBag.remove ticket >> touch ticket) (EventSubscribed (eventSubscribedHeightRef $ _eventSubscription_subscribed parentSub) (toAny subscribed)), occ)
360+ { eventSubscribedHeightRef = eventSubscribedHeightRef $ _eventSubscription_subscribed $ _cacheSubscribed_parent subscribed
317−
361+ , eventSubscribedRetained = toAny subscribedTicket
318−cleanupCacheSubscribed :: CacheSubscribed x a -> IO ()
362+ }
319−cleanupCacheSubscribed self = do
363+ }
320− unsubscribe $ _cacheSubscribed_parent self
364+ return (es, occ)
321− writeIORef (_cacheSubscribed_cached self) Nothing
323subscribe :: Event x a -> Subscriber x a -> EventM x (EventSubscription x)
366subscribe :: Event x a -> Subscriber x a -> EventM x (EventSubscription x)
324subscribe e s = fst <$> subscribeAndRead e s
367subscribe e s = fst <$> subscribeAndRead e s
@@ -352,7 +395,7 @@ eventDyn :: (HasSpiderTimeline x, Patch p) => Dyn x p -> Event x p
352eventDyn !j = Event $ \sub -> getDynHold j >>= \h -> subscribeHoldEvent h sub
395eventDyn !j = Event $ \sub -> getDynHold j >>= \h -> subscribeHoldEvent h sub
354{-# INLINE subscribeCoincidenceInner #-}
397{-# INLINE subscribeCoincidenceInner #-}
355−subscribeCoincidenceInner :: CanSubscribe x m => Event x a -> Height -> CoincidenceSubscribed x a -> m (Maybe a, Height, EventSubscribed x)
398+subscribeCoincidenceInner :: HasSpiderTimeline x => Event x a -> Height -> CoincidenceSubscribed x a -> EventM x (Maybe a, Height, EventSubscribed x)
356subscribeCoincidenceInner inner outerHeight subscribedUnsafe = do
399subscribeCoincidenceInner inner outerHeight subscribedUnsafe = do
357 subInner <- liftIO $ newSubscriberCoincidenceInner subscribedUnsafe
400 subInner <- liftIO $ newSubscriberCoincidenceInner subscribedUnsafe
358 (subscription@(EventSubscription _ innerSubd), innerOcc) <- subscribeAndRead inner subInner
401 (subscription@(EventSubscription _ innerSubd), innerOcc) <- subscribeAndRead inner subInner
@@ -497,6 +540,13 @@ propagate a subscribers = withIncreasedDepth $ do
497 --TODO: Should we check if nodes already have their values before propagating? Maybe we're re-doing work
540 --TODO: Should we check if nodes already have their values before propagating? Maybe we're re-doing work
498 WeakBag.traverse subscribers $ \s -> subscriberPropagate s a
541 WeakBag.traverse subscribers $ \s -> subscriberPropagate s a
543+-- | Propagate everything at the current height
544+propagateFast :: a -> FastWeakBag (Subscriber x a) -> EventM x ()
545+propagateFast a subscribers = withIncreasedDepth $ do
546+ -- 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
547+ --TODO: Should we check if nodes already have their values before propagating? Maybe we're re-doing work
548+ FastWeakBag.traverse subscribers $ \s -> subscriberPropagate s a
500--------------------------------------------------------------------------------
550--------------------------------------------------------------------------------
501-- EventSubscribed
551-- EventSubscribed
502--------------------------------------------------------------------------------
552--------------------------------------------------------------------------------
@@ -595,7 +645,7 @@ walkInvalidHeightParents s0 = do
597{-# INLINE subscribeHoldEvent #-}
647{-# INLINE subscribeHoldEvent #-}
598−subscribeHoldEvent :: CanSubscribe x m => Hold x p -> Subscriber x p -> m (EventSubscription x, Maybe p)
648+subscribeHoldEvent :: HasSpiderTimeline x => Hold x p -> Subscriber x p -> EventM x (EventSubscription x, Maybe p)
599subscribeHoldEvent = subscribeAndRead . holdEvent
649subscribeHoldEvent = subscribeAndRead . holdEvent
601--------------------------------------------------------------------------------
651--------------------------------------------------------------------------------
@@ -751,6 +801,7 @@ data EventEnv x
751 , eventEnvMergeUpdates :: !(IORef [SomeMergeUpdate x])
801 , eventEnvMergeUpdates :: !(IORef [SomeMergeUpdate x])
752 , eventEnvMergeInits :: !(IORef [SomeMergeInit x]) -- Needed for Subscribe
802 , eventEnvMergeInits :: !(IORef [SomeMergeInit x]) -- Needed for Subscribe
753 , eventEnvClears :: !(IORef [SomeClear]) -- Needed for Subscribe
803 , eventEnvClears :: !(IORef [SomeClear]) -- Needed for Subscribe
804+ , eventEnvIntClears :: !(IORef [SomeIntClear])
754 , eventEnvRootClears :: !(IORef [SomeRootClear])
805 , eventEnvRootClears :: !(IORef [SomeRootClear])
755 , eventEnvCurrentHeight :: !(IORef Height) -- Needed for Subscribe
806 , eventEnvCurrentHeight :: !(IORef Height) -- Needed for Subscribe
756 , eventEnvResetCoincidences :: !(IORef [SomeResetCoincidence x]) -- Needed for Subscribe
807 , eventEnvResetCoincidences :: !(IORef [SomeResetCoincidence x]) -- Needed for Subscribe
@@ -835,6 +886,14 @@ instance HasSpiderTimeline x => Defer SomeClear (EventM x) where
835scheduleClear :: Defer SomeClear m => IORef (Maybe a) -> m ()
886scheduleClear :: Defer SomeClear m => IORef (Maybe a) -> m ()
836scheduleClear r = defer $ SomeClear r
887scheduleClear r = defer $ SomeClear r
889+instance HasSpiderTimeline x => Defer SomeIntClear (EventM x) where
890+ {-# INLINE getDeferralQueue #-}
891+ getDeferralQueue = asksEventEnv eventEnvIntClears
893+{-# INLINE scheduleIntClear #-}
894+scheduleIntClear :: Defer SomeIntClear m => IORef (IntMap a) -> m ()
895+scheduleIntClear r = defer $ SomeIntClear r
838instance HasSpiderTimeline x => Defer SomeRootClear (EventM x) where
897instance HasSpiderTimeline x => Defer SomeRootClear (EventM x) where
839 {-# INLINE getDeferralQueue #-}
898 {-# INLINE getDeferralQueue #-}
840 getDeferralQueue = asksEventEnv eventEnvRootClears
899 getDeferralQueue = asksEventEnv eventEnvRootClears
@@ -870,7 +929,7 @@ hold v0 e = do
872{-# INLINE getHoldEventSubscription #-}
931{-# INLINE getHoldEventSubscription #-}
873−getHoldEventSubscription :: forall m p x. (Defer (SomeAssignment x) m, Defer (SomeHoldInit x) m, Defer SomeClear m, Patch p, HasCurrentHeight x m, Defer (SomeMergeInit x) m, Defer (SomeResetCoincidence x) m, HasSpiderTimeline x, m ~ EventM x) => Hold x p -> m (EventSubscription x)
932+getHoldEventSubscription :: forall p x. (HasSpiderTimeline x, Patch p) => Hold x p -> EventM x (EventSubscription x)
874getHoldEventSubscription h = do
933getHoldEventSubscription h = do
875 ep <- liftIO $ readIORef $ holdParent h
934 ep <- liftIO $ readIORef $ holdParent h
876 case ep of
935 case ep of
@@ -970,21 +1029,6 @@ newtype SomeMergeInit x = SomeMergeInit { unSomeMergeInit :: EventM x () }
970-- EventM can do everything BehaviorM can, plus create holds
1029-- EventM can do everything BehaviorM can, plus create holds
971newtype 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
1030newtype 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
973−newtype ComputeM x a = ComputeM (ReaderT (IORef [SomeHoldInit x], IORef [SomeDynInit x]) IO a) deriving (Functor, Applicative, Monad, MonadIO, MonadFix)
975−instance Defer (SomeHoldInit x) (ComputeM x) where
976− getDeferralQueue = ComputeM $ asks fst
978−instance Defer (SomeDynInit x) (ComputeM x) where
979− getDeferralQueue = ComputeM $ asks snd
981−{-# INLINABLE runComputeM #-}
982−runComputeM :: (Defer (SomeHoldInit x) m, Defer (SomeDynInit x) m) => ComputeM x a -> m a
983−runComputeM (ComputeM a) = do
984− holds <- getDeferralQueue
985− dyns <- getDeferralQueue
986− liftIO $ runReaderT a (holds, dyns)
988newtype MergeSubscribedParent x a = MergeSubscribedParent { unMergeSubscribedParent :: EventSubscription x }
1032newtype MergeSubscribedParent x a = MergeSubscribedParent { unMergeSubscribedParent :: EventSubscription x }
990data MergeSubscribedParentWithMove x k a = MergeSubscribedParentWithMove
1034data MergeSubscribedParentWithMove x k a = MergeSubscribedParentWithMove
@@ -1111,7 +1155,7 @@ instance HasSpiderTimeline x => FunctorMaybe (Event x) where
1112instance HasSpiderTimeline x => Align (Event x) where
1156instance HasSpiderTimeline x => Align (Event x) where
1113 nil = eventNever
1157 nil = eventNever
1114− align ea eb = fmapMaybe dmapToThese $ merge $ dynamicConst $ DMap.fromList [LeftTag :=> ea, RightTag :=> eb]
1158+ align ea eb = fmapMaybe dmapToThese $ merge $ dynamicConst $ DMap.fromDistinctAscList [LeftTag :=> ea, RightTag :=> eb]
1116newtype Dyn x p = Dyn { unDyn :: IORef (Either (BehaviorM x (PatchTarget p), Event x p) (Hold x p)) }
1160newtype Dyn x p = Dyn { unDyn :: IORef (Either (BehaviorM x (PatchTarget p), Event x p) (Hold x p)) }
@@ -1230,9 +1274,11 @@ scheduleMerge' initialHeight heightRef a = scheduleMerge initialHeight $ do
1230 GT -> scheduleMerge' height heightRef a -- The height has been increased (by a coincidence event; TODO: is this the only way?)
1274 GT -> scheduleMerge' height heightRef a -- The height has been increased (by a coincidence event; TODO: is this the only way?)
1233−data SomeClear = forall a. SomeClear (IORef (Maybe a))
1277+data SomeClear = forall a. SomeClear {-# UNPACK #-} !(IORef (Maybe a))
1279+data SomeIntClear = forall a. SomeIntClear {-# UNPACK #-} !(IORef (IntMap a))
1235−data SomeRootClear = forall k. SomeRootClear (IORef (DMap k Identity))
1281+data SomeRootClear = forall k. SomeRootClear {-# UNPACK #-} !(IORef (DMap k Identity))
1237data SomeAssignment x = forall a. SomeAssignment {-# UNPACK #-} !(IORef a) {-# UNPACK #-} !(IORef [Weak (Invalidator x)]) a
1283data SomeAssignment x = forall a. SomeAssignment {-# UNPACK #-} !(IORef a) {-# UNPACK #-} !(IORef [Weak (Invalidator x)]) a
@@ -1363,16 +1409,6 @@ getDynHold d = do
1363zeroRef :: IORef Height
1409zeroRef :: IORef Height
1364zeroRef = unsafePerformIO $ newIORef zeroHeight
1410zeroRef = unsafePerformIO $ newIORef zeroHeight
1366−type CanSubscribe x m =
1367− ( HasSpiderTimeline x
1368− , HasCurrentHeight x m
1369− , Defer (SomeHoldInit x) m, Defer (SomeMergeInit x) m -- The things we need to lazily initialize state
1370− , Defer SomeClear m -- Wiping out event firings
1371− , Defer (SomeAssignment x) m -- Updating state
1372− , Defer (SomeResetCoincidence x) m
1376getRootSubscribed :: GCompare k => k a -> Root x k -> Subscriber x a -> IO (WeakBagTicket, RootSubscribed x a, Maybe a)
1412getRootSubscribed :: GCompare k => k a -> Root x k -> Subscriber x a -> IO (WeakBagTicket, RootSubscribed x a, Maybe a)
1377getRootSubscribed k r sub = do
1413getRootSubscribed k r sub = do
1378 mSubscribed <- readIORef $ rootSubscribed r
1414 mSubscribed <- readIORef $ rootSubscribed r
@@ -1428,8 +1464,71 @@ cleanupRootSubscribed self@RootSubscribed { rootSubscribedKey = k, rootSubscribe
1428subscribeRootSubscribed :: RootSubscribed x a -> Subscriber x a -> IO WeakBagTicket
1464subscribeRootSubscribed :: RootSubscribed x a -> Subscriber x a -> IO WeakBagTicket
1429subscribeRootSubscribed subscribed sub = WeakBag.insert sub (rootSubscribedSubscribers subscribed) (rootSubscribedWeakSelf subscribed) cleanupRootSubscribed
1465subscribeRootSubscribed subscribed sub = WeakBag.insert sub (rootSubscribedSubscribers subscribed) (rootSubscribedWeakSelf subscribed) cleanupRootSubscribed
1467+newtype EventSelectorInt x a = EventSelectorInt { selectInt :: Int -> Event x a }
1469+data FanInt x a = FanInt
1470+ { _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
1471+ , _fanInt_subscriptionRef :: {-# UNPACK #-} !(IORef (EventSubscription x)) -- This should have a valid subscription iff subscribers is non-empty
1472+ , _fanInt_occRef :: {-# UNPACK #-} !(IORef (IntMap a))
1475+newFanInt :: IO (FanInt x a)
1477+ 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
1478+ subscriptionRef <- newIORef $ error "fanInt: no subscription"
1479+ occRef <- newIORef $ error "fanInt: no occurrence"
1481+ { _fanInt_subscribers = subscribers
1482+ , _fanInt_subscriptionRef = subscriptionRef
1483+ , _fanInt_occRef = occRef
1486+{-# NOINLINE unsafeNewFanInt #-}
1487+unsafeNewFanInt :: b -> FanInt x a
1488+unsafeNewFanInt b = unsafePerformIO $ do
1492+fanInt :: HasSpiderTimeline x => Event x (IntMap a) -> EventSelectorInt x a
1494+ let self = unsafeNewFanInt p
1495+ in EventSelectorInt $ \k -> Event $ \sub -> do
1496+ isEmpty <- liftIO $ FastMutableIntMap.isEmpty (_fanInt_subscribers self)
1497+ when isEmpty $ do -- This is the first subscriber, so we need to subscribe to our input
1498+ (subscription, parentOcc) <- subscribeAndRead p $ Subscriber
1499+ { subscriberPropagate = \m -> do
1500+ liftIO $ writeIORef (_fanInt_occRef self) m
1501+ scheduleIntClear $ _fanInt_occRef self
1502+ FastMutableIntMap.forIntersectionWithImmutable_ (_fanInt_subscribers self) m $ \b v -> do --TODO: Do we need to know that no subscribers are being added as we traverse?
1503+ FastWeakBag.traverse b $ \s -> do
1504+ subscriberPropagate s v
1505+ , subscriberInvalidateHeight = \old -> do
1506+ FastMutableIntMap.for_ (_fanInt_subscribers self) $ \b -> do
1507+ FastWeakBag.traverse b $ \s -> do
1508+ subscriberInvalidateHeight s old
1509+ , subscriberRecalculateHeight = \new -> do
1510+ FastMutableIntMap.for_ (_fanInt_subscribers self) $ \b -> do
1511+ FastWeakBag.traverse b $ \s -> do
1512+ subscriberRecalculateHeight s new
1515+ writeIORef (_fanInt_subscriptionRef self) subscription
1516+ writeIORef (_fanInt_occRef self) $ fromMaybe IntMap.empty parentOcc
1517+ scheduleIntClear $ _fanInt_occRef self
1519+ b <- FastMutableIntMap.lookup (_fanInt_subscribers self) k >>= \case
1521+ b <- FastWeakBag.empty
1522+ FastMutableIntMap.insert (_fanInt_subscribers self) k b
1525+ t <- liftIO $ FastWeakBag.insert sub b
1526+ currentOcc <- readIORef (_fanInt_occRef self)
1527+ (EventSubscription _ (EventSubscribed !heightRef _)) <- readIORef (_fanInt_subscriptionRef self)
1528+ return (EventSubscription (FastWeakBag.remove t) $! EventSubscribed heightRef $! toAny (_fanInt_subscriptionRef self, t), IntMap.lookup k currentOcc)
1431{-# INLINABLE getFanSubscribed #-}
1530{-# INLINABLE getFanSubscribed #-}
1432−getFanSubscribed :: (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)
1531+getFanSubscribed :: (HasSpiderTimeline x, GCompare k) => k a -> Fan x k -> Subscriber x a -> EventM x (WeakBagTicket, FanSubscribed x k, Maybe a)
1433getFanSubscribed k f sub = do
1532getFanSubscribed k f sub = do
1434 mSubscribed <- liftIO $ readIORef $ fanSubscribed f
1533 mSubscribed <- liftIO $ readIORef $ fanSubscribed f
1435 case mSubscribed of
1534 case mSubscribed of
@@ -1488,7 +1587,7 @@ subscribeFanSubscribed k subscribed sub = do
1488 Just (FanSubscribedChildren list _ weakSelf) -> {-# SCC "hitSubscribeFanSubscribed" #-} WeakBag.insert sub list weakSelf cleanupFanSubscribed
1587 Just (FanSubscribedChildren list _ weakSelf) -> {-# SCC "hitSubscribeFanSubscribed" #-} WeakBag.insert sub list weakSelf cleanupFanSubscribed
1490{-# INLINABLE getSwitchSubscribed #-}
1589{-# INLINABLE getSwitchSubscribed #-}
1491−getSwitchSubscribed :: (Defer (SomeHoldInit x) m, Defer (SomeAssignment x) m, Defer (SomeMergeInit x) m, Defer SomeClear m, Defer (SomeResetCoincidence x) m, HasCurrentHeight x m, HasSpiderTimeline x, m ~ EventM x) => Switch x a -> Subscriber x a -> m (WeakBagTicket, SwitchSubscribed x a, Maybe a)
1590+getSwitchSubscribed :: HasSpiderTimeline x => Switch x a -> Subscriber x a -> EventM x (WeakBagTicket, SwitchSubscribed x a, Maybe a)
1492getSwitchSubscribed s sub = do
1591getSwitchSubscribed s sub = do
1493 mSubscribed <- liftIO $ readIORef $ switchSubscribed s
1592 mSubscribed <- liftIO $ readIORef $ switchSubscribed s
1494 case mSubscribed of
1593 case mSubscribed of
@@ -1544,7 +1643,7 @@ subscribeSwitchSubscribed :: SwitchSubscribed x a -> Subscriber x a -> IO WeakBa
1544subscribeSwitchSubscribed subscribed sub = WeakBag.insert sub (switchSubscribedSubscribers subscribed) (switchSubscribedWeakSelf subscribed) cleanupSwitchSubscribed
1643subscribeSwitchSubscribed subscribed sub = WeakBag.insert sub (switchSubscribedSubscribers subscribed) (switchSubscribedWeakSelf subscribed) cleanupSwitchSubscribed
1546{-# INLINABLE getCoincidenceSubscribed #-}
1645{-# INLINABLE getCoincidenceSubscribed #-}
1547−getCoincidenceSubscribed :: forall x a m. (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) => Coincidence x a -> Subscriber x a -> m (WeakBagTicket, CoincidenceSubscribed x a, Maybe a)
1646+getCoincidenceSubscribed :: forall x a. HasSpiderTimeline x => Coincidence x a -> Subscriber x a -> EventM x (WeakBagTicket, CoincidenceSubscribed x a, Maybe a)
1548getCoincidenceSubscribed c sub = do
1647getCoincidenceSubscribed c sub = do
1549 mSubscribed <- liftIO $ readIORef $ coincidenceSubscribed c
1648 mSubscribed <- liftIO $ readIORef $ coincidenceSubscribed c
1550 case mSubscribed of
1649 case mSubscribed of
@@ -1607,7 +1706,7 @@ mergeWithMove d = cacheEvent (mergeCheapWithMove d)
1608{-# INLINE [1] mergeCheap #-}
1707{-# INLINE [1] mergeCheap #-}
1609mergeCheap :: forall k x. (HasSpiderTimeline x, GCompare k) => Dynamic x (PatchDMap k (Event x)) -> Event x (DMap k Identity)
1708mergeCheap :: forall k x. (HasSpiderTimeline x, GCompare k) => Dynamic x (PatchDMap k (Event x)) -> Event x (DMap k Identity)
1610−mergeCheap = mergeCheap' getInitialSubscribers updateMe destroy DMap.size
1709+mergeCheap = mergeCheap' getInitialSubscribers updateMe destroy
1612 updateMe :: MergeUpdateFunc k x (PatchDMap k (Event x)) (MergeSubscribedParent x)
1711 updateMe :: MergeUpdateFunc k x (PatchDMap k (Event x)) (MergeSubscribedParent x)
1613 updateMe subscriber heightBagRef oldParents (PatchDMap p) = do
1712 updateMe subscriber heightBagRef oldParents (PatchDMap p) = do
@@ -1642,7 +1741,7 @@ mergeCheap = mergeCheap' getInitialSubscribers updateMe destroy DMap.size
1643{-# INLINE [1] mergeCheapWithMove #-}
1742{-# INLINE [1] mergeCheapWithMove #-}
1644mergeCheapWithMove :: forall k x. (HasSpiderTimeline x, GCompare k) => Dynamic x (PatchDMapWithMove k (Event x)) -> Event x (DMap k Identity)
1743mergeCheapWithMove :: forall k x. (HasSpiderTimeline x, GCompare k) => Dynamic x (PatchDMapWithMove k (Event x)) -> Event x (DMap k Identity)
1645−mergeCheapWithMove = mergeCheap' getInitialSubscribers updateMe destroy DMap.size
1744+mergeCheapWithMove = mergeCheap' getInitialSubscribers updateMe destroy
1647 updateMe :: MergeUpdateFunc k x (PatchDMapWithMove k (Event x)) (MergeSubscribedParentWithMove x k)
1746 updateMe :: MergeUpdateFunc k x (PatchDMapWithMove k (Event x)) (MergeSubscribedParentWithMove x k)
1648 updateMe subscriber heightBagRef oldParents p = do
1747 updateMe subscriber heightBagRef oldParents p = do
@@ -1700,78 +1799,108 @@ type MergeDestroyFunc k s
1700 = DMap k s
1799 = DMap k s
1701 -> IO ()
1800 -> IO ()
1802+data Merge x k s = Merge
1803+ { _merge_parentsRef :: {-# UNPACK #-} !(IORef (DMap k s))
1804+ , _merge_heightBagRef :: {-# UNPACK #-} !(IORef HeightBag)
1805+ , _merge_heightRef :: {-# UNPACK #-} !(IORef Height)
1806+ , _merge_sub :: {-# UNPACK #-} !(Subscriber x (DMap k Identity))
1807+ , _merge_accumRef :: {-# UNPACK #-} !(IORef (DMap k Identity))
1810+invalidateMergeHeight :: Merge x k s -> IO ()
1811+invalidateMergeHeight m = invalidateMergeHeight' (_merge_heightRef m) (_merge_sub m)
1813+invalidateMergeHeight' :: IORef Height -> Subscriber x a -> IO ()
1814+invalidateMergeHeight' heightRef sub = do
1815+ oldHeight <- readIORef heightRef
1816+ 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
1817+ writeIORef heightRef $! invalidHeight
1818+ subscriberInvalidateHeight sub oldHeight
1821+revalidateMergeHeight :: Merge x k s -> IO ()
1822+revalidateMergeHeight m = do
1823+ currentHeight <- readIORef $ _merge_heightRef m
1824+ when (currentHeight == invalidHeight) $ do -- revalidateMergeHeight may be called multiple times; perhaps the's a way to finesse it to avoid this check
1825+ heights <- readIORef $ _merge_heightBagRef m
1826+ parents <- readIORef $ _merge_parentsRef m
1827+ -- When the number of heights in the bag reaches the number of parents, we should have a valid height
1828+ case heightBagSize heights `compare` DMap.size parents of
1831+ let height = succHeight $ heightBagMax heights
1832+ when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: height: " <> show height
1833+ writeIORef (_merge_heightRef m) $! height
1834+ subscriberRecalculateHeight (_merge_sub m) height
1835+ GT -> error $ "revalidateMergeHeight: more heights (" <> show (heightBagSize heights) <> ") than parents (" <> show (DMap.size parents) <> ") for Merge"
1837+scheduleMergeSelf :: HasSpiderTimeline x => Merge x k s -> Height -> EventM x ()
1838+scheduleMergeSelf m height = scheduleMerge' height (_merge_heightRef m) $ do
1839+ vals <- liftIO $ readIORef $ _merge_accumRef m
1840+ 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
1841+ --TODO: Assert that m is not empty
1842+ subscriberPropagate (_merge_sub m) vals
1844+mergeSubscriber :: forall x k s a. (HasSpiderTimeline x, GCompare k) => Merge x k s -> EventM x (k a) -> Subscriber x a
1845+mergeSubscriber m getKey = Subscriber
1846+ { subscriberPropagate = \a -> do
1847+ oldM <- liftIO $ readIORef $ _merge_accumRef m
1849+ let newM = DMap.insertWith (error $ "Same key fired multiple times for Merge") k (Identity a) oldM
1850+ tracePropagate (Proxy :: Proxy x) $ " DMap.size oldM = " <> show (DMap.size oldM) <> "; DMap.size newM = " <> show (DMap.size newM)
1851+ liftIO $ writeIORef (_merge_accumRef m) $! newM
1852+ when (DMap.null oldM) $ do -- Only schedule the firing once
1853+ height <- liftIO $ readIORef $ _merge_heightRef m
1854+ --TODO: assertions about height
1855+ currentHeight <- getCurrentHeight
1856+ when (height <= currentHeight) $ do
1857+ if height /= invalidHeight
1859+ myStack <- liftIO $ whoCreatedIORef undefined --TODO
1860+ error $ "Height (" ++ show height ++ ") is not greater than current height (" ++ show currentHeight ++ ")\n" ++ unlines (reverse myStack)
1863+ nodesInvolvedInCycle <- walkInvalidHeightParents $ eventSubscribedMerge subscribed
1864+ stacks <- forM nodesInvolvedInCycle $ \(Some.This es) -> whoCreatedEventSubscribed es
1865+ let cycleInfo = ":\n" <> drawForest (listsToForest stacks)
1869+ error $ "Causality loop found" <> cycleInfo
1870+ scheduleMergeSelf m height
1871+ , subscriberInvalidateHeight = \old -> do --TODO: When removing a parent doesn't actually change the height, maybe we can avoid invalidating
1872+ modifyIORef' (_merge_heightBagRef m) $ heightBagRemove old
1873+ invalidateMergeHeight m
1874+ , subscriberRecalculateHeight = \new -> do
1875+ modifyIORef' (_merge_heightBagRef m) $ heightBagAdd new
1876+ revalidateMergeHeight m
1879+--TODO: Be able to run as much of this as possible promptly
1880+updateMerge :: (HasSpiderTimeline x, GCompare k) => Merge x k s -> MergeUpdateFunc k x p s -> p -> SomeMergeUpdate x
1881+updateMerge m updateFunc p = SomeMergeUpdate updateMe (invalidateMergeHeight m) (revalidateMergeHeight m)
1882+ where updateMe = do
1883+ oldParents <- liftIO $ readIORef $ _merge_parentsRef m
1884+ (subscriptionsToKill, newParents) <- updateFunc (mergeSubscriber m) (_merge_heightBagRef m) oldParents p
1885+ liftIO $ writeIORef (_merge_parentsRef m) $! newParents
1886+ return subscriptionsToKill
1703{-# INLINE mergeCheap' #-}
1888{-# INLINE mergeCheap' #-}
1704−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)
1889+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)
1705−mergeCheap' getInitialSubscribers updateFunc destroy numParents d = Event $ \sub -> do
1890+mergeCheap' getInitialSubscribers updateFunc destroy d = Event $ \sub -> do
1706 initialParents <- readBehaviorUntracked $ dynamicCurrent d
1891 initialParents <- readBehaviorUntracked $ dynamicCurrent d
1707 accumRef <- liftIO $ newIORef $ error "merge: accumRef not yet initialized"
1892 accumRef <- liftIO $ newIORef $ error "merge: accumRef not yet initialized"
1708 heightRef <- liftIO $ newIORef $ error "merge: heightRef not yet initialized"
1893 heightRef <- liftIO $ newIORef $ error "merge: heightRef not yet initialized"
1709 heightBagRef <- liftIO $ newIORef $ error "merge: heightBagRef not yet initialized"
1894 heightBagRef <- liftIO $ newIORef $ error "merge: heightBagRef not yet initialized"
1710 parentsRef :: IORef (DMap k s) <- liftIO $ newIORef $ error "merge: parentsRef not yet initialized"
1895 parentsRef :: IORef (DMap k s) <- liftIO $ newIORef $ error "merge: parentsRef not yet initialized"
1711− let scheduleSelf height = scheduleMerge' height heightRef $ do
1896+ let m = Merge
1712− m <- liftIO $ readIORef accumRef
1897+ { _merge_parentsRef = parentsRef
1713− liftIO $ writeIORef accumRef $! DMap.empty
1898+ , _merge_heightBagRef = heightBagRef
1714− --TODO: Assert that m is not empty
1899+ , _merge_heightRef = heightRef
1715− subscriberPropagate sub m
1900+ , _merge_sub = sub
1716− invalidateMyHeight = do
1901+ , _merge_accumRef = accumRef
1717− oldHeight <- readIORef heightRef
1718− 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
1719− writeIORef heightRef $! invalidHeight
1720− subscriberInvalidateHeight sub oldHeight
1721− revalidateMyHeight = do
1722− currentHeight <- readIORef heightRef
1723− when (currentHeight == invalidHeight) $ do -- revalidateMergeHeight may be called multiple times; perhaps the's a way to finesse it to avoid this check
1724− heights <- readIORef heightBagRef
1725− parents <- readIORef parentsRef
1726− -- When the number of heights in the bag reaches the number of parents, we should have a valid height
1727− case heightBagSize heights `compare` numParents parents of
1730− let height = succHeight $ heightBagMax heights
1731− when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: height: " <> show height
1732− writeIORef heightRef $! height
1733− subscriberRecalculateHeight sub height
1734− GT -> error $ "revalidateMergeHeight: more heights (" <> show (heightBagSize heights) <> ") than parents (" <> show (numParents parents) <> ") for Merge"
1735− subscriber :: forall a. EventM x (k a) -> Subscriber x a
1736− subscriber getKey = Subscriber
1737− { subscriberPropagate = \a -> do
1738− oldM <- liftIO $ readIORef accumRef
1740− let newM = DMap.insertWith (error $ "Same key fired multiple times for Merge") k (Identity a) oldM
1741− tracePropagate (Proxy :: Proxy x) $ " DMap.size oldM = " <> show (DMap.size oldM) <> "; DMap.size newM = " <> show (DMap.size newM)
1742− liftIO $ writeIORef accumRef $! newM
1743− when (DMap.null oldM) $ do -- Only schedule the firing once
1744− height <- liftIO $ readIORef heightRef
1745− --TODO: assertions about height
1746− currentHeight <- getCurrentHeight
1747− when (height <= currentHeight) $ do
1748− if height /= invalidHeight
1750− myStack <- liftIO $ whoCreatedIORef undefined --TODO
1751− error $ "Height (" ++ show height ++ ") is not greater than current height (" ++ show currentHeight ++ ")\n" ++ unlines (reverse myStack)
1754− nodesInvolvedInCycle <- walkInvalidHeightParents $ eventSubscribedMerge subscribed
1755− stacks <- forM nodesInvolvedInCycle $ \(Some.This es) -> whoCreatedEventSubscribed es
1756− let cycleInfo = ":\n" <> drawForest (listsToForest stacks)
1760− error $ "Causality loop found" <> cycleInfo
1761− scheduleSelf height
1762− , subscriberInvalidateHeight = \old -> do --TODO: When removing a parent doesn't actually change the height, maybe we can avoid invalidating
1763− modifyIORef' heightBagRef $ heightBagRemove old
1765− , subscriberRecalculateHeight = \new -> do
1766− modifyIORef' heightBagRef $ heightBagAdd new
1769− let updateMe x = do
1770− oldParents <- liftIO $ readIORef parentsRef
1771− (subscriptionsToKill, newParents) <- updateFunc subscriber heightBagRef oldParents x
1772− liftIO $ writeIORef parentsRef $! newParents
1773− return subscriptionsToKill
1774− (dm, heights, initialParentState) <- getInitialSubscribers initialParents subscriber
1903+ (dm, heights, initialParentState) <- getInitialSubscribers initialParents $ mergeSubscriber m
1775 let myHeightBag = heightBagFromList $ filter (/= invalidHeight) heights
1904 let myHeightBag = heightBagFromList $ filter (/= invalidHeight) heights
1776 myHeight = if invalidHeight `elem` heights
1905 myHeight = if invalidHeight `elem` heights
1777 then invalidHeight
1906 then invalidHeight
@@ -1781,9 +1910,7 @@ mergeCheap' getInitialSubscribers updateFunc destroy numParents d = Event $ \sub
1781 then (if DMap.null dm then Nothing else Just dm, DMap.empty)
1910 then (if DMap.null dm then Nothing else Just dm, DMap.empty)
1782 else (Nothing, dm)
1911 else (Nothing, dm)
1783 unless (DMap.null accum) $ do
1912 unless (DMap.null accum) $ do
1784− scheduleSelf myHeight
1913+ scheduleMergeSelf m myHeight
1785− occRef <- liftIO $ newIORef occ
1786− when (isJust occ) $ scheduleClear occRef
1787 liftIO $ writeIORef accumRef $! accum
1914 liftIO $ writeIORef accumRef $! accum
1788 liftIO $ writeIORef heightRef $! myHeight
1915 liftIO $ writeIORef heightRef $! myHeight
1789 liftIO $ writeIORef heightBagRef $! myHeightBag
1916 liftIO $ writeIORef heightBagRef $! myHeightBag
@@ -1793,12 +1920,12 @@ mergeCheap' getInitialSubscribers updateFunc destroy numParents d = Event $ \sub
1793 let s = Subscriber
1920 let s = Subscriber
1794 { subscriberPropagate = \a -> {-# SCC "traverseMergeChange" #-} do
1921 { subscriberPropagate = \a -> {-# SCC "traverseMergeChange" #-} do
1795 tracePropagate (Proxy :: Proxy x) $ "SubscriberMerge/Change"
1922 tracePropagate (Proxy :: Proxy x) $ "SubscriberMerge/Change"
1796− defer $ SomeMergeUpdate (updateMe a) invalidateMyHeight revalidateMyHeight
1923+ defer $ updateMerge m updateFunc a
1797 , subscriberInvalidateHeight = \_ -> return ()
1924 , subscriberInvalidateHeight = \_ -> return ()
1798 , subscriberRecalculateHeight = \_ -> return ()
1925 , subscriberRecalculateHeight = \_ -> return ()
1800 (changeSubscription, change) <- subscribeAndRead (dynamicUpdated d) s
1927 (changeSubscription, change) <- subscribeAndRead (dynamicUpdated d) s
1801− forM_ change $ \c -> defer $ SomeMergeUpdate (updateMe c) invalidateMyHeight revalidateMyHeight
1928+ forM_ change $ \c -> defer $ updateMerge m updateFunc c
1802 -- We explicitly hold on to the unsubscribe function from subscribing to the update event.
1929 -- We explicitly hold on to the unsubscribe function from subscribing to the update event.
1803 -- If we don't do this, there are certain cases where mergeCheap will fail to properly retain
1930 -- If we don't do this, there are certain cases where mergeCheap will fail to properly retain
1804 -- its subscription.
1931 -- its subscription.
@@ -1808,6 +1935,106 @@ mergeCheap' getInitialSubscribers updateFunc destroy numParents d = Event $ \sub
1938+mergeInt :: forall x a. (HasSpiderTimeline x) => Dynamic x (PatchIntMap (Event x a)) -> Event x (IntMap a)
1939+mergeInt = cacheEvent . mergeIntCheap
1941+{-# INLINABLE mergeIntCheap #-}
1942+mergeIntCheap :: forall x a. (HasSpiderTimeline x) => Dynamic x (PatchIntMap (Event x a)) -> Event x (IntMap a)
1943+mergeIntCheap d = Event $ \sub -> do
1944+ initialParents <- readBehaviorUntracked $ dynamicCurrent d
1945+ accum <- liftIO $ FastMutableIntMap.newEmpty
1946+ heightRef <- liftIO $ newIORef zeroHeight
1947+ heightBagRef <- liftIO $ newIORef heightBagEmpty
1948+ parents <- liftIO $ FastMutableIntMap.newEmpty
1949+ let scheduleSelf = do
1950+ height <- liftIO $ readIORef $ heightRef
1951+ scheduleMerge' height heightRef $ do
1952+ vals <- liftIO $ FastMutableIntMap.getFrozenAndClear accum
1953+ subscriberPropagate sub vals
1954+ invalidateMyHeight = do
1955+ invalidateMergeHeight' heightRef sub
1956+ recalculateMyHeight = do
1957+ currentHeight <- readIORef heightRef
1958+ 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?
1959+ heights <- readIORef heightBagRef
1960+ numParents <- FastMutableIntMap.size parents
1961+ case heightBagSize heights `compare` numParents of
1964+ let height = succHeight $ heightBagMax heights
1965+ when debugInvalidateHeight $ putStrLn $ "recalculateSubscriberHeight: height: " <> show height
1966+ writeIORef heightRef $! height
1967+ subscriberRecalculateHeight sub height
1968+ GT -> error $ "revalidateMergeHeight: more heights (" <> show (heightBagSize heights) <> ") than parents (" <> show numParents <> ") for Merge"
1969+ mySubscriber k = Subscriber
1970+ { subscriberPropagate = \a -> do
1971+ wasEmpty <- liftIO $ FastMutableIntMap.isEmpty accum
1972+ liftIO $ FastMutableIntMap.insert accum k a
1973+ when wasEmpty scheduleSelf
1974+ , subscriberInvalidateHeight = \old -> do
1975+ modifyIORef' heightBagRef $ heightBagRemove old
1977+ , subscriberRecalculateHeight = \new -> do
1978+ modifyIORef' heightBagRef $ heightBagAdd new
1979+ recalculateMyHeight
1981+ forM_ (IntMap.toList initialParents) $ \(k, p) -> do
1982+ (subscription@(EventSubscription _ parentSubd), parentOcc) <- subscribeAndRead p $ mySubscriber k
1984+ forM_ parentOcc $ FastMutableIntMap.insert accum k
1985+ FastMutableIntMap.insert parents k subscription
1986+ height <- getEventSubscribedHeight parentSubd
1987+ if height == invalidHeight
1988+ then writeIORef heightRef invalidHeight
1990+ modifyIORef' heightBagRef $ heightBagAdd height
1991+ modifyIORef' heightRef $ \oldHeight ->
1992+ if oldHeight == invalidHeight
1994+ else max (succHeight height) oldHeight
1995+ myHeight <- liftIO $ readIORef heightRef
1996+ currentHeight <- getCurrentHeight
1997+ isEmpty <- liftIO $ FastMutableIntMap.isEmpty accum
1998+ occ <- if currentHeight >= myHeight -- If we should have fired by now
2000+ then return Nothing
2001+ else liftIO $ Just <$> FastMutableIntMap.getFrozenAndClear accum
2002+ else do when (not isEmpty) scheduleSelf -- We have things accumulated, but we shouldn't have fired them yet
2004+ changeSubdRef <- liftIO $ newIORef $ error "getMergeSubscribed: changeSubdRef not yet initialized"
2005+ defer $ SomeMergeInit $ do
2006+ let updateMe a = SomeMergeUpdate u invalidateMyHeight recalculateMyHeight
2009+ let f k newParent = do
2010+ subscription@(EventSubscription _ subd) <- subscribe newParent $ mySubscriber k
2011+ newParentHeight <- liftIO $ getEventSubscribedHeight subd
2012+ liftIO $ modifyIORef' heightBagRef $ heightBagAdd newParentHeight
2013+ return subscription
2014+ newSubscriptions <- FastMutableIntMap.traverseIntMapPatchWithKey f a
2015+ oldParents <- liftIO $ FastMutableIntMap.applyPatch parents newSubscriptions
2016+ liftIO $ for_ oldParents $ \oldParent -> do
2017+ oldParentHeight <- getEventSubscribedHeight $ _eventSubscription_subscribed oldParent
2018+ modifyIORef' heightBagRef $ heightBagRemove oldParentHeight
2019+ return $ IntMap.elems oldParents
2021+ { subscriberPropagate = \a -> {-# SCC "traverseMergeChange" #-} do
2022+ tracePropagate (Proxy :: Proxy x) $ "SubscriberMergeInt/Change"
2024+ , subscriberInvalidateHeight = \_ -> return ()
2025+ , subscriberRecalculateHeight = \_ -> return ()
2027+ (changeSubscription, change) <- subscribeAndRead (dynamicUpdated d) s
2028+ forM_ change $ \c -> defer $ updateMe c
2029+ -- We explicitly hold on to the unsubscribe function from subscribing to the update event.
2030+ -- If we don't do this, there are certain cases where mergeCheap will fail to properly retain
2031+ -- its subscription.
2032+ liftIO $ writeIORef changeSubdRef (s, changeSubscription)
2033+ let unsubscribeAll = traverse_ unsubscribe =<< FastMutableIntMap.getFrozenAndClear parents
2034+ return ( EventSubscription unsubscribeAll $ EventSubscribed heightRef $ toAny (parents, changeSubdRef)
1811newtype EventSelector x k = EventSelector { select :: forall a. k a -> Event x a }
2038newtype EventSelector x k = EventSelector { select :: forall a. k a -> Event x a }
1813fan :: (HasSpiderTimeline x, GCompare k) => Event x (DMap k Identity) -> EventSelector x k
2040fan :: (HasSpiderTimeline x, GCompare k) => Event x (DMap k Identity) -> EventSelector x k
@@ -1847,13 +2074,14 @@ newEventEnv = do
1847 mergeInitRef <- newIORef []
2074 mergeInitRef <- newIORef []
1848 heightRef <- newIORef zeroHeight
2075 heightRef <- newIORef zeroHeight
1849 toClearRef <- newIORef []
2076 toClearRef <- newIORef []
2077+ toClearIntRef <- newIORef []
1850 toClearRootRef <- newIORef []
2078 toClearRootRef <- newIORef []
1851 coincidenceInfosRef <- newIORef []
2079 coincidenceInfosRef <- newIORef []
1852 delayedRef <- newIORef IntMap.empty
2080 delayedRef <- newIORef IntMap.empty
1853− return $ EventEnv toAssignRef holdInitRef dynInitRef mergeUpdateRef mergeInitRef toClearRef toClearRootRef heightRef coincidenceInfosRef delayedRef
2081+ return $ EventEnv toAssignRef holdInitRef dynInitRef mergeUpdateRef mergeInitRef toClearRef toClearIntRef toClearRootRef heightRef coincidenceInfosRef delayedRef
1855clearEventEnv :: EventEnv x -> IO ()
2083clearEventEnv :: EventEnv x -> IO ()
1856−clearEventEnv (EventEnv toAssignRef holdInitRef dynInitRef mergeUpdateRef mergeInitRef toClearRef toClearRootRef heightRef coincidenceInfosRef delayedRef) = do
2084+clearEventEnv (EventEnv toAssignRef holdInitRef dynInitRef mergeUpdateRef mergeInitRef toClearRef toClearIntRef toClearRootRef heightRef coincidenceInfosRef delayedRef) = do
1857 writeIORef toAssignRef []
2085 writeIORef toAssignRef []
1858 writeIORef holdInitRef []
2086 writeIORef holdInitRef []
1859 writeIORef dynInitRef []
2087 writeIORef dynInitRef []
@@ -1861,6 +2089,7 @@ clearEventEnv (EventEnv toAssignRef holdInitRef dynInitRef mergeUpdateRef mergeI
1861 writeIORef mergeInitRef []
2089 writeIORef mergeInitRef []
1862 writeIORef heightRef zeroHeight
2090 writeIORef heightRef zeroHeight
1863 writeIORef toClearRef []
2091 writeIORef toClearRef []
2092+ writeIORef toClearIntRef []
1864 writeIORef toClearRootRef []
2093 writeIORef toClearRootRef []
1865 writeIORef coincidenceInfosRef []
2094 writeIORef coincidenceInfosRef []
1866 writeIORef delayedRef IntMap.empty
2095 writeIORef delayedRef IntMap.empty
@@ -1876,6 +2105,8 @@ runFrame a = SpiderHost $ ask >>= \_ -> lift $ do
1876 result <- runEventM go
2105 result <- runEventM go
1877 toClear <- readIORef $ eventEnvClears env
2106 toClear <- readIORef $ eventEnvClears env
1878 forM_ toClear $ \(SomeClear ref) -> {-# SCC "clear" #-} writeIORef ref Nothing
2107 forM_ toClear $ \(SomeClear ref) -> {-# SCC "clear" #-} writeIORef ref Nothing
2108+ toClearInt <- readIORef $ eventEnvIntClears env
2109+ forM_ toClearInt $ \(SomeIntClear ref) -> {-# SCC "intClear" #-} writeIORef ref $! IntMap.empty
1879 toClearRoot <- readIORef $ eventEnvRootClears env
2110 toClearRoot <- readIORef $ eventEnvRootClears env
1880 forM_ toClearRoot $ \(SomeRootClear ref) -> {-# SCC "rootClear" #-} writeIORef ref $! DMap.empty
2111 forM_ toClearRoot $ \(SomeRootClear ref) -> {-# SCC "rootClear" #-} writeIORef ref $! DMap.empty
1881 toAssign <- readIORef $ eventEnvAssignments env
2112 toAssign <- readIORef $ eventEnvAssignments env
@@ -2070,6 +2301,8 @@ withSpiderTimeline k = do
2071newtype SpiderPullM x a = SpiderPullM (BehaviorM x a) deriving (Functor, Applicative, Monad, MonadIO, MonadFix)
2302newtype SpiderPullM x a = SpiderPullM (BehaviorM x a) deriving (Functor, Applicative, Monad, MonadIO, MonadFix)
2304+type ComputeM = EventM
2073newtype SpiderPushM x a = SpiderPushM (ComputeM x a) deriving (Functor, Applicative, Monad, MonadIO, MonadFix)
2306newtype SpiderPushM x a = SpiderPushM (ComputeM x a) deriving (Functor, Applicative, Monad, MonadIO, MonadFix)
2075#ifndef SPECIALIZE_TO_SPIDERTIMELINE_GLOBAL
2308#ifndef SPECIALIZE_TO_SPIDERTIMELINE_GLOBAL
@@ -2122,6 +2355,10 @@ instance HasSpiderTimeline x => R.Reflex (SpiderTimeline x) where
2122 eventCoercion Coercion = Coercion
2355 eventCoercion Coercion = Coercion
2123 behaviorCoercion Coercion = Coercion
2356 behaviorCoercion Coercion = Coercion
2124 dynamicCoercion = unsafeCoerce --TODO: How can we avoid this unsafeCoerce? This is safe only because we know how Identity works as a Patch instance
2357 dynamicCoercion = unsafeCoerce --TODO: How can we avoid this unsafeCoerce? This is safe only because we know how Identity works as a Patch instance
2358+ {-# INLINABLE mergeIntIncremental #-}
2359+ mergeIntIncremental = SpiderEvent . mergeInt . (unsafeCoerce :: Dynamic x (PatchIntMap (R.Event (SpiderTimeline x) a)) -> Dynamic x (PatchIntMap (Event x a))) . unSpiderIncremental
2360+ {-# INLINABLE fanInt #-}
2361+ fanInt e = R.EventSelectorInt $ SpiderEvent . selectInt (fanInt (unSpiderEvent e))
2127data RootTrigger x a = forall k. GCompare k => RootTrigger (WeakBag (Subscriber x a), IORef (DMap k Identity), k a)
2364data RootTrigger x a = forall k. GCompare k => RootTrigger (WeakBag (Subscriber x a), IORef (DMap k Identity), k a)