packages feed

concurrent-machines (empty) → 0.1.0.0

raw patch · 13 files changed

+1025/−0 lines, 13 filesdep +asyncdep +basedep +concurrent-machinessetup-changed

Dependencies added: async, base, concurrent-machines, containers, lifted-async, machines, monad-control, semigroups, tasty, tasty-hunit, time, transformers, transformers-base

Files

+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2014, Anthony Cowley++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Anthony Cowley nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ concurrent-machines.cabal view
@@ -0,0 +1,72 @@+name:                concurrent-machines+version:             0.1.0.0+synopsis:            Concurrent networked stream transducers++description: A simple use-case for this library is to run the stages+             of a pipelined streaming computation concurrently. If+             data is streaming through multiple processing stages, you+             might build a machine like+             .+             @+             step1 >~> step2 >~> step3+             @+             .+             The @>~>@ operator connects the machines on+             either side with a one-element buffer. This means that+             data is pulled from upstream sources eagerly (perhaps+             pulling one more value than will be consumed by+             downstream), but it also means that each stage can be+             working simultaneously, increasing throughput of the+             entire pipeline.+             .+             A few small examples are available in the @examples@+             directory of the source repository.++license:             BSD3+license-file:        LICENSE+author:              Anthony Cowley+maintainer:          acowley@gmail.com+copyright:           Copyright (C) 2014 Anthony Cowley+category:            Concurrency, Control+build-type:          Simple+-- extra-source-files:  +cabal-version:       >=1.10++source-repository head+  type:     git+  location: http://github.com/acowley/concurrent-machines.git++library+  exposed-modules:     Data.Machine.Concurrent,+                       Data.Machine.Fanout,+                       Data.Machine.Regulated,+                       Data.Machine.Concurrent.AsyncStep,+                       Data.Machine.Concurrent.Buffer,+                       Data.Machine.Concurrent.Fanout,+                       Data.Machine.Concurrent.Scatter,+                       Data.Machine.Concurrent.Tee,+                       Data.Machine.Concurrent.Wye+  -- other-modules:       +  other-extensions:    GADTs, FlexibleContexts, RankNTypes, TupleSections, +                       ScopedTypeVariables+  build-depends:       base >= 4.6 && < 5, +                       monad-control >= 1.0 && < 1.1,+                       transformers >= 0.4 && < 0.5,+                       time >= 1.4 && < 1.6,+                       containers >= 0.5 && < 0.6,+                       transformers-base >= 0.4 && < 0.5,+                       machines >= 0.5 && < 0.6,+                       async >= 2.0.1 && < 2.1,+                       lifted-async >= 0.1 && < 0.8,+                       semigroups >= 0.8 && < 0.17+  hs-source-dirs:      src+  default-language:    Haskell2010++test-suite tests+  type: exitcode-stdio-1.0+  hs-source-dirs: tests+  main-is: AllTests.hs+  ghc-options: -Wall -O0+  default-language: Haskell2010+  build-depends: base >= 4.6 && < 5, concurrent-machines, machines,+                 tasty, tasty-hunit, transformers, time
+ src/Data/Machine/Concurrent.hs view
@@ -0,0 +1,125 @@+{-# LANGUAGE GADTs, FlexibleContexts, RankNTypes, ScopedTypeVariables,+             TupleSections #-}+-- | The primary use of concurrent machines is to establish a+-- pipelined architecture that can boost overall throughput by running+-- each stage of the pipeline at the same time. The processing, or+-- production, rate of each stage may not be identical, so facilities+-- are provided to loosen the temporal coupling between pipeline+-- stages using buffers.+--+-- This architecture also lends itself to operations where multiple+-- workers are available for procesisng inputs. If each worker is to+-- process the same set of inputs, consider 'fanout' and+-- 'fanoutSteps'. If each worker is to process a disjoint set of+-- inputs, consider 'scatter'.+module Data.Machine.Concurrent (module Data.Machine,+                                -- * Concurrent connection+                                (>~>), (<~<),+                                -- * Buffered machines+                                buffer, rolling,+                                bufferConnect, rollingConnect,+                                -- * Concurrent processing of shared inputs+                                fanout, fanoutSteps,+                                -- * Concurrent multiple-input machines+                                wye, tee, scatter, splitSum, mergeSum, +                                splitProd) where+import Control.Applicative+import Control.Concurrent.Async.Lifted+import Control.Monad (join)+import Control.Monad.Trans.Control+import Data.Machine hiding (tee, wye)+import Data.Machine.Concurrent.AsyncStep+import Data.Machine.Concurrent.Buffer+import Data.Machine.Concurrent.Fanout+import Data.Machine.Concurrent.Scatter+import Data.Machine.Concurrent.Wye+import Data.Machine.Concurrent.Tee++-- | Build a new 'Machine' by adding a 'Process' to the output of an+-- old 'Machine'. The upstream machine is run concurrently with+-- downstream with the aim that upstream will have a yielded value+-- ready as soon as downstream awaits. This effectively creates a+-- buffer between upstream and downstream, or source and sink, that+-- can contain up to one value.+--+-- @+-- ('<~<') :: 'Process' b c -> 'Process' a b -> 'Process' a c+-- ('<~<') :: 'Process' c d -> 'Data.Machine.Tee.Tee' a b c -> 'Data.Machine.Tee.Tee' a b d+-- ('<~<') :: 'Process' b c -> 'Machine' k b -> 'Machine' k c+-- @+(<~<) :: MonadBaseControl IO m+     => ProcessT m b c -> MachineT m k b -> MachineT m k c+mp <~< ma = racers ma mp++-- | Flipped ('<~<').+(>~>) :: MonadBaseControl IO m+     => MachineT m k b -> ProcessT m b c -> MachineT m k c+ma >~> mp = mp <~< ma++infixl 7 >~>++-- | We want the first available response.+waitEither' :: MonadBaseControl IO m +            => Maybe (Async (StM m a)) -> Async (StM m b)+            -> m (Either a b)+waitEither' Nothing y = Right <$> wait y+waitEither' (Just x) y = waitEither x y++-- | Let a source and a sink chase each other, providing an effective+-- one-element buffer between the two. The idea is to run both+-- concurrently at all times so that as soon as the sink 'Await's, we+-- have a source-yielded value to provide it. This, of course,+-- involves eagerly running the source, percolating its 'Await's up+-- the chain as soon as possible.+racers :: forall m k a b. MonadBaseControl IO m+       => MachineT m k a -> ProcessT m a b -> MachineT m k b+racers src snk = MachineT . join $+                 go <$> (Just <$> asyncRun src) <*> asyncRun snk+  where go :: MonadBaseControl IO m+           => Maybe (AsyncStep m k a)+           -> AsyncStep m (Is a) b+           -> m (MachineStep m k b)+        go srcA snkA =+          waitEither' srcA snkA >>= \n -> case n of+            Left (Stop :: MachineStep m k a) -> go Nothing snkA+            Left (Yield o k) -> wait snkA >>= \m -> case m of+              (Stop :: MachineStep m (Is a) b) -> return Stop+              Yield o' k' -> return . Yield o' . MachineT . flushDown k' $+                             \f -> join $ go <$> (Just <$> asyncRun k)+                                             <*> asyncRun (f o)+              Await f Refl _ -> join $ go <$> (Just <$> asyncRun k)+                                          <*> asyncRun (f o)+            Left (Await g kg fg) -> asyncAwait g kg fg $+                                    MachineT . flip go snkA . Just+            Right (Stop :: MachineStep m (Is a) b) -> return Stop+            Right (Yield o k) -> asyncRun k >>=+                                 return . Yield o . MachineT . go srcA+            Right (Await f Refl ff) -> case srcA of+              Nothing -> asyncRun ff >>= go Nothing+              Just src' -> wait src' >>= \m -> case m of+                Stop -> return Stop+                Yield o k -> join $ go <$> (Just <$> asyncRun k)+                                       <*> asyncRun (f o)+                a -> feedUp (encased a) $ \o k -> join $+                       go <$> (Just <$> asyncRun k) <*> asyncRun (f o)+        -- If we have an upstream source value ready, we must flush+        -- all available values yielded by downstream until it awaits.+        flushDown :: Monad m+                  => ProcessT m a b+                  -> ((a -> ProcessT m a b) -> m (MachineStep m k b))+                  -> m (MachineStep m k b)+        flushDown m k = runMachineT m >>= \s -> case s of+          Stop -> return Stop+          Yield o m' -> return . Yield o . MachineT $ flushDown m' k+          Await f Refl _ -> k f+        -- If downstream is awaiting an input, we must pull in all+        -- necessary upstream awaits until we have a yielded value to+        -- push downstream.+        feedUp :: MonadBaseControl IO m+               => MachineT m k a+               -> (a -> MachineT m k a -> m (MachineStep m k b))+               -> m (MachineStep m k b)+        feedUp m k = runMachineT m >>= \s -> case s of+          Stop -> return Stop+          Yield o m' -> k o m'+          Await g kg fg -> return $ awaitStep g kg fg (MachineT . flip feedUp k)
+ src/Data/Machine/Concurrent/AsyncStep.hs view
@@ -0,0 +1,61 @@+{-# LANGUAGE FlexibleContexts, GADTs, RankNTypes, ScopedTypeVariables #-}+-- | Internal helpers for taking asynchronous machine steps.+module Data.Machine.Concurrent.AsyncStep where+import Control.Concurrent.Async.Lifted (Async, async, wait)+import Control.Monad.Trans.Control (MonadBaseControl, StM)+import Data.Machine++-- | Slightly more compact notation for a 'Step'.+type MachineStep m k o = Step k o (MachineT m k o)++-- | Compact notation for a 'Step' taken asynchronously.+type AsyncStep m k o = Async (StM m (MachineStep m k o))++-- | Build an 'Await' step given a continuation that provides+-- subsequent steps. @awaitStep f sel ff k@ is like applying the+-- 'Await' constructor directly, but the continuation @k@ is used to+-- continue the machine. +-- +-- @awaitStep f sel ff k = Await (k . f) sel (k ff)@+awaitStep :: (a -> d) -> k' a -> d -> (d -> r) -> Step k' b r+awaitStep f sel ff k = Await (k . f) sel (k ff)++-- | Run one step of a machine as an 'Async' operation.+asyncRun :: MonadBaseControl IO m => MachineT m k o -> m (AsyncStep m k o)+asyncRun = async . runMachineT++-- | Satisfy a downstream Await by blocking on an upstream step.+stepAsync :: forall m k k' a' d b.+             MonadBaseControl IO m+           => (forall c. k c -> k' c)+           -> AsyncStep m k a'+           -> (a' -> d)+           -> d+           -> d+           -> (AsyncStep m k a' -> d -> MachineT m k' b)+           -> MachineT m k' b+stepAsync sel src f def prev go = MachineT $ wait src >>= \u -> case u of+  Stop -> go' stopped def+  Yield a k -> go' k (f a)+  Await g kg fg -> return $ awaitStep g (sel kg) fg (MachineT . flip go' prev)+  where go' :: MachineT m k a' -> d -> m (MachineStep m k' b)+        go' k d = asyncRun k >>= runMachineT . flip go d++-- | @asyncEncased f x@ launches @x@ and provides the resulting+-- 'AsyncStep' to @f@. Turn a function on 'AsyncStep' to a funciton on+-- 'MachineT'.+asyncEncased :: MonadBaseControl IO m+             => (AsyncStep m k1 o1 -> MachineT m k o)+             -> MachineT m k1 o1+             -> MachineT m k o+asyncEncased f x = MachineT $ asyncRun x >>= runMachineT . f++-- | Similar to 'awaitStep', but for continuations that want their inputs+-- to be run asynchronously.+asyncAwait :: MonadBaseControl IO m+           => (a -> MachineT m k o)+           -> k' a+           -> MachineT m k o+           -> (AsyncStep m k o -> MachineT m k1 o1)+           -> m (Step k' b (MachineT m k1 o1))+asyncAwait f sel ff = return . awaitStep f sel ff . asyncEncased
+ src/Data/Machine/Concurrent/Buffer.hs view
@@ -0,0 +1,145 @@+{-# LANGUAGE FlexibleContexts, GADTs, ScopedTypeVariables, TupleSections #-}+-- | Place buffers between two machines. This is most useful with+-- irregular production rates.+module Data.Machine.Concurrent.Buffer (+  -- * Blocking buffers+  bufferConnect, buffer,+  -- * Non-blocking (rolling) buffers+  rollingConnect, rolling,+  -- * Internal helpers+  mediatedConnect, BufferRoom(..)+  ) where+import Control.Applicative ((<$>), (<*>))+import Control.Concurrent.Async.Lifted (wait, waitEither)+import Control.Monad.Trans.Control (MonadBaseControl)+import Control.Monad (join, (>=>))+import Data.Machine.Concurrent.AsyncStep+import Data.Machine+import Data.Sequence (ViewL(..), (|>))+import qualified Data.Sequence as S+import Data.Traversable (traverse)++-- | Drain downstream until it awaits a value, then pass the awaiting+-- step to the given function.+drain :: (Functor m, Monad m)+      => MachineStep m k a+      -> (MachineStep m k a -> m (MachineStep m k' a))+      -> m (MachineStep m k' a)+drain z k = go z+  where go Stop = return Stop+        go (Yield o kd) = Yield o . MachineT . go <$> runMachineT kd+        go aStep = k aStep++-- | Feed upstream until it yields a value, then pass the yielded+-- value and next step to the given function.+feedToBursting :: Monad m+               => MachineStep m k a+               -> (Maybe (a, MachineT m k a) -> m (MachineStep m k b))+               -> m (MachineStep m k b)+feedToBursting z k = go z+  where go Stop = k Nothing+        go (Await f kf ff) = return $+          Await (\a -> go' (f a)) kf (go' ff)+        go (Yield o kk) = k $ Just (o, kk)+        go' step = MachineT $ runMachineT step >>= go++-- | Mediate a 'MachineT' and a 'ProcessT' with a bounded capacity+-- buffer. The source machine runs concurrently with the sink process,+-- and is only blocked when the buffer is full.+bufferConnect :: MonadBaseControl IO m+              => Int -> MachineT m k b -> ProcessT m b c -> MachineT m k c+bufferConnect n = mediatedConnect S.empty snoc view+  where snoc acc x = (if S.length acc < n - 1 then Vacancy else NoVacancy) $+                       acc |> x+        view acc = case S.viewl acc of+                     EmptyL -> Nothing+                     x :< acc' -> Just (x, acc')++-- | Mediate a 'MachineT' and a 'ProcessT' with a rolling buffer. The+-- source machine runs concurrently with the sink process and is never+-- blocked. If the sink process can not keep up with upstream, yielded+-- values will be dropped.+rollingConnect :: MonadBaseControl IO m+              => Int -> MachineT m k b -> ProcessT m b c -> MachineT m k c+rollingConnect n = mediatedConnect S.empty snoc view+  where snoc acc x = Vacancy $ S.take (n-1) acc |> x+        view acc = case S.viewl acc of+                     EmptyL -> Nothing+                     x :< acc' -> Just (x, acc')++-- | Eagerly request values from the wrapped machine. Values are+-- placed in a buffer of the given size. When the buffer is full+-- (i.e. downstream is running behind), we stop pumping the wrapped+-- machine.+buffer :: MonadBaseControl IO m => Int -> MachineT m k o -> MachineT m k o+buffer n src = bufferConnect n src echo++-- | Eagerly request values from the wrapped machine. Values are+-- placed in a rolling buffer of the given size. If downstream can not+-- catch up, values yielded by the wrapped machine will be dropped.+rolling :: MonadBaseControl IO m => Int -> MachineT m k o -> MachineT m k o+rolling n src = rollingConnect n src echo++-- | Indication if the payload value is "full" or not.+data BufferRoom a = NoVacancy a | Vacancy a deriving (Eq, Ord, Show)++-- | Mediate a 'MachineT' and a 'ProcessT' with a buffer. +--+-- @mediatedConnect z snoc view source sink@ pipes @source@ into+-- @sink@ through a buffer initialized to @z@ and updated with+-- @snoc@. Upstream is blocked if @snoc@ indicates that the buffer is+-- full after adding a new element. Downstream blocks if @view@+-- indicates that the buffer is empty. Otherwise, @view@ is expected+-- to return the next element to process and an updated buffer.+mediatedConnect :: forall m t b k c. MonadBaseControl IO m+                => t -> (t -> b -> BufferRoom t) -> (t -> Maybe (b,t))+                -> MachineT m k b -> ProcessT m b c -> MachineT m k c+mediatedConnect z snoc view src0 snk0 = +  MachineT $ do srcFuture <- asyncRun src0+                snkFuture <- asyncRun snk0+                go z (Just srcFuture) snkFuture+  where -- Wait for the next available step+        go :: t+           -> Maybe (AsyncStep m k b)+           -> AsyncStep m (Is b) c+           -> m (MachineStep m k c)+        go acc src snk = maybe (Left <$> wait snk) (waitEither snk) src >>=+                           goStep acc . either (Right . (,src)) (Left . (,snk))++        -- Kick off the next step of both the source and the sink+        goAsync :: t+                -> Maybe (MachineT m k b)+                -> ProcessT m b c+                -> m (MachineStep m k c)+        goAsync acc src snk = +          join $ go acc <$> traverse asyncRun src <*> asyncRun snk++        -- Handle whichever step is ready first+        goStep :: t  -> Either (MachineStep m k b, AsyncStep m (Is b) c)+                               (MachineStep m (Is b) c, Maybe (AsyncStep m k b))+               -> m (MachineStep m k c)+        goStep acc step = case step of+          -- @src@ stepped first+          Left (Stop, snk) -> go acc Nothing snk+          Left (Await g kg fg, snk) -> +            asyncAwait g kg fg (MachineT . flip (go acc) snk . Just)+          Left (Yield o k, snk) -> case snoc acc o of+            -- add it to the right end of the buffer+            Vacancy acc' -> asyncRun k >>= flip (go acc') snk . Just+            -- buffer was full+            NoVacancy acc' -> +              let go' snk' = do src' <- asyncRun k+                                goStep acc' (Right (snk', Just src'))+              in wait snk >>= flip drain go'++          -- @snk@ stepped first+          Right (Stop, _) -> return Stop+          Right (Yield o k, src) -> +            return $ Yield o (MachineT $ asyncRun k >>= go acc src)+          Right (Await f Refl ff, src) -> +            case view acc of+              Nothing -> maybe (goAsync acc Nothing ff) (wait >=> demandSrc) src+              Just (x, acc') -> asyncRun (f x) >>= go acc' src+            where demandSrc = flip feedToBursting go'+                  go' Nothing = goAsync acc Nothing ff+                  go' (Just (o, k)) = goAsync acc (Just k) (f o)
+ src/Data/Machine/Concurrent/Fanout.hs view
@@ -0,0 +1,76 @@+{-# LANGUAGE FlexibleContexts, GADTs, ScopedTypeVariables #-}+-- | Provide a notion of fanout wherein a single input is passed to+-- several consumers. The consumers are run concurrently.+module Data.Machine.Concurrent.Fanout (fanout, fanoutSteps) where+import Control.Arrow (second)+import Control.Concurrent.Async.Lifted (Async, async, wait)+import Control.Monad (foldM)+import Control.Monad.Trans.Control (MonadBaseControl, StM)+import Data.Machine (Step(..), MachineT(..), encased, ProcessT, Is(..))+import Data.Machine.Concurrent.AsyncStep (MachineStep)+import Data.Maybe (catMaybes)+import Data.Monoid (Monoid, mempty, mconcat)+import Data.Semigroup (Semigroup(sconcat))+import Data.List.NonEmpty (NonEmpty((:|)))++-- | Feed a value to a 'ProcessT' at an 'Await' 'Step'. If the+-- 'ProcessT' is awaiting a value, then its next step is+-- returned. Otherwise, the original process is returned.+feed :: forall m a b. MonadBaseControl IO m+     => a -> ProcessT m a b+     -> m (Async (StM m ([b], Maybe (MachineStep m (Is a) b))))+feed x m = async $ runMachineT m >>= \(v :: MachineStep m (Is a) b) ->+             case v of+               Await f Refl _ -> runMachineT (f x) >>= flushYields+               s -> return ([]::[b], Just s)++-- | Like 'Data.List.mapAccumL' but with a monadic accumulating+-- function.+mapAccumLM :: (Functor m, MonadBaseControl IO m)+           => (acc -> x -> m (acc, y)) -> acc -> [Async (StM m x)]+           -> m (acc, [y])+mapAccumLM f z = fmap (second ($ [])) . foldM aux (z,id)+  where aux (acc,ys) x = do (yielded, nxt) <- wait x >>= f acc+                            return $ (yielded, (nxt:) . ys)++-- | Exhaust a sequence of all successive 'Yield' steps taken by a+-- 'MachineT'. Returns the list of yielded values and the next+-- (non-Yield) step of the machine.+flushYields :: Monad m+            => Step k o (MachineT m k o) -> m ([o], Maybe (MachineStep m k o))+flushYields = go id+  where go rs (Yield o s) = runMachineT s >>= go ((o:) . rs)+        go rs Stop = return (rs [], Nothing)+        go rs s = return (rs [], Just s)++-- | Share inputs with each of a list of processes in lockstep. Any+-- values yielded by the processes for a given input are combined into+-- a single yield from the composite process.+fanout :: (Functor m, MonadBaseControl IO m, Semigroup r)+       => [ProcessT m a r] -> ProcessT m a r+fanout xs = encased $ Await (MachineT . aux) Refl (fanout xs)+  where aux y = do (rs,xs') <- mapM (feed y) xs >>= mapAccumLM yields []+                   let nxt = fanout . map encased $ catMaybes xs'+                   case rs of+                     [] -> runMachineT nxt+                     (r:rs') -> return $ Yield (sconcat $ r :| rs') nxt+        yields rs (rs', Nothing) = return (rs' ++ rs, Nothing)+        yields rs (rs', Just s) = return (rs' ++ rs, Just s)++-- | Share inputs with each of a list of processes in lockstep. If+-- none of the processes yields a value, the composite process will+-- itself yield 'mempty'. The idea is to provide a handle on steps+-- only executed for their side effects. For instance, if you want to+-- run a collection of 'ProcessT's that await but don't yield some+-- number of times, you can use 'fanOutSteps . map (fmap (const ()))'+-- followed by a 'taking' process.+fanoutSteps :: (Functor m, MonadBaseControl IO m, Monoid r)+            => [ProcessT m a r] -> ProcessT m a r+fanoutSteps xs = encased $ Await (MachineT . aux) Refl (fanoutSteps xs)+  where aux y = do (rs,xs') <- mapM (feed y) xs >>= mapAccumLM yields []+                   let nxt = fanoutSteps . map encased $ catMaybes xs'+                   if null rs+                   then return $ Yield mempty nxt+                   else return $ Yield (mconcat rs) nxt+        yields rs (rs', Nothing) = return (rs' ++ rs,Nothing)+        yields rs (rs', Just s) = return (rs'++rs, Just s)
+ src/Data/Machine/Concurrent/Scatter.hs view
@@ -0,0 +1,239 @@+{-# LANGUAGE FlexibleContexts, GADTs, TupleSections, RankNTypes,+             ScopedTypeVariables #-}+-- | Routing for splitting and merging processing pipelines.+module Data.Machine.Concurrent.Scatter (+  scatter, mergeSum, splitSum, splitProd+  ) where+import Control.Arrow ((***))+import Control.Concurrent.Async (Async, waitAny)+import Control.Concurrent.Async.Lifted (wait, waitEither, waitBoth)+import Control.Monad ((>=>))+import Control.Monad.Base (liftBase)+import Control.Monad.Trans.Control (MonadBaseControl, restoreM, StM)+import Data.Machine+import Data.Machine.Concurrent.AsyncStep++holes :: [a] -> [[a]]+holes = go id+  where go _ [] = []+        go x (y:ys) = x ys : go (x . (y:)) ys++diff :: [a] -> [(a,[a])]+diff xs = zip xs (holes xs)++waitAnyHole :: MonadBaseControl IO m => [(Async (StM m a), [b])] -> m (a, [b])+waitAnyHole xs = do (_,(s,b)) <- liftBase $ waitAny xs'+                    fmap (,b) (restoreM s)+  where xs' = map (\(a,b) -> fmap (,b) a) xs++-- | Produces values from whichever source 'MachineT' yields+-- first. This operation may also be viewed as a /gather/ operation in+-- that all values produced by the given machines are interleaved when+-- fed downstream. Note that inputs are /not/ shared. The composite+-- machine will await an input when any constituent machine awaits an+-- input. That input will be supplied to the awaiting constituent and+-- no other.+--+-- Some examples of more specific useful types @scatter@ may be used+-- at,+-- +-- @+-- scatter :: [ProcessT m a b] -> ProcessT m a b+-- scatter :: [SourceT m a] -> SourceT m a+-- @+--+-- The former may be used to stream data through a collection of+-- worker 'Process'es, the latter may be used to intersperse values+-- from a collection of sources.+scatter :: MonadBaseControl IO m => [MachineT m k o] -> MachineT m k o+scatter [] = stopped+scatter sinks = MachineT $ mapM asyncRun sinks+                 >>= waitAnyHole . diff+                 >>= uncurry go+  where go :: MonadBaseControl IO m+           => MachineStep m k o+           -> [AsyncStep m k o]+           -> m (MachineStep m k o)+        go Stop [] = return Stop+        go Stop sinks' = waitAnyHole (diff sinks') >>= uncurry go+        go (Yield o k) sinks' = +          asyncRun k >>= return . Yield o . MachineT . goWait . (:sinks')+        go (Await f fk ff) sinks' =+          asyncAwait f fk ff (MachineT . goWait . (:sinks'))+        goWait :: MonadBaseControl IO m+               => [AsyncStep m k o]+               -> m (MachineStep m k o)+        goWait = waitAnyHole . diff >=> uncurry go++-- | Similar to 'Control.Arrow.|||': split the input between two+-- processes and merge their outputs.+--+-- Connect two processes to the downstream tails of a 'Machine' that+-- produces 'Either's. The two downstream consumers are run+-- concurrently when possible. When one downstream consumer stops, the+-- other is allowed to run until it stops or the upstream source+-- yields a value the remaining consumer can not handle.+--+-- @mergeSum sinkL sinkR@ produces a topology like this,+--+-- @+--                                 sinkL+--                                /      \+--                              a          \+--                             /            \+--    source -- Either a b -->                -- r -->+--                             \\            /+--                              b          /+--                               \\       /+--                                 sinkR +-- @+mergeSum :: forall m a b r. MonadBaseControl IO m+         => ProcessT m a r -> ProcessT m b r -> ProcessT m (Either a b) r+mergeSum snkL snkR = MachineT $ do sl <- asyncRun snkL+                                   sr <- asyncRun snkR+                                   go sl sr+  where go :: MonadBaseControl IO m+           => AsyncStep m (Is a) r+           -> AsyncStep m (Is b) r+           -> m (MachineStep m (Is (Either a b)) r)+        go sl sr = waitEither sl sr >>= +                   \(s :: Either (MachineStep m (Is a) r)+                                 (MachineStep m (Is b) r)) -> case s of+          Left Stop -> wait sr >>= runMachineT . rightOnly . encased+          Right Stop -> wait sl >>= runMachineT . leftOnly . encased++          Left (Yield o k) -> +            return . Yield o . MachineT $ asyncRun k >>= flip go sr+          Right (Yield o k) -> +            return . Yield o . MachineT $ asyncRun k >>= go sl+                               +          Left (Await f Refl ff) ->+            return $ +            Await (\u -> case u of+                           Left a -> MachineT $ asyncRun (f a) >>= flip go sr+                           Right b -> MachineT $ +                                      wait sr >>= forceFeed (go sl) b . encased)+                  Refl+                  (MachineT $ asyncRun ff >>= flip go sr)+          Right (Await g Refl gg) -> return $+            Await (\u -> case u of+                           Left a -> +                             MachineT $+                             wait sl >>= forceFeed (flip go sr) a . encased+                           Right b -> MachineT $ asyncRun (g b) >>= go sl)+                  Refl+                  (MachineT $ asyncRun gg >>= go sl)++-- | Similar to 'Control.Arrow.+++': split the input between two+-- processes, retagging and merging their outputs.+--+-- The two processes are run concurrently whenever possible.+splitSum :: forall m a b c d. MonadBaseControl IO m+         => ProcessT m a b -> ProcessT m c d -> ProcessT m (Either a c) (Either b d)+splitSum snkL snkR = MachineT $ do sl <- asyncRun (fmap lft snkL)+                                   sr <- asyncRun (fmap rgt snkR)+                                   go sl sr+  where lft :: b -> Either b d+        lft = Left+        rgt :: d -> Either b d+        rgt = Right+        go :: MonadBaseControl IO m+           => AsyncStep m (Is a) (Either b d)+           -> AsyncStep m (Is c) (Either b d)+           -> m (MachineStep m (Is (Either a c)) (Either b d))+        go sl sr = waitEither sl sr >>=+                   \(s :: Either (MachineStep m (Is a) (Either b d))+                                 (MachineStep m (Is c) (Either b d))) -> case s of+          Left Stop -> wait sr >>= runMachineT . rightOnly . encased+          Right Stop -> wait sl >>= runMachineT . leftOnly . encased++          Left (Yield o k) -> +            return . Yield o . MachineT $ asyncRun k >>= flip go sr+          Right (Yield o k) -> +            return . Yield o . MachineT $ asyncRun k >>= go sl+                               +          Left (Await f Refl ff) ->+            return $ +            Await (\u -> case u of+                           Left a -> MachineT $ asyncRun (f a) >>= flip go sr+                           Right b -> MachineT $ +                                      wait sr >>= forceFeed (go sl) b . encased)+                  Refl+                  (MachineT $ asyncRun ff >>= flip go sr)+          Right (Await g Refl gg) -> return $+            Await (\u -> case u of+                           Left a -> +                             MachineT $+                             wait sl >>= forceFeed (flip go sr) a . encased+                           Right b -> MachineT $ asyncRun (g b) >>= go sl)+                  Refl+                  (MachineT $ asyncRun gg >>= go sl)++-- | @forceFeed k x p@ runs machine @p@ until it awaits, at which+-- point it is fed @x@. The result of that feeding is asynchronously+-- run, and supplied to the continuation @k@.+forceFeed :: forall m a k b. MonadBaseControl IO m+          => (AsyncStep m (Is a) b -> m (MachineStep m k b))+          -> a+          -> ProcessT m a b+          -> m (MachineStep m k b)+forceFeed go x = aux+  where aux p = runMachineT p >>= \v -> case v of+          -- Stop -> asyncRun stopped >>= go+          Stop -> return Stop+          Yield o k -> return . Yield o . MachineT $ aux k+          Await f Refl _ -> asyncRun (f x) >>= go++-- | We have a sink for the Right output of a source, so we want to+-- keep running it as long as upstream does not yield a 'Left' which+-- we can not handle. When upstream yields a 'Left', we 'stop'.+rightOnly :: Monad m => ProcessT m b r -> ProcessT m (Either a b) r+rightOnly snk = repeatedly (await >>= either (const stop) yield) ~> snk++-- | We have a sink for the Left output of a source, so we want to+-- keep running it as long as upstream does not yield a 'Right' which+-- we can not handle. When upstream yields a 'Right', we 'stop'.+leftOnly :: Monad m => ProcessT m a r -> ProcessT m (Either a b) r+leftOnly snk = repeatedly (await >>= either yield (const stop)) ~> snk++-- | Connect two processes to the downstream tails of a 'Machine' that+-- produces tuples. The two downstream consumers are run+-- concurrently. When one downstream consumer stops, the entire+-- pipeline is stopped.+--+-- @splitProd sink1 sink2@ produces a topology like this,+--+-- @+--                            sink1+--                           /      \+--                         a          \+--                        /            \+--    source -- (a,b) -->               -- r -->+--                        \\            /+--                         b         /+--                           \\     /+--                            sink2 +-- @+splitProd :: forall m a b r. MonadBaseControl IO m+          => ProcessT m a r -> ProcessT m b r -> ProcessT m (a,b) r+splitProd snk1 snk2 = MachineT $ do s1 <- asyncRun snk1+                                    s2 <- asyncRun snk2+                                    go s1 s2+  where go :: AsyncStep m (Is a) r+           -> AsyncStep m (Is b) r+           -> m (MachineStep m (Is (a,b)) r)+        go s1 s2 = waitBoth s1 s2 >>= +                   \(ss :: (MachineStep m (Is a) r, MachineStep m (Is b) r)) -> case ss of+          (Stop, _) -> return Stop+          (_, Stop) -> return Stop+          (Yield o1 k1, Yield o2 k2) -> +            return . Yield o1 . encased $ Yield o2 $ MachineT $+            do k1' <- asyncRun k1+               k2' <- asyncRun k2+               go k1' k2'+          (Yield o k, _) ->+            return . Yield o . MachineT $ asyncRun k >>= flip go s2+          (_, Yield o k) ->+            return . Yield o . MachineT $ asyncRun k >>= go s1+          (Await f Refl ff, Await g Refl gg) ->+            return $ Await (uncurry splitProd . (f***g)) Refl (splitProd ff gg)
+ src/Data/Machine/Concurrent/Tee.hs view
@@ -0,0 +1,36 @@+{-# LANGUAGE FlexibleContexts, GADTs, ScopedTypeVariables #-}+-- | Support for machines with two inputs from which input may be+-- drawn deterministically. In contrast to "Data.Machine.Tee", the two+-- inputs are eagerly run concurrently in this implementation.+module Data.Machine.Concurrent.Tee where+import Control.Concurrent.Async.Lifted (wait)+import Control.Monad.Trans.Control (MonadBaseControl)+import Data.Machine+import Data.Machine.Concurrent.AsyncStep++-- | Compose a pair of pipes onto the front of a Tee.+tee :: forall m a a' b b' c. MonadBaseControl IO m+    => ProcessT m a a' -> ProcessT m b b' -> TeeT m a' b' c -> TeeT m a b c+tee ma mb m = MachineT $ do srcL <- asyncRun ma+                            srcR <- asyncRun mb+                            go m (Just srcL) (Just srcR)+  where go :: MonadBaseControl IO m+           => TeeT m a' b' c+           -> Maybe (AsyncStep m (Is a) a')+           -> Maybe (AsyncStep m (Is b) b')+           -> m (MachineStep m (T a b) c)+        go snk srcL srcR = runMachineT snk >>= \v -> case v of+          Stop -> return Stop+          Yield o k -> return . Yield o . MachineT $ go k srcL srcR+          Await f L ff -> maybe (return Stop) wait srcL >>= +                          \(u :: MachineStep m (Is a) a') -> case u of+            Stop            -> go ff Nothing srcR+            Yield a k       -> asyncRun k >>= flip (go (f a)) srcR . Just+            Await g Refl fg -> +              asyncAwait g L fg $ MachineT . flip (go (encased v)) srcR . Just+          Await f R ff -> maybe (return Stop) wait srcR >>= +                          \(u :: MachineStep m (Is b) b') -> case u of+            Stop            -> go ff srcL Nothing+            Yield b k       -> asyncRun k >>= go (f b) srcL . Just+            Await g Refl fg -> +              asyncAwait g R fg $ MachineT . go (encased v) srcL . Just
+ src/Data/Machine/Concurrent/Wye.hs view
@@ -0,0 +1,111 @@+{-# LANGUAGE GADTs, FlexibleContexts, RankNTypes, ScopedTypeVariables,+             TupleSections #-}+-- | Support for machines with two inputs from which input may be+-- drawn deterministically or non-deterministically. In contrast to+-- "Data.Machine.Wye", the two inputs are eagerly run concurrently in+-- this implementation.+module Data.Machine.Concurrent.Wye (wye) where+import Control.Applicative+import Control.Concurrent.Async.Lifted (wait, waitEither)+import Control.Monad.Trans.Control (MonadBaseControl, StM)+import Data.Machine hiding (wye, (~>), (<~))+import Data.Machine.Concurrent.AsyncStep++isX :: Is a c -> Y a b c+isX Refl = X++isY :: Is b c -> Y a b c+isY Refl = Y++-- | Only the 'X' input of a 'Wye' is not yet stopped, so we may employ+-- simpler dispatch logic.+wyeOnlyX :: forall a a' b b' c m. MonadBaseControl IO m+         => AsyncStep m (Is a) a' -> WyeT m a' b' c -> WyeT m a b c+wyeOnlyX src snk = MachineT $ runMachineT snk >>= \v -> case v of+  Stop -> return Stop+  Yield o k -> return $ Yield o (wyeOnlyX src k)+  Await _ Y ff -> runMachineT $ wye stopped stopped ff+  Await f X ff -> runMachineT $ stepAsync isX src f ff (encased v) wyeOnlyX+  Await f Z ff -> runMachineT $ +    stepAsync isX src (f . Left) ff (encased v) wyeOnlyX++-- | Only the 'Y' input of a 'Wye' is not yet stopped, so we may+-- employ simpler dispatch logic.+wyeOnlyY :: MonadBaseControl IO m+         => AsyncStep m (Is b) b' -> WyeT m a' b' c -> WyeT m a b c+wyeOnlyY src m = MachineT $ runMachineT m >>= \v -> case v of+  Stop -> return Stop+  Yield o k -> return $ Yield o (wyeOnlyY src k)+  Await _ X ff -> runMachineT $ wye stopped stopped ff+  Await f Y ff -> runMachineT $ stepAsync isY src f ff (encased v) wyeOnlyY+  Await f Z ff -> +    runMachineT $ stepAsync isY src (f . Right) ff (encased v) wyeOnlyY++-- | Precompose a 'Process' onto each input of a 'Wye' (or 'WyeT').+--+-- When the choice of input is free (using the 'Z' input descriptor)+-- the two sources will be interleaved.+wye :: forall m a a' b b' c.+       (MonadBaseControl IO m)+    => ProcessT m a a' -> ProcessT m b b' -> WyeT m a' b' c -> WyeT m a b c+wye ma mb m = MachineT $ do srcL <- asyncRun ma+                            srcR <- asyncRun mb+                            go True m srcL srcR+  where go :: Bool+           -> WyeT m a' b' c+           -> AsyncStep m (Is a) a'+           -> AsyncStep m (Is b) b'+           -> m (MachineStep m (Y a b) c)+        go fair snk srcL srcR = runMachineT snk >>= \v -> case v of+          Stop         -> return Stop+          Yield o k    -> return . Yield o . MachineT $ go fair k srcL srcR+          Await f X ff -> wait srcL >>=+                          \(u :: MachineStep m (Is a) a') -> case u of+            Stop -> runMachineT $ wyeOnlyY srcR ff+            Yield a k -> asyncRun k >>= flip (go fair (f a)) srcR+            Await g Refl fg -> +              asyncAwait g X fg $ MachineT . flip (go fair (encased v)) srcR+          Await f Y ff -> wait srcR >>=+                          \(u :: MachineStep m (Is b) b') -> case u of+            Stop -> runMachineT $ wyeOnlyX srcL ff+            Yield b k -> asyncRun k >>= go fair (f b) srcL+            Await h Refl fh -> +              asyncAwait h Y fh $ MachineT . go fair (encased v) srcL++          -- Wait for whoever yields first+          Await f Z _  -> +            waitFair fair srcL srcR+            >>= \(u :: Either (MachineStep m (Is a) a')+                              (MachineStep m (Is b) b')) -> case u of+            Left (Yield a k) -> +              asyncRun k >>= \srcL' -> go (not fair) (f $ Left a) srcL' srcR+            Right (Yield b k) -> +              asyncRun k >>= \srcR' -> go (not fair) (f $ Right b) srcL srcR'+            Left Stop -> runMachineT $ wyeOnlyY srcR (encased v)+            Right Stop -> runMachineT $ wyeOnlyX srcL (encased v)++            -- The first source to respond wants to await, see what+            -- the other source has to offer.+            Left la@(Await g Refl fg) -> +              wait srcR >>= \(w :: MachineStep m (Is b) b') -> case w of+                Stop -> asyncAwait g X fg $ \l' -> wyeOnlyX l' (encased v)+                Yield b k -> runMachineT $ wye (encased la) k (f $ Right b)+                ra@(Await h Refl fh) -> return $+                  Await (\c -> case c of+                                 Left a -> wye (g a) (encased ra) (encased v)+                                 Right b -> wye (encased la) (h b) (encased v))+                        Z+                        (wye fg fh $ encased v)+            Right ra@(Await h Refl fh) -> +              wait srcL >>= \(w :: MachineStep m (Is a) a') -> case w of+                Stop -> asyncAwait h Y fh $ \r' -> wyeOnlyY r' (encased v)+                Yield a k -> runMachineT $ wye k (encased ra) (f $ Left a)+                la@(Await g Refl fg) -> return $+                  Await (\c -> case c of+                                 Left a -> wye (g a) (encased ra) (encased v)+                                 Right b -> wye (encased la) (h b) (encased v))+                        Z+                        (wye fg fh $ encased v)+          where waitFair True l r = waitEither l r+                waitFair False l r = either Right Left <$> waitEither r l+
+ src/Data/Machine/Fanout.hs view
@@ -0,0 +1,72 @@+{-# LANGUAGE GADTs #-}+-- | Provide a notion of fanout wherein a single input is passed to+-- several consumers. The consumers are run sequentially.+module Data.Machine.Fanout (fanout, fanoutSteps) where+import Control.Applicative+import Control.Arrow+import Control.Monad (foldM)+import Data.Machine+import Data.Maybe (catMaybes)+import Data.Monoid+import Data.Semigroup (Semigroup(sconcat))+import Data.List.NonEmpty (NonEmpty((:|)))++-- | Feed a value to a 'ProcessT' at an 'Await' 'Step'. If the+-- 'ProcessT' is awaiting a value, then its next step is+-- returned. Otherwise, the original process is returned.+feed :: Monad m => a -> ProcessT m a b -> m (Step (Is a) b (ProcessT m a b))+feed x m = runMachineT m >>= \v ->+            case v of+              Await f Refl _ -> runMachineT (f x)+              s -> return s++-- | Like 'Data.List.mapAccumL' but with a monadic accumulating+-- function.+mapAccumLM :: (Functor m, Monad m)+           => (acc -> x -> m (acc, y)) -> acc -> [x] -> m (acc, [y])+mapAccumLM f z = fmap (second ($ [])) . foldM aux (z,id)+  where aux (acc,ys) x = second ((. ys) . (:)) <$> f acc x++-- | Exhaust a sequence of all successive 'Yield' steps taken by a+-- 'MachineT'. Returns the list of yielded values and the next+-- (non-Yield) step of the machine.+flushYields :: Monad m+            => Step k o (MachineT m k o) -> m ([o], Maybe (MachineT m k o))+flushYields = go id+  where go rs (Yield o s) = runMachineT s >>= go ((o:) . rs)+        go rs Stop = return (rs [], Nothing)+        go rs s = return (rs [], Just $ encased s)++-- | Share inputs with each of a list of processes in lockstep. Any+-- values yielded by the processes for a given input are combined into+-- a single yield from the composite process.+fanout :: (Functor m, Monad m, Semigroup r)+       => [ProcessT m a r] -> ProcessT m a r+fanout xs = encased $ Await (MachineT . aux) Refl (fanout xs)+  where aux y = do (rs,xs') <- mapM (feed y) xs >>= mapAccumLM yields []+                   let nxt = fanout $ catMaybes xs'+                   case rs of+                     [] -> runMachineT nxt+                     (r:rs') -> return $ Yield (sconcat $ r :| rs') nxt+        yields rs Stop = return (rs,Nothing)+        yields rs y@(Yield _ _) = first (++ rs) <$> flushYields y+        yields rs a@(Await _ _ _) = return (rs, Just $ encased a)++-- | Share inputs with each of a list of processes in lockstep. If+-- none of the processes yields a value, the composite process will+-- itself yield 'mempty'. The idea is to provide a handle on steps+-- only executed for their side effects. For instance, if you want to+-- run a collection of 'ProcessT's that await but don't yield some+-- number of times, you can use 'fanOutSteps . map (fmap (const ()))'+-- followed by a 'taking' process.+fanoutSteps :: (Functor m, Monad m, Monoid r)+            => [ProcessT m a r] -> ProcessT m a r+fanoutSteps xs = encased $ Await (MachineT . aux) Refl (fanoutSteps xs)+  where aux y = do (rs,xs') <- mapM (feed y) xs >>= mapAccumLM yields []+                   let nxt = fanoutSteps $ catMaybes xs'+                   if null rs+                   then return $ Yield mempty nxt+                   else return $ Yield (mconcat rs) nxt+        yields rs Stop = return (rs,Nothing)+        yields rs y@(Yield _ _) = first (++rs) <$> flushYields y+        yields rs a@(Await _ _ _) = return (rs, Just $ encased a)
+ src/Data/Machine/Regulated.hs view
@@ -0,0 +1,22 @@+-- | Slow producers down to run at desired rates.+module Data.Machine.Regulated where+import Control.Concurrent (threadDelay)+import Control.Monad (when)+import Control.Monad.IO.Class (MonadIO(..))+import Data.Machine.Plan+import Data.Machine.Process+import Data.Machine.Type+import Data.Time.Clock (getCurrentTime, diffUTCTime)++-- | A pass-through process rate-limited to the given inter-step+-- period in seconds. This may be used to slow down an upstream+-- producer; it can not speed things up.+regulated :: MonadIO m => Double -> ProcessT m a a+regulated target = construct $ liftIO getCurrentTime >>= go 0+  where go dt prevT =+          do await >>= yield+             t <- liftIO getCurrentTime+             let e = target - realToFrac (diffUTCTime t prevT)+                 dt' = dt + 0.5 * e+             when (dt' > 0) (liftIO . threadDelay . round $ dt' * 1000000)+             go dt' t
+ tests/AllTests.hs view
@@ -0,0 +1,34 @@+import Data.Time.Clock (getCurrentTime, diffUTCTime)+import Control.Concurrent (threadDelay)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Data.Machine.Concurrent+import Test.Tasty+import Test.Tasty.HUnit++worker :: String -> Double -> ProcessT IO () ()+worker _name dt = repeatedly $ do _ <- await+                                  liftIO $ threadDelay dt'+                                  yield ()+  where dt' = floor $ dt * 10000++timed :: MonadIO m => m a -> m (a, Double)+timed m = do t1 <- liftIO getCurrentTime+             r <- m+             t2 <- liftIO getCurrentTime+             return (r, realToFrac $ t2 `diffUTCTime` t1)++pipeline :: TestTree+pipeline = testCaseSteps "pipeline" $ \step -> do+  (r,dt) <- timed . runT . supply (repeat ()) $+            worker "A" 1 ~> worker "B" 1 ~> worker "C" 1 ~> taking 10+  (r',dt') <- timed . runT . supply (repeat ()) $+              worker "A" 1 >~> worker "B" 1 >~> worker "C" 1 >~> taking 10+  step "Consistent results"+  assertEqual "Results" r r'+  step "Parallelism"+  assertBool ("Pipeline faster than sequential" ++ show (dt',dt)) (dt' * 2 < dt)++main :: IO ()+main = defaultMain $ +       testGroup "concurrent-machines"+       [ pipeline ]