diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,11 @@ +# 1.4.2 + +* Add `withTrivialScheduler` +* Add `Results` data type as well as corresponding functions: + * `withSchedulerR` + * `withSchedulerWSR` + * `withTrivialSchedulerR` + # 1.4.1 * Add functions: `replicateWork` diff --git a/scheduler.cabal b/scheduler.cabal --- a/scheduler.cabal +++ b/scheduler.cabal @@ -1,5 +1,5 @@ name: scheduler -version: 1.4.1 +version: 1.4.2 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 @@ -26,12 +26,12 @@ , Control.Scheduler.Internal , Control.Scheduler.Computation , Control.Scheduler.Queue - build-depends: base >= 4.9 && < 5 + build-depends: base >= 4.9 && < 5 , atomic-primops , deepseq , exceptions , unliftio-core - , primitive + , primitive >= 0.5.2.1 default-language: Haskell2010 ghc-options: -Wall @@ -45,6 +45,7 @@ , Control.SchedulerSpec build-depends: base , deepseq + , genvalidity-hspec , scheduler , hspec , QuickCheck diff --git a/src/Control/Scheduler.hs b/src/Control/Scheduler.hs --- a/src/Control/Scheduler.hs +++ b/src/Control/Scheduler.hs @@ -3,6 +3,7 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE PatternSynonyms #-} +{-# LANGUAGE RankNTypes #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} -- | @@ -17,12 +18,20 @@ ( -- * Scheduler Scheduler , SchedulerWS - , trivialScheduler_ + , Results(..) + -- ** Regular , withScheduler , withScheduler_ + , withSchedulerR + -- ** Stateful workers , withSchedulerWS , withSchedulerWS_ + , withSchedulerWSR , unwrapSchedulerWS + -- ** Trivial (no parallelism) + , trivialScheduler_ + , withTrivialScheduler + , withTrivialSchedulerR -- * Scheduling computation , scheduleWork , scheduleWork_ @@ -58,14 +67,16 @@ import Control.Exception import Control.Monad import Control.Monad.IO.Unlift +import Control.Monad.Primitive (PrimMonad) 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 qualified Data.Foldable as F (foldl', traverse_, toList) import Data.IORef import Data.Maybe (catMaybes) import Data.Primitive.Array +import Data.Primitive.MutVar import Data.Traversable #if !MIN_VERSION_primitive(0,6,2) import Control.Monad.ST @@ -121,6 +132,27 @@ workerStatesComp :: WorkerStates s -> Comp workerStatesComp = _workerStatesComp +withSchedulerWSInternal :: + MonadUnliftIO m + => (Comp -> (Scheduler m a -> t) -> m b) + -> WorkerStates s + -> (SchedulerWS s m a -> t) + -> m b +withSchedulerWSInternal withScheduler' 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. Throws `MutexException` if an attempt is made -- to concurrently use the same `WorkerStates` with another `SchedulerWS`. -- @@ -159,27 +191,22 @@ -- -- @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) +withSchedulerWS = withSchedulerWSInternal withScheduler -- | 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 +withSchedulerWS_ = withSchedulerWSInternal withScheduler_ +-- | Same as `withSchedulerWS`, except instead of a list it produces `Results`, which +-- allows for distinguishing between the ways computation was terminated. +-- +-- @since 1.4.2 +withSchedulerWSR :: MonadUnliftIO m => WorkerStates s -> (SchedulerWS s m a -> m b) -> m (Results a) +withSchedulerWSR = withSchedulerWSInternal withSchedulerR + -- | Schedule a job that will get a worker state passed as an argument -- -- @since 1.4.0 @@ -212,9 +239,10 @@ 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 --- managed by this scheduler and collect whatever results have been computed, with supplied --- element guaranteed to being the last one. +-- | As soon as possible try to terminate any computation that is being performed by all +-- workers managed by this scheduler and collect whatever results have been computed, with +-- supplied element guaranteed to being the last one. In case when `Results` is the return +-- type this function will cause the scheduler to produce `FinishedEarly` -- -- /Important/ - With `Seq` strategy this will not stop other scheduled tasks from being computed, -- although it will make sure their results are discarded. @@ -224,7 +252,9 @@ terminate = _terminate -- | Same as `terminate`, but returning a single element list containing the supplied --- argument. This can be very useful for parallel search algorithms. +-- argument. This can be very useful for parallel search algorithms. In case when +-- `Results` is the return type this function will cause the scheduler to produce +-- `FinishedEarlyWith` -- -- /Important/ - Same as with `terminate`, when `Seq` strategy is used, this will not prevent -- computation from continuing, but the scheduler will return only the result supplied to this @@ -260,9 +290,11 @@ -- -- @since 1.4.1 replicateWork :: Applicative m => Int -> Scheduler m a -> m a -> m () -replicateWork !n scheduler f - | n <= 0 = pure () - | otherwise = scheduleWork scheduler f *> replicateWork (pred n) scheduler f +replicateWork !n scheduler f = go n + where + go !k + | k <= 0 = pure () + | otherwise = scheduleWork scheduler f *> go (k - 1) -- | Similar to `terminate`, but for a `Scheduler` that does not keep any results of computation. -- @@ -285,6 +317,40 @@ } +-- | This trivial scheduler will behave in the same way as `withScheduler` with `Seq` +-- computation strategy, except it is restricted to `PrimMonad`, instead of `MonadUnliftIO`. +-- +-- @since 1.4.2 +withTrivialScheduler :: PrimMonad m => (Scheduler m a -> m b) -> m [a] +withTrivialScheduler action = F.toList <$> withTrivialSchedulerR action + +-- | This trivial scheduler will behave in a similar way as `withSchedulerR` with `Seq` +-- computation strategy, except it is restricted to `PrimMonad`, instead of +-- `MonadUnliftIO` and the work isn't scheduled, but rather computed immediately. +-- +-- @since 1.4.2 +withTrivialSchedulerR :: PrimMonad m => (Scheduler m a -> m b) -> m (Results a) +withTrivialSchedulerR action = do + resVar <- newMutVar [] + finResVar <- newMutVar Nothing + _ <- action $ Scheduler + { _numWorkers = 1 + , _scheduleWorkId = \f -> do + r <- f (WorkerId 0) + modifyMutVar' resVar (r:) + , _terminate = \r -> do + rs <- readMutVar resVar + writeMutVar finResVar (Just (FinishedEarly rs r)) + pure r + , _terminateWith = \r -> do + writeMutVar finResVar (Just (FinishedEarlyWith r)) + pure r + } + readMutVar finResVar >>= \case + Just rs -> pure $ reverseResults rs + Nothing -> Finished . Prelude.reverse <$> readMutVar resVar + + -- | This is generally a faster way to traverse while ignoring the result rather than using `mapM_`. -- -- @since 1.0.0 @@ -304,7 +370,7 @@ transList xs' = snd . mapAccumL withR xs' where withR (x:xs) _ = (xs, x) - withR _ _ = error "Impossible - Mismatched sizes" + withR _ _ = errorWithoutStackTrace "Impossible - Mismatched sizes" -- | Just like `traverseConcurrently`, but restricted to `Foldable` and discards the results of -- computation. @@ -418,9 +484,21 @@ -> (Scheduler m a -> m b) -- ^ Action that will be scheduling all the work. -> m [a] -withScheduler comp = withSchedulerInternal comp scheduleJobs readResults reverse +withScheduler comp = withSchedulerInternal comp scheduleJobs readResults (reverse . resultsToList) +-- | Same as `withScheduler`, except instead of a list it produces `Results`, which allows +-- for distinguishing between the ways computation was terminated. +-- +-- @since 1.4.2 +withSchedulerR :: + MonadUnliftIO m + => Comp -- ^ Computation strategy + -> (Scheduler m a -> m b) + -- ^ Action that will be scheduling all the work. + -> m (Results a) +withSchedulerR comp = withSchedulerInternal comp scheduleJobs readResults reverseResults + -- | Same as `withScheduler`, but discards results of submitted jobs. -- -- @since 1.0.0 @@ -430,17 +508,17 @@ -> (Scheduler m a -> m b) -- ^ Action that will be scheduling all the work. -> m () -withScheduler_ comp = void . withSchedulerInternal comp scheduleJobs_ (const (pure [])) id +withScheduler_ comp = void . withSchedulerInternal comp scheduleJobs_ (const (pure [])) (const ()) withSchedulerInternal :: MonadUnliftIO m => Comp -- ^ Computation strategy -> (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 + -> (Results a -> c) -- ^ Adjust results in some way -> (Scheduler m a -> m b) -- ^ Action that will be scheduling all the work. - -> m [a] + -> m c withSchedulerInternal comp submitWork collect adjust onScheduler = do jobsNumWorkers <- getCompWorkers comp sWorkersCounterRef <- liftIO $ newIORef jobsNumWorkers @@ -455,12 +533,14 @@ , _terminate = \a -> do mas <- collect jobsQueue - let as = adjust (a : catMaybes mas) - liftIO $ void $ tryPutMVar workDoneMVar $ SchedulerTerminatedEarly as + let as = catMaybes mas + liftIO $ + void $ tryPutMVar workDoneMVar $ SchedulerTerminatedEarly (FinishedEarly as a) pure a , _terminateWith = \a -> do - liftIO $ void $ tryPutMVar workDoneMVar $ SchedulerTerminatedEarly [a] + liftIO $ + void $ tryPutMVar workDoneMVar $ SchedulerTerminatedEarly (FinishedEarlyWith a) pure a } onRetire = @@ -489,18 +569,32 @@ terminateWorkers = liftIO . traverse_ (`throwTo` SomeAsyncException WorkerTerminateException) doWork tids = do when (comp == Seq) $ runWorker 0 jobsQueue onRetire - mExc <- liftIO $ readMVar workDoneMVar - -- \ wait for all worker to finish. If any one of them had a problem this MVar will - -- contain an exception + mExc <- liftIO (readMVar workDoneMVar) + -- \ wait for all worker to finish. If any one of the workers had a problem, then + -- this MVar will contain an exception case mExc of - SchedulerFinished -> adjust . catMaybes <$> collect jobsQueue + SchedulerFinished -> adjust . Finished . catMaybes <$> collect jobsQueue -- \ Now we are sure all workers have done their job we can safely read all of the -- IORefs with results - SchedulerTerminatedEarly as -> terminateWorkers tids >> pure as + SchedulerTerminatedEarly as -> terminateWorkers tids >> pure (adjust as) SchedulerWorkerException (WorkerException exc) -> liftIO $ throwIO exc -- \ Here we need to unwrap the legit worker exception and rethrow it, so the main thread -- will think like it's his own safeBracketOnError spawnWorkers terminateWorkers doWork + + +resultsToList :: Results a -> [a] +resultsToList = \case + Finished rs -> rs + FinishedEarly rs r -> r:rs + FinishedEarlyWith r -> [r] + + +reverseResults :: Results a -> Results a +reverseResults = \case + Finished rs -> Finished (reverse rs) + FinishedEarly rs r -> FinishedEarly (reverse rs) r + res -> res -- | Specialized exception handler for the work scheduler. diff --git a/src/Control/Scheduler/Internal.hs b/src/Control/Scheduler/Internal.hs --- a/src/Control/Scheduler/Internal.hs +++ b/src/Control/Scheduler/Internal.hs @@ -1,3 +1,4 @@ +{-# LANGUAGE LambdaCase #-} {-# OPTIONS_HADDOCK hide, not-home #-} {-# LANGUAGE Unsafe #-} -- | @@ -13,6 +14,7 @@ , WorkerStates(..) , SchedulerWS(..) , Jobs(..) + , Results(..) , SchedulerOutcome(..) , WorkerException(..) , WorkerTerminateException(..) @@ -25,7 +27,45 @@ import Data.IORef import Data.Primitive.Array +-- | Computated outcome of scheduled jobs. +-- +-- @since 1.4.2 +data Results a + = Finished ![a] + -- ^ Finished normally with all scheduled jobs completed + | FinishedEarly ![a] !a + -- ^ Finished early by the means of `Control.Scheduler.terminate`. + | FinishedEarlyWith !a + -- ^ Finished early by the means of `Control.Scheduler.terminateWith`. + deriving (Show, Read, Eq) +instance Functor Results where + fmap f = + \case + Finished xs -> Finished (fmap f xs) + FinishedEarly xs x -> FinishedEarly (fmap f xs) (f x) + FinishedEarlyWith x -> FinishedEarlyWith (f x) + +instance Foldable Results where + foldr f acc = + \case + Finished xs -> foldr f acc xs + FinishedEarly xs x -> foldr f (f x acc) xs + FinishedEarlyWith x -> f x acc + foldr1 f = + \case + Finished xs -> foldr1 f xs + FinishedEarly xs x -> foldr f x xs + FinishedEarlyWith x -> x + +instance Traversable Results where + traverse f = + \case + Finished xs -> Finished <$> traverse f xs + FinishedEarly xs x -> FinishedEarly <$> traverse f xs <*> f x + FinishedEarlyWith x -> FinishedEarlyWith <$> f x + + data Jobs m a = Jobs { jobsNumWorkers :: {-# UNPACK #-} !Int , jobsQueue :: !(JQueue m a) @@ -68,7 +108,7 @@ data SchedulerOutcome a = SchedulerFinished - | SchedulerTerminatedEarly ![a] + | SchedulerTerminatedEarly !(Results a) | SchedulerWorkerException WorkerException @@ -82,7 +122,7 @@ data WorkerTerminateException = WorkerTerminateException - -- ^ When a brother worker dies of some exception, all the other ones will be terminated + -- ^ When a co-worker dies of some exception, all the other ones will be terminated -- asynchronously with this one. deriving (Show) 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 @@ -45,7 +45,7 @@ -- @since 1.4.0 newtype WorkerId = WorkerId { getWorkerId :: Int - } deriving (Show, Eq, Ord, Enum, Num) + } deriving (Show, Read, Eq, Ord, Enum, Num) popQueue :: Queue m a -> Maybe (Job m a, Queue m a) diff --git a/tests/Control/SchedulerSpec.hs b/tests/Control/SchedulerSpec.hs --- a/tests/Control/SchedulerSpec.hs +++ b/tests/Control/SchedulerSpec.hs @@ -1,5 +1,6 @@ {-# LANGUAGE CPP #-} {-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TypeApplications #-} module Control.SchedulerSpec ( spec ) where @@ -11,15 +12,20 @@ import Control.Exception.Base (ArithException(DivideByZero), AsyncException(ThreadKilled)) import Control.Monad -import qualified Data.Foldable as F (traverse_) -import Control.Scheduler +import Control.Scheduler as S import Data.Bits (complement) +import qualified Data.Foldable as F (toList, traverse_) import Data.IORef -import Data.List (sort) +import Data.List (groupBy, sort, sortOn) import Test.Hspec import Test.QuickCheck import Test.QuickCheck.Function import Test.QuickCheck.Monadic +import Test.Validity.Eq +import Test.Validity.Functor +import Test.Validity.Monoid +import Test.Validity.Ord +import Test.Validity.Show import UnliftIO.Async import UnliftIO.Exception hiding (assert) #if !MIN_VERSION_base(4,11,0) @@ -89,15 +95,21 @@ where f' = pure . apply f --- prop_ReplicateM :: Comp -> [Int] -> Fun Int Int -> Property --- prop_ReplicateM comp xs f = --- monadicIO $ run $ do +replicateSeq :: (Int -> IO Int -> IO [Int]) -> Int -> Fun Int Int -> Property +replicateSeq justAs n f = + concurrentProperty $ do + iRef <- newIORef 0 + jRef <- newIORef 0 + let g ref = atomicModifyIORef' ref (\i -> (apply f i, i + 1)) + (===) <$> S.replicateConcurrently Seq n (g jRef) <*> justAs n (g iRef) --- (===) <$> traverse f' xs <*> traverseConcurrently comp f' xs --- where --- f' = pure . apply f +prop_ReplicateM :: Int -> Fun Int Int -> Property +prop_ReplicateM = replicateSeq replicateM +prop_ReplicateWorkSeq :: Int -> Fun Int Int -> Property +prop_ReplicateWorkSeq = replicateSeq (\ n g -> withScheduler Seq (\s -> replicateWork n s g)) + prop_ArbitraryCompNested :: [(Comp, Int)] -> Property prop_ArbitraryCompNested xs = concurrentProperty $ do @@ -205,25 +217,27 @@ scheduleWork_ scheduler (terminate_ scheduler >> yield >> threadDelay 10000 >> writeIORef ref False) - property <$> readIORef ref + counterexample "Scheduler did not terminate early" <$> readIORef ref prop_FinishEarly :: Comp -> Property prop_FinishEarly comp = concurrentProperty $ do - res <- - withScheduler comp $ \scheduler -> do - scheduleWork scheduler (pure (2 :: Int)) - scheduleWork scheduler (threadDelay 10000 >> terminate scheduler 3 >> pure 1) - pure (res === [2, 3]) + let scheduleJobs scheduler = do + scheduleWork scheduler (pure (2 :: Int)) + scheduleWork scheduler (threadDelay 10000 >> terminate scheduler 3 >> pure 1) + res <- withScheduler comp scheduleJobs + res' <- withSchedulerR comp scheduleJobs + pure (res === [2, 3] .&&. res' === FinishedEarly [2] 3) prop_FinishEarlyWith :: Comp -> Int -> Property prop_FinishEarlyWith comp n = concurrentProperty $ do - res <- - withScheduler comp $ \scheduler -> do - scheduleWork scheduler $ pure (complement (n + 1)) - scheduleWork scheduler $ terminateWith scheduler n >> pure (complement n) - pure (res === [n]) + let scheduleJobs scheduler = do + scheduleWork scheduler $ pure (complement (n + 1)) + scheduleWork scheduler $ terminateWith scheduler n >> pure (complement n) + res <- withScheduler comp scheduleJobs + res' <- withSchedulerR comp scheduleJobs + pure (res === [n] .&&. res' === FinishedEarlyWith n) prop_FinishBeforeStarting :: Comp -> Property prop_FinishBeforeStarting comp = @@ -260,6 +274,49 @@ ys <- readIORef yRefs pure (nSame .&&. xs === ys) +prop_SameAsTrivialScheduler :: Comp -> [Int] -> Fun Int Int -> Property +prop_SameAsTrivialScheduler comp zs f = + concurrentProperty $ do + let schedule scheduler = forM_ zs (scheduleWork scheduler . pure . apply f) + xs <- withScheduler comp schedule + ys <- withTrivialScheduler schedule + pure (xs === ys) + +prop_Terminate :: + (Show a, Eq a) + => ((Scheduler IO Int -> IO ()) -> IO a) + -> (Scheduler IO Int -> Int -> IO Int) + -> ([Int] -> Int -> a) + -> [Int] + -> Int + -> [Int] + -> Expectation +prop_Terminate withSchedulerR' term expected xs x ys = do + rs <- withSchedulerR' $ \ scheduler -> do + forM_ xs (scheduleWork scheduler . pure) + _ <- scheduleWork scheduler $ term scheduler x + forM_ ys (scheduleWork scheduler . pure) + rs `shouldBe` expected xs x + +-- prop_TerminateSeq :: +-- ((Scheduler IO Int -> IO ()) -> IO (Results Int)) -> [Int] -> Int -> [Int] -> Expectation +-- prop_TerminateSeq withSchedulerR' xs x ys = do +-- rs <- withSchedulerR' $ \ scheduler -> do +-- forM_ xs (scheduleWork scheduler . pure) +-- _ <- scheduleWork scheduler $ terminate scheduler x +-- forM_ ys (scheduleWork scheduler . pure) +-- rs `shouldBe` FinishedEarly xs x + +-- prop_TerminateWithSeq :: +-- ((Scheduler IO Int -> IO ()) -> IO (Results Int)) -> [Int] -> Int -> [Int] -> Expectation +-- prop_TerminateWithSeq withSchedulerR' xs x ys = do +-- rs <- withSchedulerR' $ \ scheduler -> do +-- forM_ xs (scheduleWork scheduler . pure) +-- _ <- scheduleWork scheduler $ terminateWith scheduler x +-- forM_ ys (scheduleWork scheduler . pure) +-- rs `shouldBe` FinishedEarlyWith x + + newtype Elem = Elem Int deriving (Eq, Show) instance Exception Elem @@ -291,14 +348,35 @@ 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_WorkerStateExclusive :: Comp -> NonNegative Int -> Expectation +prop_WorkerStateExclusive comp (NonNegative n) = do + state <- initWorkerStates comp (\wid -> (,) wid <$> newIORef (0 :: Int)) + workerStatesComp state `shouldBe` comp + nWorkers <- getCompWorkers comp + let scheduleJobs schedulerWS = do + replicateM n $ + scheduleWorkState schedulerWS $ \(wid, ref) -> do + counter <- readIORef ref + writeIORef ref (counter + 1) + pure (wid, counter) + gather = map (sortOn snd) . groupBy (\x y -> fst x == fst y) . sortOn fst + isMonotonicStartingAt _ [] = True + isMonotonicStartingAt k (k':ks) = k == k' && isMonotonicStartingAt (k + 1) ks + baseIds = [(wid, -1) | wid <- [0 .. WorkerId nWorkers - 1]] + ids <- withSchedulerWS state scheduleJobs + length ids `shouldBe` n + let gathered = gather (ids ++ baseIds) + map (map snd) gathered `shouldSatisfy` all (isMonotonicStartingAt (-1)) + ids' <- withSchedulerWSR state scheduleJobs + length ids' `shouldBe` n + let gathered' = gather (baseIds ++ ids ++ F.toList ids') + map (map snd) gathered' `shouldSatisfy` all (isMonotonicStartingAt (-1)) + withSchedulerWS_ state $ \schedulerWS -> do + numWorkers (unwrapSchedulerWS schedulerWS) `shouldBe` nWorkers + replicateM (10 * n) $ + scheduleWorkState_ schedulerWS $ \(wid, ref) -> do + counter <- readIORef ref + when (counter > 0) $ snd (last (gathered' !! getWorkerId wid)) `shouldBe` pred counter prop_MutexException :: Comp -> Property prop_MutexException comp = @@ -308,7 +386,6 @@ scheduleWorkState_ schedulerWS $ \_s -> withSchedulerWS_ state $ \_s' -> pure () - spec :: Spec spec = do describe "Comp" $ do @@ -319,27 +396,71 @@ property $ \(x :: Comp) y z -> x <> (y <> z) === (x <> y) <> z it "mconcat = foldr '(<>)' mempty" $ property $ \(xs :: [Comp]) -> mconcat xs === foldr (<>) mempty xs + eqSpecOnArbitrary @Comp + monoidSpecOnArbitrary @Comp describe "Show" $ do it "show == showsPrec 0" $ property $ \(x :: Comp) -> x `deepseq` show x === showsPrec 0 x "" it "(show) == showsPrec 1" $ property $ \(x :: Comp) (Positive n) -> x /= Seq && x /= Par ==> ("(" <> show x <> ")" === showsPrec n x "") + describe "Results" $ do + eqSpecOnArbitrary @(Results Int) + functorSpecOnArbitrary @Results + showReadSpecOnArbitrary @(Results Int) + it "Traversable" $ property $ \(rs :: Results Int) (f :: Fun Int (Maybe Int)) -> + traverse (apply f) (F.toList rs) === fmap F.toList (traverse (apply f) rs) + describe "WorkerId" $ do + eqSpecOnArbitrary @WorkerId + ordSpecOnArbitrary @WorkerId + it "MaxMin" $ property $ \x y -> + conjoin [ max (WorkerId x) (WorkerId y) === WorkerId (max x y) + , min (WorkerId x) (WorkerId y) === WorkerId (min x y) + ] + showReadSpecOnArbitrary @WorkerId + describe "Enum" $ do + it "toEnumFromEnum" $ property $ \ wid@(WorkerId i) -> + toEnum (getWorkerId wid) === wid .&&. fromEnum wid === i + it "succ . pred" $ property $ \ wid@(WorkerId i) -> + i /= minBound && i /= maxBound ==> + succ (pred wid) === wid .&&. pred (succ wid) === wid + describe "Trivial" $ do + it "WorkerIdIsZero" $ do + scheduleWorkId trivialScheduler_ (`shouldBe` 0) + withTrivialScheduler (`scheduleWorkId` pure) `shouldReturn` [0] + it "TerminateDoesNothing" $ do + terminate_ trivialScheduler_ `shouldReturn` () + terminate trivialScheduler_ () `shouldReturn` () + terminateWith trivialScheduler_ () `shouldReturn` () + it "TerminateSeq" $ timed $ prop_Terminate withTrivialScheduler terminate (\xs x -> xs ++ [x]) + it "TerminateWithSeq" $ timed $ prop_Terminate withTrivialScheduler terminateWith (\_ x -> [x]) + it "TerminateSeqR" $ timed $ prop_Terminate withTrivialSchedulerR terminate FinishedEarly + it "TerminateWithSeqR" $ + timed $ prop_Terminate withTrivialSchedulerR terminateWith (const FinishedEarlyWith) describe "Seq" $ do it "SameList" $ timed $ prop_SameList Seq it "Recursive" $ timed $ prop_Recursive Seq it "Nested" $ timed $ prop_Nested Seq it "Serially" $ timed $ prop_Serially Seq - it "Trivial" $ timed prop_TrivialSchedulerSameAsSeq_ + it "TrivialAsSeq_" $ timed prop_TrivialSchedulerSameAsSeq_ + it "replicateConcurrently == replicateM" $ timed prop_ReplicateM + it "replicateConcurrently == replicateWork" $ timed prop_ReplicateWorkSeq + it "WorkerIdIsZero" $ + withScheduler Seq (`scheduleWorkId` pure) `shouldReturn` [0] + it "TerminateSeq" $ timed $ prop_Terminate (withScheduler Seq) terminate (\xs x -> xs ++ [x]) + it "TerminateWithSeq" $ timed $ prop_Terminate (withScheduler Seq) terminateWith (\_ x -> [x]) + it "TerminateSeqR" $ timed $ prop_Terminate (withSchedulerR Seq) terminate FinishedEarly + it "TerminateWithSeqR" $ + timed $ prop_Terminate (withSchedulerR Seq) terminateWith (const FinishedEarlyWith) describe "ParOn" $ do it "SameList" $ timed $ \cs -> prop_SameList (ParOn cs) it "Recursive" $ timed $ \cs -> prop_Recursive (ParOn cs) it "Nested" $ timed $ \cs -> prop_Nested (ParOn cs) it "Serially" $ timed $ \cs -> prop_Serially (ParOn cs) describe "Arbitrary Comp" $ do + it "Trivial" $ timed prop_SameAsTrivialScheduler it "ArbitraryNested" $ timed prop_ArbitraryCompNested it "AllJobsProcessed" $ timed prop_AllJobsProcessed it "traverseConcurrently == traverse" $ timed prop_Traverse - --it "replicateConcurrently == replicateM" $ timed prop_ReplicateM describe "Exceptions" $ do it "CatchDivideByZero" $ timed prop_CatchDivideByZero it "CatchDivideByZeroNested" $ timed prop_CatchDivideByZeroNested @@ -357,8 +478,19 @@ it "FinishBeforeStarting" $ timed prop_FinishBeforeStarting it "FinishWithBeforeStarting" $ timed prop_FinishWithBeforeStarting describe "WorkerState" $ do - it "ReturnsState" $ timed prop_ReturnsState it "MutexException" $ timed prop_MutexException + it "WorkerStateExclusive" $ timed prop_WorkerStateExclusive + +instance Arbitrary WorkerId where + arbitrary = WorkerId <$> arbitrary + +instance Arbitrary a => Arbitrary (Results a) where + arbitrary = + oneof + [ Finished <$> arbitrary + , FinishedEarly <$> arbitrary <*> arbitrary + , FinishedEarlyWith <$> arbitrary + ] timed :: Testable prop => prop -> Property timed = within 2000000