monad-schedule 0.1.2.2 → 0.2

raw patch · 12 files changed

+392/−53 lines, 12 filesdep +generic-arbitrarydep +operationaldep +timedep ~QuickCheckdep ~basePVP ok

version bump matches the API change (PVP)

Dependencies added: generic-arbitrary, operational, time

Dependency ranges changed: QuickCheck, base

API changes (from Hackage documentation)

Files

• CHANGELOG.md +6/−0
• monad-schedule.cabal +17/−6
• src/Control/Monad/Schedule/Class.hs +19/−12
• src/Control/Monad/Schedule/FreeAsync.hs +235/−0
• src/Control/Monad/Schedule/OSThreadPool.hs +1/−2
• src/Control/Monad/Schedule/Sequence.hs +0/−1
• src/Control/Monad/Schedule/Trans.hs +1/−4
• src/Control/Monad/Schedule/Yield.hs +0/−1
• test/FreeAsync.hs +106/−0
• test/Main.hs +3/−19
• test/Trans.hs +2/−6
• test/Yield.hs +2/−2

@@ -1,5 +1,11 @@ # Revision history for monad-schedule +## 0.2++* Added FreeAsync+* Dropped support below GHC 9.2+* Documentation fixes+ ## 0.1.2.2  * Compatibility with GHC 9.8

@@ -1,6 +1,6 @@ cabal-version:   2.4 name:            monad-schedule-version:         0.1.2.2+version:         0.2 license:         MIT license-file:    LICENSE author:          Manuel Bärenz@@ -18,9 +18,6 @@ extra-doc-files: CHANGELOG.md  tested-with:-  GHC == 8.8.4-  GHC == 8.10.7-  GHC == 9.0.2   GHC == 9.2.5   GHC == 9.4.5   GHC == 9.6.2@@ -32,16 +29,22 @@  common deps   build-depends:-    , base          >=4.13.0 && <4.20.0+    , base          >=4.16.0 && <4.20.0     , free          >=5.1 && < 5.3     , stm           ^>=2.5     , time-domain   ^>=0.1     , transformers  >=0.5 && < 0.7+    , operational   ^>=0.2.4+  if flag(dev)+    ghc-options: -Werror+  ghc-options:+    -W  library   import:           deps   exposed-modules:     Control.Monad.Schedule.Class+    Control.Monad.Schedule.FreeAsync     Control.Monad.Schedule.OSThreadPool     Control.Monad.Schedule.RoundRobin     Control.Monad.Schedule.Sequence@@ -56,6 +59,7 @@   type:             exitcode-stdio-1.0   main-is:          Main.hs   other-modules:+    FreeAsync     Trans     Yield @@ -63,9 +67,16 @@   build-depends:     , HUnit                       ^>=1.6     , monad-schedule-    , QuickCheck                  ^>=2.14+    , QuickCheck                  >=2.14 && <2.16     , test-framework              ^>=0.8     , test-framework-hunit        ^>=0.3     , test-framework-quickcheck2  ^>=0.3+    , generic-arbitrary           ^>=1.0+    , time                        >=1.9    default-language: Haskell2010++flag dev+  description: Enable warnings as errors. Active on ci.+  default: False+  manual: True

