scheduler 1.1.0 → 1.2.0

raw patch · 5 files changed

+83/−33 lines, 5 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +15/−0
• README.md +16/−0
• scheduler.cabal +2/−1
• src/Control/Scheduler.hs +43/−25
• src/Control/Scheduler/Queue.hs +7/−7

@@ -0,0 +1,15 @@+# 1.2.0++* Addition of `scheduleWorkId` and `scheduleWorkId_`++# 1.1.0++* Add functions: `replicateConcurrently` and `replicateConcurrently_`+* Made `traverseConcurrently_` lazy, thus making it possible to apply to infinite lists and other such+  foldables.+* Fix `Monoid` instance for `Comp`+* Addition of `Par'` pattern++# 1.0.0++Initial release.

@@ -71,6 +71,22 @@ [11,22,33,44,55] ``` +### Identify workers doing the work++Since version `scheduler-1.2.0` it is possible to identify which worker is doing the+job. This is especially useful for limiting resources to particular workers that should+not be shared between separate threads.++```haskell+λ> let scheduleId = (`scheduleWorkId` (\ i -> threadDelay 100000 >> pure i))+λ> withScheduler (ParOn [4,7,5]) $ \s -> scheduleId s >> scheduleId s >> scheduleId s+[4,7,5]+λ> withScheduler (ParN 3) $ \s -> scheduleId s >> scheduleId s >> scheduleId s+[1,2,0]+λ> withScheduler (ParN 3) $ \s -> scheduleId s >> scheduleId s >> scheduleId s+[0,1,2]+```+ ### Exceptions  Whenever any of the scheduled jobs result in an exception, all of the workers will be killed and the

@@ -1,5 +1,5 @@ name:                scheduler-version:             1.1.0+version:             1.2.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@@ -11,6 +11,7 @@ category:            Parallelism, Concurrency build-type:          Custom extra-source-files:  README.md+                   , CHANGELOG.md cabal-version:       >=1.10  custom-setup

@@ -19,6 +19,8 @@   , numWorkers   , scheduleWork   , scheduleWork_+  , scheduleWorkId+  , scheduleWorkId_   , terminate   , terminate_   , terminateWith@@ -48,8 +50,8 @@ import Data.Atomics (atomicModifyIORefCAS, atomicModifyIORefCAS_) import qualified Data.Foldable as F (foldl', traverse_) import Data.IORef-import Data.Traversable import Data.Maybe (catMaybes)+import Data.Traversable   data Jobs m a = Jobs@@ -63,16 +65,17 @@ -- -- @since 1.0.0 data Scheduler m a = Scheduler-  { numWorkers    :: {-# UNPACK #-} !Int+  { 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.+  , 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.0.0-  , terminate     :: a -> m a+  -- @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.@@ -81,7 +84,7 @@   -- although it will make sure their results are discarded.   --   -- @since 1.1.0-  , terminateWith :: a -> m a+  , 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.   --@@ -92,12 +95,26 @@   -- @since 1.1.0   } --- | Same as `scheduleWork`, but only for a `Scheduler` that doesn't keep the result.++-- | 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)++-- | Same as `scheduleWork`, but only for a `Scheduler` that doesn't keep the results.+-- -- @since 1.1.0 scheduleWork_ :: Scheduler m () -> m () -> m () scheduleWork_ = scheduleWork +-- | Same as `scheduleWorkId`, but only for a `Scheduler` that doesn't keep the results.+--+-- @since 1.2.0+scheduleWorkId_ :: Scheduler m () -> (Int -> m ()) -> m ()+scheduleWorkId_ = scheduleWorkId+ -- | 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.@@ -113,7 +130,7 @@ trivialScheduler_ :: Applicative f => Scheduler f () trivialScheduler_ = Scheduler   { numWorkers = 1-  , scheduleWork = id+  , scheduleWorkId = \f -> f 0   , terminate = const $ pure ()   , terminateWith = const $ pure ()   }@@ -172,21 +189,21 @@   withScheduler_ comp $ \s -> scheduleWork s $ replicateM_ n (scheduleWork s $ void f)  -scheduleJobs :: MonadIO m => Jobs m a -> m a -> m ()+scheduleJobs :: MonadIO m => Jobs m a -> (Int -> 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)+scheduleJobs_ :: MonadIO m => Jobs m a -> (Int -> m b) -> m ()+scheduleJobs_ = scheduleJobsWith (\job -> pure (Job_ (void . job))) -scheduleJobsWith :: MonadIO m => (m b -> m (Job m a)) -> Jobs m a -> m b -> m ()+scheduleJobsWith :: MonadIO m => ((Int -> m b) -> m (Job m a)) -> Jobs m a -> (Int -> m b) -> m () scheduleJobsWith mkJob' jobs action = do   liftIO $ atomicModifyIORefCAS_ (jobsCountRef jobs) (+ 1)   job <--    mkJob' $ do-      res <- action+    mkJob' $ \ i -> do+      res <- action i       res `seq`         dropCounterOnZero (jobsCountRef jobs) $         retireWorkersN (jobsQueue jobs) (jobsNumWorkers jobs)@@ -213,14 +230,15 @@ -- | 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+             Int+          -> JQueue m a           -> m () -- ^ Action to run upon retirement           -> m ()-runWorker jQueue onRetire = go+runWorker wId jQueue onRetire = go   where     go =       popJQueue jQueue >>= \case-        Just job -> job >> go+        Just job -> job wId >> go         Nothing -> onRetire  @@ -275,7 +293,7 @@ withSchedulerInternal ::      MonadUnliftIO m   => Comp -- ^ Computation strategy-  -> (Jobs m a -> m a -> m ()) -- ^ How to schedule work+  -> (Jobs m a -> (Int -> m a) -> m ()) -- ^ How to schedule work   -> (JQueue m a -> m [Maybe a]) -- ^ How to collect results   -> ([a] -> [a]) -- ^ Adjust results in some way   -> (Scheduler m a -> m b)@@ -291,7 +309,7 @@       scheduler =         Scheduler           { numWorkers = jobsNumWorkers-          , scheduleWork = submitWork jobs+          , scheduleWorkId = submitWork jobs           , terminate =               \a -> do                 mas <- collect jobsQueue@@ -311,24 +329,24 @@   _ <- onScheduler scheduler   -- / Ensure at least something gets scheduled, so retirement can be triggered   jc <- liftIO $ readIORef jobsCountRef-  when (jc == 0) $ scheduleJobs_ jobs (pure ())+  when (jc == 0) $ scheduleJobs_ jobs (\_ -> pure ())   let spawnWorkersWith fork ws =         withRunInIO $ \run ->           forM ws $ \w ->             fork w $ \unmask ->               catch-                (unmask $ run $ runWorker jobsQueue onRetire)+                (unmask $ run $ runWorker w jobsQueue onRetire)                 (run . handleWorkerException jobsQueue workDoneMVar jobsNumWorkers)       spawnWorkers =         case comp of           Seq -> return []             -- \ no need to fork threads for a sequential computation-          Par -> spawnWorkersWith forkOnWithUnmask [1 .. jobsNumWorkers]+          Par -> spawnWorkersWith forkOnWithUnmask [0 .. jobsNumWorkers - 1]           ParOn ws -> spawnWorkersWith forkOnWithUnmask ws-          ParN _ -> spawnWorkersWith (\_ -> forkIOWithUnmask) [1 .. jobsNumWorkers]+          ParN _ -> spawnWorkersWith (\_ -> forkIOWithUnmask) [0 .. jobsNumWorkers - 1]       terminateWorkers = liftIO . traverse_ (`throwTo` SomeAsyncException WorkerTerminateException)       doWork tids = do-        when (comp == Seq) $ runWorker jobsQueue onRetire+        when (comp == Seq) $ runWorker 0 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

@@ -43,17 +43,17 @@         y:ys -> Just (y, queue {qQueue = ys, qStack = []})  data Job m a-  = Job !(IORef (Maybe a)) !(m a)-  | Job_ !(m ())+  = Job !(IORef (Maybe a)) (Int -> m a)+  | Job_ (Int -> m ())   | Retire  -mkJob :: MonadIO m => m a -> m (Job m a)+mkJob :: MonadIO m => (Int -> m a) -> m (Job m a) mkJob action = do   resRef <- liftIO $ newIORef Nothing   return $!-    Job resRef $ do-      res <- action+    Job resRef $ \ i -> do+      res <- action i       liftIO $ writeIORef resRef $ Just res       return $! res @@ -85,7 +85,7 @@            , liftIO $ putMVar qBaton ()))  -popJQueue :: MonadIO m => JQueue m a -> m (Maybe (m ()))+popJQueue :: MonadIO m => JQueue m a -> m (Maybe (Int -> m ())) popJQueue (JQueue jQueueRef) = liftIO inner   where     inner =@@ -96,7 +96,7 @@           Just (job, newQueue) ->             ( newQueue             , case job of-                Job _ action -> return $ Just (void action)+                Job _ action -> return $ Just (void . action)                 Job_ action_ -> return $ Just action_                 Retire       -> return Nothing)