diff --git a/Control/Broccoli.hs b/Control/Broccoli.hs
--- a/Control/Broccoli.hs
+++ b/Control/Broccoli.hs
@@ -5,21 +5,22 @@
 module Control.Broccoli (
   X,
   E,
-  Output(..),
-  runProgram,
-  edge,
-  accumulate,
+  never,
   snapshot,
   snapshot_,
-  out,
-  filterE,
+  accumulate,
+  edge,
   justE,
   maybeE,
-  never,
-  debugX,
-  debugE,
+  filterE,
+  Setup,
+  runProgram,
   newX,
-  newE
+  newE,
+  input,
+  output,
+  debugX,
+  debugE
 ) where
 
 import Control.Applicative
@@ -38,7 +39,7 @@
   PureX :: a -> X a
   FmapX :: forall a b . (b -> a) -> X b -> X a
   ApplX :: forall a b . X (b -> a) -> X b -> X a
-  PortX :: TVar a -> X a
+  PortX :: MVar [ThreadId] -> TVar a -> X a
 
 -- | An event that carries values of type a when it occurs.
 data E a where
@@ -48,10 +49,7 @@
   ProductE  :: (b -> c -> a) -> E b -> E c -> E a
   SnapshotE :: E b -> X a -> E a
   JustE     :: E (Maybe a) -> E a
-  PortE     :: TChan a -> E a
-
--- | A special IO action which will react to events. See 'out'.
-newtype Output = Output (IO ())
+  PortE     :: MVar [ThreadId] -> TChan a -> E a
 
 instance Functor X where
   fmap f x = FmapX f x
@@ -67,6 +65,47 @@
   mempty = NeverE
   mappend e1 e2 = MappendE e1 e2
 
+-- | A monad for hooking up inputs and outputs to a program.
+data Setup a = Setup (MVar [ThreadId] -> IO a)
+
+instance Monad Setup where
+  return x = Setup (\_ -> return x)
+  (Setup r) >>= f = Setup r' where
+    r' mv = do
+      x <- r mv
+      let Setup r'' = f x
+      r'' mv
+
+instance Applicative Setup where
+  pure = return
+  (<*>) = ap
+
+instance Functor Setup where
+  fmap f (Setup io) = Setup (\mv -> f <$> io mv)
+
+setupIO :: IO a -> Setup a
+setupIO io = Setup (\_ -> io)
+
+getThreads :: Setup (MVar [ThreadId])
+getThreads = Setup (\mv -> return mv)
+
+getThreadsE :: E a -> Maybe (MVar [ThreadId])
+getThreadsE e = case e of
+  NeverE -> Nothing
+  FmapE _ e' -> getThreadsE e'
+  MappendE e1 e2 -> getFirst $ First (getThreadsE e1) <> First (getThreadsE e2)
+  ProductE _ e1 e2 -> getFirst $ First (getThreadsE e1) <> First (getThreadsE e2)
+  SnapshotE e' x -> getFirst $ First (getThreadsE e') <> First (getThreadsX x)
+  JustE e' -> getThreadsE e'
+  PortE mv _ -> Just mv
+
+getThreadsX :: X a -> Maybe (MVar [ThreadId])
+getThreadsX x = case x of
+  PureX _ -> Nothing
+  FmapX _ x' -> getThreadsX x'
+  ApplX x1 x2 -> getFirst $ First (getThreadsX x1) <> First (getThreadsX x2)
+  PortX mv _ -> Just mv
+
 dupE :: E a -> IO (E a)
 dupE e = case e of
   NeverE -> return NeverE
@@ -77,9 +116,6 @@
     e1' <- dupE e1
     e2' <- dupE e2
     return (MappendE e1' e2')
