packages feed

monad-schedule (empty) → 0.1.0.0

raw patch · 8 files changed

+566/−0 lines, 8 filesdep +basedep +freedep +stm

Dependencies added: base, free, stm, time-domain, transformers

Files

+ CHANGELOG.md view
@@ -0,0 +1,5 @@+# Revision history for monad-schedule++## 0.1.0.0 -- YYYY-mm-dd++* First version. Released on an unsuspecting world.
+ LICENSE view
@@ -0,0 +1,20 @@+Copyright (c) 2021 Manuel Bärenz++Permission is hereby granted, free of charge, to any person obtaining+a copy of this software and associated documentation files (the+"Software"), to deal in the Software without restriction, including+without limitation the rights to use, copy, modify, merge, publish,+distribute, sublicense, and/or sell copies of the Software, and to+permit persons to whom the Software is furnished to do so, subject to+the following conditions:++The above copyright notice and this permission notice shall be included+in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ monad-schedule.cabal view
@@ -0,0 +1,34 @@+cabal-version:      2.4+name:               monad-schedule+version:            0.1.0.0+license:            MIT+license-file:       LICENSE+author:             Manuel Bärenz+maintainer:         programming@manuelbaerenz.de+synopsis:           A new, simple, composable concurrency abstraction.+description:        A monad @m@ is said to allow scheduling if you can pass a number of actions @m a@ to it,+                    and those can be executed at the same time concurrently.+                    You can observe the result of the actions after some time:+                    Some actions will complete first, and the results of these are returned then as a list @'NonEmpty' a@.+                    Other actions are still running, and for these you will receive continuations of type @m a@,+                    which you can further run or schedule to completion as you like.+category:           Concurrency+++extra-source-files: CHANGELOG.md++library+  exposed-modules:+      Control.Monad.Schedule.Class+      Control.Monad.Schedule.RoundRobin+      Control.Monad.Schedule.Sequence+      Control.Monad.Schedule.Trans+      Control.Monad.Schedule.OSThreadPool+  build-depends:+      base >= 4.13.0 && <= 4.17+    , stm >= 2.5+    , transformers >= 0.5+    , free >= 5.1+    , time-domain >= 0.1+  hs-source-dirs:   src+  default-language: Haskell2010
+ src/Control/Monad/Schedule/Class.hs view
@@ -0,0 +1,214 @@+{-# LANGUAGE ApplicativeDo #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE OverloadedLists #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeOperators #-}+module Control.Monad.Schedule.Class where+++-- base+import Control.Arrow+import Control.Concurrent+import Data.Either+import Data.Foldable (fold, forM_)+import Data.List.NonEmpty hiding (length)+import Data.Function+import Data.Kind (Type)+import Data.Void++-- transformers+import Control.Monad.Trans.Accum+import Control.Monad.Trans.Class+import Control.Monad.Trans.Writer+import Control.Monad.Trans.Reader+import qualified Data.List.NonEmpty as NonEmpty+import Control.Monad.Trans.Cont+import Control.Monad (void)+import Unsafe.Coerce (unsafeCoerce)+import Data.Functor.Identity+import Data.Maybe (fromJust)+import Prelude hiding (map, zip)+import Control.Monad.IO.Class+import Control.Monad.Trans.Except+import Control.Monad.Trans.Maybe++{- | 'Monad's in which actions can be scheduled concurrently.++@'schedule' actions@ is expected to run @actions@ concurrently,+whatever that means for a particular monad @m@.+'schedule' does not return before at least one value has finished,+and the returned values @'NonEmpty' a@ are all those that finish first.+The actions @[m a]@ (possibly empty) are the remaining, still running ones.+Executing any of them is expected to be blocking,+and awaits the return of the corresponding action.++A lawful instance is considered to satisfy these conditions:++  * The set of returned values is invariant under scheduling.+    In other words, @sequence@ will result in the same set of values as @scheduleAndFinish@.+'schedule' thus can be thought of as a concurrency-utilizing version of 'sequence'.+-}+class MonadSchedule m where+  -- | Run the actions concurrently,+  --   and return the result of the first finishers,+  --   together with completions for the unfinished actions.+  schedule :: NonEmpty (m a) -> m (NonEmpty a, [m a])++-- | Keeps 'schedule'ing actions until all are finished.+--   Returns the same set of values as 'sequence',+--   but utilises concurrency and may thus change the order of the values.+scheduleAndFinish :: (Monad m, MonadSchedule m) => NonEmpty (m a) -> m (NonEmpty a)+scheduleAndFinish actions = do+  (finishedFirst, running) <- schedule actions+  case running of+    [] -> return finishedFirst+    (a : as) -> do+      finishedLater <- scheduleAndFinish $ a :| as+      return $ finishedFirst <> finishedLater++-- | Uses 'scheduleAndFinish' to execute all actions concurrently,+--   then orders them again.+--   Thus it behaves semantically like 'sequence',+--   but leverages concurrency.+sequenceScheduling :: (Monad m, MonadSchedule m) => NonEmpty (m a) -> m (NonEmpty a)+sequenceScheduling+  =   zip [1..]+  >>> map strength+  >>> scheduleAndFinish+  >>> fmap (sortWith fst >>> map snd)+  where+    strength :: Functor m => (a, m b) -> m (a, b)+    strength (a, mb) = (a, ) <$> mb++{- |+Fork all actions concurrently in separate threads and wait for the first one to complete.++Many monadic actions complete at nondeterministic times+(such as event listeners),+and it is thus impossible to schedule them deterministically+with most other actions.+Using concurrency, they can still be scheduled with all other actions in 'IO',+by running them in separate GHC threads.+-}+instance MonadSchedule IO where+  schedule as = do+    var <- newEmptyMVar+    forM_ as $ \action -> forkIO $ putMVar var =<< action+    a <- takeMVar var+    as' <- drain var+    let remaining = replicate (length as - 1 - length as') $ takeMVar var+    return (a :| as', remaining)+      where+        drain :: MVar a -> IO [a]+        drain var = do+          aMaybe <- tryTakeMVar var+          case aMaybe of+            Just a -> do+              as' <- drain var+              return $ a : as'+            Nothing -> return []++-- TODO Needs dependency+-- instance MonadSchedule STM where++-- | Write in the order of scheduling:+--   The first actions to return write first.+instance (Monoid w, Functor m, MonadSchedule m) => MonadSchedule (WriterT w m) where+  schedule = fmap runWriterT+    >>> schedule+    >>> fmap (first (fmap fst &&& (fmap snd >>> fold)) >>> assoc >>> first (second $ fmap WriterT))+    >>> WriterT+    where+      assoc :: ((a, w), c) -> ((a, c), w)+      assoc ((a, w), c) = ((a, c), w)++-- | Broadcast the same environment to all actions.+--   The continuations keep this initial environment.+instance (Monad m, MonadSchedule m) => MonadSchedule (ReaderT r m) where+  schedule actions = ReaderT $ \r+    -> fmap (`runReaderT` r) actions+    & schedule+    & fmap (second $ fmap lift)++-- | Combination of 'WriterT' and 'ReaderT'.+--   Pass the same initial environment to all actions+--   and write to the log in the order of scheduling in @m@.+instance (Monoid w, Monad m, MonadSchedule m) => MonadSchedule (AccumT w m) where+  schedule actions = AccumT $ \w+    -> fmap (`runAccumT` w) actions+    & schedule+    & fmap collectWritesAndWrap+    where+      collectWritesAndWrap ::+        Monoid w =>+        (NonEmpty (a, w), [m (a, w)]) ->+        ((NonEmpty a, [AccumT w m a]), w)+      collectWritesAndWrap (finished, running) =+        let (as, logs) = NonEmpty.unzip finished+        in ((as, AccumT . const <$> running), fold logs)++-- | Schedule all actions according to @m@ and in case of exceptions+--   throw the first exception of the immediately returning actions.+instance (Monad m, MonadSchedule m) => MonadSchedule (ExceptT e m) where+  schedule+    =   fmap runExceptT+    >>> schedule+    >>> fmap ((sequenceA *** fmap ExceptT) >>> extrudeEither)+    >>> ExceptT+    where+      extrudeEither :: (Either e a, b) -> Either e (a, b)+      extrudeEither (ea, b) = (, b) <$> ea++instance (Monad m, MonadSchedule m) => MonadSchedule (MaybeT m) where+  schedule+    =   fmap (maybeToExceptT ())+    >>> schedule+    >>> exceptToMaybeT+    >>> fmap (second $ fmap exceptToMaybeT)++-- instance (Monad m, MonadSchedule m) => MonadSchedule (ContT r m) where+--   schedule actions = ContT $ \scheduler+--     -> fmap (runContT >>> _) actions+--     & schedule+--     & _++-- | Runs two values in a 'MonadSchedule' concurrently+--   and returns the first one that yields a value+--   and a continuation for the other value.+race+  :: (Monad m, MonadSchedule m)+  => m a -> m b+  -> m (Either (a, m b) (m a, b))+race aM bM = recoverResult <$> schedule ((Left <$> aM) :| [Right <$> bM])+  where+    recoverResult :: Monad m => (NonEmpty (Either a b), [m (Either a b)]) -> Either (a, m b) (m a, b)+    recoverResult (Left a :| [], [bM']) = Left (a, fromRight e <$> bM')+    recoverResult (Right b :| [], [aM']) = Right (fromLeft e <$> aM', b)+    recoverResult (Left a :| [Right b], []) = Left (a, return b)+    recoverResult (Right b :| [Left a], []) = Right (return a, b)+    recoverResult _ = e+    e = error "race: Internal error"++-- FIXME I should only need Selective+-- | Runs both schedules concurrently and returns their results at the end.+async+  :: (Monad m, MonadSchedule m)+  => m  a -> m b+  -> m (a,     b)+async aSched bSched = do+  ab <- race aSched bSched+  case ab of+    Left  (a, bCont) -> do+      b <- bCont+      return (a, b)+    Right (aCont, b) -> do+      a <- aCont+      return (a, b)
+ src/Control/Monad/Schedule/OSThreadPool.hs view
@@ -0,0 +1,81 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE RecordWildCards #-}+module Control.Monad.Schedule.OSThreadPool where++-- base+import Control.Concurrent+import Control.Monad ( void, forM, replicateM )+import Control.Monad.IO.Class+import Data.List.NonEmpty hiding (zip, cycle)+import Data.Proxy+import GHC.TypeLits+import Prelude hiding (take)++-- stm+import Control.Concurrent.STM.TChan++-- rhine+import Control.Monad.Schedule.Class+import Control.Concurrent.STM+import Data.Either (partitionEithers)++newtype OSThreadPool (n :: Nat) a = OSThreadPool { unOSThreadPool :: IO a }+  deriving (Functor, Applicative, Monad, MonadIO)++data WorkerLink a = WorkerLink+  { jobTChan :: TChan (Maybe (IO a))+  , resultTChan :: TChan a+  }++putJob :: WorkerLink a -> OSThreadPool n a -> IO ()+putJob WorkerLink { .. } OSThreadPool { .. }+  = atomically+  $ writeTChan jobTChan+  $ Just unOSThreadPool++makeWorkerLink :: IO (WorkerLink a)+makeWorkerLink = do+  jobTChan <- atomically newTChan+  resultTChan <- atomically newTChan+  let worker = do+        job <- atomically $ readTChan jobTChan+        case job of+          Nothing -> return ()+          Just action -> do+            result <- action+            atomically $ writeTChan resultTChan result+            worker+  void $ forkOS worker+  return WorkerLink { .. }++proxyForActions :: NonEmpty (OSThreadPool n a) -> Proxy n+proxyForActions _ = Proxy++instance (KnownNat n, 1 <= n) => MonadSchedule (OSThreadPool n) where+  schedule actions = OSThreadPool $ do+    let n = natVal $ proxyForActions actions+    workerLinks <- replicateM (fromInteger n) makeWorkerLink+    backgroundActions <- forM (zip (cycle workerLinks) (toList actions))+      $ \(link, action) -> do+        putJob link action+        return $ resultTChan link+    pollPools backgroundActions+    where+      pollPools :: [TChan a] -> IO (NonEmpty a, [OSThreadPool n a])+      pollPools chans = do+        results <- traverse pollPool chans+        case partitionEithers results of+          (_, []) -> do+            threadDelay 1000+            pollPools chans+          (remainingChans, a : as) -> return+            ( a :| as+            , OSThreadPool . atomically . readTChan <$> remainingChans+            )++      pollPool :: TChan a -> IO (Either (TChan a) a)+      pollPool chan = maybe (Left chan) Right <$> atomically (tryReadTChan chan)
+ src/Control/Monad/Schedule/RoundRobin.hs view
@@ -0,0 +1,25 @@+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+module Control.Monad.Schedule.RoundRobin where++-- base+import Control.Monad.IO.Class+import qualified Data.List.NonEmpty as NonEmpty++-- transformers+import Control.Monad.Trans.Class++-- monad-schedule+import Control.Monad.Schedule.Class++-- | Any monad can be trivially scheduled by executing all actions after each other,+--   step by step.+newtype RoundRobinT m a = RoundRobinT { unRoundRobin :: m a }+  deriving (Functor, Applicative, Monad, MonadIO)++instance MonadTrans RoundRobinT where+  lift = RoundRobinT++-- | Execute only the first action, and leave the others for later, preserving the order.+instance Monad m => MonadSchedule (RoundRobinT m) where+  schedule actions = ( , NonEmpty.tail actions) <$> fmap pure (NonEmpty.head actions)
+ src/Control/Monad/Schedule/Sequence.hs view
@@ -0,0 +1,26 @@+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+module Control.Monad.Schedule.Sequence where++-- base+import Control.Arrow ((>>>))+import Control.Monad.IO.Class+import qualified Data.List.NonEmpty as NonEmpty++-- transformers+import Control.Monad.Trans.Class++-- monad-schedule+import Control.Monad.Schedule.Class++-- | Any monad can be trivially scheduled by executing all actions sequentially.+newtype SequenceT m a = SequenceT { unSequence :: m a }+  deriving (Functor, Applicative, Monad, MonadIO)++instance MonadTrans SequenceT where+  lift = SequenceT++-- | Execute all actions in sequence and return their result when all of them are done.+--   Essentially, this is 'sequenceA'.+instance Monad m => MonadSchedule (SequenceT m) where+  schedule = sequenceA >>> fmap (, [])
+ src/Control/Monad/Schedule/Trans.hs view
@@ -0,0 +1,161 @@+{- |+This module supplies a general purpose monad transformer+that adds a syntactical "delay", or "waiting" side effect.+-}++{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeSynonymInstances #-}+module Control.Monad.Schedule.Trans where++-- base+import Data.Ord (comparing)+import Control.Arrow (Arrow(second))+import Control.Concurrent+import qualified Control.Concurrent as C+import Control.Category ((>>>))+import Control.Monad (join)+import Data.Functor.Classes+import Data.List.NonEmpty as N++-- transformers+import Control.Monad.IO.Class+import Control.Monad.Trans.Class++-- free+import Control.Monad.Trans.Free++-- time-domain+import Data.TimeDomain++-- monad-schedule+import Control.Monad.Schedule.Class++-- TODO Implement Time via StateT++-- * Waiting action++-- | A functor implementing a syntactical "waiting" action.+data Wait diff a = Wait+  { getDiff :: diff+      -- ^ The duration to wait.+  , awaited :: a+      -- ^ The encapsulated value.+  }+  deriving (Functor, Eq, Show)++instance Eq diff => Eq1 (Wait diff) where+  liftEq eq (Wait diff1 a) (Wait diff2 b) = diff1 == diff2 && eq a b++-- | Compare by the time difference, regardless of the value.+compareWait :: Ord diff => Wait diff a -> Wait diff a -> Ordering+compareWait = comparing getDiff++-- * 'ScheduleT'++{- |+Values in @ScheduleT diff m@ are delayed computations with side effects in 'm'.+Delays can occur between any two side effects, with lengths specified by a 'diff' value.+These delays don't have any semantics, it can be given to them with 'runScheduleT'.+-}+type ScheduleT diff = FreeT (Wait diff)++-- | The side effect that waits for a specified amount.+wait :: Monad m => diff -> ScheduleT diff m ()+wait diff = FreeT $ return $ Free $ Wait diff $ return ()++-- | Supply a semantic meaning to 'Wait'.+--   For every occurrence of @Wait diff@ in the @ScheduleT diff m a@ value,+--   a waiting action is executed, depending on 'diff'.+runScheduleT :: Monad m => (diff -> m ()) -> ScheduleT diff m a -> m a+runScheduleT waitAction = iterT $ \(Wait n ma) -> waitAction n >> ma++-- | Run a 'ScheduleT' value in a 'MonadIO',+--   interpreting the times as milliseconds.+runScheduleIO+  :: (MonadIO m, Integral n)+  => ScheduleT n m a -> m a+runScheduleIO = runScheduleT $ liftIO . threadDelay . (* 1000) . fromIntegral++-- | Formally execute all waiting actions,+--   returning the final value and all moments when the schedule would have waited.+execScheduleT :: Monad m => ScheduleT diff m a -> m (a, [diff])+execScheduleT action = do+  free <- runFreeT action+  case free of+    Pure a -> return (a, [])+    Free (Wait diff cont) -> do+      (a, diffs) <- execScheduleT cont+      return (a, diff : diffs)++instance Ord diff => MonadSchedule (Wait diff) where+  schedule waits = let (smallestWait :| waits') = N.sortBy compareWait waits in ((, waits') . pure) <$> smallestWait++-- | Run each action one step until it is discovered which action(s) are pure, or yield next.+--   If there is a pure action, it is returned,+--   otherwise all actions are shifted to the time when the earliest action yields.+instance (Ord diff, TimeDifference diff, Monad m, MonadSchedule m) => MonadSchedule (ScheduleT diff m) where+  schedule actions = do+    (frees, delayed) <- lift $ schedule $ runFreeT <$> actions+    shiftList (sortBy compareFreeFWait frees) $ FreeT <$> delayed+    where+      compareFreeFWait+        :: Ord diff+        => FreeF (Wait diff) a b+        -> FreeF (Wait diff) a b+        -> Ordering+      compareFreeFWait (Pure _) (Pure _) = EQ+      compareFreeFWait (Pure _) (Free _) = LT+      compareFreeFWait (Free _) (Pure _) = GT+      compareFreeFWait (Free wait1) (Free wait2) = compareWait wait1 wait2++      -- Separate pure from free values+      partitionFreeF+        :: [FreeF f a b]+        -> ([a], [f b])+      partitionFreeF [] = ([], [])+      partitionFreeF (Pure a  : xs) = let (as, fbs) = partitionFreeF xs in (a : as, fbs)+      partitionFreeF (Free fb : xs) = let (as, fbs) = partitionFreeF xs in (as, fb : fbs)++      -- Shift a waiting action by some duration+      shift+        :: TimeDifference diff+        => diff+        -> Wait diff a+        -> Wait diff a+      shift diff1 (Wait diff2 a) = Wait (diff2 `difference` diff1) a++      -- Shift a list of free actions by the duration of the head+      -- (assuming the list is sorted).+      -- If the head is pure, return it with the remaining actions,+      -- otherwise wait the minimum duration, give the continuation of the head,+      -- and shift the remaining actions by that minimum duration.+      shiftListOnce+        :: TimeDifference diff+        => NonEmpty (FreeF (Wait diff) a b)+        -> Either+             (NonEmpty a, [Wait diff b]) -- Pure value has completed+             (Wait diff (b, [Wait diff b])) -- All values wait+      shiftListOnce actions = case partitionFreeF $ toList actions of+        (a : as, waits) -> Left (a :| as, waits)+        ([], Wait diff cont : waits) -> Right $ Wait diff (cont, shift diff <$> waits)++      -- Repeatedly shift the list by the smallest available waiting duration+      -- until one action returns as pure.+      -- Return its result, together with the remaining free actions.+      shiftList+        :: (TimeDifference diff, Ord diff, Monad m, MonadSchedule m)+        => NonEmpty (FreeF (Wait diff) a (ScheduleT diff m a))+        -- ^ Actionable+        -> [ScheduleT diff m a]+        -- ^ Delayed+        -> ScheduleT diff m (NonEmpty a, [ScheduleT diff m a])+      shiftList actions delayed = case shiftListOnce actions of+        -- Some actions returned. Wrap up the remaining ones.+        Left (as, waits) -> return (as, delayed ++ ((FreeT . return . Free) <$> waits))+        -- No action has returned.+        -- Wait the remaining time and start scheduling again.+        Right (Wait diff (cont, waits)) -> do+          wait diff+          schedule (cont :| delayed ++ ((FreeT . return . Free) <$> waits))