@@ -16,25 +16,28 @@
18module Asyncly.Core
18module Asyncly.Core
20− , MonadAsync
20+ MonadAsync
22− -- * Construction
22+ -- * Streams
23− , yielding
23+ , Stream (..)
26 , interleave
25 , interleave
28− -- * Asynchronous Operations
27+ -- * Concurrent Stream Vars (SVars)
29− , CtxType (..)
28+ , SVar
30− , SchedPolicy (..)
29+ , SVarSched (..)
31− , Dimension (..)
30+ , SVarTag (..)
32 , EndOfStream (..)
32 , EndOfStream (..)
37+ -- * Concurrent Streams
33 , parallel
38 , parallel
@@ -81,30 +84,35 @@ data ChildEvent a =
82-- | Conjunction is used for monadic/product style composition. Disjunction is
85-- | Conjunction is used for monadic/product style composition. Disjunction is
83-- used for fold/sum style composition. We need to distiguish the two types of
86-- 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.
87+-- SVars so that the scheduling of the two is independent.
85−data Dimension = Conjunction | Disjunction deriving Eq
88+data SVarTag = Conjunction | Disjunction deriving Eq
87-- | For fairly interleaved parallel composition the sched policy is FIFO
90-- | For fairly interleaved parallel composition the sched policy is FIFO
88-- whereas for left biased parallel composition it is LIFO.
91-- whereas for left biased parallel composition it is LIFO.
89−data SchedPolicy = LIFO | FIFO deriving Eq
92+data SVarSched = LIFO | FIFO deriving Eq
91−-- | Identify the type of the context used for parallel composition.
94+-- | Identify the type of the SVar. Two computations using the same style can
92−data CtxType = CtxType Dimension SchedPolicy deriving Eq
95+-- be bunched on the same SVar.
93−
96+data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
94−-- | A Context represents a parallel composition. It has a runqueue which holds
97+
98+-- | An SVar is a conduit to multiple streams running concurrently. It has an
95−-- the tasks that are to be picked by a pool of worker threads. It has an
99+-- associated runqueue that holds the streams to be picked by a pool of worker
96−-- output queue where the results are placed by the worker threads. A
100+-- threads. It has an associated output queue where the output stream elements
97−-- doorBell is used by the worker threads to intimate the consumer thread
101+-- are placed by the worker threads. A doorBell is used by the worker threads
98−-- about availability of new results. New work is usually enqueued as a result
102+-- to intimate the consumer thread about availability of new results in the
105+-- New work is enqueued either at the time of creation of the SVar or as a
99−-- of executing the parallel combinators i.e. '<|' and '<|>'.
106+-- result of executing the parallel combinators i.e. '<|' and '<|>' by already
100−data Context m a =
108+data SVar m a =
101− Context { outputQueue :: IORef [ChildEvent a]
109+ SVar { outputQueue :: IORef [ChildEvent a]
102 , doorBell :: MVar Bool -- wakeup mechanism for outQ
110 , doorBell :: MVar Bool -- wakeup mechanism for outQ
103 , enqueue :: Stream m a -> IO ()
111 , enqueue :: Stream m a -> IO ()
104 , runqueue :: m ()
112 , runqueue :: m ()
105 , runningThreads :: IORef (Set ThreadId)
113 , runningThreads :: IORef (Set ThreadId)
106 , queueEmpty :: m Bool
114 , queueEmpty :: m Bool
107− , ctxType :: CtxType
115+ , svarStyle :: SVarStyle
110------------------------------------------------------------------------------
118------------------------------------------------------------------------------
@@ -121,7 +129,7 @@ data Context m a =
121newtype Stream m a =
129newtype Stream m a =
123 runStream :: forall r.
131 runStream :: forall r.
124− Maybe (Context m a) -- local state
132+ Maybe (SVar m a) -- local state
125 -> m r -- stop
133 -> m r -- stop
126 -> (a -> Maybe (Stream m a) -> m r) -- yield
134 -> (a -> Maybe (Stream m a) -> m r) -- yield
@@ -131,15 +139,15 @@ newtype Stream m a =
131type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
139type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
133-- | Yield a singleton value in a stream.
141-- | Yield a singleton value in a stream.
134−yielding :: a -> Stream m a
142+yields :: a -> Stream m a
135−yielding a = Stream $ \_ _ yld -> yld a Nothing
143+yields a = Stream $ \_ _ yld -> yld a Nothing
137------------------------------------------------------------------------------
145------------------------------------------------------------------------------
138-- Semigroup
146-- Semigroup
139------------------------------------------------------------------------------
147------------------------------------------------------------------------------
141-- | '<>' concatenates two streams sequentially i.e. the first stream is
149-- | '<>' 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.
143instance Semigroup (Stream m a) where
151instance Semigroup (Stream m a) where
144 m1 <> m2 = go m1
152 m1 <> m2 = go m1
@@ -189,36 +197,36 @@ doFork action exHandler =
189-- XXX exception safety of all atomic/MVar operations
197-- XXX exception safety of all atomic/MVar operations
191{-# INLINE send #-}
199{-# INLINE send #-}
192−send :: MonadIO m => Context m a -> ChildEvent a -> m ()
200+send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
193−send ctx msg = liftIO $ do
201+send sv msg = liftIO $ do
194− atomicModifyIORefCAS_ (outputQueue ctx) $ \es -> msg : es
202+ atomicModifyIORefCAS_ (outputQueue sv) $ \es -> msg : es
195 -- XXX need a memory barrier? The wake up must happen only after the
203 -- XXX need a memory barrier? The wake up must happen only after the
196 -- store has finished otherwise we can have lost wakeup problems.
204 -- store has finished otherwise we can have lost wakeup problems.
197− void $ tryPutMVar (doorBell ctx) True
205+ void $ tryPutMVar (doorBell sv) True
199{-# INLINE sendStop #-}
207{-# INLINE sendStop #-}
200−sendStop :: MonadIO m => Context m a -> m ()
208+sendStop :: MonadIO m => SVar m a -> m ()
201−sendStop ctx = liftIO myThreadId >>= \tid -> send ctx (ChildStop tid Nothing)
209+sendStop sv = liftIO myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
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
204{-# INLINE enqueueLIFO #-}
212{-# INLINE enqueueLIFO #-}
205enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
213enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
206enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
214enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
208−runqueueLIFO :: MonadIO m => Context m a -> IORef [Stream m a] -> m ()
216+runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
209−runqueueLIFO ctx q = run
217+runqueueLIFO sv q = run
214 work <- dequeue
222 work <- dequeue
215 case work of
223 case work of
216− Nothing -> sendStop ctx
224+ Nothing -> sendStop sv
217− Just m -> (runStream m) (Just ctx) run yield
225+ Just m -> (runStream m) (Just sv) run yield
219− sendit a = send ctx (ChildYield a)
227+ sendit a = send sv (ChildYield a)
220 yield a Nothing = sendit a >> run
228 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
223 dequeue = liftIO $ atomicModifyIORefCAS q $ \ ms ->
231 dequeue = liftIO $ atomicModifyIORefCAS q $ \ ms ->
224 case ms of
232 case ms of
@@ -229,19 +237,19 @@ runqueueLIFO ctx q = run
229enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
237enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
230enqueueFIFO = pushL
238enqueueFIFO = pushL
232−runqueueFIFO :: MonadIO m => Context m a -> LinkedQueue (Stream m a) -> m ()
240+runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
233−runqueueFIFO ctx q = run
241+runqueueFIFO sv q = run
238 work <- dequeue
246 work <- dequeue
239 case work of
247 case work of
240− Nothing -> sendStop ctx
248+ Nothing -> sendStop sv
241− Just m -> (runStream m) (Just ctx) run yield
249+ Just m -> (runStream m) (Just sv) run yield
243 dequeue = liftIO $ tryPopR q
251 dequeue = liftIO $ tryPopR q
244− sendit a = send ctx (ChildYield a)
252+ sendit a = send sv (ChildYield a)
245 yield a Nothing = sendit a >> run
253 yield a Nothing = sendit a >> run
246 yield a (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run
254 yield a (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run
@@ -254,68 +262,65 @@ runqueueFIFO ctx q = run
254-- 2) To know when all threads are done.
262-- 2) To know when all threads are done.
256{-# NOINLINE addThread #-}
264{-# NOINLINE addThread #-}
257−addThread :: MonadIO m => Context m a -> ThreadId -> m ()
265+addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
258−addThread ctx tid =
266+addThread sv tid =
259− liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
267+ liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.insert tid s)
261{-# INLINE delThread #-}
269{-# INLINE delThread #-}
262−delThread :: MonadIO m => Context m a -> ThreadId -> m ()
270+delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
263−delThread ctx tid =
271+delThread sv tid =
264− liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.delete tid s)
272+ liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
266{-# INLINE allThreadsDone #-}
274{-# INLINE allThreadsDone #-}
267−allThreadsDone :: MonadIO m => Context m a -> m Bool
275+allThreadsDone :: MonadIO m => SVar m a -> m Bool
268−allThreadsDone ctx = liftIO $ do
276+allThreadsDone sv = liftIO $ do
269− readIORef (runningThreads ctx) >>= return . S.null
277+ readIORef (runningThreads sv) >>= return . S.null
271{-# NOINLINE handleChildException #-}
279{-# NOINLINE handleChildException #-}
272−handleChildException :: MonadIO m => Context m a -> SomeException -> m ()
280+handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
273−handleChildException ctx e = do
281+handleChildException sv e = do
274 tid <- liftIO myThreadId
282 tid <- liftIO myThreadId
275− send ctx (ChildStop tid (Just e))
283+ send sv (ChildStop tid (Just e))
277{-# NOINLINE pushWorker #-}
285{-# NOINLINE pushWorker #-}
278−pushWorker :: MonadAsync m => Context m a -> m ()
286+pushWorker :: MonadAsync m => SVar m a -> m ()
279−pushWorker ctx =
287+pushWorker sv =
280− doFork (runqueue ctx) (handleChildException ctx) >>= addThread ctx
288+ doFork (runqueue sv) (handleChildException sv) >>= addThread sv
282-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
290-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
283-- Also, if a worker blocks on the output queue we can decide if we want to
291-- Also, if a worker blocks on the output queue we can decide if we want to
284-- block or make it go away entirely, depending on the number of workers and
292-- block or make it go away entirely, depending on the number of workers and
285-- the type of the queue.
293-- the type of the queue.
286{-# INLINE sendWorkerWait #-}
294{-# INLINE sendWorkerWait #-}
287−sendWorkerWait :: MonadAsync m => Context m a -> m ()
295+sendWorkerWait :: MonadAsync m => SVar m a -> m ()
288−sendWorkerWait ctx = do
296+sendWorkerWait sv = do
289− case ctxType ctx of
297+ case svarStyle sv of
290− CtxType _ LIFO -> liftIO $ threadDelay 200
298+ SVarStyle _ LIFO -> liftIO $ threadDelay 200
291− CtxType _ FIFO -> liftIO $ threadDelay 0
299+ SVarStyle _ FIFO -> liftIO $ threadDelay 0
293− output <- liftIO $ readIORef (outputQueue ctx)
301+ output <- liftIO $ readIORef (outputQueue sv)
294 when (null output) $ do
302 when (null output) $ do
295− done <- queueEmpty ctx
303+ done <- queueEmpty sv
296 if (not done)
304 if (not done)
297− then (pushWorker ctx) >> sendWorkerWait ctx
305+ then (pushWorker sv) >> sendWorkerWait sv
298− else void (liftIO $ takeMVar (doorBell ctx))
306+ else void (liftIO $ takeMVar (doorBell sv))
300-- | An 'async' stream has finished but is still being used.
308-- | An 'async' stream has finished but is still being used.
301data EndOfStream = EndOfStream deriving Show
309data EndOfStream = EndOfStream deriving Show
302instance Exception EndOfStream
310instance Exception EndOfStream
304−-- | Pull a stream from a context
312+-- | Pull a stream from an SVar.
305−{-# NOINLINE pullFromCtx #-}
313+{-# NOINLINE streamSVar #-}
306−pullFromCtx :: MonadAsync m => Context m a -> Stream m a
314+streamSVar :: MonadAsync m => SVar m a -> Stream m a
307−pullFromCtx ctx = Stream $ \_ stp yld -> do
315+streamSVar sv = 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
316+ -- XXX if reading the IORef is costly we can use a flag in the SVar to
312 -- indicate we are done.
317 -- indicate we are done.
313− done <- allThreadsDone ctx
318+ done <- allThreadsDone sv
314 when done $ throwM EndOfStream
319 when done $ throwM EndOfStream
316− res <- liftIO $ tryTakeMVar (doorBell ctx)
321+ res <- liftIO $ tryTakeMVar (doorBell sv)
317− when (isNothing res) $ sendWorkerWait ctx
322+ when (isNothing res) $ sendWorkerWait sv
318− list <- liftIO $ atomicModifyIORefCAS (outputQueue ctx) $ \x -> ([], x)
323+ list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
319 -- To avoid lock overhead we read all events at once instead of reading one
324 -- To avoid lock overhead we read all events at once instead of reading one
320 -- at a time. We just reverse the list to process the events in the order
325 -- at a time. We just reverse the list to process the events in the order
321 -- they arrived. Maybe we can use a queue instead?
326 -- they arrived. Maybe we can use a queue instead?
@@ -324,16 +329,16 @@ pullFromCtx ctx = Stream $ \_ stp yld -> do
326 handleException e tid = do
331 handleException e tid = do
327− delThread ctx tid
332+ delThread sv tid
328 -- XXX implement kill async exception handling
333 -- XXX implement kill async exception handling
329− -- liftIO $ readIORef (runningThreads ctx) >>= mapM_ killThread
334+ -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
332 {-# INLINE processEvents #-}
337 {-# INLINE processEvents #-}
333 processEvents [] = Stream $ \_ stp yld -> do
338 processEvents [] = Stream $ \_ stp yld -> do
334− done <- allThreadsDone ctx
339+ done <- allThreadsDone sv
335 if not done
340 if not done
336− then (runStream (pullFromCtx ctx)) Nothing stp yld
341+ then (runStream (streamSVar sv)) Nothing stp yld
339 processEvents (ev : es) = Stream $ \_ stp yld -> do
344 processEvents (ev : es) = Stream $ \_ stp yld -> do
@@ -344,91 +349,85 @@ pullFromCtx ctx = Stream $ \_ stp yld -> do
344 ChildYield a -> yield a
349 ChildYield a -> yield a
345 ChildStop tid e ->
350 ChildStop tid e ->
346 case e of
351 case e of
347− Nothing -> delThread ctx tid >> continue
352+ Nothing -> delThread sv tid >> continue
348 Just ex -> handleException ex tid
353 Just ex -> handleException ex tid
350−getFifoCtx :: MonadIO m => CtxType -> IO (Context m a)
355+getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
351−getFifoCtx ctype = do
356+getFifoSVar ctype = do
352 outQ <- newIORef []
357 outQ <- newIORef []
353 outQMv <- newEmptyMVar
358 outQMv <- newEmptyMVar
354 running <- newIORef S.empty
359 running <- newIORef S.empty
355 q <- newQ
360 q <- newQ
356− let ctx =
361+ let sv =
357− Context { outputQueue = outQ
362+ SVar { outputQueue = outQ
358 , doorBell = outQMv
363 , doorBell = outQMv
359 , runningThreads = running
364 , runningThreads = running
360− , runqueue = runqueueFIFO ctx q
365+ , runqueue = runqueueFIFO sv q
361 , enqueue = pushL q
366 , enqueue = pushL q
362 , queueEmpty = liftIO $ nullQ q
367 , queueEmpty = liftIO $ nullQ q
363− , ctxType = ctype
368+ , svarStyle = ctype
365− in return ctx
370+ in return sv
367−getLifoCtx :: MonadIO m => CtxType -> IO (Context m a)
372+getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
368−getLifoCtx ctype = do
373+getLifoSVar ctype = do
369 outQ <- newIORef []
374 outQ <- newIORef []
370 outQMv <- newEmptyMVar
375 outQMv <- newEmptyMVar
371 running <- newIORef S.empty
376 running <- newIORef S.empty
372 q <- newIORef []
377 q <- newIORef []
373 let checkEmpty = liftIO (readIORef q) >>= return . null
378 let checkEmpty = liftIO (readIORef q) >>= return . null
374− let ctx =
379+ let sv =
375− Context { outputQueue = outQ
380+ SVar { outputQueue = outQ
376 , doorBell = outQMv
381 , doorBell = outQMv
377 , runningThreads = running
382 , runningThreads = running
378− , runqueue = runqueueLIFO ctx q
383+ , runqueue = runqueueLIFO sv q
379 , enqueue = enqueueLIFO q
384 , enqueue = enqueueLIFO q
380 , queueEmpty = checkEmpty
385 , queueEmpty = checkEmpty
381− , ctxType = ctype
386+ , svarStyle = ctype
383− in return ctx
388+ in return sv
385−-- | Split the original computation in a pull-push pair. The original
390+-- | Create a new SVar and enqueue one stream computation on it.
386−-- computation pulls from a Channel while m1 and m2 push to the channel.
391+newSVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
392+newSVar1 style m = do
395+ SVarStyle _ FIFO -> do
396+ c <- getFifoSVar style
398+ SVarStyle _ LIFO -> do
399+ c <- getLifoSVar style
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
404+ liftIO $ (enqueue sv) m
408+-- | Create a new SVar and enqueue two stream computations on it.
387−pushPairToCtx :: MonadAsync m
409+newSVar2 :: MonadAsync m
388− => CtxType -> Stream m a -> Stream m a -> m (Context m a)
410+ => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
389−pushPairToCtx ctype m1 m2 = do
411+newSVar2 style m1 m2 = do
390 -- Note: We must have all the work on the queue before sending the
412 -- Note: We must have all the work on the queue before sending the
391 -- pushworker, otherwise the pushworker may exit before we even get a
413 -- pushworker, otherwise the pushworker may exit before we even get a
392 -- chance to push.
414 -- chance to push.
393− ctx <- liftIO $
415+ sv <- liftIO $
394− case ctype of
416+ case style of
395− CtxType _ FIFO -> do
417+ SVarStyle _ FIFO -> do
396− c <- getFifoCtx ctype
418+ c <- getFifoSVar style
397 (enqueue c) m1 >> (enqueue c) m2
419 (enqueue c) m1 >> (enqueue c) m2
399− CtxType _ LIFO -> do
421+ SVarStyle _ LIFO -> do
400− c <- getLifoCtx ctype
422+ c <- getLifoSVar style
401 (enqueue c) m2 >> (enqueue c) m1
423 (enqueue c) m2 >> (enqueue c) m1
403− pushWorker ctx
425+ pushWorker sv
404− return ctx
426+ return sv
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
413− CtxType _ FIFO -> do
414− c <- getFifoCtx ctype
416− CtxType _ LIFO -> do
417− c <- getLifoCtx ctype
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
422− liftIO $ (enqueue ctx) m
426−{-# NOINLINE pushPullFork #-}
428+------------------------------------------------------------------------------
427−pushPullFork :: MonadAsync m
429+-- Running streams concurrently
428− => CtxType -> Stream m a -> Stream m a -> Stream m a
430+------------------------------------------------------------------------------
429−pushPullFork ct m1 m2 = Stream $ \_ stp yld -> do
430− ctx <- pushPairToCtx ct m1 m2
431− (runStream (pullFromCtx ctx)) Nothing stp yld
433-- Concurrency rate control. Our objective is to create more threads on demand
432-- Concurrency rate control. Our objective is to create more threads on demand
434-- if the consumer is running faster than us. As soon as we encounter an
433-- 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
458-- TBD for pure work (when we are not in the IO monad) we can divide it into
457-- TBD for pure work (when we are not in the IO monad) we can divide it into
459-- just the number of CPUs.
458-- just the number of CPUs.
461−-- XXX to rate control left folded structrues we will have to return the
460+{-# NOINLINE makeAsync #-}
462−-- residual work back to the dispatcher. It will also consume a lot of
461+makeAsync :: MonadAsync m
463−-- memory due to queueing of all the work before execution starts.
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
465-- | Compose two streams in parallel using a scheduling policy specified by
467-- | Compose two streams in parallel using a scheduling policy specified by
466−-- 'CtxType'. Note: This is designed to scale for right associated
468+-- 'SVarStyle'. Note: This is designed to scale for right associated
467−-- compositions, therefore always use a right fold for folding bigger
469+-- compositions, therefore always use a right fold for folding large or
468−-- structures.
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.
469{-# INLINE parallel #-}
473{-# INLINE parallel #-}
470−parallel :: MonadAsync m => CtxType -> Stream m a -> Stream m a -> Stream m a
474+parallel :: MonadAsync m => SVarStyle -> Stream m a -> Stream m a -> Stream m a
471−parallel ct m1 m2 = Stream $ \ctx stp yld -> do
475+parallel style m1 m2 = Stream $ \st stp yld -> do
472− case ctx of
476+ case st of
473− Nothing -> (runStream (pushPullFork ct m1 m2)) Nothing stp yld
477+ Nothing -> (runStream (makeAsync style m1 m2)) Nothing stp yld
474− Just c | ctxType c /= ct ->
478+ Just sv | svarStyle sv /= style ->
475− (runStream (pushPullFork ct m1 m2)) Nothing stp yld
479+ (runStream (makeAsync style m1 m2)) Nothing stp yld
476− Just c -> liftIO ((enqueue c) m2) >> (runStream m1) ctx stp yld
480+ Just sv -> liftIO ((enqueue sv) m2) >> (runStream m1) st stp yld
478------------------------------------------------------------------------------
482------------------------------------------------------------------------------
479-- Semigroup and Monoid style compositions for parallel actions
483-- Semigroup and Monoid style compositions for parallel actions
@@ -494,12 +498,12 @@ parAhead = undefined
494-- | Same as '<|>'.
498-- | Same as '<|>'.
495{-# INLINE parAlt #-}
499{-# INLINE parAlt #-}
496parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
500parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
497−parAlt = parallel (CtxType Disjunction FIFO)
501+parAlt = parallel (SVarStyle Disjunction FIFO)
499-- | Same as '<|'.
503-- | Same as '<|'.
500{-# INLINE parLeft #-}
504{-# INLINE parLeft #-}
501parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
505parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
502−parLeft = parallel (CtxType Disjunction LIFO)
506+parLeft = parallel (SVarStyle Disjunction LIFO)
504-------------------------------------------------------------------------------
508-------------------------------------------------------------------------------
505-- Instances (only used for deriving newtype instances)
509-- Instances (only used for deriving newtype instances)
@@ -562,18 +566,18 @@ instance MonadThrow m => MonadThrow (Stream m) where
562-- XXX handle and test cross thread state transfer
566-- XXX handle and test cross thread state transfer
563instance MonadError e m => MonadError e (Stream m) where
567instance MonadError e m => MonadError e (Stream m) where
564 throwError = lift . throwError
568 throwError = lift . throwError
565− catchError m h = Stream $ \ctx stp yld ->
569+ catchError m h = Stream $ \st stp yld ->
566− let handle r = r `catchError` \e -> (runStream (h e)) ctx stp yld
570+ let handle r = r `catchError` \e -> (runStream (h e)) st stp yld
567 yield a Nothing = yld a Nothing
571 yield a Nothing = yld a Nothing
568 yield a (Just r) = yld a (Just (catchError r h))
572 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
571instance MonadReader r m => MonadReader r (Stream m) where
575instance MonadReader r m => MonadReader r (Stream m) where
572 ask = lift ask
576 ask = lift ask
573− local f m = Stream $ \ctx stp yld ->
577+ local f m = Stream $ \st stp yld ->
574 let yield a Nothing = local f $ yld a Nothing
578 let yield a Nothing = local f $ yld a Nothing
575 yield a (Just r) = local f $ yld a (Just (local f r))
579 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
578instance MonadState s m => MonadState s (Stream m) where
582instance MonadState s m => MonadState s (Stream m) where
579 get = lift get
583 get = lift get