@@ -13,24 +13,17 @@ -- base import Control.Arrow import Control.Concurrent-import Control.Monad (void)-import Control.Monad.IO.Class import Data.Either import Data.Foldable (fold, forM_) import Data.Function import Data.Functor.Identity-import Data.Kind (Type) import Data.List.NonEmpty hiding (length) import qualified Data.List.NonEmpty as NonEmpty-import Data.Maybe (fromJust)-import Data.Void-import Unsafe.Coerce (unsafeCoerce) import Prelude hiding (map, zip)  -- transformers import Control.Monad.Trans.Accum import Control.Monad.Trans.Class-import Control.Monad.Trans.Cont import Control.Monad.Trans.Except import Control.Monad.Trans.Identity import Control.Monad.Trans.Maybe@@ -49,10 +42,9 @@ 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:+A lawful instance is considered to preserve pure values.+Applying 'schedule' to values like @'pure' a@ will eventually return exactly these values. -  * 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@@ -93,8 +85,7 @@ instance MonadSchedule Identity where   schedule as = (,[]) <$> sequence as -{- |-Fork all actions concurrently in separate threads and wait for the first one to complete.+{- | 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),@@ -102,6 +93,11 @@ with most other actions. Using concurrency, they can still be scheduled with all other actions in 'IO', by running them in separate GHC threads.++Caution: Using 'schedule' repeatedly on the returned continuations of a previous 'schedule' call+will add a layer of indirection to the continuation every time,+eventually slowing down performance and building up memory.+For a monad that doesn't have this problem, see 'Control.Monad.Schedule.FreeAsync.FreeAsyncT'. -} instance MonadSchedule IO where   schedule as = do@@ -260,3 +256,14 @@     Right (aCont, b) -> do       a <- aCont       return (a, b)++{- | Run both actions concurrently and apply the first result to the second.++Use as a concurrent replacement for '<*>' from 'Applicative'.+-}+apSchedule :: (MonadSchedule m, Monad m) => m (a -> b) -> m a -> m b+apSchedule f a = uncurry id <$> async f a++-- | Concurrent replacement for '*>' from 'Applicative' or '>>' from 'Monad'.+scheduleWith :: (MonadSchedule m, Monad m) => m a -> m b -> m b+scheduleWith a b = (id <$ a) `apSchedule` b

@@ -0,0 +1,235 @@+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE NamedFieldPuns #-}++-- | Asynchronicity implementation using 'MVar's and free monads.+module Control.Monad.Schedule.FreeAsync (+  -- * 'FreeAsyncT'+  FreeAsyncT (..),+  FreeAsync,+  freeAsync,+  asyncMVar,+  runFreeAsync,+  runFreeAsyncT,++  -- * Concurrent 'Applicative' interface+  ConcurrentlyT (..),+  Concurrently,+  concurrently,+  concurrentlyMVar,+  lift',+  runConcurrentlyT,+  runConcurrently,+)+where++-- base+import Control.Arrow (second, (>>>))+import Control.Concurrent (MVar, forkIO, newEmptyMVar, putMVar, takeMVar, threadDelay, tryTakeMVar, yield)+import Control.Monad.IO.Class+import Data.Either (partitionEithers)+import Data.List.NonEmpty (NonEmpty (..), appendList, toList)++-- transformers+import Control.Monad.Trans.Class++-- operational+import Control.Monad.Operational (ProgramT, ProgramViewT (..), interpretWithMonadT, singleton, unviewT, viewT)++-- monad-schedule+import Control.Monad.Schedule.Class (MonadSchedule (..), apSchedule)++{- | An 'IO'-like monad with the capability of async/await-style futures.++Synchronous (blocking) computations in this monad can be created using 'lift' and 'liftIO'.+Asynchronous computations that can run in the background are created with 'freeAsync' or 'asyncMVar'.++To leverage the asynchronicity, you can schedule computations with 'MonadSchedule' operations such as 'schedule' or 'race'.++Caution: Composing computations with 'Applicative' or 'Monad' operations like '<*>', '>>=' and @do@-notation+will force all but the final computation in order:+When running @a '*>' b '*>' c@, @b@ will not be started before @a@ has completed.+To start all operations and run them concurrently, use e.g. 'Control.Monad.Schedule.Class.scheduleWith'.+To use an 'Applicative' interface for concurrency, have a look at 'ConcurrentlyT'.+-}+newtype FreeAsyncT m a = FreeAsyncT {getFreeAsyncT :: ProgramT MVar m a}+  deriving newtype (Functor, Applicative, Monad, MonadTrans)++type FreeAsync = FreeAsyncT IO++-- FIXME MFunctor & MMonad instances pending https://github.com/HeinrichApfelmus/operational/pull/28/++-- | Lifts into 'FreeAsyncT' /without/ concurrency. See 'freeAsync'.+instance (MonadIO m) => MonadIO (FreeAsyncT m) where+  liftIO = lift . liftIO++{- | Run an 'IO' computation in the background.++This returns a "promise", or "future", which completes the computation when run.+-}+freeAsync :: (MonadIO m) => IO a -> FreeAsyncT m a+freeAsync action = FreeAsyncT $ do+  var <- liftIO newEmptyMVar+  liftIO $ forkIO $ putMVar var =<< action+  singleton var++-- | Complete all computations and remove the 'FreeAsyncT' layer.+runFreeAsyncT :: (MonadIO m) => FreeAsyncT m a -> m a+runFreeAsyncT = interpretWithMonadT (liftIO . takeMVar) . getFreeAsyncT++-- | Like 'runFreeAsyncT', but specialized to 'IO'.+runFreeAsync :: FreeAsync a -> IO a+runFreeAsync = runFreeAsyncT++{- | Asynchronously await an 'MVar'.++@'asyncMVar' var@ will attempt 'takeMVar' in a way that can be 'schedule'd concurrently with other 'asyncMVar's or 'freeAsync's.+-}+asyncMVar :: MVar a -> FreeAsyncT m a+asyncMVar = FreeAsyncT . singleton++data MVarCont m a = forall b.+  MVarCont+  { mvar :: MVar b+  , cont :: b -> ProgramT MVar m a+  }++embedMVarCont :: (Monad m) => MVarCont m a -> FreeAsyncT m a+embedMVarCont MVarCont {mvar, cont} = FreeAsyncT $ unviewT $ mvar :>>= cont++{- | Concurrently wait for the completion of 'IO' actions.+Has a slight runtime overhead over the direct @'MonadSchedule' 'IO'@ instance, but better fairness.+-}+instance (MonadIO m) => MonadSchedule (FreeAsyncT m) where+  schedule actions = retryForever $ getFreeAsyncT <$> actions+    where+      retryForever :: (MonadIO m) => NonEmpty (ProgramT MVar m a) -> FreeAsyncT m (NonEmpty a, [FreeAsyncT m a])+      retryForever actions = do+        -- Look at all actions+        views <- lift (mapM viewT actions)+        -- Have some of them finished?+        case partitionNonEmpty $ viewToEither <$> views of+          -- All have finished+          Left (as, []) -> return (as, [])+          -- Some have finished, some are waiting for MVars+          Left (as, cont : conts) -> do+            -- Peek at the MVars+            progressed <- lift $ tryProgresses $ cont :| conts+            -- Are some MVars present already?+            case progressed of+              -- Yes, some were present, step the corresponding actions+              Left (actions, conts) -> do+                -- Look at the progressed actions+                views <- lift (mapM viewT actions)+                -- Have some of them returned now?+                case partitionNonEmpty $ viewToEither <$> views of+                  -- Yes. Return those as well+                  Left (as', conts') -> return (as <> as', embedMVarCont <$> (conts ++ conts'))+                  -- No, they are blocked on other MVars now+                  Right conts' -> return (as, embedMVarCont <$> toList conts' <> conts)+              -- All MVars are still blocked+              Right conts -> return (as, embedMVarCont <$> toList conts)+          -- All actions are waiting for MVars+          Right conts -> do+            -- Retry until some MVars get unblocked+            (progressed, conts) <- lift $ retryProgresses conts+            -- Some MVars are unblocked, start over.+            retryForever $ appendList progressed $ getFreeAsyncT . embedMVarCont <$> conts++      viewToEither :: ProgramViewT MVar m a -> Either a (MVarCont m a)+      viewToEither (Return a) = Left a+      viewToEither (mvar :>>= cont) = Right MVarCont {mvar, cont}++      partitionNonEmpty :: NonEmpty (Either a b) -> Either (NonEmpty a, [b]) (NonEmpty b)+      partitionNonEmpty (Left a :| abs) = let (as, bs) = partitionEithers abs in Left (a :| as, bs)+      partitionNonEmpty (Right b :| abs) = case partitionEithers abs of+        ([], bs) -> Right $ b :| bs+        (a : as, bs) -> Left (a :| as, b : bs)++      tryProgress :: (MonadIO m) => MVarCont m a -> m (Either (ProgramT MVar m a) (MVarCont m a))+      tryProgress mvarcont@MVarCont {mvar, cont} = do+        result <- liftIO $ tryTakeMVar mvar+        return $ maybe (Right mvarcont) (Left . cont) result++      tryProgresses :: (MonadIO m) => NonEmpty (MVarCont m a) -> m (Either (NonEmpty (ProgramT MVar m a), [MVarCont m a]) (NonEmpty (MVarCont m a)))+      tryProgresses conts = do+        result <- partitionNonEmpty <$> mapM tryProgress conts+        case result of+          Left (progressed, []) -> return $ Left (progressed, [])+          Left (progressed, cont : conts) -> do+            inner <- tryProgresses $ cont :| conts+            case inner of+              Left (progressed', finalConts) -> return $ Left (progressed <> progressed', finalConts)+              Right finalConts -> return $ Left (progressed, toList finalConts)+          Right conts -> return $ Right conts++      retryProgresses :: (MonadIO m) => NonEmpty (MVarCont m a) -> m (NonEmpty (ProgramT MVar m a), [MVarCont m a])+      retryProgresses conts = do+        result <- tryProgresses conts+        case result of+          Left progress -> return progress+          Right _ -> do+            liftIO $ yield >> threadDelay 100+            retryProgresses conts++{- | Like 'FreeAsyncT', but leverages concurrency in the 'Applicative' interface.++The central difference to 'FreeAsyncT' is the 'Applicative' instance:+@concurrently a *> concurrently b *> concurrently c@ will launch all three actions immediately+and return when all actions have completed.+On the other hand, @concurrently a >>= f@ has to compute sequentially.++For more readable syntax, it can be useful to switch the @ApplicativeDo@ extension on.++The downside of this 'Applicative' instance is that 'ConcurrentlyT' can't be an instance of 'MonadTrans'.+As a drop-in replacement, the function 'lift'' is supplied.++Caution: To lift an 'IO' action concurrently, you need to use 'concurrently' and not 'liftIO'.+-}+newtype ConcurrentlyT m a = ConcurrentlyT {getConcurrentlyT :: FreeAsyncT m a}+  deriving newtype (Functor, Monad, MonadIO)++type Concurrently = ConcurrentlyT IO++{- | Lift an 'IO' action such that it can be run concurrently.++See 'freeAsync'.+-}+concurrently :: (MonadIO m) => IO a -> ConcurrentlyT m a+concurrently = ConcurrentlyT . freeAsync++-- | Like 'asyncMVar'+concurrentlyMVar :: MVar a -> ConcurrentlyT m a+concurrentlyMVar = ConcurrentlyT . asyncMVar++{- | Lift a computation to 'ConcurrentlyT'.++This replaces the missing 'MonadTrans' instance.++Caution: Computations lifted with this function cannot be scheduled concurrently!+If this is your intention, 'concurrently' needs to be used instead.+-}+lift' :: (Monad m) => m a -> ConcurrentlyT m a+lift' = ConcurrentlyT . lift++-- | Run a 'ConcurrentlyT' computation to completion, removing the newtype layers.+runConcurrentlyT :: (MonadIO m) => ConcurrentlyT m a -> m a+runConcurrentlyT = runFreeAsyncT . getConcurrentlyT++-- | Like 'runConcurrently', but specialised to 'IO'.+runConcurrently :: Concurrently a -> IO a+runConcurrently = runConcurrentlyT++instance (MonadIO m) => Applicative (ConcurrentlyT m) where+  pure = ConcurrentlyT . pure+  (<*>) = apSchedule++-- | Like 'FreeAsyncT', but executes actions composed via the 'Applicative' interface concurrently.+instance (MonadIO m) => MonadSchedule (ConcurrentlyT m) where+  schedule =+    fmap getConcurrentlyT+      >>> schedule+      >>> fmap (second (map ConcurrentlyT))+      >>> ConcurrentlyT

@@ -18,9 +18,8 @@  -- stm import Control.Concurrent.STM-import Control.Concurrent.STM.TChan --- rhine+-- monad-schedule import Control.Monad.Schedule.Class  newtype OSThreadPool (n :: Nat) a = OSThreadPool {unOSThreadPool :: IO a}

@@ -7,7 +7,6 @@ import Control.Arrow ((>>>)) import Control.Monad.IO.Class import Data.Functor.Identity-import qualified Data.List.NonEmpty as NonEmpty  -- transformers import Control.Monad.Trans.Class

@@ -9,11 +9,7 @@ module Control.Monad.Schedule.Trans where  -- base-import Control.Arrow (Arrow (second))-import Control.Category ((>>>)) import Control.Concurrent-import qualified Control.Concurrent as C-import Control.Monad (join) import Data.Functor.Classes import Data.Functor.Identity import Data.List (partition)@@ -156,6 +152,7 @@       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)+        ([], []) -> error "ScheduleT.shiftListOnce: Internal error. Please report as a bug: https://github.com/turion/monad-schedule/issues/new"        -- Repeatedly shift the list by the smallest available waiting duration       -- until one action returns as pure.

