Concurrential 0.2.1.0 → 0.3.0.0
raw patch · 4 files changed
+96/−217 lines, 4 filesdep −stm
Dependencies removed: stm
Files
- Concurrential.cabal +3/−4
- Control/Concurrent/Concurrential.hs +93/−104
- Control/Concurrent/Concurrential/Safely.hs +0/−51
- Control/Concurrent/Except.hs +0/−58
Concurrential.cabal view
@@ -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
Control/Concurrent/Concurrential.hs view
@@ -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)
− Control/Concurrent/Concurrential/Safely.hs
@@ -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
− Control/Concurrent/Except.hs
@@ -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