scheduler 1.2.0 → 1.3.0

raw patch · 4 files changed

+76/−43 lines, 4 files

Files

• CHANGELOG.md +7/−0
• scheduler.cabal +1/−1
• src/Control/Scheduler.hs +53/−39
• src/Control/Scheduler/Computation.hs +15/−3

@@ -1,3 +1,10 @@+# 1.3.0++* Make sure internal `Scheduler` accessor functions are no longer exported, they only+  cause breakage.+* Make sure number of capabilities does not change through out the program execution, as+  far as `scheduler` is concerned.+ # 1.2.0  * Addition of `scheduleWorkId` and `scheduleWorkId_`

@@ -1,5 +1,5 @@ name:                scheduler-version:             1.2.0+version:             1.3.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

@@ -65,43 +65,57 @@ -- -- @since 1.0.0 data Scheduler m a = Scheduler-  { 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-  , scheduleWorkId :: (Int -> m a) -> m ()-  -- ^ Schedule an action to be picked up and computed by a worker from a pool of-  -- jobs. Argument supplied to the job will be the id of the worker doing the job.-  ---  -- @since 1.2.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+  { _numWorkers     :: {-# UNPACK #-} !Int+  , _scheduleWorkId :: (Int -> m a) -> m ()+  , _terminate      :: a -> m a+  , _terminateWith  :: a -> m a   }  +-- ^ 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+numWorkers :: Scheduler m a -> Int+numWorkers = _numWorkers++ -- | Schedule an action to be picked up and computed by a worker from a pool of+-- jobs. Argument supplied to the job will be the id of the worker doing the job.+--+-- @since 1.2.0+scheduleWorkId :: Scheduler m a -> (Int -> m a) -> m ()+scheduleWorkId =_scheduleWorkId++-- | As soon as possible try to terminate any computation that is being performed by all workers+-- managed by this scheduler and collect whatever results have been computed, with supplied+-- element guaranteed to being the last one.+--+-- /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+terminate :: Scheduler m a -> a -> m a+terminate = _terminate++-- | 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+terminateWith :: Scheduler m a -> a -> m a+terminateWith = _terminateWith+++-- | Schedule an action to be picked up and computed by a worker from a pool of -- jobs. Similar to `scheduleWorkId`, except the job doesn't get the worker id. -- -- @since 1.0.0 scheduleWork :: Scheduler m a -> m a -> m ()-scheduleWork scheduler f = scheduleWorkId scheduler (const f)+scheduleWork scheduler f = _scheduleWorkId scheduler (const f)  -- | Same as `scheduleWork`, but only for a `Scheduler` that doesn't keep the results. --@@ -113,7 +127,7 @@ -- -- @since 1.2.0 scheduleWorkId_ :: Scheduler m () -> (Int -> m ()) -> m ()-scheduleWorkId_ = scheduleWorkId+scheduleWorkId_ = _scheduleWorkId  -- | Similar to `terminate`, but for a `Scheduler` that does not keep any results of computation. --@@ -121,7 +135,7 @@ -- -- @since 1.1.0 terminate_ :: Scheduler m () -> m ()-terminate_ = (`terminateWith` ())+terminate_ = (`_terminateWith` ())  -- | The most basic scheduler that simply runs the task instead of scheduling it. Early termination -- requests are simply ignored.@@ -129,10 +143,10 @@ -- @since 1.1.0 trivialScheduler_ :: Applicative f => Scheduler f () trivialScheduler_ = Scheduler-  { numWorkers = 1-  , scheduleWorkId = \f -> f 0-  , terminate = const $ pure ()-  , terminateWith = const $ pure ()+  { _numWorkers = 1+  , _scheduleWorkId = \f -> f 0+  , _terminate = const $ pure ()+  , _terminateWith = const $ pure ()   }  @@ -308,15 +322,15 @@   let jobs = Jobs {..}       scheduler =         Scheduler-          { numWorkers = jobsNumWorkers-          , scheduleWorkId = submitWork jobs-          , terminate =+          { _numWorkers = jobsNumWorkers+          , _scheduleWorkId = submitWork jobs+          , _terminate =               \a -> do                 mas <- collect jobsQueue                 let as = adjust (a : catMaybes mas)                 liftIO $ void $ tryPutMVar workDoneMVar $ SchedulerTerminatedEarly as                 pure a-          , terminateWith =+          , _terminateWith =               \a -> do                 liftIO $ void $ tryPutMVar workDoneMVar $ SchedulerTerminatedEarly [a]                 pure a

@@ -20,7 +20,9 @@ #if !MIN_VERSION_base(4,11,0) import Data.Semigroup #endif+import Data.IORef import Data.Word+import System.IO.Unsafe (unsafePerformIO)  -- | Computation strategy to use when scheduling work. data Comp@@ -97,14 +99,24 @@         ParN n2 -> ParN (max n1 n2) {-# NOINLINE joinComp #-} --- | Figure out how many workers will this computation strategy create+numCapsRef :: IORef Int+numCapsRef = unsafePerformIO $ do+  caps <- getNumCapabilities+  newIORef caps+{-# NOINLINE numCapsRef #-}++-- | Figure out how many workers will this computation strategy create. --+-- /Note/ - If at any point during program execution global number of capabilities gets+-- changed with `Control.Concurrent.setNumCapabilities`, it will have no affect on this+-- function, unless it hasn't yet been called with `Par` or `ParN` 0 arguments.+-- -- @since 1.1.0 getCompWorkers :: MonadIO m => Comp -> m Int getCompWorkers =   \case     Seq -> return 1-    Par -> liftIO getNumCapabilities+    Par -> liftIO (readIORef numCapsRef)     ParOn ws -> return $ length ws-    ParN 0 -> liftIO getNumCapabilities+    ParN 0 -> liftIO (readIORef numCapsRef)     ParN n -> return $ fromIntegral n