-  PortE ch -> do
-    ch' <- atomically (dupTChan ch)
-    return (PortE ch')
   ProductE f e1 e2 -> do
     e1' <- dupE e1
     e2' <- dupE e2
@@ -90,11 +126,13 @@
   JustE e' -> do
     e'' <- dupE e'
     return (JustE e'')
+  PortE mv ch -> do
+    ch' <- atomically (dupTChan ch)
+    return (PortE mv ch')
 
 readE :: E a -> IO a
 readE e = case e of
   NeverE -> hang
-  PortE ch -> atomically (readTChan ch)
   MappendE e1 e2 -> race (readE e1) (readE e2)
   FmapE f e' -> f <$> readE e'
   ProductE f e1 e2 -> do
@@ -109,6 +147,7 @@
     case m of
       Nothing -> loop
       Just x  -> return x
+  PortE _ ch -> atomically (readTChan ch)
 
 readX :: X a -> STM a
 readX x = case x of
@@ -118,7 +157,7 @@
     f <- readX ff
     x <- readX xx
     return (f x)
-  PortX tv -> readTVar tv
+  PortX _ tv -> readTVar tv
 
 hang :: IO a
 hang = do
@@ -130,31 +169,8 @@
   e <- dupE e0
   readE e
 
-runEvent :: E a -> (a -> IO ()) -> IO ()
-runEvent e0 act = do
-  e <- dupE e0
-  forever (readE e >>= act)
-
 ---
 
--- | Create a new signal with an initial value. Get back an IO action to
--- change the value of the signal. This is the only way the signal will
--- change.
-newX :: a -> IO (a -> IO (), X a)
-newX v0 = do
-  tv <- newTVarIO v0
-  return
-    ( \x -> atomically (writeTVar tv x)
-    , PortX tv )
-
--- | Create a new event and the associated IO action to trigger the event.
-newE :: IO (a -> IO (), E a)
-newE = do
-  ch <- newBroadcastTChanIO
-  return
-    ( \x -> atomically (writeTChan ch x)
-    , PortE ch )
-
 -- | An event which gets the value of a signal when another event occurs.
 snapshot :: E a -> X b -> E (a,b)
 snapshot e x = ProductE (,) e (SnapshotE e x)
@@ -179,70 +195,114 @@
 never :: E a
 never = mempty
 
--- | Events will occur when an "edge" is detected in a signal. The detection
--- algorithm checks the two values before and after a discrete change in
--- the signal.
+-- | Sum over events using an initial state and a state transition function.
+accumulate :: E a -> s -> (a -> s -> s) -> X s
+accumulate e0 s0 trans = case getThreadsE e0 of
+  Nothing -> pure s0
+  Just mv -> PortX mv tv where
+    tv = unsafePerformIO $ do
+      state <- newTVarIO s0
+      threadId <- forkIO $ do
+        putStrLn "accum forked"
+        e <- dupE e0
+        forever $ do
+          x <- readE e
+          atomically $ do
+            s <- readTVar state
+            let s' = trans x s
+            writeTVar state s'
+      modifyMVar_ mv (return . (threadId:))
+      return state
+
+-- | An event that occurs when an edge is detected in a signal. The edge test
+-- is applied to values before and after a discrete transition in the signal.
+-- The test should return Nothing when the two values are the same.
 edge :: X a -> (a -> a -> Maybe b) -> E b
-edge x diff = PortE ch where
-  ch = unsafePerformIO $ do
-    out <- newBroadcastTChanIO
-    forkIO $ do
-      v0 <- atomically (readX x)
-      ref <- newIORef v0
-      forever $ do
-        v <- readIORef ref
-        (d, v') <- atomically $ do
-          v' <- readX x
-          case diff v v' of
-            Just d  -> return (d, v')
-            Nothing -> retry
-        writeIORef ref v'
-        atomically (writeTChan out d)
-    return out
+edge x diff = case getThreadsX x of
+  Nothing -> never
+  Just mv -> PortE mv ch where
+    ch = unsafePerformIO $ do
+      out <- newBroadcastTChanIO
+      threadId <- forkIO $ do
+        putStrLn "edge forked"
+        v0 <- atomically (readX x)
+        ref <- newIORef v0
+        forever $ do
+          v <- readIORef ref
+          (d, v') <- atomically $ do
+            v' <- readX x
+            case diff v v' of
+              Just d  -> return (d, v')
+              Nothing -> retry
+          writeIORef ref v'
+          atomically (writeTChan out d)
+      modifyMVar_ mv (return . (threadId:))
+      return out
 
--- | Create a signal out of an input event and a state machine.
-accumulate :: E a -> s -> (a -> s -> s) -> X s
-accumulate e0 s0 trans = PortX tv where
-  tv = unsafePerformIO $ do
-    state <- newTVarIO s0
-    forkIO $ do
-      e <- dupE e0
-      forever $ do
-        x <- readE e
-        atomically $ do
-          s <- readTVar state
-          let s' = trans x s
-          writeTVar state s'
-    return state
 
--- | A handler for events.
-out :: E a -> (a -> IO ()) -> Output
-out e0 act = Output (runEvent e0 act)
+-- | Creates a new event and an IO action to trigger it.
+newE :: Setup (E a, a -> IO ())
+newE = do
+  mv <- getThreads
+  bch <- setupIO newBroadcastTChanIO
+  return (PortE mv bch, atomically . writeTChan bch)
 
--- | Prints the values of a signal as they change.
-debugX :: (Eq a, Show a) => X a -> Output
-debugX x = Output $ do
-  v0 <- atomically (readX x)
-  print v0
-  let diff a b = if a == b then Nothing else Just b
-  runEvent (edge x diff) print
+-- | Creates a new signal and an IO action to update it. The argument is
+-- the initial value of the signal.
+newX :: a -> Setup (X a, a -> IO ())
+newX v = do
+  mv <- getThreads
+  tv <- setupIO (newTVarIO v)
+  return (PortX mv tv, atomically . writeTVar tv)
 
--- | Prints the values of events as they occur.
-debugE :: (Show a) => E a -> Output
-debugE e = out e print
+-- | Spawn a thread to execute an action for each event occurrence.
+output :: E a -> (a -> IO ()) -> Setup ()
+output e0 act = do
+  mv <- getThreads
+  setupIO $ do
+    e <- dupE e0
+    tid <- (forkIO . forever) (readE e >>= act)
+    modifyMVar_ mv (return . (tid:))
+    return ()
 
--- | Run a set of Outputs. This spawns several threads then waits for an
--- event. The output threads will then be killed to stop further processing.
--- However other threads which no longer have any effect will probably remain,
--- taking up resources. After threads have been spawned but before waiting,
--- the given "boot" action will be executed.
-runProgram :: IO () -- ^ action to execute after initialization
-           -> E ()  -- ^ event that will shutdown the system
-           -> [Output] -- ^ set of output event handlers to run
-           -> IO ()
-runProgram notifyBoot stop outs = do
-  tids <- forM outs (\(Output io) -> forkIO io)
-  threadDelay 5000
-  notifyBoot
-  waitE stop
-  forM_ tids killThread
+-- | Spawn an input thread to generate source signals and events.
+input :: IO () -> Setup ()
+input handler = do
+  mv <- getThreads
+  setupIO $ do
+    tid <- forkIO handler
+    modifyMVar_ mv (return . (tid:))
+    return ()
+
+-- | Run the setup action to create input and output threads. The returned IO
+-- action will be executed when setup is complete. runProgram blocks until
+-- the returned event occurs, at which time it kills all the threads and
+-- returns.
+runProgram :: Setup (IO (), E ()) -> IO ()
+runProgram (Setup setup) = do
+  mv <- newMVar []
+  (boot, exit) <- setup mv
+  --threadDelay 5000
+  boot
+  waitE exit
+  withMVar mv (mapM killThread)
+  return ()
+
+-- | Print out events as they occur. Only for debugging purposes.
+debugE :: Show a => E a -> E a
+debugE e = unsafePerformIO $ do
+  e' <- dupE e
+  (forkIO . forever) (readE e' >>= print)
+  return e
+
+-- | Print out transitions in a signal. Only for debugging purposes.
+debugX :: (Eq a, Show a) => X a -> X a
+debugX x =
+  let diff a b = if a == b then Nothing else Just (a,b) in
+  let e = edge x diff in
+  unsafePerformIO $ do
+    forkIO $ do
+      putStrLn "edge forked"
+      e' <- dupE e
+      forever (readE e' >>= print)
+    return x
diff --git a/broccoli.cabal b/broccoli.cabal
--- a/broccoli.cabal
+++ b/broccoli.cabal
@@ -2,7 +2,7 @@
 -- documentation, see http://haskell.org/cabal/users-guide/
 
 name:                broccoli
-version:             0.1.0.0
+version:             0.2.0.0
 synopsis:            Small library for interactive functional programs.
 description:         Small library for interactive functional programs.
 license:             BSD3
@@ -32,4 +32,4 @@
 source-repository this
   type: git
   location: https://github.com/evanrinehart/broccoli
-  tag: 0.1.0.0
+  tag: 0.2.0.0