scheduler 1.4.2.2 → 1.4.2.3
raw patch · 6 files changed
+22/−204 lines, 6 filesdep −asyncdep −criteriondep −monad-pardep ~basedep ~primitivedep ~unliftiosetup-changedPVP ok
version bump matches the API change (PVP)
Dependencies removed: async, criterion, monad-par, parallel, streamly
Dependency ranges changed: base, primitive, unliftio
API changes (from Hackage documentation)
Files
- Setup.hs +0/−28
- bench/Scheduler.hs +0/−103
- scheduler.cabal +4/−27
- src/Control/Scheduler.hs +15/−37
- src/Control/Scheduler/Internal.hs +2/−2
- tests/doctests.hs +1/−7
Setup.hs view
@@ -1,33 +1,5 @@-{-# 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
− bench/Scheduler.hs
@@ -1,103 +0,0 @@-module Main where--import qualified Control.Concurrent.Async as A (mapConcurrently, replicateConcurrently)-import Control.Monad (replicateM)-import Control.Monad.Par (IVar, Par, get, newFull_, runParIO) --, parMapM)-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 =- 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.drain $ 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.drain $ 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 ++ ")"
scheduler.cabal view
@@ -1,7 +1,7 @@ name: scheduler-version: 1.4.2.2+version: 1.4.2.3 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.+description: A work stealing scheduler that is primarily developed for [massiv](https://github.com/lehins/massiv) array library, but it is general enough to be useful for any computation that fits the model of few workers and many jobs. homepage: https://github.com/lehins/haskell-scheduler license: BSD3 license-file: LICENSE@@ -9,17 +9,11 @@ maintainer: alexey@kuleshevi.ch copyright: 2018-2019 Alexey Kuleshevich category: Parallelism, Concurrency-build-type: Custom+build-type: Simple extra-source-files: README.md , CHANGELOG.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@@ -31,7 +25,7 @@ , deepseq , exceptions , unliftio-core- , primitive >= 0.5.2.1+ , primitive >= 0.6.4 default-language: Haskell2010 ghc-options: -Wall@@ -64,23 +58,6 @@ , scheduler , template-haskell , vector- 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- , monad-par- , scheduler- , parallel- , unliftio >= 0.2.10- , streamly >= 0.6.1 default-language: Haskell2010 source-repository head
src/Control/Scheduler.hs view
@@ -75,12 +75,9 @@ import qualified Data.Foldable as F (foldl', traverse_, toList) import Data.IORef import Data.Maybe (catMaybes)-import Data.Primitive.Array+import Data.Primitive.SmallArray import Data.Primitive.MutVar import Data.Traversable-#if !MIN_VERSION_primitive(0,6,2)-import Control.Monad.ST-#endif -- | Get the underlying `Scheduler`, which cannot access `WorkerStates`. --@@ -95,36 +92,17 @@ initWorkerStates :: MonadIO m => Comp -> (WorkerId -> m s) -> m (WorkerStates s) initWorkerStates comp initState = do nWorkers <- getCompWorkers comp- workerStates <- mapM (initState . WorkerId) [0 .. nWorkers - 1]+ arr <- liftIO $ newSmallArray nWorkers (error "Uninitialized")+ let go i = when (i < nWorkers) $ do+ state <- initState (WorkerId i)+ liftIO $ writeSmallArray arr i state+ go (i + 1)+ go 0+ workerStates <- liftIO $ unsafeFreezeSmallArray arr mutex <- liftIO $ newIORef False pure WorkerStates- { _workerStatesComp = comp- , _workerStatesArray = arrayFromListN nWorkers workerStates- , _workerStatesMutex = mutex- }--arrayFromListN :: Int -> [a] -> Array a-#if MIN_VERSION_primitive(0,6,2)-arrayFromListN = fromListN-#else--- Modified copy from primitive-0.7.0.0-arrayFromListN n l =- runST $ do- ma <- newArray n (error "initWorkerStates: uninitialized element")- let go !ix [] =- if ix == n- then return ()- else error "initWorkerStates: list length less than specified size"- go !ix (x:xs) =- if ix < n- then do- writeArray ma ix x- go (ix + 1) xs- else error "initWorkerStates: list length greater than specified size"- go 0 l- unsafeFreezeArray ma-#endif+ {_workerStatesComp = comp, _workerStatesArray = workerStates, _workerStatesMutex = mutex} -- | Get the computation strategy the states where initialized with. --@@ -213,7 +191,7 @@ scheduleWorkState :: SchedulerWS s m a -> (s -> m a) -> m () scheduleWorkState schedulerS withState = scheduleWorkId (_getScheduler schedulerS) $ \(WorkerId i) ->- withState (indexArray (_workerStatesArray (_workerStates schedulerS)) i)+ withState (indexSmallArray (_workerStatesArray (_workerStates schedulerS)) i) -- | Same as `scheduleWorkState`, but dont' keep the result of computation. --@@ -221,7 +199,7 @@ scheduleWorkState_ :: SchedulerWS s m () -> (s -> m ()) -> m () scheduleWorkState_ schedulerS withState = scheduleWorkId_ (_getScheduler schedulerS) $ \(WorkerId i) ->- withState (indexArray (_workerStatesArray (_workerStates schedulerS)) i)+ withState (indexSmallArray (_workerStatesArray (_workerStates schedulerS)) i) -- | Get the number of workers. Will mainly depend on the computation strategy and/or number of@@ -463,10 +441,10 @@ -- * 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 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.+-- * __Warning__ It is pretty dangerous to schedule jobs that can block, because it might+-- lead to a potential 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. -- -- >>> withScheduler (ParOn [1,2]) $ \s -> newEmptyMVar >>= (\ mv -> scheduleWork s (readMVar mv) >> scheduleWork s (putMVar mv ())) -- [(),()]
src/Control/Scheduler/Internal.hs view
@@ -25,7 +25,7 @@ import Control.Scheduler.Computation import Control.Scheduler.Queue import Data.IORef-import Data.Primitive.Array+import Data.Primitive.SmallArray -- | Computed outcome of scheduled jobs. --@@ -101,7 +101,7 @@ -- @since 1.4.0 data WorkerStates s = WorkerStates { _workerStatesComp :: !Comp- , _workerStatesArray :: !(Array s)+ , _workerStatesArray :: !(SmallArray s) , _workerStatesMutex :: !(IORef Bool) }
tests/doctests.hs view
@@ -1,12 +1,6 @@ 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+main = doctest ["src"]