git.iko.soy/streamly
on master help sign in
More cleanup in RunAsync
504800bHarendra Kumar
src/Asyncly/RunAsync.hs+25−101
compare with master
unifiedsplit
@@ -48,6 +48,11 @@ import GHC.Prim (Any)
48import Asyncly.Threads
48import Asyncly.Threads
49import Asyncly.AsyncT
49import Asyncly.AsyncT
50
50
51+-- This transformer runs AsyncT under a state to manage the threads.
52+-- Separating the state from the pure ListT transformer is cleaner but it
53+-- results in 2x performance degradation. At some point if that performance is
54+-- really needed we can combine the two.
55+
51newtype AsynclyT m a = AsynclyT { runAsynclyT :: AsyncT (StateT Context m) a }
56newtype AsynclyT m a = AsynclyT { runAsynclyT :: AsyncT (StateT Context m) a }
52
57
53deriving instance MonadAsync m => Functor (AsynclyT m)
58deriving instance MonadAsync m => Functor (AsynclyT m)
@@ -58,6 +63,10 @@ deriving instance MonadAsync m => MonadIO (AsynclyT m)
58instance MonadTrans (AsynclyT) where
63instance MonadTrans (AsynclyT) where
59 lift mx = AsynclyT $ AsyncT $ \s k -> lift mx >>= (\a -> (k a Nothing))
64 lift mx = AsynclyT $ AsyncT $ \s k -> lift mx >>= (\a -> (k a Nothing))
60
65
66+------------------------------------------------------------------------------
67+-- Running the monad
68+------------------------------------------------------------------------------
69+
61getContext :: Maybe (IORef [Recording]) -> IO Context
70getContext :: Maybe (IORef [Recording]) -> IO Context
62getContext lref = do
71getContext lref = do
63 childChan <- atomically newTChan
72 childChan <- atomically newTChan
@@ -65,9 +74,9 @@ getContext lref = do
65 credit <- newIORef maxBound
74 credit <- newIORef maxBound
66 return $ initContext childChan pendingRef credit lref
75 return $ initContext childChan pendingRef credit lref
67
76
68−-- | Run an 'AsyncT m' computation, wait for it to finish and discard the
77+-- | Run an 'AsynclyT m' computation, wait for it to finish and discard the
69-- results.
78-- results.
70−{-# SPECIALIZE wait_ :: AsynclyT IO a -> IO () #-}
79+{-# INLINABLE wait_ #-}
71wait_ :: forall m a. (MonadAsync m, MonadCatch m) => AsynclyT m a -> m ()
80wait_ :: forall m a. (MonadAsync m, MonadCatch m) => AsynclyT m a -> m ()
72wait_ (AsynclyT m) = do
81wait_ (AsynclyT m) = do
73 ctx <- liftIO $ getContext Nothing
82 ctx <- liftIO $ getContext Nothing
@@ -76,31 +85,35 @@ wait_ (AsynclyT m) = do
76
85
77 where
86 where
78
87
88+ stop = return ()
79− run m = (runAsyncT m) (return ())
89+ run m = (runAsyncT m) stop (\_ r -> maybe stop run r)
80− (\a r -> maybe (return ()) (\x -> run x) r)
81
90
82data Step a r = Stop | Done a | Yield a r
91data Step a r = Stop | Done a | Yield a r
83
92
84−{-# SPECIALIZE wait :: AsynclyT IO a -> IO [a] #-}
93+{-# INLINABLE wait #-}
85−{-# INLINE wait #-}
94+-- | Run an 'AsynclyT m' computation and collect the results generated by each
95+-- thread of the computation in a list.
86wait :: forall m a. (MonadAsync m, MonadCatch m) => AsynclyT m a -> m [a]
96wait :: forall m a. (MonadAsync m, MonadCatch m) => AsynclyT m a -> m [a]
87wait (AsynclyT m) = liftIO (getContext Nothing) >>= run m
97wait (AsynclyT m) = liftIO (getContext Nothing) >>= run m
88
98
89 where
99 where
90
100
101+ stop = return Stop
102+ done a = return (Done a)
103+ yield a x = return $ Yield a x
104+
91 run ma ctx = do
105 run ma ctx = do
92− (res, ctx') <- runStateT ((runAsyncT ma) (return Stop)
106+ (res, ctx') <- runStateT
93− (\a r -> maybe (return (Done a)) (\x -> return $ Yield a x) r)) ctx
107+ ((runAsyncT ma) stop (\a r -> maybe (done a) (\x -> yield a x) r))
108+ ctx
94 case res of
109 case res of
95 Yield x mb -> liftM (x :) (run mb ctx')
110 Yield x mb -> liftM (x :) (run mb ctx')
96− -- XXX pull from children here
97 Stop -> return []
111 Stop -> return []
98 Done x -> return (x : [])
112 Done x -> return (x : [])
99
113
100-- scatter
114-- scatter
101−{-# SPECIALIZE each :: [a] -> AsynclyT IO a #-}
115+{-# INLINABLE each #-}
102−each :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
103− => [a] -> AsynclyT m a
116+each :: MonadAsync m => [a] -> AsynclyT m a
104each xs = foldr (<|>) empty $ map return xs
117each xs = foldr (<|>) empty $ map return xs
105
118
106------------------------------------------------------------------------------
119------------------------------------------------------------------------------
@@ -114,95 +127,6 @@ each xs = foldr (<|>) empty $ map return xs
114-- threads n action = AsynclyT $ AsyncT $ threadCtl n (runAsyncT $ runAsynclyT action)
127-- threads n action = AsynclyT $ AsyncT $ threadCtl n (runAsyncT $ runAsynclyT action)
115
128
116{-
129{-
117−wait_ :: AsyncT IO a -> IO ()
118−wait_ m = do
119− ctx <- liftIO $ getContext (return :: a -> AsyncT IO a) Nothing
120− _ <- runStateT (run m) ctx
121− return ()
122−
123− where
124−
125− run m = runAsyncT m >>= \x ->
126− case x of
127− Yield _ r -> run r
128− -- XXX pull from children here
129− _ -> return ()
130−
131−wait :: forall m a. (MonadAsync m, MonadCatch m) => AsyncT m a -> m [a]
132−wait m = liftIO (getContext (return :: a -> AsyncT m a) Nothing) >>= run m
133−
134− where
135−
136− run ma ctx = do
137− (a, ctx') <- runStateT (runAsyncT ma) ctx
138− case a of
139− Yield x mb -> liftM (x :) (run mb ctx')
140− -- XXX pull from children here
141− Stop -> return []
142−-}
143−
144−{-
145−------------------------------------------------------------------------------
146−-- Running the monad
147−------------------------------------------------------------------------------
148−
149−-- | Run an 'AsyncT m' computation and collect the results generated by each
150−-- thread of the computation in a list.
151−{-# SPECIALIZE waitAsync :: (a -> StateT Context IO (Maybe a))
152− -> Maybe (IORef [Recording]) -> AsyncT IO a -> IO () #-}
153−waitAsync :: (MonadAsync m, MonadCatch m)
154− => (a -> StateT Context m (Maybe a))
155− -> Maybe (IORef [Recording])
156− -> AsyncT m a -> m ()
157−waitAsync finalizer lref m = do
158− childChan <- liftIO $ atomically newTChan
159− pendingRef <- liftIO $ newIORef []
160− credit <- liftIO $ newIORef maxBound
161−
162− let ctx = initContext childChan pendingRef credit finalizer lref
163−
164− r <- try $ runStateT comp ctx
165− e <- handleResult ctx r
166− maybe (return ()) throwM e
167−
168− where
169− comp = do
170− x <- runAsyncT m
171− maybe (return Nothing) finalizer x
172−
173−-- TBD throttling of producer based on consumption rate.
174−
175−-- | Invoked to store the result of the computation in the context and finish
176−-- the computation when the computation is done
177−{-# SPECIALIZE gatherResult :: IORef [a] -> a -> StateT Context IO (Maybe a) #-}
178−gatherResult :: MonadAsync m => IORef [a] -> a -> StateT Context m (Maybe a)
179−gatherResult ref r = do
180− liftIO $ atomicModifyIORefCAS ref $ \rs -> (r : rs, ())
181− return Nothing
182−
183−gather :: (MonadAsync m, MonadCatch m)
184− => AsyncT m a
185− -> StateT Context m (Maybe [a])
186−gather m = do
187− resultsRef <- liftIO $ newIORef []
188− lift $ waitAsync (gatherResult resultsRef) Nothing m
189− r <- liftIO $ readIORef resultsRef
190− return $ Just r
191−
192−-- | Run an 'AsyncT m' computation and collect the results generated by each
193−-- thread of the computation in a list.
194−{-# SPECIALIZE wait :: AsyncT IO a -> IO [a] #-}
195−wait :: (MonadAsync m, MonadCatch m) => AsyncT m a -> m [a]
196−wait m = do
197− resultsRef <- liftIO $ newIORef []
198− waitAsync (gatherResult resultsRef) Nothing m
199− liftIO $ readIORef resultsRef
200−
201−-- | Run an 'AsyncT m' computation, wait for it to finish and discard the
202−-- results.
203−wait_ :: (MonadAsync m, MonadCatch m) => AsyncT m a -> m ()
204−wait_ m = waitAsync (const (return Nothing)) Nothing m
205−
206------------------------------------------------------------------------------
130------------------------------------------------------------------------------
207-- Logging
131-- Logging
208------------------------------------------------------------------------------
132------------------------------------------------------------------------------