e5421a4 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Implement isolation of concurrency policies

e5421a4Harendra Kumar9 years ago

src/Asyncly/AsyncT.hs+24−14

compare with master

unifiedsplit

@@ -97,6 +97,10 @@ data ChildEvent a =

97-- State threaded around the monad for thread management

98------------------------------------------------------------------------------

100+data Dimension = Conjunction | Disjunction deriving Eq

101+data SchedPolicy = LIFO | FIFO deriving Eq

102+data CtxType = CtxType Dimension SchedPolicy deriving Eq

103+

100data Context m a =

104data Context m a =

101 Context { outputQueue :: IORef [ChildEvent a]

105 Context { outputQueue :: IORef [ChildEvent a]

102 , doorBell :: MVar Bool -- wakeup mechanism for outQ

106 , doorBell :: MVar Bool -- wakeup mechanism for outQ

@@ -104,6 +108,7 @@ data Context m a =

104 , runqueue :: m ()

108 , runqueue :: m ()

105 , runningThreads :: IORef (Set ThreadId)

109 , runningThreads :: IORef (Set ThreadId)

106 , queueEmpty :: m Bool

110 , queueEmpty :: m Bool

111+ , ctxType :: CtxType

107 }

112 }

108

113

109-- | Represents a monadic stream of values of type 'a' resulting from actions

114-- | Represents a monadic stream of values of type 'a' resulting from actions

@@ -224,15 +229,15 @@ instance Monad m => Monad (AsyncT m) where

224(>->) = bindWith (<=>)

229(>->) = bindWith (<=>)

225

230

226-- | Execute a monadic action for each element in the stream, running nested

231-- | Execute a monadic action for each element in the stream, running nested

227−-- loops in parallel, but with a depth first preference.

232+-- loops in parallel, but giving preference to inner loops.

228(>>|) :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b

233(>>|) :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b

229−(>>|) = bindWith (<|)

234+(>>|) = bindWith (parallel (CtxType Conjunction LIFO))

230

235

231-- | Execute a monadic action for each element in the stream, running nested

236-- | Execute a monadic action for each element in the stream, running nested

232-- loops in parallel, but in a fairly interleaved manner i.e. both inner and

237-- loops in parallel, but in a fairly interleaved manner i.e. both inner and

233-- outer loops yield alternately.

238-- outer loops yield alternately.

234(>|>) :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b

239(>|>) :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b

235−(>|>) = bindWith (<|>)

240+(>|>) = bindWith (parallel (CtxType Conjunction FIFO))

236

241

237-- TBD Ideally we would want parallelism to be completely hidden from the user,

242-- TBD Ideally we would want parallelism to be completely hidden from the user,

238-- for example we can automatically decide where to start an action in parallel

243-- for example we can automatically decide where to start an action in parallel

@@ -408,8 +413,9 @@ pullWorker ctx = AsyncT $ \pctx stp yld -> do

408-- | Split the original computation in a pull-push pair. The original

413-- | Split the original computation in a pull-push pair. The original

409-- computation pulls from a Channel while m1 and m2 push to the channel.

414-- computation pulls from a Channel while m1 and m2 push to the channel.

410{-# NOINLINE pullFork #-}

415{-# NOINLINE pullFork #-}

416+pullFork :: MonadAsync m

411−pullFork :: MonadAsync m => AsyncT m a -> AsyncT m a -> Bool -> AsyncT m a

417+ => CtxType -> AsyncT m a -> AsyncT m a -> AsyncT m a

412−pullFork m1 m2 fifo = AsyncT $ \_ stp yld -> do

418+pullFork ct m1 m2 = AsyncT $ \_ stp yld -> do

413 ctx <- liftIO $ newContext

419 ctx <- liftIO $ newContext

414 pushWorker ctx >> (runAsyncT (pullWorker ctx)) Nothing stp yld

420 pushWorker ctx >> (runAsyncT (pullWorker ctx)) Nothing stp yld

415

421

@@ -420,8 +426,8 @@ pullFork m1 m2 fifo = AsyncT $ \_ stp yld -> do

420 outQMv <- newEmptyMVar

426 outQMv <- newEmptyMVar

421 running <- newIORef S.empty

427 running <- newIORef S.empty

422

428

423− case fifo of

429+ case ct of

424− True -> do

430+ CtxType _ FIFO -> do

425 q <- newQ

431 q <- newQ

426 pushL q m1 >> pushL q m2

432 pushL q m1 >> pushL q m2

427 let ctx =

433 let ctx =

@@ -431,9 +437,10 @@ pullFork m1 m2 fifo = AsyncT $ \_ stp yld -> do

431 , runqueue = runqueueFIFO ctx q

437 , runqueue = runqueueFIFO ctx q

432 , enqueue = pushL q

438 , enqueue = pushL q

433 , queueEmpty = liftIO $ nullQ q

439 , queueEmpty = liftIO $ nullQ q

440+ , ctxType = ct

434 }

441 }

