diff --git a/Concurrential.cabal b/Concurrential.cabal --- a/Concurrential.cabal +++ b/Concurrential.cabal @@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: Concurrential -version: 0.2.1.0 +version: 0.3.0.0 synopsis: Mix concurrent and sequential computation -- description: homepage: http://github.com/avieth/Concurrential @@ -18,8 +18,7 @@ library exposed-modules: Control.Concurrent.Concurrential - , Control.Concurrent.Concurrential.Safely - , Control.Concurrent.Except + -- other-modules: other-extensions: GADTs , DeriveDataTypeable @@ -28,6 +27,6 @@ , DeriveFunctor , ScopedTypeVariables - build-depends: base >=4.7 && <4.8, async >=2.0 && <2.1, stm >= 2.0 + build-depends: base >=4.7 && <4.8, async >=2.0 && <2.1 -- hs-source-dirs: default-language: Haskell2010 diff --git a/Control/Concurrent/Concurrential.hs b/Control/Concurrent/Concurrential.hs --- a/Control/Concurrent/Concurrential.hs +++ b/Control/Concurrent/Concurrential.hs @@ -25,8 +25,8 @@ Concurrential - , Retractor - , Injector + , Runner + , Joiner , runConcurrential , runConcurrentialSimple @@ -34,6 +34,9 @@ , sequentially , concurrently + , wait + -- ^ From Async + ) where import Control.Applicative @@ -42,7 +45,22 @@ import Control.Exception import Data.Typeable --- | Description of the way in which a monadic term should be carried out. +-- | Our own Identity functor, so that we don't have to depend upon some +-- other package. +newtype Identity a = Identity { + runIdentity :: a + } deriving (Functor) + +instance Applicative Identity where + pure = Identity + f <*> x = Identity $ (runIdentity f) (runIdentity x) + +instance Monad Identity where + return = Identity + x >>= k = Identity $ (runIdentity . k) (runIdentity x) + +-- | Description of the way in which a monadic term's evaluation should be +-- carried out. data Choice m t = Sequential (m t) | Concurrent (m t) deriving (Typeable) @@ -52,7 +70,7 @@ Concurrent io -> Concurrent $ fmap f io -- | Description of computation which is composed of sequential and concurrent --- parts in some monad. +-- parts in some monad @m@. data Concurrential m t where SCAtom :: Choice m t -> Concurrential m t SCBind :: Concurrential m s -> (s -> Concurrential m t) -> Concurrential m t @@ -73,69 +91,90 @@ return = pure (>>=) = SCBind --- | This corresponds to the notion of a monad transformer; there is some --- monad g, and then its associated transformer f. If you have an +-- | This corresponds to the notion of a common type of monad transformer: +-- there is some monad g, and then its associated transformer type f, for +-- instance MaybeT = f and Maybe = g +-- If we have an -- --- f m a +-- @ +-- f m a +-- @ -- --- then you can get an +-- then we can get an -- --- m (g a) +-- @ +-- m (g a) +-- @ -- --- just by the definition of what it means to be a monad transformer. -- Here we're interested in the special case where we can achieve IO (g a). -- This does not mean we have to be dealing with an f IO a, it could mean -- that the IO is buried deeper in the transformer stack! -type Injector f g = forall a . f a -> IO (g a) +-- +-- Motivation: @Async@ functions work with @IO@ and only @IO@, but the @m@ +-- parameter of a Concurrential may be some other monad which is capable of +-- performing @IO@, like @Either String IO@ for instance. In order to run +-- computations in this moand through @Async@, we need to know how to get a +-- hold of an @IO@. That's what the runner does. +type Runner f g = forall a . f a -> IO (g a) -- | A witness of this type proves that g is in some sense compatible with IO: -- we can bind through it. -- TBD would it suffice to give the simpler type -- forall a . g (IO a) -> IO (g a) -- ? -type Retractor g = forall a . g (IO (g a)) -> IO (g a) +type Joiner g = forall a . g (IO (g a)) -> IO (g a) -- | Run a Concurrential term with a continuation. We choose CPS here because -- it allows us to explot @withAsync@, giving us a guarantee that an -- exception in a spawning thread will kill spawned threads. +-- +-- TBD generalize the IO to any MonadIO? +-- Maybe not! runConcurrentialK will always run your monad @f@ down to its +-- IO base; it has to, in order to do concurrency. runConcurrentialK - :: (Functor m, Applicative m, Monad m) - => Retractor m - -> Injector f m - -> Concurrential f t + :: (Functor f, Applicative f, Monad f) + => Joiner f + -> Runner m f + -> Concurrential m t -- ^ The computation to run. - -> Async (m s) + -> Async (f s) -- ^ The sequential part. - -> (forall s . (Async (m s), Async (m t)) -> IO (m r)) + -> (forall s . (Async (f s), Async (f t)) -> IO r) -- ^ The continuation; fst is sequential part, snd is value part. -- We use the rank 2 type for s because we really don't care what the -- value of the sequential part it, we just need to wait for it and then -- continue with >>. - -> IO (m r) -runConcurrentialK retractor injector sc sequentialPart k = case sc of + -> IO r +runConcurrentialK joiner runner sc sequentialPart k = case sc of SCAtom choice -> case choice of -- The async created becomes the sequential part and the value -- part. So when another Sequential is encountered, its value part -- will have to wait for this computation to complete. Sequential em -> withAsync - (wait sequentialPart >> injector em) + (wait sequentialPart >> runner em) (\async -> k (async, async)) -- The async created is the value part, but the sequential part -- remains the same. Concurrent em -> withAsync - (injector em) + (runner em) (\async -> k (sequentialPart, async)) SCBind sc next -> - runConcurrentialK retractor injector sc sequentialPart $ \(sequentialPart, asyncS) -> + runConcurrentialK joiner runner sc sequentialPart $ \(sequentialPart, asyncS) -> let waitAndContinue = do s <- wait asyncS - let k' (sequentialPart, asyncT) = wait asyncT - let continue = \x -> runConcurrentialK retractor injector (next x) sequentialPart k' - retractor (fmap continue s) + let continue = \x -> + runConcurrentialK + joiner + runner + (next x) + sequentialPart + (wait . snd) + let unretracted = fmap continue s + joiner unretracted in withAsync waitAndContinue (\async -> k (sequentialPart, async)) SCAp left right -> - runConcurrentialK retractor injector left sequentialPart $ \(sequentialPart, asyncF) -> - runConcurrentialK retractor injector right sequentialPart $ \(sequentialPart, asyncX) -> + runConcurrentialK joiner runner left sequentialPart $ \(sequentialPart, asyncF) -> + runConcurrentialK joiner runner right sequentialPart $ \(sequentialPart, asyncX) -> let waitAndApply = do f <- wait asyncF x <- wait asyncX @@ -145,34 +184,34 @@ -- | Run a Concurrential term, realizing the effects of the IO-like terms which -- compose it. runConcurrential - :: (Functor m, Applicative m, Monad m) - => Retractor m - -> Injector f m - -> Concurrential f t - -> IO (m t) -runConcurrential retractIO injectIO c = do - -- I believe it is safe to supply the async in this way, without using - -- withAsync, because the computation is trivial, and we need not worry - -- about this thread dangling. - sequentialPart <- async $ return (return ()) - runConcurrentialK retractIO injectIO c sequentialPart (wait . snd) + :: (Functor f, Applicative f, Monad f) + => Joiner f + -> Runner m f + -> Concurrential m t + -> (Async (f t) -> IO r) + -- ^ Similar contract to withAsync; the Async argument is useless outside of + -- this function. + -> IO r +runConcurrential joiner runner c k = do + let action = \sequentialPart -> + runConcurrentialK joiner runner c sequentialPart (k . snd) + withAsync (return (return ())) action -runConcurrentialSimple :: Concurrential IO t -> IO t -runConcurrentialSimple = join . runConcurrential retractor injector +runConcurrentialSimple :: Concurrential IO t -> (Async t -> IO r) -> IO r +runConcurrentialSimple c k = runConcurrential simpleJoiner simpleRunner c (continue k) + where - retractor :: Retractor IO - retractor = join - injector :: Injector IO IO - injector io = io >>= return . return - -- Note that if we chose injector = return we would lose concurrency! - -- This is very subtle and I don't understand it well. - -- My best explanation: the injector must bring the effect held in the - -- term "to the front" so that it would be realized by, for instance, a - -- withAsync call. If we leave it as just @return@ then runConcurrential - -- will concurrently build up the term which will ultimately be run - -- sequentially. --- | Create an IO which must be run sequentially. + continue :: (Async t -> IO r) -> (Async (Identity t) -> IO r) + continue k = \async -> k $ fmap runIdentity async + + simpleJoiner :: Joiner Identity + simpleJoiner = runIdentity + + simpleRunner :: Runner IO Identity + simpleRunner = fmap Identity + +-- | Create an effect which must be run sequentially. -- If a @sequentially io@ appears in a @Concurrential t@ term then it will -- always be run to completion before any later sequential part of the term -- is run. Consider the following terms: @@ -194,7 +233,7 @@ sequentially :: m t -> Concurrential m t sequentially = SCAtom . Sequential --- | Create an IO which is run concurrently where possible, i.e. whenever it +-- | Create an effect which is run concurrently where possible, i.e. whenever it -- combined applicatively with other terms. For instance: -- -- @ @@ -207,53 +246,3 @@ -- been used. concurrently :: m t -> Concurrential m t concurrently = SCAtom . Concurrent - --- So how can I accomplish my goal now? How does shared state come in to play? --- Perhaps it remains a transformer? Ok, sure, but how do we hook up some --- "on exception" callbacks? That has to be part of an Extender/Retractor pair. --- Ah yes, we can factor that into the SharedState transformer's runner! --- --- Hm but yet another problem lurks... every bare IO will get an exception --- handler, sure, but how will I know what to do with the exception, when it --- lacks any context? In the desired use case I need to remember, in the --- exception handler, the resource descriptor for which the thread was working. --- That's lost in the general `runExceptionSafe` manner! --- But then, do we really need the context? The important part is that every --- thread works to completion or exception, and we have that. --- On the other hand, in the solution that I have here, the programmer is simply --- not allowed to say what to do on exception. That seems wrong. --- So perhaps we add an SCCatch term --- --- SCCatch :: Concurrential t -> (SomeException -> Concurrential t) -> Concurrential t --- --- but this would make the work that I just did redundant: it shifts from --- offering after-the-fact handling to up-front handling... is it not enough to --- handle the exceptions in the IOs that you give to concurrently or --- sequentially? If all of these things are exception safe, then it's all --- good. --- And then there's the point that brought us here: if some thread does go --- wrong, no new threads should be created, and computation should be abandoned. --- Thus the interface is: if you can't carry on, throw an exception, and we've --- got your back. --- Yeah, I favour not allowing the programmer to write up exception handling in --- Concurrential (do it in the IOs) since it's just simpler. But is it too --- restrictive?!?!? --- --- What if we assert that all embedded IOs must be IO (m t) for some monad m? --- In fact, all we need is some MonadIO, rather than IO itself. This allows --- the exception handling via --- in :: IO t -> ExceptT SomeException IO t --- in io = (liftIO io) `catch` (\(e :: SomeException) -> throwE e) --- Yeah, why not this? We can skip the class and just use a rank 2 type --- featuring --- (forall a . IO a -> m a) --- but of course, runConcurrentialK needs to give its results in IO, for it --- spawns threads, no? Indeed no, liftIO should suffice. --- withAsync :: IO a -> (Async a -> IO b) -> IO b --- we can use that with liftIO to get... --- liftWithAsync :: m a -> (Async a -> m b) -> m b --- liftWithAsync x k = --- hm no this is not what we want: we wish to use withAsync to do the entire --- monadic computation in another thread, and then bind through its result. --- I think what we really need is --- (forall a . m a -> IO a) diff --git a/Control/Concurrent/Concurrential/Safely.hs b/Control/Concurrent/Concurrential/Safely.hs deleted file mode 100644 --- a/Control/Concurrent/Concurrential/Safely.hs +++ /dev/null @@ -1,51 +0,0 @@ -{-| -Module : Control.Concurrent.Concurrential.Safely -Description : Handle all exceptions in Concurrential computation. -Copyright : (c) Alexander Vieth, 2015 -Licence : BSD3 -Maintainer : aovieth@gmail.com -Stability : experimental -Portability : non-portable (GHC only) --} - -{-# LANGUAGE ScopedTypeVariables #-} - -module Control.Concurrent.Concurrential.Safely ( - - safely - , runSafely - - ) where - -import Control.Applicative -import Control.Monad -import Control.Exception -import Control.Concurrent.Except -import Control.Concurrent.Concurrential - -injector :: Injector (ExceptT SomeException IO) (Either SomeException) -injector term = runExceptT term >>= return - -retractor :: Retractor (Either SomeException) -retractor term = case term of - Left e -> return $ Left e - Right v -> v - --- | Make an arbitrary IO suitable for use with @sequentially@ or @concurrently@ --- so as to produce a term that can be run by @runSafely@: --- --- let a = concurrently . safely $ dangerousComputation1 --- b = concurrently . safely $ dangerousComputation2 --- in runSafely $ a *> b --- -safely :: IO a -> ExceptT SomeException IO a -safely io = ExceptT ((Right <$> io) `catch` (\(e :: SomeException) -> return $ Left e)) - --- | Run a term such that computation is halted as soon as an exception is --- encountered, but any pending threads are waited on. The first exception --- to be thown (in term-order, not necessarily temporal order) is given as --- Left, and a Right is given if no exception is encountered. -runSafely - :: Concurrential (ExceptT SomeException IO) a - -> IO (Either SomeException a) -runSafely = runConcurrential retractor injector diff --git a/Control/Concurrent/Except.hs b/Control/Concurrent/Except.hs deleted file mode 100644 --- a/Control/Concurrent/Except.hs +++ /dev/null @@ -1,58 +0,0 @@ -{-| -Module : Control.Concurrent.Except -Description : Just like ExceptT from transformers but with a different Applicative - instance. -Copyright : (c) Alexander Vieth, 2015 -Licence : BSD3 -Maintainer : aovieth@gmail.com -Stability : experimental -Portability : non-portable (GHC only) --} - -{-# LANGUAGE DeriveDataTypeable #-} - -module Control.Concurrent.Except ( - - ExceptT(..) - , injectE - , throwE - , catchE - - ) where - -import Control.Applicative -import Data.Typeable - -data ExceptT e m a = ExceptT { - runExceptT :: m (Either e a) - } deriving (Typeable) - -instance Functor m => Functor (ExceptT e m) where - fmap f term = ExceptT $ (fmap . fmap) f (runExceptT term) - -instance Applicative m => Applicative (ExceptT e m) where - pure = ExceptT . pure . pure - f <*> x = ExceptT $ (<*>) <$> runExceptT f <*> runExceptT x - -instance Monad m => Monad (ExceptT e m) where - return = ExceptT . return . return - x >>= k = ExceptT $ do - outcome <- runExceptT x - case outcome of - Left e -> return $ Left e - Right x -> runExceptT $ k x - -injectE :: Applicative m => Either e a -> ExceptT e m a -injectE x = case x of - Left e -> throwE e - Right v -> pure v - -throwE :: Applicative m => e -> ExceptT e m a -throwE = ExceptT . pure . Left - -catchE :: Monad m => ExceptT e m a -> (e -> ExceptT e' m a) -> ExceptT e' m a -catchE exceptT handler = ExceptT $ do - outcome <- runExceptT exceptT - case outcome of - Left exception -> runExceptT $ handler exception - Right value -> return $ Right value