diff --git a/Setup.hs b/Setup.hs --- a/Setup.hs +++ b/Setup.hs @@ -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 diff --git a/bench/Scheduler.hs b/bench/Scheduler.hs deleted file mode 100644 --- a/bench/Scheduler.hs +++ /dev/null @@ -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 ++ ")" diff --git a/scheduler.cabal b/scheduler.cabal --- a/scheduler.cabal +++ b/scheduler.cabal @@ -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 diff --git a/src/Control/Scheduler.hs b/src/Control/Scheduler.hs --- a/src/Control/Scheduler.hs +++ b/src/Control/Scheduler.hs @@ -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 ())) -- [(),()] diff --git a/src/Control/Scheduler/Internal.hs b/src/Control/Scheduler/Internal.hs --- a/src/Control/Scheduler/Internal.hs +++ b/src/Control/Scheduler/Internal.hs @@ -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) } diff --git a/tests/doctests.hs b/tests/doctests.hs --- a/tests/doctests.hs +++ b/tests/doctests.hs @@ -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"]