@@ -6,7 +6,6 @@ import Data.Functor.Identity (Identity (runIdentity))  -- monad-schedule-import Control.Monad.Schedule.Class import Control.Monad.Schedule.Trans  -- * 'YieldT'

@@ -0,0 +1,106 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingVia #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -fconstraint-solver-iterations=10 #-}++module FreeAsync where++-- base+import Control.Concurrent (newMVar, takeMVar)+import Control.Monad (replicateM_)+import Control.Monad.IO.Class+import Data.IORef (newIORef, readIORef, writeIORef)+import Data.List.NonEmpty (fromList)+import GHC.Conc (threadDelay)+import GHC.Generics (Generic)++-- time+import Data.Time (diffUTCTime, getCurrentTime)++-- test-framework+import Test.Framework (testGroup)++-- test-framework-hunit+import Test.Framework.Providers.HUnit (testCase)++-- test-framework-QuickCheck2+import Test.Framework.Providers.QuickCheck2 (testProperty)++-- QuickCheck2+import Test.QuickCheck (Arbitrary, genericShrink, ioProperty, (.&&.), (===))+import Test.QuickCheck.Arbitrary.Generic (Arbitrary (..), genericArbitrary)++-- hunit+import Test.HUnit ((@?=))+import Test.HUnit.Base (assert)++-- monad-schedule+import Control.Monad.Schedule.Class (async, scheduleAndFinish)+import Control.Monad.Schedule.FreeAsync (FreeAsync, asyncMVar, concurrently, freeAsync, runConcurrently, runFreeAsync)++tests =+  testGroup+    "FreeAsync"+    [ testCase "Can lift and run an IORef action" $ do+        ref <- newIORef ""+        runFreeAsync $ liftIO $ writeIORef ref "Hello"+        result <- readIORef ref+        result @?= "Hello"+    , testProperty "Can schedule two values" $ \(v1 :: Value) (v2 :: Value) -> ioProperty $ do+        (result1, result2) <- runFreeAsync $ async (interpretValue 23 v1) (interpretValue 42 v2)+        return $ result1 === 23 .&&. result2 === 42+    , testCase "Can schedule two MVars" $ do+        var1 <- newMVar 23+        var2 <- newMVar 42+        result <- runFreeAsync $ async (asyncMVar var1) (asyncMVar var2)+        result @?= (23, 42)+    , testCase "Can schedule lifted retrieval of two MVars" $ do+        var1 <- newMVar 23+        var2 <- newMVar 42+        result <- runFreeAsync $ async (liftIO $ takeMVar var1) (liftIO $ takeMVar var2)+        result @?= (23, 42)+    , testCase "Sequencing adds times" $ do+        before <- getCurrentTime+        runFreeAsync $ replicateM_ 100 $ freeAsync $ threadDelay 1000+        after <- getCurrentTime+        let diff = after `diffUTCTime` before+        assert $ diff >= 0.1 && diff < 0.2+    , testCase "Scheduling does things in parallel" $ do+        before <- getCurrentTime+        runFreeAsync $ scheduleAndFinish $ fromList $ replicate 100 $ freeAsync $ threadDelay 1000+        after <- getCurrentTime+        let diff = after `diffUTCTime` before+        print diff+        assert $ diff > 0.001 && diff <= 0.005 -- Assume overhead is at most 5x at ms durations+    , testGroup+        "ConcurrentlyT"+        [ testCase "Sequencing does things in parallel" $ do+            before <- getCurrentTime+            runConcurrently $ replicateM_ 100 $ concurrently $ threadDelay 1000+            after <- getCurrentTime+            let diff = after `diffUTCTime` before+            print diff+            assert $ diff > 0.001 && diff <= 0.005 -- Assume overhead is at most 5x at ms durations+        ]+    ]++data Value+  = Pure+  | Lifted+  | FreeAsync_+  | AsyncMVar+  | TakeMVar+  | TakeMVarSingleLift+  deriving (Show, Generic)++instance Arbitrary Value where+  arbitrary = genericArbitrary+  shrink = genericShrink++interpretValue :: a -> Value -> FreeAsync a+interpretValue a Pure = pure a+interpretValue a Lifted = liftIO (pure a)+interpretValue a FreeAsync_ = freeAsync $ pure a+interpretValue a AsyncMVar = liftIO (newMVar a) >>= asyncMVar+interpretValue a TakeMVar = liftIO (newMVar a) >>= liftIO . takeMVar+interpretValue a TakeMVarSingleLift = liftIO (newMVar a >>= takeMVar)

