git.iko.soy/streamly
on e16a980 help sign in
streamly/src/Streamly/Streams/Parallel.hs
646 lines 24 kB history raw
1{-# LANGUAGE CPP #-}
2{-# LANGUAGE ConstraintKinds #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE GeneralizedNewtypeDeriving#-}
6{-# LANGUAGE InstanceSigs #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE UndecidableInstances #-} -- XXX
10#include "inline.hs"
12-- |
13-- Module : Streamly.Streams.Parallel
14-- Copyright : (c) 2017 Harendra Kumar
15--
16-- License : BSD3
17-- Maintainer : streamly@composewell.com
18-- Stability : experimental
19-- Portability : GHC
20--
21--
22module Streamly.Streams.Parallel
23 (
24 ParallelT
25 , Parallel
26 , parallely
27 , parallel
28 , parallelFst
29 , parallelMin
30 , tapAsync
32 -- * Function application
33 , mkParallel
34 , (|$)
35 , (|&)
36 , (|$.)
37 , (|&.)
38 )
39where
41import Control.Concurrent (myThreadId)
42import Control.Exception (SomeException(..), throwIO)
43import Control.Monad (ap)
44import Control.Monad.Base (MonadBase(..), liftBaseDefault)
45import Control.Monad.Catch (MonadThrow, throwM)
46-- import Control.Monad.Error.Class (MonadError(..))
47import Control.Monad.IO.Class (MonadIO(..))
48import Control.Monad.Reader.Class (MonadReader(..))
49import Control.Monad.State.Class (MonadState(..))
50import Control.Monad.Trans.Class (MonadTrans(lift))
51import Data.Functor (void)
52import Data.IORef (readIORef, writeIORef, newIORef)
53import Data.Maybe (fromJust)
54#if __GLASGOW_HASKELL__ < 808
55import Data.Semigroup (Semigroup(..))
56#endif
57import Prelude hiding (map)
59import qualified Data.Set as Set
61import Streamly.Streams.SVar (fromSVar, fromStreamVar)
62import Streamly.Streams.Serial (map)
63import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, foldStream,
64 foldStreamShared, adapt)
66import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)
67import Streamly.Internal.Data.Time.Clock (Clock(..), getTime)
68import Streamly.Internal.Data.SVar hiding (handleChildException)
70import qualified Streamly.Streams.StreamK as K
71import qualified Streamly.Streams.StreamD as D
72import qualified Streamly.Internal.Data.SVar as SVar
73import qualified Streamly.Internal.Data.Fold as FL
75#include "Instances.hs"
77-------------------------------------------------------------------------------
78-- Parallel
79-------------------------------------------------------------------------------
81-------------------------------------------------------------------------------
82-- StreamK based worker routines
83-------------------------------------------------------------------------------
85{-# NOINLINE runOne #-}
86runOne
87 :: MonadIO m
88 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
89runOne st m0 winfo =
90 case getYieldLimit st of
91 Nothing -> go m0
92 Just _ -> runOneLimited st m0 winfo
94 where
96 go m = do
97 liftIO $ decrementBufferLimit sv
98 foldStreamShared st yieldk single stop m
100 sv = fromJust $ streamVar st
102 stop = liftIO $ do
103 incrementBufferLimit sv
104 sendStop sv winfo
105 sendit a = liftIO $ void $ send sv (ChildYield a)
106 single a = sendit a >> (liftIO $ sendStop sv winfo)
107 yieldk a r = sendit a >> go r
109runOneLimited
110 :: MonadIO m
111 => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
112runOneLimited st m0 winfo = go m0
114 where
116 go m = do
117 yieldLimitOk <- liftIO $ decrementYieldLimit sv
118 if yieldLimitOk
119 then do
120 liftIO $ decrementBufferLimit sv
121 foldStreamShared st yieldk single stop m
122 else do
123 liftIO $ cleanupSVarFromWorker sv
124 liftIO $ sendStop sv winfo
126 sv = fromJust $ streamVar st
128 stop = liftIO $ do
129 incrementBufferLimit sv
130 incrementYieldLimit sv
131 sendStop sv winfo
132 sendit a = liftIO $ void $ send sv (ChildYield a)
133 single a = sendit a >> (liftIO $ sendStop sv winfo)
134 yieldk a r = sendit a >> go r
136-------------------------------------------------------------------------------
137-- StreamD based worker routines
138-------------------------------------------------------------------------------
140-- Using StreamD the worker stream producing code can fuse with the code to
141-- queue output to the SVar giving some perf boost.
142--
143-- Note that StreamD can only be used in limited situations, specifically, we
144-- cannot implement joinStreamVarPar using this.
146{-# INLINE_NORMAL pushWorkerParD #-}
147pushWorkerParD :: MonadAsync m
148 => State t m a -> SVar t m a -> D.Stream m a -> m ()
149pushWorkerParD st sv xs =
150 if svarInspectMode sv
151 then forkWithDiag
152 else do
153 tid <-
154 case getYieldLimit st of
155 Nothing -> doFork (work Nothing)
156 (svarMrun sv)
157 (SVar.handleChildException sv)
158 Just _ -> doFork (workLim Nothing)
159 (svarMrun sv)
160 (SVar.handleChildException sv)
161 modifyThread sv tid
163 where
165 {-# NOINLINE work #-}
166 work info = (D.runFold (FL.toParallelSVar sv info) xs)
168 {-# NOINLINE workLim #-}
169 workLim info = D.runFold (FL.toParallelSVarLimited sv info) xs
171 {-# NOINLINE forkWithDiag #-}
172 forkWithDiag = do
173 -- We do not use workerCount in case of ParallelVar but still there is
174 -- no harm in maintaining it correctly.
175 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
176 recordMaxWorkers sv
177 -- This allocation matters when significant number of workers are being
178 -- sent. We allocate it only when needed. The overhead increases by 4x.
179 winfo <-
180 case yieldRateInfo sv of
181 Nothing -> return Nothing
182 Just _ -> liftIO $ do
183 cntRef <- newIORef 0
184 t <- getTime Monotonic
185 lat <- newIORef (0, t)
186 return $ Just WorkerInfo
187 { workerYieldMax = 0
188 , workerYieldCount = cntRef
189 , workerLatencyStart = lat
190 }
191 tid <-
192 case getYieldLimit st of
193 Nothing -> doFork (work winfo)
194 (svarMrun sv)
195 (SVar.handleChildException sv)
196 Just _ -> doFork (workLim winfo)
197 (svarMrun sv)
198 (SVar.handleChildException sv)
199 modifyThread sv tid
201-------------------------------------------------------------------------------
202-- Consing and appending a stream in parallel style
203-------------------------------------------------------------------------------
205-- Note that consing and appending requires StreamK as it would not scale well
206-- with StreamD unless we are only consing a very small number of streams or
207-- elements in a stream. StreamK allows us to manipulate control flow in a way
208-- which StreamD cannot allow. StreamK can make a jump without having to
209-- remember the past state.
211{-# NOINLINE forkSVarPar #-}
212forkSVarPar :: (IsStream t, MonadAsync m)
213 => SVarStopStyle -> t m a -> t m a -> t m a
214forkSVarPar ss m r = mkStream $ \st yld sng stp -> do
215 sv <- newParallelVar ss st
216 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
217 case ss of
218 StopBy -> liftIO $ do
219 set <- readIORef (workerThreads sv)
220 writeIORef (svarStopBy sv) $ Set.elemAt 0 set
221 _ -> return ()
222 pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream r)
223 foldStream st yld sng stp (fromSVar sv)
225{-# INLINE joinStreamVarPar #-}
226joinStreamVarPar :: (IsStream t, MonadAsync m)
227 => SVarStyle -> SVarStopStyle -> t m a -> t m a -> t m a
228joinStreamVarPar style ss m1 m2 = mkStream $ \st yld sng stp ->
229 case streamVar st of
230 Just sv | svarStyle sv == style && svarStopStyle sv == ss -> do
231 pushWorkerPar sv (runOne st $ toStream m1)
232 foldStreamShared st yld sng stp m2
233 _ -> foldStreamShared st yld sng stp (forkSVarPar ss m1 m2)
235-------------------------------------------------------------------------------
236-- User facing APIs
237-------------------------------------------------------------------------------
239-- | XXX we can implement it more efficienty by directly implementing instead
240-- of combining streams using parallel.
241{-# INLINE consMParallel #-}
242{-# SPECIALIZE consMParallel :: IO a -> ParallelT IO a -> ParallelT IO a #-}
243consMParallel :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a
244consMParallel m r = fromStream $ K.yieldM m `parallel` (toStream r)
246-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'
247-- Merges two streams concurrently.
248--
249-- @since 0.2.0
250{-# INLINE parallel #-}
251parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
252parallel = joinStreamVarPar ParallelVar StopNone
254-- This is a co-parallel like combinator for streams, where first stream is the
255-- main stream and the rest are just supporting it, when the first ends
256-- everything ends.
257--
258-- | Like `parallel` but stops the output as soon as the first stream stops.
259--
260-- @since 0.7.0
261{-# INLINE parallelFst #-}
262parallelFst :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
263parallelFst = joinStreamVarPar ParallelVar StopBy
265-- This is a race like combinator for streams.
266--
267-- | Like `parallel` but stops the output as soon as any of the two streams
268-- stops.
269--
270-- @since 0.7.0
271{-# INLINE parallelMin #-}
272parallelMin :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
273parallelMin = joinStreamVarPar ParallelVar StopAny
275------------------------------------------------------------------------------
276-- Convert a stream to parallel
277------------------------------------------------------------------------------
279_mkParallelK :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
280_mkParallelK m = do
281 sv <- newParallelVar StopNone defState
282 pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)
283 return $ fromSVar sv
285-- We may have to use it in higher order functions like concatMap, so use the
286-- Normal phase for inlining.
287{-# INLINE_NORMAL mkParallel #-}
288mkParallel :: (IsStream t, MonadAsync m)
289 => State Stream m a -> t m a -> m (t m a)
290mkParallel st m = do
291 sv <- newParallelVar StopNone defState
292 pushWorkerParD st sv (D.toStreamD m)
293 return $ fromSVar sv
295------------------------------------------------------------------------------
296-- Stream to stream concurrent function application
297------------------------------------------------------------------------------
299{-# INLINE _applyParallelK #-}
300_applyParallelK :: (IsStream t, MonadAsync m)
301 => (t m a -> t m b) -> t m a -> t m b
302_applyParallelK f m = mkStream $ \st yld sng stp -> do
303 sv <- newParallelVar StopNone (adaptState st)
304 pushWorkerPar sv (runOne st{streamVar = Just sv} (toStream m))
305 foldStream st yld sng stp $ f $ fromSVar sv
307-- XXX need to implement fromSVar in direct style for fused generation from
308-- SVar.
309{-# INLINE applyParallelD #-}
310applyParallelD :: (IsStream t, MonadAsync m)
311 => (t m a -> t m b) -> t m a -> t m b
312applyParallelD f m = mkStream $ \st yld sng stp -> do
313 sv <- newParallelVar StopNone (adaptState st)
314 pushWorkerParD (adaptState st) sv (D.toStreamD m)
315 foldStream st yld sng stp $ f $ fromSVar sv
317{-# INLINE applyParallel #-}
318applyParallel :: (IsStream t, MonadAsync m)
319 => (t m a -> t m b) -> t m a -> t m b
320applyParallel = applyParallelD
322------------------------------------------------------------------------------
323-- Stream runner concurrent function application
324------------------------------------------------------------------------------
326{-# INLINE _foldParallelK #-}
327_foldParallelK :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
328_foldParallelK f m = do
329 sv <- newParallelVar StopNone defState
330 pushWorkerPar sv (runOne defState{streamVar = Just sv} $ toStream m)
331 f $ fromSVar sv
333-- XXX need to implement fromSVar in direct style for fused generation from
334-- SVar.
335{-# INLINE foldParallelD #-}
336foldParallelD :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
337foldParallelD f m = do
338 sv <- newParallelVar StopNone defState
339 pushWorkerParD defState sv (D.toStreamD m)
340 f $ fromSVar sv
342{-# INLINE foldParallel #-}
343foldParallel :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
344foldParallel = foldParallelD
346------------------------------------------------------------------------------
347-- Parallel tee
348------------------------------------------------------------------------------
350{-# INLINE teeToSVar #-}
351teeToSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> t m a
352teeToSVar svr m = mkStream $ \st yld sng stp -> do
353 foldStreamShared st yld sng stp (go svr m)
355 where
357 go sv m0 = mkStream $ \st yld sng stp -> do
358 let stop = do
359 liftIO $ do
360 incrementBufferLimit sv
361 sendStop sv Nothing
362 -- XXX wait for Stop response event on exception channel,
363 -- drain the exception channel.
364 stp
365 sendit a = liftIO $ do
366 -- XXX check for exceptions before decrement so that we do not
367 -- block forever if the child already exited with an exception.
368 decrementBufferLimit sv
369 void $ send sv (ChildYield a)
370 single a = sendit a >> (liftIO $ sendStop sv Nothing) >> sng a
371 yieldk a r = sendit a >> yld a (go sv r)
372 in foldStreamShared st yieldk single stop m0
374-- In case of folds the roles of worker and parent on an SVar are reversed. The
375-- parent stream pushes values to an SVar instead of pulling from it and a
376-- worker thread running the fold pulls from the SVar and folds the stream. We
377-- keep a separate channel for pushing exceptions in the reverse direction i.e.
378-- from the fold to the parent stream.
379--
380-- NOTE: If we use fromSVar here it will kill the main computation (the parent)
381-- when the SVar goes away so we use fromStreamVar instead.
383{-# NOINLINE handleChildException #-}
384handleChildException :: SVar t m a -> SomeException -> IO ()
385handleChildException _sv e = do
386 -- tid <- myThreadId
387 -- void $ sendReverse sv (ChildStop tid (Just e))
388 throwIO e
390-- | Redirect a copy of the stream to a supplied fold and run it concurrently
391-- in an independent thread. The fold may buffer some elements. The buffer size
392-- is determined by the prevailing 'maxBuffer' setting.
393--
394-- @
395-- Stream m a -> m b
396-- |
397-- -----stream m a ---------------stream m a-----
398--
399-- @
400--
401-- @
402-- > S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
403-- 1
404-- 2
405-- @
406--
407-- Exceptions from the concurrently running fold are propagated to the current
408-- computation. Note that, because of buffering in the fold, exceptions may be
409-- delayed and may not correspond to the current element being processed in the
410-- parent stream, but we guarantee that the tap finishes and all exceptions
411-- from it are drained before the parent stream stops.
412--
413--
414-- Compare with 'tap'.
415--
416-- @since 0.7.0
417{-# INLINE tapAsync #-}
418tapAsync :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> t m a
419tapAsync f m = mkStream $ \st yld sng stp -> do
420 -- Buffer size for the SVar is derived from the current state
421 sv <- newParallelVar StopNone (adaptState st)
422 -- XXX exception handling
423 -- XXX if we terminate due to an exception, do we need to actively
424 -- terminate the fold?
425 liftIO myThreadId >>= modifyThread sv
426 void $ doFork (void $ f $ fromStream $ fromStreamVar sv)
427 (svarMrun sv) (handleChildException sv)
428 foldStreamShared st yld sng stp (teeToSVar sv m)
430------------------------------------------------------------------------------
431-- Concurrent Application
432------------------------------------------------------------------------------
434infixr 0 |$
435infixr 0 |$.
437infixl 1 |&
438infixl 1 |&.
440-- | Parallel function application operator for streams; just like the regular
441-- function application operator '$' except that it is concurrent. The
442-- following code prints a value every second even though each stage adds a 1
443-- second delay.
444--
445--
446-- @
447-- drain $
448-- S.mapM (\\x -> threadDelay 1000000 >> print x)
449-- |$ S.repeatM (threadDelay 1000000 >> return 1)
450-- @
451--
452-- /Concurrent/
453--
454-- @since 0.3.0
455{-# INLINE (|$) #-}
456(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
457f |$ x = applyParallel f x
459-- | Parallel reverse function application operator for streams; just like the
460-- regular reverse function application operator '&' except that it is
461-- concurrent.
462--
463-- @
464-- drain $
465-- S.repeatM (threadDelay 1000000 >> return 1)
466-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
467-- @
468--
469-- /Concurrent/
470--
471-- @since 0.3.0
472{-# INLINE (|&) #-}
473(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
474x |& f = f |$ x
476-- | Parallel function application operator; applies a @run@ or @fold@ function
477-- to a stream such that the fold consumer and the stream producer run in
478-- parallel. A @run@ or @fold@ function reduces the stream to a value in the
479-- underlying monad. The @.@ at the end of the operator is a mnemonic for
480-- termination of the stream.
481--
482-- @
483-- S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
484-- |$. S.repeatM (threadDelay 1000000 >> return 1)
485-- @
486--
487-- /Concurrent/
488--
489-- @since 0.3.0
490{-# INLINE (|$.) #-}
491(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
492f |$. x = foldParallel f x
494-- | Parallel reverse function application operator for applying a run or fold
495-- functions to a stream. Just like '|$.' except that the operands are reversed.
496--
497-- @
498-- S.repeatM (threadDelay 1000000 >> return 1)
499-- |&. S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
500-- @
501--
502-- /Concurrent/
503--
504-- @since 0.3.0
505{-# INLINE (|&.) #-}
506(|&.) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b
507x |&. f = f |$. x
509------------------------------------------------------------------------------
510-- ParallelT
511------------------------------------------------------------------------------
513-- | Async composition with strict concurrent execution of all streams.
514--
515-- The 'Semigroup' instance of 'ParallelT' executes both the streams
516-- concurrently without any delay or without waiting for the consumer demand
517-- and /merges/ the results as they arrive. If the consumer does not consume
518-- the results, they are buffered upto a configured maximum, controlled by the
519-- 'maxBuffer' primitive. If the buffer becomes full the concurrent tasks will
520-- block until there is space in the buffer.
521--
522-- Both 'WAsyncT' and 'ParallelT', evaluate the constituent streams fairly in a
523-- round robin fashion. The key difference is that 'WAsyncT' might wait for the
524-- consumer demand before it executes the tasks whereas 'ParallelT' starts
525-- executing all the tasks immediately without waiting for the consumer demand.
526-- For 'WAsyncT' the 'maxThreads' limit applies whereas for 'ParallelT' it does
527-- not apply. In other words, 'WAsyncT' can be lazy whereas 'ParallelT' is
528-- strict.
529--
530-- 'ParallelT' is useful for cases when the streams are required to be
531-- evaluated simultaneously irrespective of how the consumer consumes them e.g.
532-- when we want to race two tasks and want to start both strictly at the same
533-- time or if we have timers in the parallel tasks and our results depend on
534-- the timers being started at the same time. If we do not have such
535-- requirements then 'AsyncT' or 'AheadT' are recommended as they can be more
536-- efficient than 'ParallelT'.
537--
538-- @
539-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
540-- @
541-- @
542-- [1,3,2,4]
543-- @
544--
545-- When streams with more than one element are merged, it yields whichever
546-- stream yields first without any bias, unlike the 'Async' style streams.
547--
548-- Any exceptions generated by a constituent stream are propagated to the
549-- output stream. The output and exceptions from a single stream are guaranteed
550-- to arrive in the same order in the resulting stream as they were generated
551-- in the input stream. However, the relative ordering of elements from
552-- different streams in the resulting stream can vary depending on scheduling
553-- and generation delays.
554--
555-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations
556-- of the loop concurrently.
557--
558-- @
559-- import "Streamly"
560-- import qualified "Streamly.Prelude" as S
561-- import Control.Concurrent
562--
563-- main = 'drain' . 'parallely' $ do
564-- n <- return 3 \<\> return 2 \<\> return 1
565-- S.yieldM $ do
566-- threadDelay (n * 1000000)
567-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
568-- @
569-- @
570-- ThreadId 40: Delay 1
571-- ThreadId 39: Delay 2
572-- ThreadId 38: Delay 3
573-- @
574--
575-- Note that parallel composition can only combine a finite number of
576-- streams as it needs to retain state for each unfinished stream.
577--
578-- /Since: 0.7.0 (maxBuffer applies to ParallelT streams)/
579--
580-- /Since: 0.1.0/
581newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
582 deriving (MonadTrans)
584-- | A parallely composing IO stream of elements of type @a@.
585-- See 'ParallelT' documentation for more details.
586--
587-- @since 0.2.0
588type Parallel = ParallelT IO
590-- | Fix the type of a polymorphic stream as 'ParallelT'.
591--
592-- @since 0.1.0
593parallely :: IsStream t => ParallelT m a -> t m a
594parallely = adapt
596instance IsStream ParallelT where
597 toStream = getParallelT
598 fromStream = ParallelT
600 {-# INLINE consM #-}
601 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
602 consM = consMParallel
604 {-# INLINE (|:) #-}
605 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
606 (|:) = consM
608------------------------------------------------------------------------------
609-- Semigroup
610------------------------------------------------------------------------------
612{-# INLINE mappendParallel #-}
613{-# SPECIALIZE mappendParallel :: ParallelT IO a -> ParallelT IO a -> ParallelT IO a #-}
614mappendParallel :: MonadAsync m => ParallelT m a -> ParallelT m a -> ParallelT m a
615mappendParallel m1 m2 = fromStream $ parallel (toStream m1) (toStream m2)
617instance MonadAsync m => Semigroup (ParallelT m a) where
618 (<>) = mappendParallel
620------------------------------------------------------------------------------
621-- Monoid
622------------------------------------------------------------------------------
624instance MonadAsync m => Monoid (ParallelT m a) where
625 mempty = K.nil
626 mappend = (<>)
628------------------------------------------------------------------------------
629-- Monad
630------------------------------------------------------------------------------
632{-# INLINE bindParallel #-}
633{-# SPECIALIZE bindParallel :: ParallelT IO a -> (a -> ParallelT IO b) -> ParallelT IO b #-}
634bindParallel :: MonadAsync m => ParallelT m a -> (a -> ParallelT m b) -> ParallelT m b
635bindParallel m f = fromStream $ K.bindWith parallel (K.adapt m) (\a -> K.adapt $ f a)
637instance MonadAsync m => Monad (ParallelT m) where
638 return = pure
639 (>>=) = bindParallel
641------------------------------------------------------------------------------
642-- Other instances
643------------------------------------------------------------------------------
645MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
646MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)