diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,9 @@
+## 0.0.7.0
+
+Add `withEffToIO'`, `race`, `connectCoroutines`, `receiveStream`,
+`insertManySecond`, `useImplWithin`, `unsafeProvideIO`, and
+`Bluefin.Internal.Pipes`
+
 ## 0.0.6.1
 
 * Documentation improvements
diff --git a/bluefin-internal.cabal b/bluefin-internal.cabal
--- a/bluefin-internal.cabal
+++ b/bluefin-internal.cabal
@@ -1,6 +1,6 @@
 cabal-version:      3.0
 name:               bluefin-internal
-version:            0.0.6.1
+version:            0.0.7.0
 license:            MIT
 license-file:       LICENSE
 author:             Tom Ellis
@@ -77,6 +77,7 @@
     default-language: Haskell2010
     hs-source-dirs: src
     build-depends:
+      async,
       base >= 4.12 && < 4.21,
       unliftio-core < 0.3,
       transformers < 0.7,
@@ -85,7 +86,8 @@
     ghc-options: -Wall
     exposed-modules:
       Bluefin.Internal,
-      Bluefin.Internal.Examples
+      Bluefin.Internal.Examples,
+      Bluefin.Internal.Pipes
 
 test-suite bluefin-test
     import:           defaults
diff --git a/src/Bluefin/Internal.hs b/src/Bluefin/Internal.hs
--- a/src/Bluefin/Internal.hs
+++ b/src/Bluefin/Internal.hs
@@ -9,6 +9,8 @@
 
 module Bluefin.Internal where
 
+import qualified Control.Concurrent.Async as Async
+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)
 import Control.Exception (throwIO, tryJust)
 import qualified Control.Exception
 import Control.Monad.Base (MonadBase (liftBase))
@@ -64,6 +66,14 @@
   Eff es a
 withEffToIO k io = effIO io (k (\f -> unsafeUnEff (f MkIOE)))
 
+withEffToIO' ::
+  (e2 :> es) =>
+  -- | Continuation with the unlifting function in scope.
+  IOE e2 ->
+  ((forall r. (forall e1. IOE e1 -> Eff (e1 :& es) r) -> IO r) -> IO a) ->
+  Eff es a
+withEffToIO' io k = withEffToIO k io
+
 -- We don't try to do anything sophisticated here.  I haven't thought
 -- through all the consequences.
 instance (e :> es) => MonadUnliftIO (EffReader (IOE e) es) where
@@ -86,6 +96,53 @@
             )
       )
 
+race ::
+  (e2 :> es) =>
+  (forall e. IOE e -> Eff (e :& es) a) ->
+  (forall e. IOE e -> Eff (e :& es) a) ->
+  IOE e2 ->
+  Eff es a
+race x y io = do
+  r <- withEffToIO' io $ \toIO ->
+    Async.race (toIO x) (toIO y)
+
+  pure $ case r of
+    Left a -> a
+    Right a -> a
+
+-- | Connect two coroutines.  Their execution is interleaved by
+-- exchanging @a@s and @b@s. When the first yields its first @a@ it
+-- starts the second (which is waiting to receive an @a@).
+connectCoroutines ::
+  forall es a b r.
+  (forall e. Coroutine a b e -> Eff (e :& es) r) ->
+  (forall e. a -> Coroutine b a e -> Eff (e :& es) r) ->
+  -- | ͘
+  Eff es r
+connectCoroutines m1 m2 = unsafeProvideIO $ \io -> do
+  av <- effIO io newEmptyMVar
+  bv <- effIO io newEmptyMVar
+
+  let t1 :: forall e. IOE e -> Eff (e :& es) r
+      t1 io' = forEach (useImplWithin m1) $ \a -> effIO io' $ do
+        putMVar av a
+        takeMVar bv
+
+  let t2 :: forall e. IOE e -> Eff (e :& es) r
+      t2 io' = do
+        ainit <- effIO io' (takeMVar av)
+        forEach (useImplWithin (m2 ainit)) $ \b_ -> effIO io' $ do
+          putMVar bv b_
+          takeMVar av
+
+  race (useImplWithin t1) (useImplWithin t2) io
+
+receiveStream ::
+  (forall e. Coroutine () a e -> Eff (e :& es) r) ->
+  (forall e. Stream a e -> Eff (e :& es) r) ->
+  Eff es r
+receiveStream r s = connectCoroutines r (\() -> s)
+
 instance (e :> es) => MonadBase IO (EffReader (IOE e) es) where
   liftBase = liftIO
 
