diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,10 @@
 # Revision history for Churros
 
+## 0.1.4.1 -- 2020-10-20
+
+* Adding `processes` function that runs a set of processes over the input.
+* Adding `withChurro` helper function.
+
 ## 0.1.4.0 -- 2020-10-18
 
 * Generalising list type to Foldable/Traversible where possible (`sources`).
diff --git a/churros.cabal b/churros.cabal
--- a/churros.cabal
+++ b/churros.cabal
@@ -1,6 +1,6 @@
 cabal-version:       2.2
 name:                churros
-version:             0.1.4.0
+version:             0.1.4.1
 license-file:        LICENSE
 author:              Lyndon Maydwell
 maintainer:          lyndon@sordina.net
diff --git a/src/Control/Churro/Prelude.hs b/src/Control/Churro/Prelude.hs
--- a/src/Control/Churro/Prelude.hs
+++ b/src/Control/Churro/Prelude.hs
@@ -21,14 +21,15 @@
 import           Control.Arrow            (arr)
 import           Control.Category         (id, (.), (>>>))
 import           Control.Concurrent       (threadDelay)
-import           Control.Concurrent.Async (cancel, Async, wait)
+import           Control.Concurrent.Async (async, cancel, Async, wait)
 import           Control.Exception        (Exception, SomeException, try)
 import           Control.Monad            (replicateM_, when)
