diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -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 diff --git a/scheduler.cabal b/scheduler.cabal --- a/scheduler.cabal +++ b/scheduler.cabal @@ -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 diff --git a/src/Control/Scheduler.hs b/src/Control/Scheduler.hs --- a/src/Control/Scheduler.hs +++ b/src/Control/Scheduler.hs @@ -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 diff --git a/src/Control/Scheduler/Computation.hs b/src/Control/Scheduler/Computation.hs --- a/src/Control/Scheduler/Computation.hs +++ b/src/Control/Scheduler/Computation.hs @@ -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 diff --git a/src/Control/Scheduler/Internal.hs b/src/Control/Scheduler/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Control/Scheduler/Internal.hs @@ -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 +-- 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" diff --git a/src/Control/Scheduler/Queue.hs b/src/Control/Scheduler/Queue.hs --- a/src/Control/Scheduler/Queue.hs +++ b/src/Control/Scheduler/Queue.hs @@ -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 = diff --git a/tests/Control/SchedulerSpec.hs b/tests/Control/SchedulerSpec.hs --- a/tests/Control/SchedulerSpec.hs +++ b/tests/Control/SchedulerSpec.hs @@ -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