scheduler 1.0.0 → 1.1.0

raw patch · 7 files changed

+749/−159 lines, 7 filesdep +asyncdep +criteriondep +fibdep ~basedep ~unliftioPVP ok

version bump matches the API change (PVP)

Dependencies added: async, criterion, fib, monad-par, parallel, streamly

Dependency ranges changed: base, unliftio

API changes (from Hackage documentation)

Files

• README.md +229/−2
• bench/Scheduler.hs +103/−0
• scheduler.cabal +20/−2
• src/Control/Scheduler.hs +150/−60
• src/Control/Scheduler/Computation.hs +33/−9
• src/Control/Scheduler/Queue.hs +48/−56
• tests/Control/SchedulerSpec.hs +166/−30

@@ -187,9 +187,236 @@ Done ``` -## Avoiding deadlocks+## Premature termination +It is possible to make all of the workers stop whatever they are doing and get either their progress+thus far or simply return an element we were looking for.++For example, we would like to find the 10th letter in the English alphabet in parallel using 8+threads. The way we do it is we schedule 26 tasks, and the first one that will find the letter with+such index will terminate all the workers and return the result:++```haskell+λ> let f sch i c = scheduleWork sch $ if i == 10 then putChar '-' >> terminateWith sch c else threadDelay 1000 >> putChar c >> pure c+λ> withScheduler (ParN 8) (\ scheduler -> zipWithM (f scheduler) [1 :: Int ..] ['a'..'z'])+ab-dec"j"+```++++## Benchmarks++It is always good to see some benchmarks. Below is a very simple comparison of:++* [`traverseConcurrently`](https://hackage.haskell.org/package/scheduler-1.0.0/docs/Control-Scheduler.html#v:traverseConcurrently) from `scheduler`+* [`pooledMapConcurrently`](https://hackage.haskell.org/package/unliftio-0.2.10/docs/UnliftIO-Async.html#v:pooledMapConcurrently) from `unliftio`+* [`mapM`](https://hackage.haskell.org/package/streamly-0.6.1/docs/Streamly-Prelude.html#v:mapM) from `streamly`+* [`parMapM`](http://hackage.haskell.org/package/monad-par-extras-0.3.3/docs/Control-Monad-Par-Combinator.html) from `monad-par`+* [`mapConcurrently`](http://hackage.haskell.org/package/async-2.2.1/docs/Control-Concurrent-Async.html#v:mapConcurrently) from `async`+* `traverse` from `base` with+  [`par`](http://hackage.haskell.org/package/parallel-3.2.2.0/docs/Control-Parallel.html#v:par) from+  `parallel`+* Regular sequential `traverse` as a basepoint++Similar functions are used for `replicateM` functionality from the corresponding libraries.++Benchmarked functions `sum` (sum of elements in a list) and `fib` (fibonacci number) are pretty+straightforward, we simply `replicateM` them many times or `mapM` same functions over a+list. Although `scheduler` is already pretty good at it, it does look like there might be some room+for improvement. `pooled*` functions from `unliftio` have the best performance, and I don't think+they can be made any faster for such workloads, but they do not allow submitting more work during+computation, as such nested parallelism while reusing the same workers is impossible, thus+functionally they are inferior.++Benchmarks can be reproduced with `stack bench` inside this repo.++```+scheduler-1.1.0: benchmarks+Running 1 benchmarks...+Benchmark scheduler: RUNNING...+benchmarking replicate/Fib(1000/10000)/scheduler/replicateConcurrently+time                 10.16 ms   (9.235 ms .. 11.09 ms)+                     0.954 R²   (0.899 R² .. 0.982 R²)+mean                 10.42 ms   (9.965 ms .. 11.26 ms)+std dev              1.732 ms   (982.5 μs .. 2.587 ms)+variance introduced by outliers: 76% (severely inflated)++benchmarking replicate/Fib(1000/10000)/unliftio/pooledReplicateConcurrently+time                 8.476 ms   (8.034 ms .. 8.867 ms)+                     0.986 R²   (0.972 R² .. 0.994 R²)+mean                 8.671 ms   (8.410 ms .. 9.090 ms)+std dev              909.4 μs   (634.5 μs .. 1.279 ms)+variance introduced by outliers: 58% (severely inflated)++benchmarking replicate/Fib(1000/10000)/streamly/replicateM+time                 11.90 ms   (11.18 ms .. 12.46 ms)+                     0.976 R²   (0.935 R² .. 0.994 R²)+mean                 12.72 ms   (12.24 ms .. 14.27 ms)+std dev              1.897 ms   (625.2 μs .. 3.881 ms)+variance introduced by outliers: 69% (severely inflated)++benchmarking replicate/Fib(1000/10000)/async/replicateConcurrently+time                 21.16 ms   (19.59 ms .. 22.38 ms)+                     0.982 R²   (0.953 R² .. 0.995 R²)+mean                 22.46 ms   (21.27 ms .. 24.73 ms)+std dev              3.644 ms   (1.499 ms .. 5.373 ms)+variance introduced by outliers: 69% (severely inflated)++benchmarking replicate/Fib(1000/10000)/monad-par/replicateM+time                 31.78 ms   (30.64 ms .. 32.87 ms)+                     0.994 R²   (0.986 R² .. 0.998 R²)+mean                 32.59 ms   (31.86 ms .. 33.98 ms)+std dev              2.290 ms   (1.296 ms .. 3.640 ms)+variance introduced by outliers: 28% (moderately inflated)++benchmarking replicate/Fib(1000/10000)/base/replicateM+time                 31.35 ms   (30.91 ms .. 31.71 ms)+                     0.999 R²   (0.998 R² .. 1.000 R²)+mean                 31.56 ms   (31.06 ms .. 32.64 ms)+std dev              1.440 ms   (674.1 μs .. 2.516 ms)+variance introduced by outliers: 16% (moderately inflated)++benchmarking replicate/Sum(1000/1000)/scheduler/replicateConcurrently+time                 17.10 ms   (16.25 ms .. 17.79 ms)+                     0.984 R²   (0.960 R² .. 0.995 R²)+mean                 17.12 ms   (16.57 ms .. 17.97 ms)+std dev              1.690 ms   (858.3 μs .. 2.444 ms)+variance introduced by outliers: 46% (moderately inflated)++benchmarking replicate/Sum(1000/1000)/unliftio/pooledReplicateConcurrently+time                 15.91 ms   (15.74 ms .. 16.06 ms)+                     0.999 R²   (0.999 R² .. 1.000 R²)+mean                 15.97 ms   (15.87 ms .. 16.15 ms)+std dev              301.2 μs   (198.0 μs .. 421.2 μs)++benchmarking replicate/Sum(1000/1000)/streamly/replicateM+time                 17.51 ms   (17.06 ms .. 17.96 ms)+                     0.997 R²   (0.995 R² .. 0.999 R²)+mean                 17.99 ms   (17.74 ms .. 18.64 ms)+std dev              865.5 μs   (599.3 μs .. 1.295 ms)+variance introduced by outliers: 17% (moderately inflated)++benchmarking replicate/Sum(1000/1000)/async/replicateConcurrently+time                 28.82 ms   (26.60 ms .. 31.82 ms)+                     0.972 R²   (0.948 R² .. 0.987 R²)+mean                 33.20 ms   (31.69 ms .. 34.13 ms)+std dev              2.548 ms   (1.706 ms .. 3.197 ms)+variance introduced by outliers: 28% (moderately inflated)++benchmarking replicate/Sum(1000/1000)/monad-par/replicateM+time                 58.84 ms   (55.71 ms .. 61.47 ms)+                     0.995 R²   (0.991 R² .. 0.999 R²)+mean                 64.34 ms   (62.70 ms .. 65.83 ms)+std dev              3.150 ms   (2.697 ms .. 3.721 ms)+variance introduced by outliers: 15% (moderately inflated)++benchmarking replicate/Sum(1000/1000)/base/replicateM+time                 56.26 ms   (55.82 ms .. 56.74 ms)+                     1.000 R²   (0.999 R² .. 1.000 R²)+mean                 56.70 ms   (56.47 ms .. 57.21 ms)+std dev              618.6 μs   (331.6 μs .. 969.4 μs)++benchmarking map/Fib(2000)/scheduler/traverseConcurrently+time                 12.23 ms   (11.51 ms .. 12.90 ms)+                     0.968 R²   (0.933 R² .. 0.989 R²)+mean                 13.65 ms   (12.87 ms .. 14.50 ms)+std dev              1.985 ms   (1.455 ms .. 2.696 ms)+variance introduced by outliers: 68% (severely inflated)++benchmarking map/Fib(2000)/unliftio/pooledTraverseConcurrently+time                 6.843 ms   (6.435 ms .. 7.201 ms)+                     0.973 R²   (0.939 R² .. 0.992 R²)+mean                 7.223 ms   (6.946 ms .. 8.280 ms)+std dev              1.240 ms   (537.0 μs .. 2.535 ms)+variance introduced by outliers: 82% (severely inflated)++benchmarking map/Fib(2000)/streamly/mapM+time                 21.89 ms   (21.01 ms .. 23.13 ms)+                     0.991 R²   (0.985 R² .. 0.996 R²)+mean                 21.44 ms   (20.16 ms .. 24.26 ms)+std dev              4.518 ms   (1.773 ms .. 8.475 ms)+variance introduced by outliers: 80% (severely inflated)++benchmarking map/Fib(2000)/async/mapConcurrently+time                 37.37 ms   (32.21 ms .. 41.57 ms)+                     0.966 R²   (0.936 R² .. 0.993 R²)+mean                 41.32 ms   (38.94 ms .. 47.96 ms)+std dev              7.065 ms   (2.678 ms .. 12.36 ms)+variance introduced by outliers: 65% (severely inflated)++benchmarking map/Fib(2000)/par/mapM+time                 7.381 ms   (6.934 ms .. 7.848 ms)+                     0.975 R²   (0.953 R² .. 0.987 R²)+mean                 7.519 ms   (7.256 ms .. 8.123 ms)+std dev              1.178 ms   (677.4 μs .. 2.015 ms)+variance introduced by outliers: 78% (severely inflated)++benchmarking map/Fib(2000)/monad-par/mapM+time                 24.57 ms   (23.52 ms .. 25.30 ms)+                     0.994 R²   (0.986 R² .. 0.998 R²)+mean                 26.15 ms   (25.48 ms .. 27.60 ms)+std dev              2.016 ms   (1.267 ms .. 3.175 ms)+variance introduced by outliers: 30% (moderately inflated)++benchmarking map/Fib(2000)/base/mapM+time                 23.79 ms   (22.66 ms .. 24.75 ms)+                     0.993 R²   (0.987 R² .. 0.999 R²)+mean                 24.66 ms   (23.96 ms .. 26.84 ms)+std dev              2.532 ms   (891.4 μs .. 4.701 ms)+variance introduced by outliers: 46% (moderately inflated)++benchmarking map/Sum(2000)/scheduler/traverseConcurrently+time                 36.13 ms   (35.31 ms .. 36.92 ms)+                     0.998 R²   (0.996 R² .. 0.999 R²)+mean                 36.93 ms   (36.03 ms .. 38.43 ms)+std dev              2.456 ms   (1.045 ms .. 4.098 ms)+variance introduced by outliers: 24% (moderately inflated)++benchmarking map/Sum(2000)/unliftio/pooledTraverseConcurrently+time                 31.99 ms   (31.60 ms .. 32.39 ms)+                     0.998 R²   (0.993 R² .. 1.000 R²)+mean                 32.87 ms   (32.43 ms .. 33.90 ms)+std dev              1.454 ms   (913.4 μs .. 2.241 ms)+variance introduced by outliers: 12% (moderately inflated)++benchmarking map/Sum(2000)/streamly/mapM+time                 39.56 ms   (36.74 ms .. 43.24 ms)+                     0.988 R²   (0.977 R² .. 0.999 R²)+mean                 43.58 ms   (42.12 ms .. 44.80 ms)+std dev              2.595 ms   (1.621 ms .. 3.785 ms)+variance introduced by outliers: 20% (moderately inflated)++benchmarking map/Sum(2000)/async/mapConcurrently+time                 48.82 ms   (45.75 ms .. 51.32 ms)+                     0.992 R²   (0.983 R² .. 0.998 R²)+mean                 54.91 ms   (52.92 ms .. 57.00 ms)+std dev              3.800 ms   (2.823 ms .. 5.142 ms)+variance introduced by outliers: 22% (moderately inflated)++benchmarking map/Sum(2000)/par/mapM+time                 57.28 ms   (56.62 ms .. 58.04 ms)+                     1.000 R²   (0.999 R² .. 1.000 R²)+mean                 55.69 ms   (54.85 ms .. 56.29 ms)+std dev              1.282 ms   (994.2 μs .. 1.599 ms)++benchmarking map/Sum(2000)/monad-par/mapM+time                 237.4 ms   (198.5 ms .. 264.7 ms)+                     0.988 R²   (0.966 R² .. 1.000 R²)+mean                 257.7 ms   (242.2 ms .. 263.0 ms)+std dev              11.57 ms   (384.2 μs .. 14.53 ms)+variance introduced by outliers: 16% (moderately inflated)++benchmarking map/Sum(2000)/base/mapM+time                 147.1 ms   (141.4 ms .. 150.9 ms)+                     0.999 R²   (0.996 R² .. 1.000 R²)+mean                 152.6 ms   (150.1 ms .. 155.1 ms)+std dev              3.676 ms   (2.725 ms .. 5.066 ms)+variance introduced by outliers: 12% (moderately inflated)++```++## Beware of Demons+ Any sort of concurrency primitives such as mutual exclusion, semaphores, etc. can easily lead to deadlocks, starvation and other common problems. Try to avoid them and be careful if you do end up using them.-

@@ -0,0 +1,103 @@+module Main where++import qualified Control.Concurrent.Async as A (mapConcurrently, replicateConcurrently)+import Control.Monad (replicateM)+import Control.Monad.Par (IVar, Par, get, newFull_, parMapM, runParIO)+import Control.Parallel (par)+import Control.Scheduler+import Criterion.Main+import Control.DeepSeq+import Data.Foldable as F+import Data.IORef+import Fib+import Streamly (asyncly)+import qualified Streamly.Prelude as S+import UnliftIO.Async (pooledMapConcurrently, pooledReplicateConcurrently)++main :: IO ()+main = do+  defaultMain+    ([mkBenchReplicate "Fib" n x fibIORef fibParVar | n <- [1000], x <- [10000]] +++     [mkBenchReplicate "Sum" n x sumIORef sumParVar | n <- [1000], x <- [1000]] +++     [mkBenchMap "Fib" n fibIO fibParIO fibPar | n <- [2000]] +++     [mkBenchMap "Sum" n sumIO sumParIO sumPar | n <- [2000]])+  where+    fibIO :: Int -> IO Integer+    fibIO x = do+      let y = fib $ toInteger x+      y `seq` pure y+    fibParIO :: Int -> IO Integer+    fibParIO x = do+      let y = fib $ toInteger x+      y `par` pure y+    fibPar :: Int -> Par Integer+    fibPar x = do+      let y = fib $ toInteger x+      y `seq` pure y+    sumIO :: Int -> IO Int+    sumIO x = do+      let y = F.foldl' (+) 0 [x .. 100*x]+      y `seq` pure y+    sumParIO :: Int -> IO Int+    sumParIO x = do+      let y = F.foldl' (+) 0 [x .. 100*x]+      y `par` pure y+    sumPar :: Int -> Par Int+    sumPar x = do+      let y = F.foldl' (+) 0 [x .. 100*x]+      y `seq` pure y+    fibIORef :: IORef Int -> IO Integer+    fibIORef xRef = readIORef xRef >>= fibIO+    fibParVar :: IVar Int -> Par Integer+    fibParVar ivar = get ivar >>= fibPar+    sumIORef :: IORef Int -> IO Int+    sumIORef xRef = readIORef xRef >>= sumIO+    sumParVar :: IVar Int -> Par Int+    sumParVar ivar = get ivar >>= sumPar++mkBenchReplicate ::+     NFData a+  => String+  -> Int -- ^ Number of tasks+  -> Int -- ^ Opaque Int a function should be applied to+  -> (IORef Int -> IO a)+  -> (IVar Int -> Par a)+  -> Benchmark+mkBenchReplicate name n x fxIO fxPar =+  bgroup+    ("replicate/" <> name <> str)+    [ bench "scheduler/replicateConcurrently" $+      nfIO $ replicateConcurrently Par n (newIORef x >>= fxIO)+    , bench "unliftio/pooledReplicateConcurrently" $+      nfIO $ pooledReplicateConcurrently n (newIORef x >>= fxIO)+    , bench "streamly/replicateM" $+      nfIO $ S.runStream $ asyncly $ S.replicateM n (newIORef x >>= fxIO)+    , bench "async/replicateConcurrently" $ nfIO $ A.replicateConcurrently n (newIORef x >>= fxIO)+    , bench "monad-par/replicateM" $ nfIO $ runParIO $ replicateM n (newFull_ x >>= fxPar)+    , bench "base/replicateM" $ nfIO $ replicateM n (newIORef x >>= fxIO)+    ]+  where+    str = "(" ++ show n ++ "/" ++ show x ++ ")"+++mkBenchMap ::+     NFData a+  => String+  -> Int -- ^ Number of tasks+  -> (Int -> IO a)+  -> (Int -> IO a)+  -> (Int -> Par a)+  -> Benchmark+mkBenchMap name n fxIO fxParIO fxPar =+  bgroup+    ("map/" <> name <> str)+    [ bench "scheduler/traverseConcurrently" $ nfIO $ traverseConcurrently Par fxIO [1 .. n]+    , bench "unliftio/pooledTraverseConcurrently" $ nfIO $ pooledMapConcurrently fxIO [1 .. n]+    , bench "streamly/mapM" $ nfIO $ S.runStream $ asyncly $ S.mapM fxIO $ S.enumerateFromTo 1 n+    , bench "async/mapConcurrently" $ nfIO $ A.mapConcurrently fxIO [1 .. n]+    , bench "par/mapM" $ nfIO $ mapM fxParIO [1 .. n]+    , bench "monad-par/mapM" $ nfIO $ runParIO $ mapM fxPar [1 .. n]+    , bench "base/mapM" $ nfIO $ mapM fxIO [1 .. n]+    ]+  where+    str = "(" ++ show n ++ ")"

@@ -1,5 +1,5 @@ name:                scheduler-version:             1.0.0+version:             1.1.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@@ -61,6 +61,24 @@                , template-haskell   default-language:    Haskell2010 ++benchmark scheduler+  type:                exitcode-stdio-1.0+  hs-source-dirs:      bench+  main-is:             Scheduler.hs+  ghc-options:         -threaded -O2 -rtsopts -with-rtsopts=-N+  build-depends:       base+                     , async+                     , criterion+                     , deepseq+                     , fib+                     , monad-par+                     , scheduler+                     , parallel+                     , unliftio >= 0.2.10+                     , streamly >= 0.6.1 && < 0.7+  default-language:    Haskell2010+ source-repository head   type:     git-  location: https://github.com/lehins/massiv+  location: https://github.com/lehins/haskell-scheduler

@@ -13,13 +13,25 @@ -- Portability : non-portable -- module Control.Scheduler-  ( -- * Scheduler and strategies-    Comp(..)-  , Scheduler(..)+  (+    -- * Scheduler+    Scheduler+  , numWorkers+  , scheduleWork+  , scheduleWork_+  , terminate+  , terminate_+  , terminateWith   -- * Initialize Scheduler   , withScheduler   , withScheduler_+  , trivialScheduler_+  -- * Computation strategies+  , Comp(..)+  , getCompWorkers   -- * Useful functions+  , replicateConcurrently+  , replicateConcurrently_   , traverseConcurrently   , traverseConcurrently_   , traverse_@@ -29,32 +41,91 @@  import Control.Concurrent import Control.Exception-import Control.Scheduler.Computation-import Control.Scheduler.Queue import Control.Monad import Control.Monad.IO.Unlift+import Control.Scheduler.Computation+import Control.Scheduler.Queue import Data.Atomics (atomicModifyIORefCAS, atomicModifyIORefCAS_)-import Data.Foldable as F (foldl')+import qualified Data.Foldable as F (foldl', traverse_) import Data.IORef import Data.Traversable+import Data.Maybe (catMaybes) + 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 the only ways to get--- access to such data type.+-- | Main type for scheduling work. See `withScheduler` or `withScheduler_` for ways to construct+-- and use this data type. -- -- @since 1.0.0 data Scheduler m a = Scheduler-  { numWorkers   :: {-# UNPACK #-} !Int-  -- ^ Get the number of workers.-  , scheduleWork :: m a -> m ()-  -- ^ Schedule an action to be picked up and computed by a worker from a pool.+  { numWorkers    :: {-# UNPACK #-} !Int+  -- ^ 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+  , scheduleWork  :: m a -> m ()+  -- ^ Schedule an action to be picked up and computed by a worker from a pool of jobs.+  --+  -- @since 1.0.0+  , terminate     :: a -> m a+  -- ^ 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.+  --+  -- /Important/ - With `Seq` strategy this will not stop other scheduled tasks from being computed,+  -- although it will make sure their results are discarded.+  --+  -- @since 1.1.0+  , terminateWith :: a -> m a+  -- ^ Same as `terminate`, but returning a single element list containing the supplied+  -- argument. This can be very useful for parallel search algorithms.+  --+  -- /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+  -- function.+  --+  -- @since 1.1.0   } +-- | Same as `scheduleWork`, but only for a `Scheduler` that doesn't keep the result.+--+-- @since 1.1.0+scheduleWork_ :: Scheduler m () -> m () -> m ()+scheduleWork_ = scheduleWork++-- | Similar to `terminate`, but for a `Scheduler` that does not keep any results of computation.+--+-- /Important/ - In case of `Seq` computation strategy this function has no affect.+--+-- @since 1.1.0+terminate_ :: Scheduler m () -> m ()+terminate_ = (`terminateWith` ())++-- | The most basic scheduler that simply runs the task instead of scheduling it. Early termination+-- requests are simply ignored.+--+-- @since 1.1.0+trivialScheduler_ :: Applicative f => Scheduler f ()+trivialScheduler_ = Scheduler+  { numWorkers = 1+  , scheduleWork = id+  , 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@@ -82,8 +153,25 @@ -- -- @since 1.0.0 traverseConcurrently_ :: (MonadUnliftIO m, Foldable t) => Comp -> (a -> m b) -> t a -> m ()-traverseConcurrently_ comp f xs = withScheduler_ comp $ \s -> traverse_ (scheduleWork s . f) xs+traverseConcurrently_ comp f xs =+  withScheduler_ comp $ \s -> scheduleWork s $ F.traverse_ (scheduleWork s . void . f) xs +-- | Replicate an action @n@ times and schedule them acccording to the supplied computation+-- strategy.+--+-- @since 1.1.0+replicateConcurrently :: MonadUnliftIO m => Comp -> Int -> m a -> m [a]+replicateConcurrently comp n f =+  withScheduler comp $ \s -> replicateM_ n $ scheduleWork s f++-- | Just like `replicateConcurrently`, but discards the results of computation.+--+-- @since 1.1.0+replicateConcurrently_ :: MonadUnliftIO m => Comp -> Int -> m a -> m ()+replicateConcurrently_ comp n f =+  withScheduler_ comp $ \s -> scheduleWork s $ replicateM_ n (scheduleWork s $ void f)++ scheduleJobs :: MonadIO m => Jobs m a -> m a -> m () scheduleJobs = scheduleJobsWith mkJob @@ -94,14 +182,16 @@ scheduleJobs_ = scheduleJobsWith (return . Job_ . void)  scheduleJobsWith :: MonadIO m => (m b -> m (Job m a)) -> Jobs m a -> m b -> m ()-scheduleJobsWith mkJob' Jobs {jobsQueue, jobsCountRef, jobsNumWorkers} action = do-  liftIO $ atomicModifyIORefCAS_ jobsCountRef (+ 1)+scheduleJobsWith mkJob' jobs action = do+  liftIO $ atomicModifyIORefCAS_ (jobsCountRef jobs) (+ 1)   job <-     mkJob' $ do       res <- action-      res `seq` dropCounterOnZero jobsCountRef $ retireWorkersN jobsQueue jobsNumWorkers+      res `seq`+        dropCounterOnZero (jobsCountRef jobs) $+        retireWorkersN (jobsQueue jobs) (jobsNumWorkers jobs)       return res-  pushJQueue jobsQueue job+  pushJQueue (jobsQueue jobs) job  -- | Helper function to place required number of @Retire@ instructions on the job queue. retireWorkersN :: MonadIO m => JQueue m a -> Int -> m ()@@ -111,7 +201,8 @@ dropCounterOnZero :: MonadIO m => IORef Int -> m () -> m () dropCounterOnZero counterRef onZero = do   jc <--    liftIO $ atomicModifyIORefCAS+    liftIO $+    atomicModifyIORefCAS       counterRef       (\ !i' ->          let !i = i' - 1@@ -119,7 +210,7 @@   when (jc == 0) onZero  --- | Runs the worker until the job queue is exhausted, at which point it will exhecute the final task+-- | 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 =>              JQueue m a@@ -146,8 +237,8 @@ -- * It is ok to initialize multiple schedulers at the same time, although that will likely result --   in suboptimal performance, unless workers are pinned to different capabilities. ----- * __Warning__ It is very dangerous to schedule jobs that do blocking `IO`, since it can lead to a---   deadlock very quickly, if you are not careful. Consider this example. First execution works fine,+-- * __Warning__ It is pretty dangerous to schedule jobs that do blocking `IO`, since it can easily+--   lead to deadlock, if you are not careful. Consider this example. First execution works fine, --   since there are two scheduled workers, and one can unblock the other, but the second scenario --   immediately results in a deadlock. --@@ -167,7 +258,7 @@   -> (Scheduler m a -> m b)      -- ^ Action that will be scheduling all the work.   -> m [a]-withScheduler comp = withSchedulerInternal comp scheduleJobs flushResults+withScheduler comp = withSchedulerInternal comp scheduleJobs readResults reverse   -- | Same as `withScheduler`, but discards results of submitted jobs.@@ -179,34 +270,44 @@   -> (Scheduler m a -> m b)      -- ^ Action that will be scheduling all the work.   -> m ()-withScheduler_ comp = withSchedulerInternal comp scheduleJobs_ (const (pure ()))-+withScheduler_ comp = void . withSchedulerInternal comp scheduleJobs_ (const (pure [])) id  withSchedulerInternal ::      MonadUnliftIO m   => Comp -- ^ Computation strategy   -> (Jobs m a -> m a -> m ()) -- ^ How to schedule work-  -> (JQueue m a -> m c) -- ^ How to collect results+  -> (JQueue m a -> m [Maybe a]) -- ^ How to collect results+  -> ([a] -> [a]) -- ^ Adjust results in some way   -> (Scheduler m a -> m b)      -- ^ Action that will be scheduling all the work.-  -> m c-withSchedulerInternal comp submitWork collect onScheduler = do-  jobsNumWorkers <--    case comp of-      Seq      -> return 1-      Par      -> liftIO getNumCapabilities-      ParOn ws -> return $ length ws-      ParN 0   -> liftIO getNumCapabilities-      ParN n   -> return $ fromIntegral n+  -> m [a]+withSchedulerInternal comp submitWork collect adjust onScheduler = do+  jobsNumWorkers <- getCompWorkers comp   sWorkersCounterRef <- liftIO $ newIORef jobsNumWorkers   jobsQueue <- newJQueue   jobsCountRef <- liftIO $ newIORef 0   workDoneMVar <- liftIO newEmptyMVar   let jobs = Jobs {..}-      scheduler = Scheduler {numWorkers = jobsNumWorkers, scheduleWork = submitWork jobs}-      onRetire = dropCounterOnZero sWorkersCounterRef $ liftIO (putMVar workDoneMVar Nothing)-  -- / Wait for the initial jobs to get scheduled before spawining off the workers, otherwise it would-  -- be trickier to identify the beginning and the end of a job pool.+      scheduler =+        Scheduler+          { numWorkers = jobsNumWorkers+          , scheduleWork = submitWork jobs+          , terminate =+              \a -> do+                mas <- collect jobsQueue+                let as = adjust (a : catMaybes mas)+                liftIO $ void $ tryPutMVar workDoneMVar $ SchedulerTerminatedEarly as+                pure a+          , terminateWith =+              \a -> do+                liftIO $ void $ tryPutMVar workDoneMVar $ SchedulerTerminatedEarly [a]+                pure a+          }+      onRetire =+        dropCounterOnZero sWorkersCounterRef $+        void $ liftIO (tryPutMVar workDoneMVar SchedulerFinished)+  -- / Wait for the initial jobs to get scheduled before spawining off the workers, otherwise it+  -- would be trickier to identify the beginning and the end of a job pool.   _ <- onScheduler scheduler   -- / Ensure at least something gets scheduled, so retirement can be triggered   jc <- liftIO $ readIORef jobsCountRef@@ -218,7 +319,6 @@               catch                 (unmask $ run $ runWorker jobsQueue onRetire)                 (run . handleWorkerException jobsQueue workDoneMVar jobsNumWorkers)-      {-# INLINE spawnWorkersWith #-}       spawnWorkers =         case comp of           Seq -> return []@@ -226,52 +326,44 @@           Par -> spawnWorkersWith forkOnWithUnmask [1 .. jobsNumWorkers]           ParOn ws -> spawnWorkersWith forkOnWithUnmask ws           ParN _ -> spawnWorkersWith (\_ -> forkIOWithUnmask) [1 .. jobsNumWorkers]-      {-# INLINE spawnWorkers #-}-      doWork = do+      terminateWorkers = liftIO . traverse_ (`throwTo` SomeAsyncException WorkerTerminateException)+      doWork tids = do         when (comp == Seq) $ runWorker 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         case mExc of-          Nothing                    -> collect jobsQueue+          SchedulerFinished -> adjust . catMaybes <$> collect jobsQueue           -- \ Now we are sure all workers have done their job we can safely read all of the           -- IORefs with results-          Just (WorkerException exc) -> liftIO $ throwIO exc+          SchedulerTerminatedEarly as -> terminateWorkers tids >> pure 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-      {-# INLINE doWork #-}-  safeBracketOnError-    spawnWorkers-    (liftIO . traverse_ (`throwTo` SomeAsyncException WorkerTerminateException))-    (const doWork)+  safeBracketOnError spawnWorkers terminateWorkers doWork   -- | Specialized exception handler for the work scheduler. handleWorkerException ::-  MonadIO m => JQueue m a -> MVar (Maybe WorkerException) -> Int -> SomeException -> m ()+  MonadIO m => JQueue m a -> MVar (SchedulerOutcome a) -> Int -> SomeException -> m () handleWorkerException jQueue workDoneMVar nWorkers exc =   case asyncExceptionFromException exc of     Just WorkerTerminateException -> return ()       -- \ some co-worker died, we can just move on with our death.     _ -> do-      _ <- liftIO $ tryPutMVar workDoneMVar $ Just $ WorkerException exc+      _ <- liftIO $ tryPutMVar workDoneMVar $ SchedulerWorkerException $ WorkerException exc       -- \ Main thread must know how we died       -- / Do the co-worker cleanup       retireWorkersN jQueue (nWorkers - 1)  --- | This exception should normally be not seen in the wild. The only one that could possibly pop up--- is the `WorkerAsyncException`.+-- | 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 where-  displayException workerExc =-    case workerExc of-      WorkerException exc ->-        "A worker handled a job that ended with exception: " ++ displayException exc+instance Exception WorkerException  data WorkerTerminateException =   WorkerTerminateException@@ -280,9 +372,7 @@   deriving (Show)  -instance Exception WorkerTerminateException where-  displayException WorkerTerminateException = "A worker was terminated by the scheduler"-+instance Exception WorkerTerminateException  -- Copy from unliftio: safeBracketOnError :: MonadUnliftIO m => m a -> (a -> m b) -> (a -> m c) -> m c

@@ -1,4 +1,5 @@ {-# LANGUAGE CPP #-}+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE PatternSynonyms #-} {-# LANGUAGE ScopedTypeVariables #-} -- |@@ -10,10 +11,12 @@ -- Portability : non-portable -- module Control.Scheduler.Computation-  ( Comp(.., Par)+  ( Comp(.., Par, Par'), getCompWorkers   ) where +import Control.Concurrent (getNumCapabilities) import Control.DeepSeq (NFData(..), deepseq)+import Control.Monad.IO.Class #if !MIN_VERSION_base(4,11,0) import Data.Semigroup #endif@@ -32,10 +35,19 @@   deriving Eq  -- | Parallel computation using all available cores. Same as @`ParOn` []@+--+-- @since 1.0.0 pattern Par :: Comp pattern Par <- ParOn [] where         Par =  ParOn [] +-- | Parallel computation using all available cores. Same as @`ParN` 0@+--+-- @since 1.1.0+pattern Par' :: Comp+pattern Par' <- ParN 0 where+        Par' =  ParN 0+ instance Show Comp where   show Seq        = "Seq"   show Par        = "Par"@@ -69,18 +81,30 @@   case x of     Seq -> y     Par -> Par+    ParN 0 -> ParN 0     ParOn xs ->       case y of-        Seq      -> x         Par      -> Par-        ParOn ys -> ParOn (xs ++ ys)         ParN 0   -> ParN 0-        ParN n2  -> ParN (max (fromIntegral (length xs)) n2)-    ParN 0 -> ParN 0+        ParOn ys -> ParOn (xs <> ys)+        _        -> x     ParN n1 ->       case y of-        Seq      -> x-        Par      -> Par-        ParOn ys -> ParN (max n1 (fromIntegral (length ys)))-        ParN n2  -> ParN (max n1 n2)+        Seq     -> x+        Par     -> Par+        ParOn _ -> y+        ParN 0  -> y+        ParN n2 -> ParN (max n1 n2) {-# NOINLINE joinComp #-}++-- | Figure out how many workers will this computation strategy create+--+-- @since 1.1.0+getCompWorkers :: MonadIO m => Comp -> m Int+getCompWorkers =+  \case+    Seq -> return 1+    Par -> liftIO getNumCapabilities+    ParOn ws -> return $ length ws+    ParN 0 -> liftIO getNumCapabilities+    ParN n -> return $ fromIntegral n

@@ -8,20 +8,14 @@ -- Portability : non-portable -- module Control.Scheduler.Queue-  ( -- * Queue-    -- ** Pure queue-    Queue-  , emptyQueue-  , pushQueue-  , popQueue-  -- ** Job queue-  , Job(Retire, Job_)+  (  -- * Job queue+    Job(Retire, Job_)   , mkJob   , JQueue   , newJQueue   , pushJQueue   , popJQueue-  , flushResults+  , readResults   -- * Tools   ) where @@ -31,67 +25,64 @@ import Data.Atomics (atomicModifyIORefCAS) import Data.IORef +data Queue m a = Queue+  { qQueue   :: ![Job m a]+  , qStack   :: ![Job m a]+  , qResults :: ![IORef (Maybe a)]+  , qBaton   :: !(MVar ())+  } --- | Pure functional Okasaki queue with total length-data Queue a = Queue { qQueue :: ![a]-                     , qStack :: ![a]-                     } -emptyQueue :: Queue a-emptyQueue = Queue [] []--pushQueue :: Queue a -> a -> Queue a-pushQueue queue@Queue {qStack} x = queue {qStack = x : qStack}--popQueue :: Queue a -> Maybe (a, Queue a)-popQueue queue@Queue {qQueue, qStack} =-  case qQueue of+popQueue :: Queue m a -> Maybe (Job m a, Queue m a)+popQueue queue =+  case qQueue queue of     x:xs -> Just (x, queue {qQueue = xs})     [] ->-      case reverse qStack of+      case reverse (qStack queue) of         []   -> Nothing-        y:ys -> Just (y, Queue {qQueue = ys, qStack = []})+        y:ys -> Just (y, queue {qQueue = ys, qStack = []})  data Job m a-  = Job !(IORef a) !(m a)+  = Job !(IORef (Maybe a)) !(m a)   | Job_ !(m ())   | Retire   mkJob :: MonadIO m => m a -> m (Job m a) mkJob action = do-  resRef <- liftIO $ newIORef $ error "mkJob: result is uncomputed"+  resRef <- liftIO $ newIORef Nothing   return $!     Job resRef $ do       res <- action-      liftIO $ writeIORef resRef res-      return res---newtype JQueue m a =-  JQueue (IORef (Queue (Job m a), [IORef a], MVar ()))+      liftIO $ writeIORef resRef $ Just res+      return $! res +newtype JQueue m a = JQueue (IORef (Queue m a))  newJQueue :: MonadIO m => m (JQueue m a)-newJQueue = do-  newBaton <- liftIO newEmptyMVar-  queueRef <- liftIO $ newIORef (emptyQueue, [], newBaton)-  return $ JQueue queueRef+newJQueue =+  liftIO $ do+    newBaton <- newEmptyMVar+    queueRef <- newIORef (Queue [] [] [] newBaton)+    return $ JQueue queueRef   pushJQueue :: MonadIO m => JQueue m a -> Job m a -> m ()-pushJQueue (JQueue jQueueRef) job = do-  newBaton <- liftIO newEmptyMVar-  join $-    liftIO $ atomicModifyIORefCAS-      jQueueRef-      (\(queue, resRefs, baton) ->-         ( ( pushQueue queue job-           , case job of-               Job resRef _ -> resRef : resRefs-               _            -> resRefs-           , newBaton)-         , liftIO $ putMVar baton ()))+pushJQueue (JQueue jQueueRef) job =+  liftIO $ do+    newBaton <- newEmptyMVar+    join $+      atomicModifyIORefCAS+        jQueueRef+        (\Queue {qQueue, qStack, qResults, qBaton} ->+           ( Queue+               qQueue+               (job : qStack)+               (case job of+                  Job resRef _ -> resRef : qResults+                  _            -> qResults)+               newBaton+           , liftIO $ putMVar qBaton ()))   popJQueue :: MonadIO m => JQueue m a -> m (Maybe (m ()))@@ -99,19 +90,20 @@   where     inner =       join $-      atomicModifyIORefCAS jQueueRef $ \jQueue@(queue, resRefs, baton) ->+      atomicModifyIORefCAS jQueueRef $ \queue ->         case popQueue queue of-          Nothing -> (jQueue, readMVar baton >> inner)+          Nothing -> (queue, readMVar (qBaton queue) >> inner)           Just (job, newQueue) ->-            ( (newQueue, resRefs, baton)+            ( newQueue             , case job of                 Job _ action -> return $ Just (void action)                 Job_ action_ -> return $ Just action_                 Retire       -> return Nothing) -flushResults :: MonadIO m => JQueue m a -> m [a]-flushResults (JQueue jQueueRef) =+++readResults :: MonadIO m => JQueue m a -> m [Maybe a]+readResults (JQueue jQueueRef) =   liftIO $ do-    resRefs <--      atomicModifyIORefCAS jQueueRef $ \(queue, resRefs, baton) -> ((queue, [], baton), resRefs)-    mapM readIORef $ reverse resRefs+    resRefs <- qResults <$> readIORef jQueueRef+    mapM readIORef resRefs

@@ -1,3 +1,5 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE ScopedTypeVariables #-} module Control.SchedulerSpec   ( spec   ) where@@ -8,8 +10,11 @@ import qualified Control.Exception as EUnsafe import Control.Exception.Base (ArithException(DivideByZero),                                AsyncException(ThreadKilled))-import Control.Scheduler import Control.Monad+import qualified Data.Foldable as F (traverse_)+import Control.Scheduler+import Data.Bits (complement)+import Data.IORef import Data.List (sort) import Test.Hspec import Test.QuickCheck@@ -17,7 +22,12 @@ import Test.QuickCheck.Monadic import UnliftIO.Async import UnliftIO.Exception hiding (assert)+#if !MIN_VERSION_base(4,11,0)+import Data.Semigroup+#endif +concurrentProperty :: IO Property -> Property+concurrentProperty = within 1000000 . monadicIO . run  instance Arbitrary Comp where   arbitrary =@@ -30,14 +40,13 @@  prop_SameList :: Comp -> [Int] -> Property prop_SameList comp xs =-  monadicIO $ run $ do+  concurrentProperty $ do     xs' <- withScheduler comp $ \scheduler -> mapM_ (scheduleWork scheduler . return) xs     return (xs === xs')  prop_Recursive :: Comp -> [Int] -> Property prop_Recursive comp xs =-  monadicIO $-  run $ do+  concurrentProperty $ do     xs' <- withScheduler comp (schedule xs)     return (sort xs === sort xs')   where@@ -47,8 +56,7 @@  prop_Serially :: Comp -> [Int] -> Property prop_Serially comp xs =-  monadicIO $-  run $ do+  concurrentProperty $ do     xs' <- schedule xs     return (xs === concat xs')   where@@ -60,8 +68,7 @@  prop_Nested :: Comp -> [Int] -> Property prop_Nested comp xs =-  monadicIO $-  run $ do+  concurrentProperty $ do     xs' <- schedule xs     return (sort xs === sort (concat xs'))   where@@ -69,16 +76,31 @@     schedule (y:ys) =       withScheduler comp (\s -> scheduleWork s (schedule ys >>= \ys' -> return (y : concat ys'))) -prop_Traversable :: Comp -> [Int] -> Fun Int Int -> Property-prop_Traversable comp xs f =-  monadicIO $ run $ (===) <$> traverse f' xs <*> traverseConcurrently comp f' xs+-- | Check whether all jobs have been completed (similar roprop_Traverse)+prop_AllJobsProcessed :: Comp -> [Int] -> Property+prop_AllJobsProcessed comp jobs =+  monadicIO+    ((=== jobs) <$>+     run (withScheduler comp $ \scheduler -> mapM_ (scheduleWork scheduler . pure) jobs))++prop_Traverse :: Comp -> [Int] -> Fun Int Int -> Property+prop_Traverse comp xs f =+  concurrentProperty $ (===) <$> traverse f' xs <*> traverseConcurrently comp f' xs   where     f' = pure . apply f +-- prop_ReplicateM :: Comp -> [Int] -> Fun Int Int -> Property+-- prop_ReplicateM comp xs f =+--   monadicIO $ run $ do++--   (===) <$> traverse f' xs <*> traverseConcurrently comp f' xs+--   where+--     f' = pure . apply f++ prop_ArbitraryCompNested :: [(Comp, Int)] -> Property prop_ArbitraryCompNested xs =-  monadicIO $-  run $ do+  concurrentProperty $ do     xs' <- schedule xs     return (sort (map snd xs) === sort (concat xs'))   where@@ -175,32 +197,146 @@          (withScheduler_ comp $ \scheduler ->             replicateM_ n (scheduleWork scheduler (myThreadId >>= killThread)))) +prop_FinishEarly_ :: Comp -> Property+prop_FinishEarly_ comp =+  comp /= Seq ==> concurrentProperty $ do+    ref <- newIORef True+    withScheduler_ comp $ \scheduler ->+      scheduleWork_ scheduler (terminate_ scheduler >> threadDelay 10000 >> writeIORef ref False)+    property <$> 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]) +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])++prop_FinishBeforeStarting :: Comp -> Property+prop_FinishBeforeStarting comp =+  concurrentProperty $ do+    res <-+      withScheduler comp $ \scheduler -> do+        void $ terminate scheduler 1+        scheduleWork scheduler (threadDelay 10000 >> pure 2)+    pure (res === [1 :: Int])++prop_FinishWithBeforeStarting :: Comp -> Int -> Property+prop_FinishWithBeforeStarting comp n =+  concurrentProperty $ do+    res <-+      withScheduler comp $ \scheduler -> do+        void $ terminateWith scheduler n+        scheduleWork scheduler $ pure (complement n)+    pure (res === [n])++prop_TrivialSchedulerSameAsSeq_ :: [Int] -> Property+prop_TrivialSchedulerSameAsSeq_ zs =+  concurrentProperty $ do+    let consRef xsRef x = atomicModifyIORef' xsRef $ \ xs -> (x:xs, ())+        trivial = trivialScheduler_+    nRef <- newIORef False+    xRefs <- newIORef []+    yRefs <- newIORef []+    withScheduler_ Seq $ \scheduler -> do+      writeIORef nRef (numWorkers scheduler == numWorkers trivial)+      mapM_ (scheduleWork_ scheduler . consRef xRefs) zs+    mapM_ (scheduleWork_ trivial . consRef yRefs) zs+    nSame <- readIORef nRef+    xs <- readIORef xRefs+    ys <- readIORef yRefs+    pure (nSame .&&. xs === ys)++newtype Elem = Elem Int deriving (Eq, Show)++instance Exception Elem+++-- | Check if an element is in the list with an exception+prop_TraverseConcurrently_ :: Comp -> [Int] -> Int -> Property+prop_TraverseConcurrently_ comp xs x =+  concurrentProperty $ do+    let f i+          | i == x = throwIO $ Elem x+          | otherwise = pure ()+    eRes :: Either Elem () <- try $ traverse_ f xs+    eRes' <- try $ traverseConcurrently_ comp f xs+    return (eRes === eRes')++-- | 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+prop_TraverseConcurrentlyInfinite_ comp xs x =+  comp /= Seq ==> concurrentProperty $ do+    let f i+          | i == x = throwIO $ Elem x+          | otherwise = pure ()+        xs' = xs ++ [x] -- ++ [0 ..]+    eRes :: Either Elem () <- try $ F.traverse_ f xs'+    eRes' <- try $ traverseConcurrently_ comp f xs'+    return (eRes === eRes')+ spec :: Spec spec = do+  describe "Comp" $ do+    describe "Monoid" $ do+      it "x <> mempty = x" $ property $ \(x :: Comp) -> x <> mempty === x+      it "mempty <> x = x" $ property $ \(x :: Comp) -> mempty <> x === x+      it "x <> (y <> z) = (x <> y) <> z" $+        property $ \(x :: Comp) y z -> x <> (y <> z) === (x <> y) <> z+      it "mconcat = foldr '(<>)' mempty" $+        property $ \(xs :: [Comp]) -> mconcat xs === foldr (<>) mempty xs+    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 "Seq" $ do-    it "SameList" $ property $ prop_SameList Seq-    it "Recursive" $ property $ prop_Recursive Seq-    it "Nested" $ property $ prop_Nested Seq-    it "Serially" $ property $ prop_Serially Seq+    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_   describe "ParOn" $ do-    it "SameList" $ property $ \ cs -> prop_SameList (ParOn cs)-    it "Recursive" $ property $ \ cs -> prop_Recursive (ParOn cs)-    it "Nested" $ property $ \ cs -> prop_Nested (ParOn cs)-    it "Serially" $ property $ \ cs -> prop_Serially (ParOn cs)+    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 "ArbitraryNested" $ property prop_ArbitraryCompNested-    it "traverseConcurrently == traverse" $ property prop_Traversable+    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" $ property prop_CatchDivideByZero-    it "CatchDivideByZeroNested" $ property prop_CatchDivideByZeroNested-    it "KillBlockedCoworker" $ property prop_KillBlockedCoworker-    it "KillSleepingCoworker" $ property prop_KillSleepingCoworker-    it "ExpectAsyncException" $ property prop_ExpectAsyncException-    it "WorkerCaughtAsyncException" $ property prop_WorkerCaughtAsyncException-    it "AllWorkersDied" $ property prop_AllWorkersDied+    it "CatchDivideByZero" $ timed prop_CatchDivideByZero+    it "CatchDivideByZeroNested" $ timed prop_CatchDivideByZeroNested+    it "KillBlockedCoworker" $ timed prop_KillBlockedCoworker+    it "KillSleepingCoworker" $ timed prop_KillSleepingCoworker+    it "ExpectAsyncException" $ timed prop_ExpectAsyncException+    it "WorkerCaughtAsyncException" $ timed prop_WorkerCaughtAsyncException+    it "AllWorkersDied" $ timed prop_AllWorkersDied+    it "traverseConcurrently_" $ timed prop_TraverseConcurrently_+    it "traverseConcurrentlyInfinite_" $ property prop_TraverseConcurrentlyInfinite_+  describe "Premature" $ do+    it "FinishEarly" $ timed prop_FinishEarly+    it "FinishEarly_" $ timed prop_FinishEarly_+    it "FinishEarlyWith" $ timed prop_FinishEarlyWith+    it "FinishBeforeStarting" $ timed prop_FinishBeforeStarting+    it "FinishWithBeforeStarting" $ timed prop_FinishWithBeforeStarting +timed :: Testable prop => prop -> Property+timed = within 2000000  -- | Assert a synchronous exception assertExceptionIO :: (NFData a, Exception exc) =>