@@ -1,27 +1,10 @@ {-# LANGUAGE FlexibleInstances #-} --- base-import Control.Arrow-import Control.Monad-import Data.Functor.Identity-import Data.List.NonEmpty- -- test-framework import Test.Framework --- test-framework-hunit-import Test.Framework.Providers.HUnit---- HUnit-import Test.HUnit hiding (Test)---- test-framework-quickcheck2-import Test.Framework.Providers.QuickCheck2---- QuickCheck-import Test.QuickCheck- -- monad-schedule (test)+import qualified FreeAsync import qualified Trans import qualified Yield @@ -30,6 +13,7 @@  tests :: [Test] tests =-  [ Trans.tests+  [ FreeAsync.tests+  , Trans.tests   , Yield.tests   ]

@@ -7,21 +7,17 @@  -- base import Control.Arrow-import Control.Monad (forever, void)+import Control.Monad (forever) import Data.List (sort) import Data.List.NonEmpty (NonEmpty) import qualified Data.List.NonEmpty as NonEmpty  -- transformers import Control.Monad.Trans.Class-import Control.Monad.Trans.Writer (Writer, execWriter, runWriter, tell)---- free-import Control.Monad.Free (_Free)+import Control.Monad.Trans.Writer (Writer, execWriter, tell)  -- QuickCheck import Test.QuickCheck-import qualified Test.QuickCheck as QuickCheck  -- test-framework import Test.Framework

@@ -17,7 +17,7 @@ import Control.Monad.Trans.Writer (Writer, execWriter, runWriter, tell)  -- QuickCheck-import Test.QuickCheck (NonEmptyList (NonEmpty), counterexample, (===), (==>))+import Test.QuickCheck (counterexample, (==>))  -- test-framework import Test.Framework@@ -33,7 +33,7 @@  -- monad-schedule import Control.Monad.Schedule.Class (schedule, scheduleAndFinish)-import Control.Monad.Schedule.Trans (runScheduleIO, runScheduleT)+import Control.Monad.Schedule.Trans (runScheduleT) import Control.Monad.Schedule.Yield  sampleActions :: NonEmpty (MySchedule ())