scheduler 1.3.0 → 1.4.0
raw patch · 8 files changed
+339/−79 lines, 8 filesdep +mwc-randomdep +primitivedep +vector
Dependencies added: mwc-random, primitive, vector
Files
- CHANGELOG.md +15/−0
- README.md +3/−3
- scheduler.cabal +6/−3
- src/Control/Scheduler.hs +161/−61
- src/Control/Scheduler/Computation.hs +6/−5
- src/Control/Scheduler/Internal.hs +101/−0
- src/Control/Scheduler/Queue.hs +20/−5
- tests/Control/SchedulerSpec.hs +27/−2
CHANGELOG.md view
@@ -1,3 +1,18 @@+# 1.4.0++* Worker id has been promoted from `Int` to a `newtype` wrapper `WorkerId`.+* Addition of `SchedulerWS` and `WorkerStates` data types. As well as the+ related `MutexException`+* Functions that came along with stateful worker threads:+ * `initWorkerStates`+ * `workerStatesComp`+ * `scheduleWorkState`+ * `scheduleWorkState_`+ * `withSchedulerWS`+ * `withSchedulerWS_`+ * `unwrapSchedulerWS`+* Made internal modules accessible, but invisible.+ # 1.3.0 * Make sure internal `Scheduler` accessor functions are no longer exported, they only
README.md view
@@ -80,11 +80,11 @@ ```haskell λ> let scheduleId = (`scheduleWorkId` (\ i -> threadDelay 100000 >> pure i)) λ> withScheduler (ParOn [4,7,5]) $ \s -> scheduleId s >> scheduleId s >> scheduleId s-[4,7,5]+[WorkerId {getWorkerId = 0},WorkerId {getWorkerId = 1},WorkerId {getWorkerId = 2}] λ> withScheduler (ParN 3) $ \s -> scheduleId s >> scheduleId s >> scheduleId s-[1,2,0]+[WorkerId {getWorkerId = 1},WorkerId {getWorkerId = 2},WorkerId {getWorkerId = 0}] λ> withScheduler (ParN 3) $ \s -> scheduleId s >> scheduleId s >> scheduleId s-[0,1,2]+[WorkerId {getWorkerId = 0},WorkerId {getWorkerId = 1},WorkerId {getWorkerId = 2}] ``` ### Exceptions
scheduler.cabal view
@@ -1,5 +1,5 @@ name: scheduler-version: 1.3.0+version: 1.4.0 synopsis: Work stealing scheduler. description: A work stealing scheduler that is primarly developed for [massiv](https://github.com/lehins/massiv) array librarry, but it is general enough to be useful for any computation that fits the model of few workers many jobs. homepage: https://github.com/lehins/haskell-scheduler@@ -23,14 +23,15 @@ library hs-source-dirs: src exposed-modules: Control.Scheduler-- other-modules: Control.Scheduler.Computation+ , Control.Scheduler.Internal+ , Control.Scheduler.Computation , Control.Scheduler.Queue build-depends: base >= 4.9 && < 5 , atomic-primops , deepseq , exceptions , unliftio-core+ , primitive default-language: Haskell2010 ghc-options: -Wall@@ -58,8 +59,10 @@ main-is: doctests.hs build-depends: base , doctest >=0.15+ , mwc-random , scheduler , template-haskell+ , vector default-language: Haskell2010
src/Control/Scheduler.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE BangPatterns #-}+{-# LANGUAGE CPP #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE PatternSynonyms #-}@@ -13,21 +14,31 @@ -- Portability : non-portable -- module Control.Scheduler- (- -- * Scheduler+ ( -- * Scheduler Scheduler- , numWorkers+ , SchedulerWS+ , trivialScheduler_+ , withScheduler+ , withScheduler_+ , withSchedulerWS+ , withSchedulerWS_+ , unwrapSchedulerWS+ -- * Scheduling computation , scheduleWork , scheduleWork_ , scheduleWorkId , scheduleWorkId_+ , scheduleWorkState+ , scheduleWorkState_ , terminate , terminate_ , terminateWith- -- * Initialize Scheduler- , withScheduler- , withScheduler_- , trivialScheduler_+ -- * Workers+ , WorkerId(..)+ , WorkerStates+ , numWorkers+ , workerStatesComp+ , initWorkerStates -- * Computation strategies , Comp(..) , getCompWorkers@@ -39,6 +50,7 @@ , traverse_ -- * Exceptions -- $exceptions+ , MutexException(..) ) where import Control.Concurrent@@ -46,33 +58,145 @@ import Control.Monad import Control.Monad.IO.Unlift import Control.Scheduler.Computation+import Control.Scheduler.Internal import Control.Scheduler.Queue import Data.Atomics (atomicModifyIORefCAS, atomicModifyIORefCAS_) import qualified Data.Foldable as F (foldl', traverse_) import Data.IORef import Data.Maybe (catMaybes)+import Data.Primitive.Array import Data.Traversable+#if !MIN_VERSION_primitive(0,6,2)+import Control.Monad.ST+#endif +-- | Get the underlying `Scheduler`, which cannot access `WorkerStates`.+--+-- @since 1.4.0+unwrapSchedulerWS :: SchedulerWS s m a -> Scheduler m a+unwrapSchedulerWS = _getScheduler -data Jobs m a = Jobs- { jobsNumWorkers :: {-# UNPACK #-} !Int- , jobsQueue :: !(JQueue m a)- , jobsCountRef :: !(IORef Int)- } --- | Main type for scheduling work. See `withScheduler` or `withScheduler_` for ways to construct--- and use this data type.+-- | Initialize a separate state for each worker. ----- @since 1.0.0-data Scheduler m a = Scheduler- { _numWorkers :: {-# UNPACK #-} !Int- , _scheduleWorkId :: (Int -> m a) -> m ()- , _terminate :: a -> m a- , _terminateWith :: a -> m a- }+-- @since 1.4.0+initWorkerStates :: MonadIO m => Comp -> (WorkerId -> m s) -> m (WorkerStates s)+initWorkerStates comp initState = do+ nWorkers <- getCompWorkers comp+ workerStates <- mapM (initState . WorkerId) [0 .. nWorkers - 1]+ mutex <- liftIO $ newIORef False+ pure+ WorkerStates+ { _workerStatesComp = comp+ , _workerStatesArray = arrayFromListN nWorkers workerStates+ , _workerStatesMutex = mutex+ } +arrayFromListN :: Int -> [a] -> Array a+#if MIN_VERSION_primitive(0,6,2)+arrayFromListN = fromListN+#else+-- Modified copy from primitive-0.7.0.0+arrayFromListN n l =+ runST $ do+ ma <- newArray n (error "initWorkerStates: uninitialized element")+ let go !ix [] =+ if ix == n+ then return ()+ else error "initWorkerStates: list length less than specified size"+ go !ix (x:xs) =+ if ix < n+ then do+ writeArray ma ix x+ go (ix + 1) xs+ else error "initWorkerStates: list length greater than specified size"+ go 0 l+ unsafeFreezeArray ma+#endif --- ^ Get the number of workers. Will mainly depend on the computation strategy and/or number of+-- | Get the computation strategy the states where initialized with.+--+-- @since 1.4.0+workerStatesComp :: WorkerStates s -> Comp+workerStatesComp = _workerStatesComp++-- | Run a scheduler with stateful workers. Throws `MutexException` if an attempt is made+-- to concurrently use the same `WorkerStates` with another `SchedulerWS`.+--+-- ==== __Examples__+--+-- A good example of using stateful workers would be generation of random number in+-- parallel. A lof of times random number generators are not gonna be thread safe, so we+-- can work around this problem, by using a separate stateful generator for each of the+-- workers.+--+-- >>> import Control.Monad as M ((>=>), replicateM)+-- >>> import Control.Concurrent (yield, threadDelay)+-- >>> import Data.List (sort)+-- >>> -- ^ Above imports are used to make sure output is deterministic, which is needed for doctest+-- >>> import System.Random.MWC as MWC+-- >>> import Data.Vector.Unboxed as V (singleton)+-- >>> states <- initWorkerStates (ParN 4) (MWC.initialize . V.singleton . fromIntegral . getWorkerId)+-- >>> let scheduleGen scheduler = scheduleWorkState scheduler (MWC.uniform >=> \r -> yield >> threadDelay 200000 >> pure r)+-- >>> sort <$> withSchedulerWS states (M.replicateM 4 . scheduleGen) :: IO [Double]+-- [0.21734983682025255,0.5000843862105709,0.5759825622603018,0.8587171114177893]+-- >>> sort <$> withSchedulerWS states (M.replicateM 4 . scheduleGen) :: IO [Double]+-- [2.3598617298033475e-2,9.949679290089553e-2,0.38223134248645885,0.7408640677124702]+--+-- In the above example we use four different random number generators from+-- [`mwc-random`](https://www.stackage.org/package/mwc-random) in order to generate 4+-- numbers, all in separate threads. The subsequent call to the `withSchedulerWS` function+-- with the same @states@ is allowed to reuse the same generators, thus avoiding expensive+-- initialization.+--+-- /Side note/ - The example presented was crafted with slight trickery in order to+-- guarantee that the output is deterministic, so if you run instructions exactly the same+-- way in GHCI you will get the exact same output. Non-determinism comes from thread+-- scheduling, rather than from random number generator, because we use exactly the same+-- seed for each worker, but workers run concurrently. Exact output is not really needed,+-- except for the doctests to pass.+--+-- @since 1.4.0+withSchedulerWS :: MonadUnliftIO m => WorkerStates s -> (SchedulerWS s m a -> m b) -> m [a]+withSchedulerWS states action =+ withRunInIO $ \run -> bracket lockState unlockState (run . runSchedulerWS)+ where+ mutex = _workerStatesMutex states+ lockState = atomicModifyIORef' mutex ((,) True)+ unlockState wasLocked+ | wasLocked = pure ()+ | otherwise = writeIORef mutex False+ runSchedulerWS isLocked+ | isLocked = liftIO $ throwIO MutexException+ | otherwise =+ withScheduler (_workerStatesComp states) $ \scheduler ->+ action (SchedulerWS states scheduler)++-- | Run a scheduler with stateful workers, while discarding computation results.+--+-- @since 1.4.0+withSchedulerWS_ :: MonadUnliftIO m => WorkerStates s -> (SchedulerWS s m () -> m b) -> m ()+withSchedulerWS_ states = void . withSchedulerWS states+++-- | Schedule a job that will get a worker state passed as an argument+--+-- @since 1.4.0+scheduleWorkState :: SchedulerWS s m a -> (s -> m a) -> m ()+scheduleWorkState schedulerS withState =+ scheduleWorkId (_getScheduler schedulerS) $ \(WorkerId i) ->+ withState (indexArray (_workerStatesArray (_workerStates schedulerS)) i)++-- | Same as `scheduleWorkState`, but dont' keep the result of computation.+--+-- @since 1.4.0+scheduleWorkState_ :: SchedulerWS s m () -> (s -> m ()) -> m ()+scheduleWorkState_ schedulerS withState =+ scheduleWorkId_ (_getScheduler schedulerS) $ \(WorkerId i) ->+ withState (indexArray (_workerStatesArray (_workerStates schedulerS)) i)+++-- | Get the number of workers. Will mainly depend on the computation strategy and/or number of -- capabilities you have. Related function is `getCompWorkers`. -- -- @since 1.0.0@@ -84,7 +208,7 @@ -- jobs. Argument supplied to the job will be the id of the worker doing the job. -- -- @since 1.2.0-scheduleWorkId :: Scheduler m a -> (Int -> m a) -> m ()+scheduleWorkId :: Scheduler m a -> (WorkerId -> m a) -> m () scheduleWorkId =_scheduleWorkId -- | As soon as possible try to terminate any computation that is being performed by all workers@@ -126,7 +250,7 @@ -- | Same as `scheduleWorkId`, but only for a `Scheduler` that doesn't keep the results. -- -- @since 1.2.0-scheduleWorkId_ :: Scheduler m () -> (Int -> m ()) -> m ()+scheduleWorkId_ :: Scheduler m () -> (WorkerId -> m ()) -> m () scheduleWorkId_ = _scheduleWorkId -- | Similar to `terminate`, but for a `Scheduler` that does not keep any results of computation.@@ -138,32 +262,24 @@ terminate_ = (`_terminateWith` ()) -- | The most basic scheduler that simply runs the task instead of scheduling it. Early termination--- requests are simply ignored.+-- requests are bluntly ignored. -- -- @since 1.1.0 trivialScheduler_ :: Applicative f => Scheduler f () trivialScheduler_ = Scheduler { _numWorkers = 1- , _scheduleWorkId = \f -> f 0+ , _scheduleWorkId = \f -> f (WorkerId 0) , _terminate = const $ pure () , _terminateWith = const $ pure () } -data SchedulerOutcome a- = SchedulerFinished- | SchedulerTerminatedEarly ![a]- | SchedulerWorkerException WorkerException--- -- | This is generally a faster way to traverse while ignoring the result rather than using `mapM_`. -- -- @since 1.0.0 traverse_ :: (Applicative f, Foldable t) => (a -> f ()) -> t a -> f () traverse_ f = F.foldl' (\c a -> c *> f a) (pure ()) - -- | Map an action over each element of the `Traversable` @t@ acccording to the supplied computation -- strategy. --@@ -203,16 +319,17 @@ withScheduler_ comp $ \s -> scheduleWork s $ replicateM_ n (scheduleWork s $ void f) -scheduleJobs :: MonadIO m => Jobs m a -> (Int -> m a) -> m ()+scheduleJobs :: MonadIO m => Jobs m a -> (WorkerId -> m a) -> m () scheduleJobs = scheduleJobsWith mkJob -- | Similarly to `scheduleWork`, but ignores the result of computation, thus having less overhead. -- -- @since 1.0.0-scheduleJobs_ :: MonadIO m => Jobs m a -> (Int -> m b) -> m ()+scheduleJobs_ :: MonadIO m => Jobs m a -> (WorkerId -> m b) -> m () scheduleJobs_ = scheduleJobsWith (\job -> pure (Job_ (void . job))) -scheduleJobsWith :: MonadIO m => ((Int -> m b) -> m (Job m a)) -> Jobs m a -> (Int -> m b) -> m ()+scheduleJobsWith ::+ MonadIO m => ((WorkerId -> m b) -> m (Job m a)) -> Jobs m a -> (WorkerId -> m b) -> m () scheduleJobsWith mkJob' jobs action = do liftIO $ atomicModifyIORefCAS_ (jobsCountRef jobs) (+ 1) job <-@@ -244,7 +361,7 @@ -- | Runs the worker until the job queue is exhausted, at which point it will execute the final task -- of retirement and return runWorker :: MonadIO m =>- Int+ WorkerId -> JQueue m a -> m () -- ^ Action to run upon retirement -> m ()@@ -307,7 +424,7 @@ withSchedulerInternal :: MonadUnliftIO m => Comp -- ^ Computation strategy- -> (Jobs m a -> (Int -> m a) -> m ()) -- ^ How to schedule work+ -> (Jobs m a -> (WorkerId -> m a) -> m ()) -- ^ How to schedule work -> (JQueue m a -> m [Maybe a]) -- ^ How to collect results -> ([a] -> [a]) -- ^ Adjust results in some way -> (Scheduler m a -> m b)@@ -346,18 +463,18 @@ when (jc == 0) $ scheduleJobs_ jobs (\_ -> pure ()) let spawnWorkersWith fork ws = withRunInIO $ \run ->- forM ws $ \w ->- fork w $ \unmask ->+ forM (zip [0 ..] ws) $ \(wId, on) ->+ fork on $ \unmask -> catch- (unmask $ run $ runWorker w jobsQueue onRetire)+ (unmask $ run $ runWorker wId jobsQueue onRetire) (run . handleWorkerException jobsQueue workDoneMVar jobsNumWorkers) spawnWorkers = case comp of Seq -> return [] -- \ no need to fork threads for a sequential computation- Par -> spawnWorkersWith forkOnWithUnmask [0 .. jobsNumWorkers - 1]+ Par -> spawnWorkersWith forkOnWithUnmask [1 .. jobsNumWorkers] ParOn ws -> spawnWorkersWith forkOnWithUnmask ws- ParN _ -> spawnWorkersWith (\_ -> forkIOWithUnmask) [0 .. jobsNumWorkers - 1]+ ParN _ -> spawnWorkersWith (\_ -> forkIOWithUnmask) [1 .. jobsNumWorkers] terminateWorkers = liftIO . traverse_ (`throwTo` SomeAsyncException WorkerTerminateException) doWork tids = do when (comp == Seq) $ runWorker 0 jobsQueue onRetire@@ -388,23 +505,6 @@ -- / Do the co-worker cleanup retireWorkersN jQueue (nWorkers - 1) ---- | This exception should normally be never seen in the wild and is for internal use only.-newtype WorkerException =- WorkerException SomeException- -- ^ One of workers experienced an exception, main thread will receive the same `SomeException`.- deriving (Show)--instance Exception WorkerException--data WorkerTerminateException =- WorkerTerminateException- -- ^ When a brother worker dies of some exception, all the other ones will be terminated- -- asynchronously with this one.- deriving (Show)---instance Exception WorkerTerminateException -- Copy from unliftio: safeBracketOnError :: MonadUnliftIO m => m a -> (a -> m b) -> (a -> m c) -> m c
src/Control/Scheduler/Computation.hs view
@@ -1,3 +1,4 @@+{-# OPTIONS_HADDOCK hide, not-home #-} {-# LANGUAGE CPP #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE PatternSynonyms #-}@@ -83,11 +84,11 @@ case x of Seq -> y Par -> Par- ParN 0 -> ParN 0+ Par' -> Par' ParOn xs -> case y of Par -> Par- ParN 0 -> ParN 0+ Par' -> Par' ParOn ys -> ParOn (xs <> ys) _ -> x ParN n1 ->@@ -95,7 +96,7 @@ Seq -> x Par -> Par ParOn _ -> y- ParN 0 -> y+ Par' -> y ParN n2 -> ParN (max n1 n2) {-# NOINLINE joinComp #-} @@ -109,7 +110,7 @@ -- -- /Note/ - If at any point during program execution global number of capabilities gets -- changed with `Control.Concurrent.setNumCapabilities`, it will have no affect on this--- function, unless it hasn't yet been called with `Par` or `ParN` 0 arguments.+-- function, unless it hasn't yet been called with `Par` or `Par'` arguments. -- -- @since 1.1.0 getCompWorkers :: MonadIO m => Comp -> m Int@@ -118,5 +119,5 @@ Seq -> return 1 Par -> liftIO (readIORef numCapsRef) ParOn ws -> return $ length ws- ParN 0 -> liftIO (readIORef numCapsRef)+ Par' -> liftIO (readIORef numCapsRef) ParN n -> return $ fromIntegral n
+ src/Control/Scheduler/Internal.hs view
@@ -0,0 +1,101 @@+{-# OPTIONS_HADDOCK hide, not-home #-}+{-# LANGUAGE Unsafe #-}+-- |+-- Module : Control.Scheduler.Internal+-- Copyright : (c) Alexey Kuleshevich 2018-2019+-- License : BSD3+-- Maintainer : Alexey Kuleshevich <lehins@yandex.ru>+-- Stability : experimental+-- Portability : non-portable+--+module Control.Scheduler.Internal+ ( Scheduler(..)+ , WorkerStates(..)+ , SchedulerWS(..)+ , Jobs(..)+ , SchedulerOutcome(..)+ , WorkerException(..)+ , WorkerTerminateException(..)+ , MutexException(..)+ ) where++import Control.Exception+import Control.Scheduler.Computation+import Control.Scheduler.Queue+import Data.IORef+import Data.Primitive.Array+++data Jobs m a = Jobs+ { jobsNumWorkers :: {-# UNPACK #-} !Int+ , jobsQueue :: !(JQueue m a)+ , jobsCountRef :: !(IORef Int)+ }++-- | Main type for scheduling work. See `Control.Scheduler.withScheduler` or+-- `Control.Scheduler.withScheduler_` for ways to construct and use this data type.+--+-- @since 1.0.0+data Scheduler m a = Scheduler+ { _numWorkers :: {-# UNPACK #-} !Int+ , _scheduleWorkId :: (WorkerId -> m a) -> m ()+ , _terminate :: a -> m a+ , _terminateWith :: a -> m a+ }++-- | This is a wrapper around `Scheduler`, but it also keeps a separate state for each+-- individual worker. See `Control.Scheduler.withSchedulerWS` or+-- `Control.Scheduler.withSchedulerWS_` for ways to construct and use this data type.+--+-- @since 1.4.0+data SchedulerWS s m a = SchedulerWS+ { _workerStates :: !(WorkerStates s)+ , _getScheduler :: !(Scheduler m a)+ }++-- | Each worker is capable of keeping it's own state, that can be share for different+-- schedulers, but not at the same time. In other words using the same `WorkerStates` on+-- `Control.Scheduler.withSchedulerS` concurrently will result in an error. Can be initialized with+-- `Control.Scheduler.initWorkerStates`+--+-- @since 1.4.0+data WorkerStates s = WorkerStates+ { _workerStatesComp :: !Comp+ , _workerStatesArray :: !(Array s)+ , _workerStatesMutex :: !(IORef Bool)+ }+++data SchedulerOutcome a+ = SchedulerFinished+ | SchedulerTerminatedEarly ![a]+ | SchedulerWorkerException WorkerException+++-- | This exception should normally be never seen in the wild and is for internal use only.+newtype WorkerException =+ WorkerException SomeException+ -- ^ One of workers experienced an exception, main thread will receive the same `SomeException`.+ deriving (Show)++instance Exception WorkerException++data WorkerTerminateException =+ WorkerTerminateException+ -- ^ When a brother worker dies of some exception, all the other ones will be terminated+ -- asynchronously with this one.+ deriving (Show)+++instance Exception WorkerTerminateException++-- | Exception that gets thrown whenever concurrent access is attempted to the `WorkerStates`+--+-- @since 1.4.0+data MutexException =+ MutexException+ deriving (Eq, Show)++instance Exception MutexException where+ displayException MutexException =+ "MutexException: WorkerStates cannot be used at the same time by different schedulers"
src/Control/Scheduler/Queue.hs view
@@ -1,4 +1,6 @@+{-# OPTIONS_HADDOCK hide, not-home #-} {-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} -- | -- Module : Control.Scheduler.Queue -- Copyright : (c) Alexey Kuleshevich 2018-2019@@ -12,11 +14,11 @@ Job(Retire, Job_) , mkJob , JQueue+ , WorkerId(..) , newJQueue , pushJQueue , popJQueue , readResults- -- * Tools ) where import Control.Concurrent.MVar@@ -25,6 +27,9 @@ import Data.Atomics (atomicModifyIORefCAS) import Data.IORef ++-- | A blocking unbounded queue that keeps the jobs in FIFO order and the results IORefs+-- in reversed data Queue m a = Queue { qQueue :: ![Job m a] , qStack :: ![Job m a]@@ -33,6 +38,16 @@ } +-- | A unique id for the worker in the `Control.Scheduler.Scheduler` context. It will+-- always be a number from @0@ up to, but not including, the number of workers a scheduler+-- has, which in turn can always be determined with `Control.Scheduler.numWorkers` function.+--+-- @since 1.4.0+newtype WorkerId = WorkerId+ { getWorkerId :: Int+ } deriving (Show, Eq, Ord, Enum, Num)++ popQueue :: Queue m a -> Maybe (Job m a, Queue m a) popQueue queue = case qQueue queue of@@ -43,12 +58,12 @@ y:ys -> Just (y, queue {qQueue = ys, qStack = []}) data Job m a- = Job !(IORef (Maybe a)) (Int -> m a)- | Job_ (Int -> m ())+ = Job !(IORef (Maybe a)) (WorkerId -> m a)+ | Job_ (WorkerId -> m ()) | Retire -mkJob :: MonadIO m => (Int -> m a) -> m (Job m a)+mkJob :: MonadIO m => (WorkerId -> m a) -> m (Job m a) mkJob action = do resRef <- liftIO $ newIORef Nothing return $!@@ -85,7 +100,7 @@ , liftIO $ putMVar qBaton ())) -popJQueue :: MonadIO m => JQueue m a -> m (Maybe (Int -> m ()))+popJQueue :: MonadIO m => JQueue m a -> m (Maybe (WorkerId -> m ())) popJQueue (JQueue jQueueRef) = liftIO inner where inner =
tests/Control/SchedulerSpec.hs view
@@ -4,7 +4,7 @@ ( spec ) where -import Control.Concurrent (killThread, myThreadId, threadDelay)+import Control.Concurrent (killThread, myThreadId, threadDelay, yield) import Control.Concurrent.MVar import Control.DeepSeq import qualified Control.Exception as EUnsafe@@ -202,7 +202,9 @@ comp /= Seq ==> concurrentProperty $ do ref <- newIORef True withScheduler_ comp $ \scheduler ->- scheduleWork_ scheduler (terminate_ scheduler >> threadDelay 10000 >> writeIORef ref False)+ scheduleWork_+ scheduler+ (terminate_ scheduler >> yield >> threadDelay 10000 >> writeIORef ref False) property <$> readIORef ref prop_FinishEarly :: Comp -> Property@@ -274,6 +276,7 @@ eRes' <- try $ traverseConcurrently_ comp f xs return (eRes === eRes') +-- TODO: fix the infinite property for single worker schedulers -- | Check if an element is in the list with an exception, where we know that list is infinite and -- element is part of that list. prop_TraverseConcurrentlyInfinite_ :: Comp -> [Int] -> Int -> Property@@ -287,6 +290,25 @@ eRes' <- try $ traverseConcurrently_ comp f xs' return (eRes === eRes') ++prop_ReturnsState :: Comp -> Property+prop_ReturnsState comp = concurrentProperty $ do+ n <- getCompWorkers comp+ state <- initWorkerStates comp (pure . getWorkerId)+ ids <- withSchedulerWS state $ \ schedulerWS ->+ replicateM (numWorkers (unwrapSchedulerWS schedulerWS)) $+ scheduleWorkState schedulerWS $ \ s -> yield >> threadDelay 30000 >> pure s+ pure (sort ids === [0..n-1])++prop_MutexException :: Comp -> Property+prop_MutexException comp =+ assertExceptionIO (== MutexException) $ do+ state <- initWorkerStates comp (pure . getWorkerId)+ withSchedulerWS_ state $ \schedulerWS ->+ scheduleWorkState_ schedulerWS $ \_s -> withSchedulerWS_ state $ \_s' -> pure ()+++ spec :: Spec spec = do describe "Comp" $ do@@ -334,6 +356,9 @@ it "FinishEarlyWith" $ timed prop_FinishEarlyWith it "FinishBeforeStarting" $ timed prop_FinishBeforeStarting it "FinishWithBeforeStarting" $ timed prop_FinishWithBeforeStarting+ describe "WorkerState" $ do+ it "ReturnsState" $ timed prop_ReturnsState+ it "MutexException" $ timed prop_MutexException timed :: Testable prop => prop -> Property timed = within 2000000