@@ -16,25 +16,28 @@
1616--
1717--
1818module Asyncly.Core
19− ( Stream (..)
20− , MonadAsync
19+ (
20+ MonadAsync
2121
22− -- * Construction
23− , yielding
24−
25− -- * Composition
22+ -- * Streams
23+ , Stream (..)
24+ , yields
2625 , interleave
2726
28− -- * Asynchronous Operations
29− , CtxType (..)
30− , SchedPolicy (..)
31− , Dimension (..)
27+ -- * Concurrent Stream Vars (SVars)
28+ , SVar
29+ , SVarSched (..)
30+ , SVarTag (..)
31+ , SVarStyle (..)
3232 , EndOfStream (..)
33+ , newSVar1
34+ , newSVar2
35+ , streamSVar
36+
37+ -- * Concurrent Streams
3338 , parallel
3439 , parAlt
3540 , parLeft
36− , pushOneToCtx
37− , pullFromCtx
3841 )
3942where
4043
@@ -81,30 +84,35 @@ data ChildEvent a =
8184
8285-- | Conjunction is used for monadic/product style composition. Disjunction is
8386-- used for fold/sum style composition. We need to distiguish the two types of
84−-- contexts so that the scheduing of the two is independent.
85−data Dimension = Conjunction | Disjunction deriving Eq
87+-- SVars so that the scheduling of the two is independent.
88+data SVarTag = Conjunction | Disjunction deriving Eq
8689
8790-- | For fairly interleaved parallel composition the sched policy is FIFO
8891-- whereas for left biased parallel composition it is LIFO.
89−data SchedPolicy = LIFO | FIFO deriving Eq
90−
91−-- | Identify the type of the context used for parallel composition.
92−data CtxType = CtxType Dimension SchedPolicy deriving Eq
93−
94−-- | A Context represents a parallel composition. It has a runqueue which holds
95−-- the tasks that are to be picked by a pool of worker threads. It has an
96−-- output queue where the results are placed by the worker threads. A
97−-- doorBell is used by the worker threads to intimate the consumer thread
98−-- about availability of new results. New work is usually enqueued as a result
99−-- of executing the parallel combinators i.e. '<|' and '<|>'.
100−data Context m a =
101− Context { outputQueue :: IORef [ChildEvent a]
92+data SVarSched = LIFO | FIFO deriving Eq
93+
94+-- | Identify the type of the SVar. Two computations using the same style can
95+-- be bunched on the same SVar.
96+data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
97+
98+-- | An SVar is a conduit to multiple streams running concurrently. It has an
99+-- associated runqueue that holds the streams to be picked by a pool of worker
100+-- threads. It has an associated output queue where the output stream elements
101+-- are placed by the worker threads. A doorBell is used by the worker threads
102+-- to intimate the consumer thread about availability of new results in the
103+-- output queue.
104+--
105+-- New work is enqueued either at the time of creation of the SVar or as a
106+-- result of executing the parallel combinators i.e. '<|' and '<|>' by already
107+-- enqueued work.
108+data SVar m a =
109+ SVar { outputQueue :: IORef [ChildEvent a]
102110 , doorBell :: MVar Bool -- wakeup mechanism for outQ
103111 , enqueue :: Stream m a -> IO ()
104112 , runqueue :: m ()
105113 , runningThreads :: IORef (Set ThreadId)
106114 , queueEmpty :: m Bool
107− , ctxType :: CtxType
115+ , svarStyle :: SVarStyle
108116 }
109117
110118------------------------------------------------------------------------------
@@ -121,7 +129,7 @@ data Context m a =
121129newtype Stream m a =
122130 Stream {
123131 runStream :: forall r.
124− Maybe (Context m a) -- local state
132+ Maybe (SVar m a) -- local state
125133 -> m r -- stop
126134 -> (a -> Maybe (Stream m a) -> m r) -- yield
127135 -> m r
@@ -131,15 +139,15 @@ newtype Stream m a =
131139type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
132140
133141-- | Yield a singleton value in a stream.
134−yielding :: a -> Stream m a
135−yielding a = Stream $ \_ _ yld -> yld a Nothing
142+yields :: a -> Stream m a
143+yields a = Stream $ \_ _ yld -> yld a Nothing
136144
137145------------------------------------------------------------------------------
138146-- Semigroup
139147------------------------------------------------------------------------------
140148
141149-- | '<>' concatenates two streams sequentially i.e. the first stream is
142−-- exhausted completely before yielding any element from the second stream.
150+-- exhausted completely before yields any element from the second stream.
143151instance Semigroup (Stream m a) where
144152 m1 <> m2 = go m1
145153 where
@@ -189,36 +197,36 @@ doFork action exHandler =
189197-- XXX exception safety of all atomic/MVar operations
190198
191199{-# INLINE send #-}
192−send :: MonadIO m => Context m a -> ChildEvent a -> m ()
193−send ctx msg = liftIO $ do
194− atomicModifyIORefCAS_ (outputQueue ctx) $ \es -> msg : es
200+send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
201+send sv msg = liftIO $ do
202+ atomicModifyIORefCAS_ (outputQueue sv) $ \es -> msg : es
195203 -- XXX need a memory barrier? The wake up must happen only after the
196204 -- store has finished otherwise we can have lost wakeup problems.
197− void $ tryPutMVar (doorBell ctx) True
205+ void $ tryPutMVar (doorBell sv) True
198206
199207{-# INLINE sendStop #-}
200−sendStop :: MonadIO m => Context m a -> m ()
201−sendStop ctx = liftIO myThreadId >>= \tid -> send ctx (ChildStop tid Nothing)
208+sendStop :: MonadIO m => SVar m a -> m ()
209+sendStop sv = liftIO myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
202210
203−-- Note: Left associated operations can grow this queue to a large size
211+-- Note: Left associated compositions can grow this queue to a large size
204212{-# INLINE enqueueLIFO #-}
205213enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
206214enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
207215
208−runqueueLIFO :: MonadIO m => Context m a -> IORef [Stream m a] -> m ()
209−runqueueLIFO ctx q = run
216+runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
217+runqueueLIFO sv q = run
210218
211219 where
212220
213221 run = do
214222 work <- dequeue
215223 case work of
216− Nothing -> sendStop ctx
217− Just m -> (runStream m) (Just ctx) run yield
224+ Nothing -> sendStop sv
225+ Just m -> (runStream m) (Just sv) run yield
218226
219− sendit a = send ctx (ChildYield a)
227+ sendit a = send sv (ChildYield a)
220228 yield a Nothing = sendit a >> run
221− yield a (Just r) = sendit a >> (runStream r) (Just ctx) run yield
229+ yield a (Just r) = sendit a >> (runStream r) (Just sv) run yield
222230
223231 dequeue = liftIO $ atomicModifyIORefCAS q $ \ ms ->
224232 case ms of
@@ -229,19 +237,19 @@ runqueueLIFO ctx q = run
229237enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
230238enqueueFIFO = pushL
231239
232−runqueueFIFO :: MonadIO m => Context m a -> LinkedQueue (Stream m a) -> m ()
233−runqueueFIFO ctx q = run
240+runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
241+runqueueFIFO sv q = run
234242
235243 where
236244
237245 run = do
238246 work <- dequeue
239247 case work of
240− Nothing -> sendStop ctx
241− Just m -> (runStream m) (Just ctx) run yield
248+ Nothing -> sendStop sv
249+ Just m -> (runStream m) (Just sv) run yield
242250
243251 dequeue = liftIO $ tryPopR q
244− sendit a = send ctx (ChildYield a)
252+ sendit a = send sv (ChildYield a)
245253 yield a Nothing = sendit a >> run
246254 yield a (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run
247255
@@ -254,68 +262,65 @@ runqueueFIFO ctx q = run
254262-- 2) To know when all threads are done.
255263
256264{-# NOINLINE addThread #-}
257−addThread :: MonadIO m => Context m a -> ThreadId -> m ()
258−addThread ctx tid =
259− liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
265+addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
266+addThread sv tid =
267+ liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.insert tid s)
260268
261269{-# INLINE delThread #-}
262−delThread :: MonadIO m => Context m a -> ThreadId -> m ()
263−delThread ctx tid =
264− liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.delete tid s)
270+delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
271+delThread sv tid =
272+ liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
265273
266274{-# INLINE allThreadsDone #-}
267−allThreadsDone :: MonadIO m => Context m a -> m Bool
268−allThreadsDone ctx = liftIO $ do
269− readIORef (runningThreads ctx) >>= return . S.null
275+allThreadsDone :: MonadIO m => SVar m a -> m Bool
276+allThreadsDone sv = liftIO $ do
277+ readIORef (runningThreads sv) >>= return . S.null
270278
271279{-# NOINLINE handleChildException #-}
272−handleChildException :: MonadIO m => Context m a -> SomeException -> m ()
273−handleChildException ctx e = do
280+handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
281+handleChildException sv e = do
274282 tid <- liftIO myThreadId
275− send ctx (ChildStop tid (Just e))
283+ send sv (ChildStop tid (Just e))
276284
277285{-# NOINLINE pushWorker #-}
278−pushWorker :: MonadAsync m => Context m a -> m ()
279−pushWorker ctx =
280− doFork (runqueue ctx) (handleChildException ctx) >>= addThread ctx
286+pushWorker :: MonadAsync m => SVar m a -> m ()
287+pushWorker sv =
288+ doFork (runqueue sv) (handleChildException sv) >>= addThread sv
281289
282290-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
283291-- Also, if a worker blocks on the output queue we can decide if we want to
284292-- block or make it go away entirely, depending on the number of workers and
285293-- the type of the queue.
286294{-# INLINE sendWorkerWait #-}
287−sendWorkerWait :: MonadAsync m => Context m a -> m ()
288−sendWorkerWait ctx = do
289− case ctxType ctx of
290− CtxType _ LIFO -> liftIO $ threadDelay 200
291− CtxType _ FIFO -> liftIO $ threadDelay 0
295+sendWorkerWait :: MonadAsync m => SVar m a -> m ()
296+sendWorkerWait sv = do
297+ case svarStyle sv of
298+ SVarStyle _ LIFO -> liftIO $ threadDelay 200
299+ SVarStyle _ FIFO -> liftIO $ threadDelay 0
292300
293− output <- liftIO $ readIORef (outputQueue ctx)
301+ output <- liftIO $ readIORef (outputQueue sv)
294302 when (null output) $ do
295− done <- queueEmpty ctx
303+ done <- queueEmpty sv
296304 if (not done)
297− then (pushWorker ctx) >> sendWorkerWait ctx
298− else void (liftIO $ takeMVar (doorBell ctx))
305+ then (pushWorker sv) >> sendWorkerWait sv
306+ else void (liftIO $ takeMVar (doorBell sv))
299307
300308-- | An 'async' stream has finished but is still being used.
301309data EndOfStream = EndOfStream deriving Show
302310instance Exception EndOfStream
303311
304−-- | Pull a stream from a context
305−{-# NOINLINE pullFromCtx #-}
306−pullFromCtx :: MonadAsync m => Context m a -> Stream m a
307−pullFromCtx ctx = Stream $ \_ stp yld -> do
308− -- When using an async handle to the context, one may keep using a stale
309− -- context even after it has been fully drained. To detect it gracefully we
310− -- raise an explicit exception.
311− -- XXX if reading the IORef is costly we can use a flag in the context to
312+-- | Pull a stream from an SVar.
313+{-# NOINLINE streamSVar #-}
314+streamSVar :: MonadAsync m => SVar m a -> Stream m a
315+streamSVar sv = Stream $ \_ stp yld -> do
316+ -- XXX if reading the IORef is costly we can use a flag in the SVar to
312317 -- indicate we are done.
313− done <- allThreadsDone ctx
318+ done <- allThreadsDone sv
314319 when done $ throwM EndOfStream
315320
316− res <- liftIO $ tryTakeMVar (doorBell ctx)
317− when (isNothing res) $ sendWorkerWait ctx
318− list <- liftIO $ atomicModifyIORefCAS (outputQueue ctx) $ \x -> ([], x)
321+ res <- liftIO $ tryTakeMVar (doorBell sv)
322+ when (isNothing res) $ sendWorkerWait sv
323+ list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
319324 -- To avoid lock overhead we read all events at once instead of reading one
320325 -- at a time. We just reverse the list to process the events in the order
321326 -- they arrived. Maybe we can use a queue instead?
@@ -324,16 +329,16 @@ pullFromCtx ctx = Stream $ \_ stp yld -> do
324329 where
325330
326331 handleException e tid = do
327− delThread ctx tid
332+ delThread sv tid
328333 -- XXX implement kill async exception handling
329− -- liftIO $ readIORef (runningThreads ctx) >>= mapM_ killThread
334+ -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
330335 throwM e
331336
332337 {-# INLINE processEvents #-}
333338 processEvents [] = Stream $ \_ stp yld -> do
334− done <- allThreadsDone ctx
339+ done <- allThreadsDone sv
335340 if not done
336− then (runStream (pullFromCtx ctx)) Nothing stp yld
341+ then (runStream (streamSVar sv)) Nothing stp yld
337342 else stp
338343
339344 processEvents (ev : es) = Stream $ \_ stp yld -> do
@@ -344,91 +349,85 @@ pullFromCtx ctx = Stream $ \_ stp yld -> do
344349 ChildYield a -> yield a
345350 ChildStop tid e ->
346351 case e of
347− Nothing -> delThread ctx tid >> continue
352+ Nothing -> delThread sv tid >> continue
348353 Just ex -> handleException ex tid
349354
350−getFifoCtx :: MonadIO m => CtxType -> IO (Context m a)
351−getFifoCtx ctype = do
355+getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
356+getFifoSVar ctype = do
352357 outQ <- newIORef []
353358 outQMv <- newEmptyMVar
354359 running <- newIORef S.empty
355360 q <- newQ
356− let ctx =
357− Context { outputQueue = outQ
361+ let sv =
362+ SVar { outputQueue = outQ
358363 , doorBell = outQMv
359364 , runningThreads = running
360− , runqueue = runqueueFIFO ctx q
365+ , runqueue = runqueueFIFO sv q
361366 , enqueue = pushL q
362367 , queueEmpty = liftIO $ nullQ q
363− , ctxType = ctype
368+ , svarStyle = ctype
364369 }
365− in return ctx
370+ in return sv
366371
367−getLifoCtx :: MonadIO m => CtxType -> IO (Context m a)
368−getLifoCtx ctype = do
372+getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
373+getLifoSVar ctype = do
369374 outQ <- newIORef []
370375 outQMv <- newEmptyMVar
371376 running <- newIORef S.empty
372377 q <- newIORef []
373378 let checkEmpty = liftIO (readIORef q) >>= return . null
374− let ctx =
375− Context { outputQueue = outQ
379+ let sv =
380+ SVar { outputQueue = outQ
376381 , doorBell = outQMv
377382 , runningThreads = running
378− , runqueue = runqueueLIFO ctx q
383+ , runqueue = runqueueLIFO sv q
379384 , enqueue = enqueueLIFO q
380385 , queueEmpty = checkEmpty
381− , ctxType = ctype
386+ , svarStyle = ctype
382387 }
383− in return ctx
384−
385−-- | Split the original computation in a pull-push pair. The original
386−-- computation pulls from a Channel while m1 and m2 push to the channel.
387−pushPairToCtx :: MonadAsync m
388− => CtxType -> Stream m a -> Stream m a -> m (Context m a)
389−pushPairToCtx ctype m1 m2 = do
388+ in return sv
389+
390+-- | Create a new SVar and enqueue one stream computation on it.
391+newSVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
392+newSVar1 style m = do
393+ sv <- liftIO $
394+ case style of
395+ SVarStyle _ FIFO -> do
396+ c <- getFifoSVar style
397+ return c
398+ SVarStyle _ LIFO -> do
399+ c <- getLifoSVar style
400+ return c
401+ -- Note: We must have all the work on the queue before sending the
402+ -- pushworker, otherwise the pushworker may exit before we even get a
403+ -- chance to push.
404+ liftIO $ (enqueue sv) m
405+ pushWorker sv
406+ return sv
407+
408+-- | Create a new SVar and enqueue two stream computations on it.
409+newSVar2 :: MonadAsync m
410+ => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
411+newSVar2 style m1 m2 = do
390412 -- Note: We must have all the work on the queue before sending the
391413 -- pushworker, otherwise the pushworker may exit before we even get a
392414 -- chance to push.
393− ctx <- liftIO $
394− case ctype of
395− CtxType _ FIFO -> do
396− c <- getFifoCtx ctype
415+ sv <- liftIO $
416+ case style of
417+ SVarStyle _ FIFO -> do
418+ c <- getFifoSVar style
397419 (enqueue c) m1 >> (enqueue c) m2
398420 return c
399− CtxType _ LIFO -> do
400− c <- getLifoCtx ctype
421+ SVarStyle _ LIFO -> do
422+ c <- getLifoSVar style
401423 (enqueue c) m2 >> (enqueue c) m1
402424 return c
403− pushWorker ctx
404− return ctx
405−
406−-- | The computation specified in the argument is pushed to a new thread and
407−-- the context is returned. The returned context can be used to pull the output
408−-- of the computation.
409−pushOneToCtx :: MonadAsync m => CtxType -> Stream m a -> m (Context m a)
410−pushOneToCtx ctype m = do
411− ctx <- liftIO $
412− case ctype of
413− CtxType _ FIFO -> do
414− c <- getFifoCtx ctype
415− return c
416− CtxType _ LIFO -> do
417− c <- getLifoCtx ctype
418− return c
419− -- Note: We must have all the work on the queue before sending the
420− -- pushworker, otherwise the pushworker may exit before we even get a
421− -- chance to push.
422− liftIO $ (enqueue ctx) m
423− pushWorker ctx
424− return ctx
425+ pushWorker sv
426+ return sv
425427
426−{-# NOINLINE pushPullFork #-}
427−pushPullFork :: MonadAsync m
428− => CtxType -> Stream m a -> Stream m a -> Stream m a
429−pushPullFork ct m1 m2 = Stream $ \_ stp yld -> do
430− ctx <- pushPairToCtx ct m1 m2
431− (runStream (pullFromCtx ctx)) Nothing stp yld
428+------------------------------------------------------------------------------
429+-- Running streams concurrently
430+------------------------------------------------------------------------------
432431
433432-- Concurrency rate control. Our objective is to create more threads on demand
434433-- if the consumer is running faster than us. As soon as we encounter an
@@ -457,23 +456,28 @@ pushPullFork ct m1 m2 = Stream $ \_ stp yld -> do
457456--
458457-- TBD for pure work (when we are not in the IO monad) we can divide it into
459458-- just the number of CPUs.
460−--
461−-- XXX to rate control left folded structrues we will have to return the
462−-- residual work back to the dispatcher. It will also consume a lot of
463−-- memory due to queueing of all the work before execution starts.
459+
460+{-# NOINLINE makeAsync #-}
461+makeAsync :: MonadAsync m
462+ => SVarStyle -> Stream m a -> Stream m a -> Stream m a
463+makeAsync style m1 m2 = Stream $ \_ stp yld -> do
464+ sv <- newSVar2 style m1 m2
465+ (runStream (streamSVar sv)) Nothing stp yld
464466
465467-- | Compose two streams in parallel using a scheduling policy specified by
466−-- 'CtxType'. Note: This is designed to scale for right associated
467−-- compositions, therefore always use a right fold for folding bigger
468−-- structures.
468+-- 'SVarStyle'. Note: This is designed to scale for right associated
469+-- compositions, therefore always use a right fold for folding large or
470+-- infinite structures. For left associated structures it will first
471+-- destructure the whole structure and then start executing, consuming memory
472+-- proportional to the size of the structure, just like a left fold.
469473{-# INLINE parallel #-}
470−parallel :: MonadAsync m => CtxType -> Stream m a -> Stream m a -> Stream m a
471−parallel ct m1 m2 = Stream $ \ctx stp yld -> do
472− case ctx of
473− Nothing -> (runStream (pushPullFork ct m1 m2)) Nothing stp yld
474− Just c | ctxType c /= ct ->
475− (runStream (pushPullFork ct m1 m2)) Nothing stp yld
476− Just c -> liftIO ((enqueue c) m2) >> (runStream m1) ctx stp yld
474+parallel :: MonadAsync m => SVarStyle -> Stream m a -> Stream m a -> Stream m a
475+parallel style m1 m2 = Stream $ \st stp yld -> do
476+ case st of
477+ Nothing -> (runStream (makeAsync style m1 m2)) Nothing stp yld
478+ Just sv | svarStyle sv /= style ->
479+ (runStream (makeAsync style m1 m2)) Nothing stp yld
480+ Just sv -> liftIO ((enqueue sv) m2) >> (runStream m1) st stp yld
477481
478482------------------------------------------------------------------------------
479483-- Semigroup and Monoid style compositions for parallel actions
@@ -494,12 +498,12 @@ parAhead = undefined
494498-- | Same as '<|>'.
495499{-# INLINE parAlt #-}
496500parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
497−parAlt = parallel (CtxType Disjunction FIFO)
501+parAlt = parallel (SVarStyle Disjunction FIFO)
498502
499503-- | Same as '<|'.
500504{-# INLINE parLeft #-}
501505parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
502−parLeft = parallel (CtxType Disjunction LIFO)
506+parLeft = parallel (SVarStyle Disjunction LIFO)
503507
504508-------------------------------------------------------------------------------
505509-- Instances (only used for deriving newtype instances)
@@ -562,18 +566,18 @@ instance MonadThrow m => MonadThrow (Stream m) where
562566-- XXX handle and test cross thread state transfer
563567instance MonadError e m => MonadError e (Stream m) where
564568 throwError = lift . throwError
565− catchError m h = Stream $ \ctx stp yld ->
566− let handle r = r `catchError` \e -> (runStream (h e)) ctx stp yld
569+ catchError m h = Stream $ \st stp yld ->
570+ let handle r = r `catchError` \e -> (runStream (h e)) st stp yld
567571 yield a Nothing = yld a Nothing
568572 yield a (Just r) = yld a (Just (catchError r h))
569− in handle $ (runStream m) ctx stp yield
573+ in handle $ (runStream m) st stp yield
570574
571575instance MonadReader r m => MonadReader r (Stream m) where
572576 ask = lift ask
573− local f m = Stream $ \ctx stp yld ->
577+ local f m = Stream $ \st stp yld ->
574578 let yield a Nothing = local f $ yld a Nothing
575579 yield a (Just r) = local f $ yld a (Just (local f r))
576− in (runStream m) ctx (local f stp) yield
580+ in (runStream m) st (local f stp) yield
577581
578582instance MonadState s m => MonadState s (Stream m) where
579583 get = lift get