435 in return ctx

442 in return ctx

436− False -> do

443+ CtxType _ LIFO -> do

437 q <- newIORef []

444 q <- newIORef []

438 enqueueLIFO q m2 >> enqueueLIFO q m1

445 enqueueLIFO q m2 >> enqueueLIFO q m1

439 let checkEmpty = liftIO (readIORef q) >>= return . null

446 let checkEmpty = liftIO (readIORef q) >>= return . null

@@ -444,6 +451,7 @@ pullFork m1 m2 fifo = AsyncT $ \_ stp yld -> do

444 , runqueue = runqueueLIFO ctx q

451 , runqueue = runqueueLIFO ctx q

445 , enqueue = enqueueLIFO q

452 , enqueue = enqueueLIFO q

446 , queueEmpty = checkEmpty

453 , queueEmpty = checkEmpty

454+ , ctxType = ct

447 }

455 }

448 in return ctx

456 in return ctx

449

457

@@ -481,11 +489,13 @@ pullFork m1 m2 fifo = AsyncT $ \_ stp yld -> do

481-- Note: This is designed to scale for right associated compositions,

489-- Note: This is designed to scale for right associated compositions,

482-- therefore always use a right fold for folding bigger structures.

490-- therefore always use a right fold for folding bigger structures.

483{-# INLINE parallel #-}

491{-# INLINE parallel #-}

484−parallel :: MonadAsync m => AsyncT m a -> AsyncT m a -> Bool -> AsyncT m a

492+parallel :: MonadAsync m => CtxType -> AsyncT m a -> AsyncT m a -> AsyncT m a

485−parallel m1 m2 fifo = AsyncT $ \ctx stp yld -> do

493+parallel ct m1 m2 = AsyncT $ \ctx stp yld -> do

486 case ctx of

494 case ctx of

487− Nothing -> (runAsyncT (pullFork m1 m2 fifo)) Nothing stp yld

495+ Nothing -> (runAsyncT (pullFork ct m1 m2)) Nothing stp yld

496+ Just c | ctxType c /= ct ->

497+ (runAsyncT (pullFork ct m1 m2)) Nothing stp yld

488− Just c -> liftIO ((enqueue c) m2) >> (runAsyncT m1) ctx stp yld

498+ Just c -> liftIO ((enqueue c) m2) >> (runAsyncT m1) ctx stp yld

489

499

490-- | `empty` represents an action that takes non-zero time to complete. Since

500-- | `empty` represents an action that takes non-zero time to complete. Since

491-- all actions take non-zero time, an `Alternative` composition ('<|>') is a

501-- all actions take non-zero time, an `Alternative` composition ('<|>') is a

@@ -496,7 +506,7 @@ instance MonadAsync m => Alternative (AsyncT m) where

496 empty = mempty

506 empty = mempty

497

507

498 {-# INLINE (<|>) #-}

508 {-# INLINE (<|>) #-}

499− m1 <|> m2 = parallel m1 m2 True

509+ (<|>) = parallel (CtxType Disjunction FIFO)

500

510

501------------------------------------------------------------------------------

511------------------------------------------------------------------------------

502-- Other monoidal compositions for parallel actions

512-- Other monoidal compositions for parallel actions

@@ -515,7 +525,7 @@ parAhead = undefined

515-- | Same as '<|'.

525-- | Same as '<|'.

516{-# INLINE parLeft #-}

526{-# INLINE parLeft #-}

517parLeft :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a

527parLeft :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a

518−parLeft m1 m2 = parallel m1 m2 False

528+parLeft = parallel (CtxType Disjunction LIFO)

519

529

520-- | Parallel composition similar to `<|>` except that it is left-biased

530-- | Parallel composition similar to `<|>` except that it is left-biased

521-- instead of being fair. Action on the left is likely to be given a chance to

531-- instead of being fair. Action on the left is likely to be given a chance to