-import           Data.Foldable            (toList, for_)
+import           Data.Foldable            (foldMap', for_)
 import           Data.Maybe               (isJust)
 import           Data.Time                (NominalDiffTime)
 import           Data.Void                (Void)
 import           GHC.Natural              (Natural)
+import Data.Traversable (for)
 
 
 -- $setup
@@ -38,6 +39,7 @@
 -- >>> :set -XBlockArguments
 -- >>> import Control.Churro.Transport
 -- >>> import Data.Time.Clock
+-- >>> import System.Timeout (timeout)
 -- 
 
 -- * Runners
@@ -134,19 +136,35 @@
 -- 
 -- Sends individual items downstream without attempting to combine them.
 -- 
--- Warning: Passing an empty list is unspecified.
+-- >>> runWaitChan $ sources [pure 1, pure 1] >>> sinkPrint
+-- 1
+-- 1
 -- 
--- TODO: Use NonEmptyList instead of []
+-- Can combine results of sources with a Monoid instance, although this isn't very useful
+-- when forming the start of a longer pipeline:
 -- 
--- >>> runWaitChan $ sources_ [pure 1, pure 1] >>> sinkPrint
+-- >>> :{
+-- do
+--   r <- runWaitChan $ sources [sourceIO \_cb -> print 1 >> return "hello ", sourceIO \_cb -> print 1 >> return "world"]
+--   print r
+-- :}
 -- 1
 -- 1
-sources :: (Transport t, Foldable f, Traversable f) => f (Churro a t Void i) -> Churro () t Void i
-sources ss = sourceIO \cb -> do
-    asyncs <- mapM (\s -> run $ s >>>> sinkIO cb) ss
-    let as = toList asyncs
-    finally' (mapM_ cancel as) do
-        mapM_ wait as
+-- "hello world"
+sources :: (Transport t, Traversable f, Monoid a) => f (Churro a t Void o) -> Churro a t Void o
+sources ss = buildChurro \_i o -> do
+    cs <- mapM runChurro ss
+    finally' (mapM_ (\(_,_,a) -> cancel a) cs) do
+        as <- for cs \(_i,o',a) ->
+            async do
+                yankAll' o' \v -> do
+                    case v of
+                        Nothing -> return ()
+                        Just x  -> yeet o (Just x)
+                wait a
+        r <- foldMap' wait as
+        yeet o Nothing
+        return r
 
 -- | Variant of `sources` with Async action of sources in argument specialised to `()`.
 -- 
@@ -247,6 +265,75 @@
     buildChurro \i o -> do
         yankAll i \x -> do mapM_ (yeet o . Just) =<< f x
         yeet o Nothing
+
+-- | Concatenates splits lists of items into individual items.
+-- 
+concatC :: Transport t => Churro () t [o] o
+concatC = processN (pure . id)
+
+-- | Run a set of churros like a work-stealing queue for its inputs.
+-- 
+-- Similar to ArrowChoice, but more straightforward due to unified output type and independent implementation.
+-- 
+-- Note: `processes` makes no judgement about the ordering of outputs corresponding to the ordering of inputs.
+-- 
+-- TODO: Figure out cancellation strategy.
+-- TODO: Consider a binary combinator and this as a folded application.
+-- 
+-- WARNING: This won't deterministically allocate work to idle workers unless a bounded channel is used.
+-- 
+-- Sanity check - All items entering should propagate, independent of the number of processes:
+-- 
+-- >>> runWaitListChan $ sourceList [1,1,1,1,1] >>> processes (replicate 3 (delay 0.1))
+-- [1,1,1,1,1]
+-- 
+-- This example creates a source of 10 values, then creates a process of 10 workers that all wait 1/2 a second.
+-- If this works, then all ten values should be consumed and propagated in 1/2 a second by distributing the load
+-- over the set of 10 workers:
+-- 
+-- >>> :{
+-- do
+--   timeout 10000000 $ runWaitListChan $ sourceList (replicate 10 1) >>> processes (replicate 1 $ delay 0.05)
+-- :}
+-- Just [1,1,1,1,1,1,1,1,1,1]
+-- 
+-- We could use different strategies such as round-robin, etc. to default to a more balanced allocation, but this wouldn't
+-- be most efficient if each worker performed at different rates of consumption.
+-- 
+processes :: (Traversable f, Transport t, Monoid a) => f (Churro a t i o) -> Churro a t i o
+processes cs = Churro do
+    (i,  o ) <- flex
+    (i', o') <- flex
+
+    let
+        worker c = async do
+            withChurro c \ci co ca -> do
+                a' <- async do
+                    c2c' co i'
+                    yeet i Nothing -- Ensure other consumers aren't blocked. FIXME: This produces once more Nothing than is required.
+                c2c id o ci
+                wait a'
+                wait ca
+
+    as <- mapM worker cs
+
+    a  <- async do
+        r <- foldMap' wait as
+        yeet i' Nothing -- Make sure to conclude the process once all the processes have finished consuming
+        return r
+
+    return (i,o',a)
+
+    where
+    -- Version of c2c that doesn't propagate Nothing once transport is consumed.
+    -- This is required here since we don't want a worker to be able to prematurely terminate the processes
+    -- While an earlier slower works still hasn't finished propagating its result.
+    c2c' o i = yankAll o (yeet i . Just)
+
+-- | Set up N worker churro processes to concurrently process the stream.
+-- 
+thief :: (Transport t, Monoid a) => Int -> Churro a t i o -> Churro a t i o
+thief n c = processes (replicate n c)
 
 -- | Extract xs from (Just x)s. Similar to `catMaybes`.
 -- 
diff --git a/src/Control/Churro/Types.hs b/src/Control/Churro/Types.hs
--- a/src/Control/Churro/Types.hs
+++ b/src/Control/Churro/Types.hs
@@ -67,8 +67,8 @@
 class Transport (t :: * -> *) where
     data In  t :: * -> *
     data Out t :: * -> *
-    flex :: IO (In t a, Out t a)  -- ^ Create a new pair of transports.
-    yank :: Out t a -> IO a       -- ^ Yank an item of the Transport
+    flex :: IO (In t a, Out t a)  -- ^ Create a new pair of Transports.
+    yank :: Out t a -> IO a       -- ^ Yank an item off the Transport
     yeet :: In t a -> a -> IO ()  -- ^ Yeet an item onto the Transport
 
 -- | Covariant functor instance for Churro - Maps over the output.
@@ -243,6 +243,15 @@
     a       <- async do cb ai ao bi
     return (ai,bo,a)
 
+-- | Helper. Finalises cancellation of async.
+-- 
+-- Use instead of runChurro unless you want to directly manage cancellation.
+-- 
+withChurro :: Churro a t i o -> (In t (Maybe i) -> Out t (Maybe o) -> Async a -> IO b) -> IO b
+withChurro c f = do
+    (i,o,a) <- runChurro c
+    finally' (cancel a) do f i o a
+
 -- | Yeet all items from a list into a raw transport.
 -- 
 -- WARNING: If you are using this to build a churro by hand make sure you yeet Nothing once you're finished.
@@ -268,7 +277,7 @@
 
 -- | Yank each raw item from a transport into a callback.
 -- 
--- The items are wrapped in Maybes and when all items are yanked, Nothing is fed to the callback.
+-- The items are wrapped in Maybes and when Nothing is yanked, Nothing is fed to the callback and `yankAll'` completes.
 -- 
 yankAll' :: (Transport t, Monoid b) => Out t (Maybe a) -> (Maybe a -> IO b) -> IO b
 yankAll' c f = do