@@ -159,6 +216,9 @@
 insertSecond :: Eff (c1 :& b) r -> Eff (c1 :& (c2 :& b)) r
 insertSecond = weakenEff (b (drop (eq (# #))))
 
+insertManySecond :: (b :> c) => Eff (c1 :& b) r -> Eff (c1 :& c) r
+insertManySecond = weakenEff (bimap has has)
+
 assoc1Eff :: Eff ((a :& b) :& c) r -> Eff (a :& (b :& c)) r
 assoc1Eff = weakenEff (assoc1 (# #))
 
@@ -184,6 +244,15 @@
   Eff es r
 useImplIn f h = inContext (f h)
 
+-- | Like 'useImplIn'
+useImplWithin ::
+  (e :> es) =>
+  (t -> Eff (e1 :& e) r) ->
+  t ->
+  -- | ͘
+  Eff (e1 :& es) r
+useImplWithin k fsh = insertManySecond (k fsh)
+
 -- | Handle to a capability to create strict mutable state handles
 data StateSource (e :: Effects) = StateSource
 
@@ -193,13 +262,13 @@
 -- | A handle to a strict mutable state of type @s@
 newtype State s (e :: Effects) = UnsafeMkState (IORef s)
 
--- | A handle to a coroutine that expects values of type @a@ and then
--- yields values of type @b@.
+-- | A handle to a coroutine that yields values of type @a@ and then
+-- expects values of type @b@.
 newtype Coroutine a b (e :: Effects) = MkCoroutine (a -> Eff e b)
 
 -- | A handle to a stream that yields values of type @a@.  It is
--- implemented as a handle to a coroutine that expects values of type
--- @()@ and then yields values of type @a@.
+-- implemented as a handle to a coroutine that yields values of type
+-- @a@ and then expects values of type @()@.
 type Stream a = Coroutine a ()
 
 -- | You can define a @Handle@ instance for your compound handles.  As
@@ -955,6 +1024,12 @@
   -- | ͘
   IO a
 runEff eff = unsafeUnEff (eff MkIOE)
+
+unsafeProvideIO ::
+  (forall e. IOE e -> Eff (e :& es) a) ->
+  -- | ͘
+  Eff es a
+unsafeProvideIO eff = unsafeRemoveEff (eff MkIOE)
 
 connect ::
   (forall e1. Coroutine a b e1 -> Eff (e1 :& es) r1) ->
diff --git a/src/Bluefin/Internal/Examples.hs b/src/Bluefin/Internal/Examples.hs
--- a/src/Bluefin/Internal/Examples.hs
+++ b/src/Bluefin/Internal/Examples.hs
@@ -4,6 +4,15 @@
 module Bluefin.Internal.Examples where
 
 import Bluefin.Internal hiding (w)
+import Bluefin.Internal.Pipes
+  ( Producer,
+    runEffect,
+    stdinLn,
+    stdoutLn,
+    takeWhile',
+    (>->),
+  )
+import qualified Bluefin.Internal.Pipes as P
 import Control.Exception (IOException)
 import qualified Control.Exception
 import Control.Monad (forever, unless, when)
@@ -646,3 +655,13 @@
   runPureEff $ do
     (_res, st) <- runState (0, False) $ \st -> try $ \e -> polymorphicBracket st (throw e 42)
     pure st
+
+pipesExample1 :: IO ()
+pipesExample1 = runEff $ \io -> runEffect (count >-> P.print io)
+  where
+    count :: (e :> es) => Producer Int e -> Eff es ()
+    count p = for_ [1 .. 5] $ \i -> P.yield p i
+
+pipesExample2 :: IO String
+pipesExample2 = runEff $ \io -> runEffect $ do
+  stdinLn io >-> takeWhile' (/= "quit") >-> stdoutLn io
diff --git a/src/Bluefin/Internal/Pipes.hs b/src/Bluefin/Internal/Pipes.hs
new file mode 100644
--- /dev/null
+++ b/src/Bluefin/Internal/Pipes.hs
@@ -0,0 +1,267 @@
+module Bluefin.Internal.Pipes where
+
+import Bluefin.Internal hiding (yield)
+import qualified Bluefin.Internal
+import Control.Monad (forever)
+import Data.Foldable (for_)
+import Data.Void (Void, absurd)
+import Prelude hiding (break, print, takeWhile)
+import qualified Prelude
+
+data Proxy a' a b' b e = MkProxy (Coroutine a' a e) (Coroutine b b' e)
+
+type Pipe a = Proxy () a ()
+
+type Producer = Proxy Void () ()
+
+type Consumer a = Pipe a Void
+
+type Effect = Producer Void
+
+infixl 7 >->
+
+(>->) ::
+  (e1 :> es) =>
+  (forall e. Proxy a' a () b e -> Eff (e :& es) r) ->
+  (forall e. Proxy () b c' c e -> Eff (e :& es) r) ->
+  Proxy a' a c' c e1 ->
+  -- | ͘
+  Eff es r
+(>->) k1 k2 (MkProxy c1 c2) =
+  receiveStream
+    (\c -> useImplIn k2 (MkProxy (mapHandle c) (mapHandle c2)))
+    (\s -> useImplIn k1 (MkProxy (mapHandle c1) (mapHandle s)))
+
+infixr 7 <-<
+
+(<-<) ::
+  (e1 :> es) =>
+  (forall e. Proxy () b c' c e -> Eff (e :& es) r) ->
+  (forall e. Proxy a' a () b e -> Eff (e :& es) r) ->
+  Proxy a' a c' c e1 ->
+  -- | ͘
+  Eff es r
+k1 <-< k2 = k2 >-> k1
+
+for ::
+  (e1 :> es) =>
+  (forall e. Proxy x' x b' b e -> Eff (e :& es) a') ->
+  (b -> forall e. Proxy x' x c' c e -> Eff (e :& es) b') ->
+  Proxy x' x c' c e1 ->
+  -- | ͘
+  Eff es a'
+for k1 k2 (MkProxy c1 c2) =
+  forEach (\bk -> useImplIn k1 (MkProxy (mapHandle c1) (mapHandle bk))) $ \b_ ->
+    useImplIn (k2 b_) (MkProxy (mapHandle c1) (mapHandle c2))
+
+infixr 4 ~>
+
+(~>) ::
+  (e1 :> es) =>
+  (a -> forall e. Proxy x' x b' b e -> Eff (e :& es) a') ->
+  (b -> forall e. Proxy x' x c' c e -> Eff (e :& es) b') ->
+  a ->
+  Proxy x' x c' c e1 ->
+  -- | ͘
+  Eff es a'
+(k1 ~> k2) a = for (k1 a) k2
+
+infixl 4 <~
+
+(<~) ::
+  (e1 :> es) =>
+  (b -> forall e. Proxy x' x c' c e -> Eff (e :& es) b') ->
+  (a -> forall e. Proxy x' x b' b e -> Eff (e :& es) a') ->
+  a ->
+  Proxy x' x c' c e1 ->
+  -- | ͘
+  Eff es a'
+k2 <~ k1 = k1 ~> k2
+
+reverseProxy :: Proxy a' a b' b e -> Proxy b b' a a' e
+reverseProxy (MkProxy c1 c2) = MkProxy c2 c1
+
+infixl 5 >~
+
+(>~) ::
+  (e1 :> es) =>
+  (forall e. Proxy a' a y' y e -> Eff (e :& es) b) ->
+  (forall e. Proxy () b y' y e -> Eff (e :& es) c) ->
+  Proxy a' a y' y e1 ->
+  -- | ͘
+  Eff es c
+(>~) k1 k2 p =
+  for
+    ( \p1 ->
+        k2 (reverseProxy p1)
+    )
+    (\() p1 -> k1 (reverseProxy p1))
+    (reverseProxy p)
+
+infixr 5 ~<
+
+(~<) ::
+  (e1 :> es) =>
+  (forall e. Proxy () b y' y e -> Eff (e :& es) c) ->
+  (forall e. Proxy a' a y' y e -> Eff (e :& es) b) ->
+  Proxy a' a y' y e1 ->
+  -- | ͘
+  Eff es c
+(~<) k1 k2 = (>~) k2 k1
+
+cat :: Pipe a a e -> Eff (e :& es) r
+cat (MkProxy c1 c2) = forever $ do
+  a <- yieldCoroutine c1 ()
+  yieldCoroutine c2 a
+
+runEffect ::
+  (forall e. Effect e -> Eff (e :& es) r) ->
+  -- | ͘
+  Eff es r
+runEffect k =
+  forEach
+    ( \c1 ->
+        forEach
+          ( \c2 ->
+              useImplIn
+                k
+                (MkProxy (mapHandle c1) (mapHandle c2))
+          )
+          absurd
+    )
+    absurd
+
+yield ::
+  (e :> es) =>
+  Proxy x1 x () a e ->
+  a ->
+  -- | ͘
+  Eff es ()
+yield (MkProxy _ c) = Bluefin.Internal.yield c
+
+await :: (e :> es) => Proxy () a y' y e -> Eff es a
+await (MkProxy c _) = yieldCoroutine c ()
+
+-- | @pipe@'s 'next' doesn't exist in Bluefin
+next :: ()
+next = ()
+
+each ::
+  (Foldable f) =>
+  f a ->
+  Proxy x' x () a e ->
+  -- | ͘
+  Eff (e :& es) ()
+each f p = for_ f (yield p)
+
+repeatM ::
+  (e :> es) =>
+  Eff es a ->
+  Proxy x' x () a e ->
+  -- | ͘
+  Eff es r
+repeatM e p = forever $ do
+  a <- e
+  yield p a
+
+replicateM ::
+  (e :> es) =>
+  Int ->
+  Eff es a ->
+  Proxy x' x () a e ->
+  -- | ͘
+  Eff es ()
+replicateM n e p = for_ [0 .. n] $ \_ -> do
+  a <- e
+  yield p a
+
+print ::
+  (e2 :> es, e1 :> es, Show a) =>
+  IOE e1 ->
+  Consumer a e2 ->
+  -- | ͘
+  Eff es r
+print io p = forever $ do
+  a <- await p
+  effIO io (Prelude.print a)
+
+unfoldr ::
+  (e :> es) =>
+  (s -> Eff es (Either r (a, s))) ->
+  s ->
+  Proxy x1 x () a e ->
+  -- | ͘
+  Eff es r
+unfoldr next_ sInit p =
+  withEarlyReturn $ \break -> evalState sInit $ \ss -> forever $ do
+    s <- get ss
+    useImpl (next_ s) >>= \case
+      Left r -> returnEarly break r
+      Right (a, s') -> do
+        put ss s'
+        yield p a
+
+mapM_ ::
+  (e :> es) =>
+  (a -> Eff es ()) ->
+  Proxy () a b b' e ->
+  -- | ͘
+  Eff es r
+mapM_ f = for cat (\a _ -> useImpl (f a))
+
+drain ::
+  (e :> es) =>
+  Proxy () b c' c e ->
+  -- | ͘
+  Eff es r
+drain = for cat (\_ _ -> pure ())
+
+map ::
+  (e :> es) =>
+  (a -> b) ->
+  Pipe a b e ->
+  -- | ͘
+  Eff es r
+map f = for cat (\a p1 -> yield p1 (f a))
+
+mapM ::
+  (e :> es) =>
+  (a -> Eff es b) ->
+  Pipe a b e ->
+  -- | ͘
+  Eff es r
+mapM f = for cat $ \a p -> do
+  b_ <- useImpl (f a)
+  yield p b_
+
+takeWhile' ::
+  (e :> es) =>
+  (r -> Bool) ->
+  Pipe r r e ->
+  -- | ͘
+  Eff es r
+takeWhile' predicate p = withEarlyReturn $ \early -> forever $ do
+  a <- await p
+  if predicate a
+    then yield p a
+    else returnEarly early a
+
+stdinLn ::
+  (e1 :> es, e2 :> es) =>
+  IOE e1 ->
+  Producer String e2 ->
+  -- | ͘
+  Eff es r
+stdinLn io c = forever $ do
+  line <- effIO io getLine
+  yield c line
+
+stdoutLn ::
+  (e1 :> es, e2 :> es) =>
+  IOE e1 ->
+  Consumer String e2 ->
+  -- | ͘
+  Eff es r
+stdoutLn io c = forever $ do
+  line <- await c
+  effIO io (putStrLn line)