scheduler 1.4.1 → 1.4.2

raw patch · 6 files changed

+349/−74 lines, 6 filesdep +genvalidity-hspecPVP ok

version bump matches the API change (PVP)

Dependencies added: genvalidity-hspec

API changes (from Hackage documentation)

Files

• CHANGELOG.md +8/−0
• scheduler.cabal +4/−3
• src/Control/Scheduler.hs +130/−36
• src/Control/Scheduler/Internal.hs +42/−2
• src/Control/Scheduler/Queue.hs +1/−1
• tests/Control/SchedulerSpec.hs +164/−32

@@ -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`

@@ -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

@@ -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<traverseConcurrently> - Mismatched sizes"+    withR _      _ = errorWithoutStackTrace "Impossible<traverseConcurrently> - 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.

@@ -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) 

@@ -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)

@@ -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