scheduler (empty) → 1.0.0
raw patch · 11 files changed
+1096/−0 lines, 11 filesdep +QuickCheckdep +atomic-primopsdep +basebuild-type:Customsetup-changed
Dependencies added: QuickCheck, atomic-primops, base, deepseq, doctest, exceptions, hspec, scheduler, template-haskell, unliftio, unliftio-core
Files
- LICENSE +30/−0
- README.md +195/−0
- Setup.hs +33/−0
- scheduler.cabal +66/−0
- src/Control/Scheduler.hs +314/−0
- src/Control/Scheduler/Computation.hs +86/−0
- src/Control/Scheduler/Queue.hs +117/−0
- tests/Control/SchedulerSpec.hs +232/−0
- tests/Main.hs +10/−0
- tests/Spec.hs +1/−0
- tests/doctests.hs +12/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Alexey Kuleshevich (c) 2019++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Alexey Kuleshevich nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,195 @@+# scheduler++This is a work stealing scheduler, which is very useful for tasks parallelization.++Whenever you have many actions you'd like to perform in parallel, but would only like to use a few+threads to do the actual computation, this package is for you.++| Language | Travis | AppVeyor | Hackage | Nightly | LTS |+|:--------:|:------:|:--------:|:-------:|:-------:|:---:|+|  | [](https://travis-ci.org/lehins/haskell-scheduler) | [](https://ci.appveyor.com/project/lehins/haskell-scheduler) | [](https://hackage.haskell.org/package/scheduler)| [](https://www.stackage.org/nightly/package/scheduler) | [](https://www.stackage.org/lts/package/scheduler) |+++## QuickStart++A few examples in order to get up and running quickly.++### Schedule simple actions++Work scheduling that does some side effecty stuff and discards the results:++```haskell+interleaveFooBar :: IO ()+interleaveFooBar = do+ withScheduler_ (ParN 2) $ \ scheduler -> do+ putStrLn "Scheduling 1st job"+ scheduleWork scheduler (putStr "foo")+ putStrLn "Scheduling 2nd job"+ scheduleWork scheduler (putStr "bar")+ putStrLn "Awaiting for jobs to be executed:"+ putStrLn "\nDone"+```++In the example above two workers will be created to handle the only two jobs that have been+scheduled. Printing with `putStr` is not thread safe, so the output that you would get with above+function is likely be interleaved:++```haskell+λ> interleaveFooBar+Scheduling 1st job+Scheduling 2nd job+Awaiting for jobs to be executed:+foboar+Done+```++Important to note that only when inner action supplied to the `withScheduler_` exits will the+scheduler start executing scheduled jobs.++### Keeping the results of computation++Another common scenario is to schedule some jobs that produce useful results. In the example below+four works will be spawned off. Due to `ParOn` each of the workers will be pinned to a particular+core.++```haskell+scheduleSums :: IO [Int]+scheduleSums =+ withScheduler (ParOn [1..4]) $ \ scheduler -> do+ scheduleWork scheduler $ pure (10 + 1)+ scheduleWork scheduler $ pure (20 + 2)+ scheduleWork scheduler $ pure (30 + 3)+ scheduleWork scheduler $ pure (40 + 4)+ scheduleWork scheduler $ pure (50 + 5)+```++Despite that the fact that sums are computed in parallel, the results of computation will appear in+the same order they've been scheduled:++```haskell+λ> scheduleSums+[11,22,33,44,55]+```++### Exceptions++Whenever any of the scheduled jobs result in an exception, all of the workers will be killed and the+exception will get re-thrown in the scheduling thread:++```haskell+infiniteJobs :: IO ()+infiniteJobs = do+ withScheduler_ (ParN 5) $ \ scheduler -> do+ scheduleWork scheduler $ putStrLn $ repeat 'a'+ scheduleWork scheduler $ putStrLn $ repeat 'b'+ scheduleWork scheduler $ putStrLn $ repeat 'c'+ scheduleWork scheduler $ pure (4 `div` (0 :: Int))+ scheduleWork scheduler $ putStrLn $ repeat 'd'+ putStrLn "\nDone"+```++Note, that if there was no exception, printing would never stop.++```haskell+λ> infiniteJobs+aaaaaaaaabcdd*** Exception: divide by zero+```++### Nested jobs++Scheduling actions can themselves schedule actions indefinitely. That of course means that order of+results produced is no longer deterministic, which is to be expected.++```haskell+nestedJobs :: IO ()+nestedJobs = do+ withScheduler_ (ParN 5) $ \ scheduler -> do+ scheduleWork scheduler $ putStr $ replicate 10 'a'+ scheduleWork scheduler $ do+ putStr $ replicate 10 'b'+ scheduleWork scheduler $ do+ putStr $ replicate 10 'c'+ scheduleWork scheduler $ putStr $ replicate 10 'e'+ scheduleWork scheduler $ putStr $ replicate 10 'd'+ scheduleWork scheduler $ putStr $ replicate 10 'f'+ putStrLn "\nDone"+```++The order in which characters appear is important, since it directly relates to the actual order in+which jobs are being scheduled and executed:++* `c`, `d` and `e` characters will always appear after `b`+* `e` will always appear after `c`++```haskell+λ> nestedJobs+abbafbafbafbafbafbafbafbafbaffcdcdcdcdcdcdcdcdcdcdeeeeeeeeee+Done+```++### Nested parallelism++Nothing really prevents you from having a scheduler within a scheduler. Of course, having multiple+schedulers at the same time seems like an unnecessary overhead, which it is, but if you do have a+use case for it, don't make me stop you, it is OK to go that route.++```haskell+nestedSchedulers :: IO ()+nestedSchedulers = do+ withScheduler_ (ParN 2) $ \ outerScheduler -> do+ scheduleWork outerScheduler $ putStr $ replicate 10 'a'+ scheduleWork outerScheduler $ do+ putStr $ replicate 10 'b'+ withScheduler_ (ParN 2) $ \ innerScheduler -> do+ scheduleWork innerScheduler $ do+ putStr $ replicate 10 'c'+ scheduleWork outerScheduler $ putStr $ replicate 10 'e'+ scheduleWork innerScheduler $ putStr $ replicate 10 'd'+ scheduleWork outerScheduler $ putStr $ replicate 10 'f'+ putStrLn "\nDone"+```++Note that the inner scheduler's job schedules a job for the outer scheduler, which is a bit crazy,+but totally safe.++```haskell+λ> nestedSchedulers+aabababababababababbffffffffffcccccccdcdcdcdddededededeeeeee+Done+```++### Single worker schedulers++If we only have one worker, than everything becomes sequential and deterministic. Consider the same+example from before, but with `Seq` computation strategy.++```haskell+nestedSequentialSchedulers :: IO ()+nestedSequentialSchedulers = do+ withScheduler_ Seq $ \ outerScheduler -> do+ scheduleWork outerScheduler $ putStr $ replicate 10 'a'+ scheduleWork outerScheduler $ do+ putStr $ replicate 10 'b'+ withScheduler_ Seq $ \ innerScheduler -> do+ scheduleWork innerScheduler $ do+ putStr $ replicate 10 'c'+ scheduleWork outerScheduler $ putStr $ replicate 10 'e'+ scheduleWork innerScheduler $ putStr $ replicate 10 'd'+ scheduleWork outerScheduler $ putStr $ replicate 10 'f'+ putStrLn "\nDone"+```++No more interleaving, everything is done in the same order each time the function is invoked.++```haskell+λ> nestedSchedulers+aaaaaaaaaabbbbbbbbbbccccccccccddddddddddffffffffffeeeeeeeeee+Done+```++## Avoiding deadlocks++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.+
+ Setup.hs view
@@ -0,0 +1,33 @@+{-# LANGUAGE CPP #-}+{-# OPTIONS_GHC -Wall #-}+module Main (main) where++#ifndef MIN_VERSION_cabal_doctest+#define MIN_VERSION_cabal_doctest(x,y,z) 0+#endif++#if MIN_VERSION_cabal_doctest(1,0,0)++import Distribution.Extra.Doctest ( defaultMainWithDoctests )+main :: IO ()+main = defaultMainWithDoctests "doctests"++#else++#ifdef MIN_VERSION_Cabal+-- If the macro is defined, we have new cabal-install,+-- but for some reason we don't have cabal-doctest in package-db+--+-- Probably we are running cabal sdist, when otherwise using new-build+-- workflow+#warning You are configuring this package without cabal-doctest installed. \+ The doctests test-suite will not work as a result. \+ To fix this, install cabal-doctest before configuring.+#endif++import Distribution.Simple++main :: IO ()+main = defaultMain++#endif
+ scheduler.cabal view
@@ -0,0 +1,66 @@+name: scheduler+version: 1.0.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+license: BSD3+license-file: LICENSE+author: Alexey Kuleshevich+maintainer: alexey@kuleshevi.ch+copyright: 2018-2019 Alexey Kuleshevich+category: Parallelism, Concurrency+build-type: Custom+extra-source-files: README.md+cabal-version: >=1.10++custom-setup+ setup-depends:+ base+ , Cabal+ , cabal-doctest >=1.0.6++library+ hs-source-dirs: src+ exposed-modules: Control.Scheduler++ other-modules: Control.Scheduler.Computation+ , Control.Scheduler.Queue+ build-depends: base >= 4.9 && < 5+ , atomic-primops+ , deepseq+ , exceptions+ , unliftio-core++ default-language: Haskell2010+ ghc-options: -Wall+++test-suite tests+ type: exitcode-stdio-1.0+ hs-source-dirs: tests+ main-is: Main.hs+ other-modules: Spec+ , Control.SchedulerSpec+ build-depends: base+ , deepseq+ , scheduler+ , hspec+ , QuickCheck+ , unliftio++ default-language: Haskell2010+ ghc-options: -Wall -fno-warn-orphans -threaded -with-rtsopts=-N++test-suite doctests+ type: exitcode-stdio-1.0+ hs-source-dirs: tests+ main-is: doctests.hs+ build-depends: base+ , doctest >=0.15+ , scheduler+ , template-haskell+ default-language: Haskell2010++source-repository head+ type: git+ location: https://github.com/lehins/massiv
+ src/Control/Scheduler.hs view
@@ -0,0 +1,314 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE PatternSynonyms #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+-- |+-- Module : Control.Scheduler+-- Copyright : (c) Alexey Kuleshevich 2018-2019+-- License : BSD3+-- Maintainer : Alexey Kuleshevich <lehins@yandex.ru>+-- Stability : experimental+-- Portability : non-portable+--+module Control.Scheduler+ ( -- * Scheduler and strategies+ Comp(..)+ , Scheduler(..)+ -- * Initialize Scheduler+ , withScheduler+ , withScheduler_+ -- * Useful functions+ , traverseConcurrently+ , traverseConcurrently_+ , traverse_+ -- * Exceptions+ -- $exceptions+ ) where++import Control.Concurrent+import Control.Exception+import Control.Scheduler.Computation+import Control.Scheduler.Queue+import Control.Monad+import Control.Monad.IO.Unlift+import Data.Atomics (atomicModifyIORefCAS, atomicModifyIORefCAS_)+import Data.Foldable as F (foldl')+import Data.IORef+import Data.Traversable++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.+--+-- @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.+ }++-- | This is generally a faster way to traverse while ignoring the result rather than using `mapM_`.+--+-- @since 1.0.0+traverse_ :: (Applicative f, Foldable t) => (a -> f ()) -> t a -> f ()+traverse_ f = F.foldl' (\c a -> c *> f a) (pure ())+++-- | Map an action over each element of the `Traversable` @t@ acccording to the supplied computation+-- strategy.+--+-- @since 1.0.0+traverseConcurrently :: (MonadUnliftIO m, Traversable t) => Comp -> (a -> m b) -> t a -> m (t b)+traverseConcurrently comp f xs = do+ ys <- withScheduler comp $ \s -> traverse_ (scheduleWork s . f) xs+ pure $ transList ys xs++transList :: Traversable t => [a] -> t b -> t a+transList xs' = snd . mapAccumL withR xs'+ where+ withR (x:xs) _ = (xs, x)+ withR _ _ = error "Impossible<traverseConcurrently> - Mismatched sizes"++-- | Just like `traverseConcurrently`, but restricted to `Foldable` and discards the results of+-- computation.+--+-- @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++scheduleJobs :: MonadIO m => Jobs m a -> m a -> m ()+scheduleJobs = scheduleJobsWith mkJob++-- | Similarly to `scheduleWork`, but ignores the result of computation, thus having less overhead.+--+-- @since 1.0.0+scheduleJobs_ :: MonadIO m => Jobs m a -> m b -> m ()+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)+ job <-+ mkJob' $ do+ res <- action+ res `seq` dropCounterOnZero jobsCountRef $ retireWorkersN jobsQueue jobsNumWorkers+ return res+ pushJQueue jobsQueue job++-- | Helper function to place required number of @Retire@ instructions on the job queue.+retireWorkersN :: MonadIO m => JQueue m a -> Int -> m ()+retireWorkersN jobsQueue n = traverse_ (pushJQueue jobsQueue) $ replicate n Retire++-- | Decrease a counter by one and perform an action when it drops down to zero.+dropCounterOnZero :: MonadIO m => IORef Int -> m () -> m ()+dropCounterOnZero counterRef onZero = do+ jc <-+ liftIO $ atomicModifyIORefCAS+ counterRef+ (\ !i' ->+ let !i = i' - 1+ in (i, i))+ when (jc == 0) onZero+++-- | Runs the worker until the job queue is exhausted, at which point it will exhecute the final task+-- of retirement and return+runWorker :: MonadIO m =>+ JQueue m a+ -> m () -- ^ Action to run upon retirement+ -> m ()+runWorker jQueue onRetire = go+ where+ go =+ popJQueue jQueue >>= \case+ Just job -> job >> go+ Nothing -> onRetire+++-- | Initialize a scheduler and submit jobs that will be computed sequentially or in parallelel,+-- which is determined by the `Comp`utation strategy.+--+-- Here are some cool properties about the `withScheduler`:+--+-- * This function will block until all of the submitted jobs have finished or at least one of them+-- resulted in an exception, which will be re-thrown at the callsite.+--+-- * It is totally fine for nested jobs to submit more jobs for the same or other scheduler+--+-- * 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,+-- since there are two scheduled workers, and one can unblock the other, but the second scenario+-- immediately results in a deadlock.+--+-- >>> withScheduler (ParOn [1,2]) $ \s -> newEmptyMVar >>= (\ mv -> scheduleWork s (readMVar mv) >> scheduleWork s (putMVar mv ()))+-- [(),()]+-- >>> import System.Timeout+-- >>> timeout 1000000 $ withScheduler (ParOn [1]) $ \s -> newEmptyMVar >>= (\ mv -> scheduleWork s (readMVar mv) >> scheduleWork s (putMVar mv ()))+-- Nothing+--+-- __Important__: In order to get work done truly in parallel, program needs to be compiled with+-- @-threaded@ GHC flag and executed with @+RTS -N -RTS@ to use all available cores.+--+-- @since 1.0.0+withScheduler ::+ MonadUnliftIO m+ => Comp -- ^ Computation strategy+ -> (Scheduler m a -> m b)+ -- ^ Action that will be scheduling all the work.+ -> m [a]+withScheduler comp = withSchedulerInternal comp scheduleJobs flushResults+++-- | Same as `withScheduler`, but discards results of submitted jobs.+--+-- @since 1.0.0+withScheduler_ ::+ MonadUnliftIO m+ => Comp -- ^ Computation strategy+ -> (Scheduler m a -> m b)+ -- ^ Action that will be scheduling all the work.+ -> m ()+withScheduler_ comp = withSchedulerInternal comp scheduleJobs_ (const (pure ()))+++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+ -> (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+ 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.+ _ <- onScheduler scheduler+ -- / Ensure at least something gets scheduled, so retirement can be triggered+ jc <- liftIO $ readIORef jobsCountRef+ when (jc == 0) $ scheduleJobs_ jobs (pure ())+ let spawnWorkersWith fork ws =+ withRunInIO $ \run ->+ forM ws $ \w ->+ fork w $ \unmask ->+ catch+ (unmask $ run $ runWorker jobsQueue onRetire)+ (run . handleWorkerException jobsQueue workDoneMVar jobsNumWorkers)+ {-# INLINE spawnWorkersWith #-}+ spawnWorkers =+ case comp of+ Seq -> return []+ -- \ no need to fork threads for a sequential computation+ Par -> spawnWorkersWith forkOnWithUnmask [1 .. jobsNumWorkers]+ ParOn ws -> spawnWorkersWith forkOnWithUnmask ws+ ParN _ -> spawnWorkersWith (\_ -> forkIOWithUnmask) [1 .. jobsNumWorkers]+ {-# INLINE spawnWorkers #-}+ doWork = 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+ -- \ 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+ -- \ 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)+++-- | Specialized exception handler for the work scheduler.+handleWorkerException ::+ MonadIO m => JQueue m a -> MVar (Maybe WorkerException) -> 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+ -- \ 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`.+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++data WorkerTerminateException =+ WorkerTerminateException+ -- ^ When a brother worker dies of some exception, all the other ones will be terminated+ -- asynchronously with this one.+ deriving (Show)+++instance Exception WorkerTerminateException where+ displayException WorkerTerminateException = "A worker was terminated by the scheduler"+++-- Copy from unliftio:+safeBracketOnError :: MonadUnliftIO m => m a -> (a -> m b) -> (a -> m c) -> m c+safeBracketOnError before after thing = withRunInIO $ \run -> mask $ \restore -> do+ x <- run before+ res1 <- try $ restore $ run $ thing x+ case res1 of+ Left (e1 :: SomeException) -> do+ _ :: Either SomeException b <-+ try $ uninterruptibleMask_ $ run $ after x+ throwIO e1+ Right y -> return y++{- $exceptions++If any one of the workers dies with an exception, even if that exceptions is asynchronous, it will be+re-thrown in the scheduling thread.++>>> let didAWorkerDie = handleJust asyncExceptionFromException (return . (== ThreadKilled)) . fmap or+>>> :t didAWorkerDie+didAWorkerDie :: Foldable t => IO (t Bool) -> IO Bool+>>> didAWorkerDie $ withScheduler Par $ \ s -> scheduleWork s $ pure False+False+>>> didAWorkerDie $ withScheduler Par $ \ s -> scheduleWork s $ myThreadId >>= killThread >> pure False+True+>>> withScheduler Par $ \ s -> scheduleWork s $ myThreadId >>= killThread >> pure False+*** Exception: thread killed++-}
+ src/Control/Scheduler/Computation.hs view
@@ -0,0 +1,86 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE PatternSynonyms #-}+{-# LANGUAGE ScopedTypeVariables #-}+-- |+-- Module : Control.Scheduler.Computation+-- Copyright : (c) Alexey Kuleshevich 2018-2019+-- License : BSD3+-- Maintainer : Alexey Kuleshevich <lehins@yandex.ru>+-- Stability : experimental+-- Portability : non-portable+--+module Control.Scheduler.Computation+ ( Comp(.., Par)+ ) where++import Control.DeepSeq (NFData(..), deepseq)+#if !MIN_VERSION_base(4,11,0)+import Data.Semigroup+#endif+import Data.Word++-- | Computation strategy to use when scheduling work.+data Comp+ = Seq -- ^ Sequential computation+ | ParOn ![Int]+ -- ^ Schedule workers to run on specific capabilities. Specifying an empty list @`ParOn` []@ or+ -- using `Par` will result in utilization of all available capabilities.+ | ParN {-# UNPACK #-} !Word16+ -- ^ Specify the number of workers that will be handling all the jobs. Difference from `ParOn` is+ -- that workers can jump between cores. Using @`ParN` 0@ will result in using all available+ -- capabilities.+ deriving Eq++-- | Parallel computation using all available cores. Same as @`ParOn` []@+pattern Par :: Comp+pattern Par <- ParOn [] where+ Par = ParOn []++instance Show Comp where+ show Seq = "Seq"+ show Par = "Par"+ show (ParOn ws) = "ParOn " ++ show ws+ show (ParN n) = "ParN " ++ show n+ showsPrec _ Seq = ("Seq" ++)+ showsPrec _ Par = ("Par" ++)+ showsPrec 0 comp = (show comp ++)+ showsPrec _ comp = (("(" ++ show comp ++ ")") ++)++instance NFData Comp where+ rnf comp =+ case comp of+ Seq -> ()+ ParOn wIds -> wIds `deepseq` ()+ ParN n -> n `deepseq` ()+ {-# INLINE rnf #-}++instance Monoid Comp where+ mempty = Seq+ {-# INLINE mempty #-}+ mappend = joinComp+ {-# INLINE mappend #-}++instance Semigroup Comp where+ (<>) = joinComp+ {-# INLINE (<>) #-}++joinComp :: Comp -> Comp -> Comp+joinComp x y =+ case x of+ Seq -> y+ Par -> Par+ 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+ ParN n1 ->+ case y of+ Seq -> x+ Par -> Par+ ParOn ys -> ParN (max n1 (fromIntegral (length ys)))+ ParN n2 -> ParN (max n1 n2)+{-# NOINLINE joinComp #-}
+ src/Control/Scheduler/Queue.hs view
@@ -0,0 +1,117 @@+{-# LANGUAGE NamedFieldPuns #-}+-- |+-- Module : Control.Scheduler.Queue+-- Copyright : (c) Alexey Kuleshevich 2018-2019+-- License : BSD3+-- Maintainer : Alexey Kuleshevich <lehins@yandex.ru>+-- Stability : experimental+-- Portability : non-portable+--+module Control.Scheduler.Queue+ ( -- * Queue+ -- ** Pure queue+ Queue+ , emptyQueue+ , pushQueue+ , popQueue+ -- ** Job queue+ , Job(Retire, Job_)+ , mkJob+ , JQueue+ , newJQueue+ , pushJQueue+ , popJQueue+ , flushResults+ -- * Tools+ ) where++import Control.Concurrent.MVar+import Control.Monad (join, void)+import Control.Monad.IO.Unlift+import Data.Atomics (atomicModifyIORefCAS)+import Data.IORef+++-- | 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+ x:xs -> Just (x, queue {qQueue = xs})+ [] ->+ case reverse qStack of+ [] -> Nothing+ y:ys -> Just (y, Queue {qQueue = ys, qStack = []})++data Job m a+ = Job !(IORef 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"+ 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 ()))+++newJQueue :: MonadIO m => m (JQueue m a)+newJQueue = do+ newBaton <- liftIO newEmptyMVar+ queueRef <- liftIO $ newIORef (emptyQueue, [], 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 ()))+++popJQueue :: MonadIO m => JQueue m a -> m (Maybe (m ()))+popJQueue (JQueue jQueueRef) = liftIO inner+ where+ inner =+ join $+ atomicModifyIORefCAS jQueueRef $ \jQueue@(queue, resRefs, baton) ->+ case popQueue queue of+ Nothing -> (jQueue, readMVar baton >> inner)+ Just (job, newQueue) ->+ ( (newQueue, resRefs, baton)+ , 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) =+ liftIO $ do+ resRefs <-+ atomicModifyIORefCAS jQueueRef $ \(queue, resRefs, baton) -> ((queue, [], baton), resRefs)+ mapM readIORef $ reverse resRefs
+ tests/Control/SchedulerSpec.hs view
@@ -0,0 +1,232 @@+module Control.SchedulerSpec+ ( spec+ ) where++import Control.Concurrent (killThread, myThreadId, threadDelay)+import Control.Concurrent.MVar+import Control.DeepSeq+import qualified Control.Exception as EUnsafe+import Control.Exception.Base (ArithException(DivideByZero),+ AsyncException(ThreadKilled))+import Control.Scheduler+import Control.Monad+import Data.List (sort)+import Test.Hspec+import Test.QuickCheck+import Test.QuickCheck.Function+import Test.QuickCheck.Monadic+import UnliftIO.Async+import UnliftIO.Exception hiding (assert)+++instance Arbitrary Comp where+ arbitrary =+ frequency+ [ (20, pure Seq)+ , (10, pure Par)+ , (35, ParOn <$> arbitrary)+ , (35, ParN . getSmall <$> arbitrary)+ ]++prop_SameList :: Comp -> [Int] -> Property+prop_SameList comp xs =+ monadicIO $ run $ 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+ xs' <- withScheduler comp (schedule xs)+ return (sort xs === sort xs')+ where+ schedule [] _ = return ()+ schedule (y:ys) scheduler = scheduleWork scheduler (schedule ys scheduler >> return y)+++prop_Serially :: Comp -> [Int] -> Property+prop_Serially comp xs =+ monadicIO $+ run $ do+ xs' <- schedule xs+ return (xs === concat xs')+ where+ schedule [] = return []+ schedule (y:ys) = do+ y' <- withScheduler comp (`scheduleWork` pure y)+ ys' <- schedule ys+ return (y':ys')++prop_Nested :: Comp -> [Int] -> Property+prop_Nested comp xs =+ monadicIO $+ run $ do+ xs' <- schedule xs+ return (sort xs === sort (concat xs'))+ where+ schedule [] = return []+ 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+ where+ f' = pure . apply f++prop_ArbitraryCompNested :: [(Comp, Int)] -> Property+prop_ArbitraryCompNested xs =+ monadicIO $+ run $ do+ xs' <- schedule xs+ return (sort (map snd xs) === sort (concat xs'))+ where+ schedule [] = return []+ schedule ((c, y):ys) =+ withScheduler c (\s -> scheduleWork s (schedule ys >>= \ys' -> return (y : concat ys')))++-- | Ensure proper exception handling.+prop_CatchDivideByZero :: Comp -> Int -> [Positive Int] -> Property+prop_CatchDivideByZero comp k xs =+ assertExceptionIO+ (== DivideByZero)+ (traverseConcurrently+ comp+ (\i -> return (k `div` i))+ (map getPositive xs ++ [0] ++ map getPositive xs))++-- | Ensure proper exception handling.+prop_CatchDivideByZeroNested :: Comp -> Int -> Positive Int -> Property+prop_CatchDivideByZeroNested comp a (Positive k) = assertExceptionIO (== DivideByZero) (schedule k)+ where+ schedule i+ | i < 0 = return []+ | otherwise =+ withScheduler comp (\s -> scheduleWork s (schedule (i - 1) >> return (a `div` i)))+++-- | Make sure one co-worker can kill another one, of course when there are at least two of.+prop_KillBlockedCoworker :: Comp -> Property+prop_KillBlockedCoworker comp =+ assertExceptionIO+ (== DivideByZero)+ (withScheduler_ comp $ \scheduler ->+ if numWorkers scheduler < 2+ then scheduleWork scheduler $ return ((1 :: Int) `div` (0 :: Int))+ else do+ mv <- newEmptyMVar+ scheduleWork scheduler $ readMVar mv+ scheduleWork scheduler $ return ((1 :: Int) `div` (0 :: Int)))++-- | Make sure one co-worker can kill another one, of course when there are at least two of.+prop_KillSleepingCoworker :: Comp -> Property+prop_KillSleepingCoworker comp =+ assertExceptionIO+ (== DivideByZero)+ (withScheduler_ comp $ \scheduler -> do+ scheduleWork scheduler $ return ((1 :: Int) `div` (0 :: Int))+ scheduleWork scheduler $ do+ threadDelay 500000+ error "This should never happen! Thread should have been killed by now.")+++prop_ExpectAsyncException :: Comp -> Property+prop_ExpectAsyncException comp =+ let didAWorkerDie =+ EUnsafe.handleJust EUnsafe.asyncExceptionFromException (return . (== EUnsafe.ThreadKilled)) .+ fmap or+ in (monadicIO . run . didAWorkerDie . withScheduler comp $ \s ->+ scheduleWork s (myThreadId >>= killThread >> pure False)) .&&.+ (monadicIO . run . fmap not . didAWorkerDie . withScheduler Par $ \s ->+ scheduleWork s $ pure False)++prop_WorkerCaughtAsyncException :: Positive Int -> Property+prop_WorkerCaughtAsyncException (Positive n) =+ assertExceptionIO (== DivideByZero) $ do+ lock <- newEmptyMVar+ result <-+ race (readMVar lock) $+ withScheduler_ (ParN 2) $ \scheduler -> do+ scheduleWork scheduler $ do+ threadDelay (n `mod` 1000000)+ EUnsafe.throwIO DivideByZero+ scheduleWork scheduler $ do+ e <- tryAny $ replicateM_ 5 $ threadDelay 1000000+ case e of+ Right _ -> throwString "Impossible, shouldn't have waited for so long"+ Left exc -> do+ putMVar lock exc+ throwString $+ "I should not have survived: " ++ displayException (exc :: SomeException)+ void $ throwString $+ case result of+ Left innerError -> "Scheduled job cought async exception: " ++ displayException innerError+ Right () -> "Scheduler terminated properly. Should not have happened"++-- | Make sure there is no problems if sub-schedules worker get killed+prop_AllWorkersDied :: Comp -> Comp -> Positive Int -> Property+prop_AllWorkersDied comp1 comp (Positive n) =+ assertAsyncExceptionIO+ (== ThreadKilled)+ (withScheduler_ comp1 $ \scheduler1 ->+ scheduleWork+ scheduler1+ (withScheduler_ comp $ \scheduler ->+ replicateM_ n (scheduleWork scheduler (myThreadId >>= killThread))))++++spec :: Spec+spec = do+ 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+ 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)+ describe "Arbitrary Comp" $ do+ it "ArbitraryNested" $ property prop_ArbitraryCompNested+ it "traverseConcurrently == traverse" $ property prop_Traversable+ 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+++-- | Assert a synchronous exception+assertExceptionIO :: (NFData a, Exception exc) =>+ (exc -> Bool) -- ^ Return True if that is the exception that was expected+ -> IO a -- ^ IO Action that should throw an exception+ -> Property+assertExceptionIO isExc action =+ monadicIO $ do+ hasFailed <-+ run $+ catch+ (do res <- action+ res `deepseq` return False) $ \exc -> displayException exc `deepseq` return (isExc exc)+ assert hasFailed++assertAsyncExceptionIO :: (Exception e, NFData a) => (e -> Bool) -> IO a -> Property+assertAsyncExceptionIO isAsyncExc action =+ monadicIO $ do+ hasFailed <-+ run $+ EUnsafe.catch+ (do res <- action+ res `deepseq` return False)+ (\exc ->+ case EUnsafe.asyncExceptionFromException exc of+ Just asyncExc+ | isAsyncExc asyncExc -> displayException asyncExc `deepseq` pure True+ _ -> EUnsafe.throwIO exc)+ assert hasFailed
+ tests/Main.hs view
@@ -0,0 +1,10 @@+module Main where++import System.IO (BufferMode(LineBuffering), hSetBuffering, stdout)+import Test.Hspec+import Spec++main :: IO ()+main = do+ hSetBuffering stdout LineBuffering+ hspec spec
+ tests/Spec.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover -optF --no-main #-}
+ tests/doctests.hs view
@@ -0,0 +1,12 @@+module Main where++import Build_doctests (flags, pkgs, module_sources)+import Data.Foldable (traverse_)+import Test.DocTest (doctest)++main :: IO ()+main = do+ traverse_ putStrLn args+ doctest args+ where+ args = flags ++ pkgs ++ module_sources