diff --git a/LICENSE b/LICENSE
new file mode 100644
--- /dev/null
+++ b/LICENSE
@@ -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.
diff --git a/Setup.hs b/Setup.hs
new file mode 100644
--- /dev/null
+++ b/Setup.hs
@@ -0,0 +1,2 @@
+import Distribution.Simple
+main = defaultMain
diff --git a/concurrent-machines.cabal b/concurrent-machines.cabal
new file mode 100644
--- /dev/null
+++ b/concurrent-machines.cabal
@@ -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
diff --git a/src/Data/Machine/Concurrent.hs b/src/Data/Machine/Concurrent.hs
new file mode 100644
--- /dev/null
+++ b/src/Data/Machine/Concurrent.hs
@@ -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)
diff --git a/src/Data/Machine/Concurrent/AsyncStep.hs b/src/Data/Machine/Concurrent/AsyncStep.hs
new file mode 100644
--- /dev/null
+++ b/src/Data/Machine/Concurrent/AsyncStep.hs
@@ -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
diff --git a/src/Data/Machine/Concurrent/Buffer.hs b/src/Data/Machine/Concurrent/Buffer.hs
new file mode 100644
--- /dev/null
+++ b/src/Data/Machine/Concurrent/Buffer.hs
@@ -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)
diff --git a/src/Data/Machine/Concurrent/Fanout.hs b/src/Data/Machine/Concurrent/Fanout.hs
new file mode 100644
--- /dev/null
+++ b/src/Data/Machine/Concurrent/Fanout.hs
@@ -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)
diff --git a/src/Data/Machine/Concurrent/Scatter.hs b/src/Data/Machine/Concurrent/Scatter.hs
new file mode 100644
--- /dev/null
+++ b/src/Data/Machine/Concurrent/Scatter.hs
@@ -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)
diff --git a/src/Data/Machine/Concurrent/Tee.hs b/src/Data/Machine/Concurrent/Tee.hs
new file mode 100644
--- /dev/null
+++ b/src/Data/Machine/Concurrent/Tee.hs
@@ -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
diff --git a/src/Data/Machine/Concurrent/Wye.hs b/src/Data/Machine/Concurrent/Wye.hs
new file mode 100644
--- /dev/null
+++ b/src/Data/Machine/Concurrent/Wye.hs
@@ -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
+
diff --git a/src/Data/Machine/Fanout.hs b/src/Data/Machine/Fanout.hs
new file mode 100644
--- /dev/null
+++ b/src/Data/Machine/Fanout.hs
@@ -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)
diff --git a/src/Data/Machine/Regulated.hs b/src/Data/Machine/Regulated.hs
new file mode 100644
--- /dev/null
+++ b/src/Data/Machine/Regulated.hs
@@ -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
diff --git a/tests/AllTests.hs b/tests/AllTests.hs
new file mode 100644
--- /dev/null
+++ b/tests/AllTests.hs
@@ -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 ]