@@ -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 #-}
@@ -62,6 +60,7 @@ import Data.Word
62import GHC.Exts
60import GHC.Exts
63import GHC.IORef (IORef (..))
61import GHC.IORef (IORef (..))
64import GHC.Stack
62import GHC.Stack
63+import Reflex.FastWeak
65import Reflex.FunctorMaybe
64import Reflex.FunctorMaybe
66import System.IO.Unsafe
65import System.IO.Unsafe
67import System.Mem.Weak
66import System.Mem.Weak
@@ -75,6 +74,8 @@ import Data.Monoid (mempty)
75import Data.Tree (Forest, Tree (..), drawForest)
74import Data.Tree (Forest, Tree (..), drawForest)
77+import Data.FastWeakBag (FastWeakBag)
78+import qualified Data.FastWeakBag as FastWeakBag
78import Data.WeakBag (WeakBag, WeakBagTicket, _weakBag_children)
79import Data.WeakBag (WeakBag, WeakBagTicket, _weakBag_children)
79import qualified Data.WeakBag as WeakBag
80import qualified Data.WeakBag as WeakBag
80#ifndef SPECIALIZE_TO_SPIDERTIMELINE_GLOBAL
81#ifndef SPECIALIZE_TO_SPIDERTIMELINE_GLOBAL
@@ -252,10 +253,9 @@ pushCheap f e = Event $ \sub -> do
252 return (subscription, occ')
253 return (subscription, occ')
254data CacheSubscribed x a
255data CacheSubscribed x a
255− = CacheSubscribed { _cacheSubscribed_subscribers :: {-# UNPACK #-} !(WeakBag (Subscriber x a))
256+ = CacheSubscribed { _cacheSubscribed_subscribers :: {-# UNPACK #-} !(FastWeakBag (Subscriber x a))
256− , _cacheSubscribed_weakSelf :: {-# UNPACK #-} !(IORef (Weak (CacheSubscribed x a)))
257 , _cacheSubscribed_parent :: {-# UNPACK #-} !(EventSubscription x)
257 , _cacheSubscribed_parent :: {-# UNPACK #-} !(EventSubscription x)
258− , _cacheSubscribed_cached :: {-# UNPACK #-} !(IORef (Maybe (CacheSubscribed x a)))
258+ , _cacheSubscribed_cached :: {-# UNPACK #-} !(IORef (Maybe (FastWeakTicket (CacheSubscribed x a))))
259 , _cacheSubscribed_occurrence :: {-# UNPACK #-} !(IORef (Maybe a))
259 , _cacheSubscribed_occurrence :: {-# UNPACK #-} !(IORef (Maybe a))
@@ -263,57 +263,59 @@ data CacheSubscribed x a
263-- repeatedly
263-- repeatedly
265--TODO: Try a caching strategy where we subscribe directly to the parent when
265--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
266+--there's only one subscriber, and then build our own FastWeakBag only when a second
267--subscriber joins
267--subscriber joins
268{-# NOINLINE cacheEvent #-}
268{-# NOINLINE cacheEvent #-}
269−cacheEvent :: HasSpiderTimeline x => Event x a -> Event x a
269+cacheEvent :: forall x a. HasSpiderTimeline x => Event x a -> Event x a
270cacheEvent e =
270cacheEvent e =
271#ifdef DEBUG_TRACE_EVENTS
271#ifdef DEBUG_TRACE_EVENTS
272 withStackOneLine $ \callSite -> Event $
272 withStackOneLine $ \callSite -> Event $
276+ let mSubscribedRef :: IORef (Maybe (FastWeakTicket (CacheSubscribed x a)))
276− let !mSubscribedRef = unsafeNewIORef e Nothing
277+ !mSubscribedRef = unsafeNewIORef e Nothing
277 in \sub -> {-# SCC "cacheEvent" #-} do
278 in \sub -> {-# SCC "cacheEvent" #-} do
278#ifdef DEBUG_TRACE_EVENTS
279#ifdef DEBUG_TRACE_EVENTS
279 unless (BS8.null callSite) $ liftIO $ BS8.hPutStrLn stderr callSite
280 unless (BS8.null callSite) $ liftIO $ BS8.hPutStrLn stderr callSite
281− mSubscribed <- liftIO $ readIORef mSubscribedRef
282+ subscribedTicket <- liftIO (readIORef mSubscribedRef) >>= \case
282− case mSubscribed of
283+ 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
284 Nothing -> do
292− weakSelf <- liftIO $ newIORef $ error "cacheEvent: weakSelf not yet intialized"
285+ subscribers <- liftIO FastWeakBag.empty
293− (subscribers, ticket) <- liftIO $ WeakBag.singleton sub weakSelf cleanupCacheSubscribed
294− occRef <- liftIO $ newIORef undefined
286+ occRef <- liftIO $ newIORef Nothing -- This should never be read prior to being set below
295 (parentSub, occ) <- subscribeAndRead e $ Subscriber
287 (parentSub, occ) <- subscribeAndRead e $ Subscriber
296 { subscriberPropagate = \a -> do
288 { subscriberPropagate = \a -> do
297 liftIO $ writeIORef occRef $ Just a
289 liftIO $ writeIORef occRef $ Just a
298 scheduleClear occRef
290 scheduleClear occRef
299− propagate a subscribers
291+ propagateFast a subscribers
300 , subscriberInvalidateHeight = \old -> do
292 , subscriberInvalidateHeight = \old -> do
301− WeakBag.traverse subscribers $ invalidateSubscriberHeight old
293+ FastWeakBag.traverse subscribers $ invalidateSubscriberHeight old
302 , subscriberRecalculateHeight = \new -> do
294 , subscriberRecalculateHeight = \new -> do
303− WeakBag.traverse subscribers $ recalculateSubscriberHeight new
295+ FastWeakBag.traverse subscribers $ recalculateSubscriberHeight new
305− liftIO $ writeIORef occRef occ
297+ when (isJust occ) $ do
298+ 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
299+ scheduleClear occRef
307 let !subscribed = CacheSubscribed
300 let !subscribed = CacheSubscribed
308 { _cacheSubscribed_subscribers = subscribers
301 { _cacheSubscribed_subscribers = subscribers
309− , _cacheSubscribed_weakSelf = weakSelf
310 , _cacheSubscribed_parent = parentSub
302 , _cacheSubscribed_parent = parentSub
311 , _cacheSubscribed_cached = mSubscribedRef
303 , _cacheSubscribed_cached = mSubscribedRef
312 , _cacheSubscribed_occurrence = occRef
304 , _cacheSubscribed_occurrence = occRef
306+ !subscribedTicket = fastWeakTicket subscribed
314− liftIO $ writeIORef mSubscribedRef $ Just subscribed
307+ liftIO $ writeIORef mSubscribedRef $ Just subscribedTicket
315− liftIO $ writeIORef weakSelf =<< evaluate =<< mkWeakPtrWithDebug subscribed "PushSubscribed"
308+ return subscribedTicket
310+ let subscribed = fastWeakTicketValue subscribedTicket
311+ subscribedWeak <- getFastWeakTicketWeak subscribedTicket
312+ ticket <- FastWeakBag.insert sub (_cacheSubscribed_subscribers subscribed) subscribedWeak cleanupCacheSubscribed
313+ occ <- readIORef $ _cacheSubscribed_occurrence subscribed
314+ let es = EventSubscribed
315+ { eventSubscribedHeightRef = eventSubscribedHeightRef $ _eventSubscription_subscribed $ _cacheSubscribed_parent subscribed
316+ , eventSubscribedRetained = toAny subscribed
316− return (EventSubscription (WeakBag.remove ticket >> touch ticket) (EventSubscribed (eventSubscribedHeightRef $ _eventSubscription_subscribed parentSub) (toAny subscribed)), occ)
318+ return (EventSubscription (FastWeakBag.remove ticket >> touch ticket) es, occ)
318cleanupCacheSubscribed :: CacheSubscribed x a -> IO ()
320cleanupCacheSubscribed :: CacheSubscribed x a -> IO ()
319cleanupCacheSubscribed self = do
321cleanupCacheSubscribed self = do
@@ -497,6 +499,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
499 --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
500 WeakBag.traverse subscribers $ \s -> subscriberPropagate s a
502+-- | Propagate everything at the current height
503+propagateFast :: a -> FastWeakBag (Subscriber x a) -> EventM x ()
504+propagateFast a subscribers = withIncreasedDepth $ do
505+ -- 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
506+ --TODO: Should we check if nodes already have their values before propagating? Maybe we're re-doing work
507+ FastWeakBag.traverse subscribers $ \s -> subscriberPropagate s a
500--------------------------------------------------------------------------------
509--------------------------------------------------------------------------------
501-- EventSubscribed
510-- EventSubscribed
502--------------------------------------------------------------------------------
511--------------------------------------------------------------------------------