b8ced4d · streamly · git.iko.soy

78import Control.Monad.Reader.Class (MonadReader(..))

79import Control.Monad.Trans.Class (MonadTrans (lift))

80import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)

81−import Data.Atomics (atomicModifyIORefCAS,

81+import Data.Atomics (casIORef, readForCAS, peekTicket

82− atomicModifyIORefCAS_)

82+ ,atomicModifyIORefCAS_)

83import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,

84 tryPopR, nullQ)

85import Data.Functor (void)

86import Data.IORef (IORef, modifyIORef, newIORef,

87− readIORef)

87+ readIORef, atomicModifyIORef)

88−import Data.Maybe (isNothing)

88+import Data.Maybe (isNothing, fromJust)

89import Data.Semigroup (Semigroup(..))

90import Data.Set (Set)

91import qualified Data.Set as S

158 , svarStyle :: SVarStyle

159 }

160

161+-- Slightly faster version of CAS. Gained some improvement by avoiding the use

162+-- of "evaluate" because we know we do not have exceptions in fn.

163+{-# INLINE atomicModifyIORefCAS #-}

164+atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b

165+atomicModifyIORefCAS ref fn = do

166+ tkt <- readForCAS ref

167+ loop tkt retries

168+

169+ where

170+

171+ retries = 30 :: Int

172+ loop _ 0 = atomicModifyIORef ref fn

173+ loop old tries = do

174+ let (new, result) = fn $ peekTicket old

175+ (success, tkt) <- casIORef ref old new

176+ if success

177+ then return result

178+ else loop tkt (tries - 1)

179+

161------------------------------------------------------------------------------

180------------------------------------------------------------------------------

162-- The stream type

181-- The stream type

163------------------------------------------------------------------------------

182------------------------------------------------------------------------------

376

395

377-- If present then delete else add. This takes care of out of order add and

396-- If present then delete else add. This takes care of out of order add and

378-- delete i.e. a delete arriving before we even added a thread.

397-- delete i.e. a delete arriving before we even added a thread.

398+-- This occurs when the forked thread is done even before the 'addThread' right

399+-- after the fork is executed.

379{-# INLINE modifyThread #-}

400{-# INLINE modifyThread #-}

380modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()

401modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()

381modifyThread sv tid = do

402modifyThread sv tid = do

382 changed <- liftIO $ atomicModifyIORefCAS (runningThreads sv) $ \old ->

403 changed <- liftIO $ atomicModifyIORefCAS (runningThreads sv) $ \old ->

383 if (S.member tid old)

404 if (S.member tid old)

384− then let new = (S.delete tid old) in (new, null new)

405+ then let new = (S.delete tid old) in (new, new)

385− else let new = (S.insert tid old) in (new, null old)

406+ else let new = (S.insert tid old) in (new, old)

386− if changed

407+ if null changed

387 then liftIO $ void $ tryPutMVar (doorBell sv) True

408 then liftIO $ void $ tryPutMVar (doorBell sv) True

388 else return ()

409 else return ()

389

410

631-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it

652-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it

632-- into just the number of CPUs.

653-- into just the number of CPUs.

633

654

634−{-# NOINLINE withNewSVar2 #-}

635−withNewSVar2 :: MonadParallel m

636− => SVarStyle -> Stream m a -> Stream m a -> Stream m a

637−withNewSVar2 style m1 m2 = Stream $ \_ stp sng yld -> do

638− sv <- newStreamVar2 style m1 m2

639− (runStream (fromStreamVar sv)) Nothing stp sng yld

640−

641-- | Join two computations on the currently running 'SVar' queue for concurrent

655-- | Join two computations on the currently running 'SVar' queue for concurrent

642-- execution. When we are using parallel composition, an SVar is passed around

656-- execution. When we are using parallel composition, an SVar is passed around

643-- as a state variable. We try to schedule a new parallel computation on the

657-- as a state variable. We try to schedule a new parallel computation on the

671 case svr of

685 case svr of

672 Just sv | svarStyle sv == style ->

686 Just sv | svarStyle sv == style ->

673 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld

687 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld

674− _ -> (runStream (withNewSVar2 style m1 m2)) Nothing stp sng yld

688+ _ -> do

689+ sv <- newStreamVar1 style (concurrently m1 m2)

690+ (runStream (fromStreamVar sv)) Nothing stp sng yld

691+ where

692+ concurrently ma mb = Stream $ \svr stp sng yld -> do

693+ liftIO $ (enqueue (fromJust svr)) mb

694+ (runStream ma) svr stp sng yld

675

695

676{-# INLINE joinStreamVarPar #-}

696{-# INLINE joinStreamVarPar #-}

677joinStreamVarPar :: MonadParallel m

697joinStreamVarPar :: MonadParallel m