diff --git a/ChangeLog.md b/ChangeLog.md
new file mode 100644
--- /dev/null
+++ b/ChangeLog.md
@@ -0,0 +1,3 @@
+# Changelog for threaded
+
+## Unreleased changes
diff --git a/LICENSE b/LICENSE
new file mode 100644
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,30 @@
+Copyright Athan Clark (c) 2020
+
+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 Athan Clark 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/README.md b/README.md
new file mode 100644
--- /dev/null
+++ b/README.md
@@ -0,0 +1,47 @@
+# threaded
+
+Aims to make managed, horizontal scaling easier - given some process that reads from a concurrent channel,
+writes to a concurrent channel, and returns when it's finished, then it should be horizontally scalable
+with respect to some thread identifier:
+
+```haskell
+main :: IO ()
+main = do
+  let mult inputs outputs = do
+        -- get first input
+        x <- atomically (readTChanRW inputs)
+        -- get second input
+        y <- atomically (readTChanRW inputs)
+        let o :: Integer
+            o = (x :: Integer) * (y :: Integer)
+
+        -- write output
+        atomically (writeTChanRW outputs o)
+        -- return
+
+  -- incoming messages for specific threads
+  incoming <- writeOnly <$> atomically newTChanRW
+
+  (mainThread, outgoing) <- threaded incoming mult
+
+  echoingThread <- async $ forever $ do
+    -- do something with each thread's output
+    (k,o) <- atomically (readTChanRW outgoing)
+    putStrLn $ show k ++ ": " ++ show o
+
+  atomically $ writeTChanRW incoming ("one",1)
+  atomically $ writeTChanRW incoming ("two",2)
+  atomically $ writeTChanRW incoming ("three",3)
+  atomically $ writeTChanRW incoming ("one",1)
+  atomically $ writeTChanRW incoming ("two",2)
+  atomically $ writeTChanRW incoming ("three",3)
+
+  threadDelay 1000000
+
+  cancel echoingThread
+  cancel mainThread
+```
+
+If the thread's identifier doesn't exist when sending an input, then the `threaded` manager will spark a new
+one. If it does exist, then it just plumbs it to its input channel. Once the process returns, the thread with
+that identifier is killed and garbage collected.
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/src/Control/Concurrent/Threaded.hs b/src/Control/Concurrent/Threaded.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Concurrent/Threaded.hs
@@ -0,0 +1,97 @@
+{-# LANGUAGE
+    DataKinds
+  , RankNTypes
+  , NamedFieldPuns
+  , FlexibleContexts
+  , ScopedTypeVariables
+  #-}
+
+module Control.Concurrent.Threaded where
+
+import Control.Concurrent.Async (Async, async, cancel)
+import Control.Concurrent.Chan.Scope (Scope (Read, Write))
+import Control.Concurrent.Chan.Extra (ChanScoped (readOnly, allowReading, writeOnly, allowWriting))
+import Control.Concurrent.STM (atomically)
+import Control.Concurrent.STM.TChan.Typed (TChanRW, newTChanRW, readTChanRW, writeTChanRW)
+import Control.Concurrent.STM.TMapMVar (TMapMVar, newTMapMVar, tryObserve, insertForce, tryLookup)
+import Control.Monad (forever)
+import Control.Monad.IO.Class (MonadIO (liftIO))
+import Control.Monad.Trans.Control.Aligned (MonadBaseControl (liftBaseWith))
+import Data.Singleton.Class (Extractable (runSingleton))
+
+
+
+data ThreadedInternal incoming outgoing = ThreadedInternal
+  { thread :: Async () -- ^ The process's execution thread
+  , outputRelay :: Async () -- ^ The thread that relays the process's output to the main output
+  , threadInput :: TChanRW 'Read incoming -- ^ The process's isolated input
+  , threadOutput :: TChanRW 'Write outgoing -- ^ The process's isolated output
+  }
+
+
+-- | Segregates concurrently operating threads by some key type @k@. Returns the
+-- thread that processes all other threads (this function is non-blocking), and the
+-- channel that dispenses the outputs from each thread.
+threaded :: forall m stM k input output
+          . Ord k
+         => Show k
+         => MonadIO m
+         => MonadBaseControl IO m stM
+         => Extractable stM
+         => -- | Incoming messages, identified by thread @k@
+            TChanRW 'Write (k, input)
+         -> -- | Process to spark in a new thread. When @m ()@ returns, the thread is considered \"dead\",
+            -- and is internally cleaned up.
+            (TChanRW 'Read input -> TChanRW 'Write output -> m ())
+         -> m (Async (), TChanRW 'Read (k, output))
+threaded incoming process = do
+  ( threads :: TMapMVar k (ThreadedInternal incoming outgoing)
+    ) <- liftIO (atomically newTMapMVar)
+  outgoing <- liftIO (atomically (readOnly <$> newTChanRW))
+
+  -- the main function that organizes the execution and plumbing of the threads
+  threadRunner <- liftBaseWith $ \runInBase -> fmap (fmap runSingleton) $ async $ runInBase $ forever $ do
+    (k, input) <- liftIO (atomically (readTChanRW (allowReading incoming)))
+
+    mThread <- liftIO (atomically (tryObserve threads k))
+    case mThread of
+      Nothing -> do
+        -- thread-specific channels
+        threadInput' <- liftIO $ atomically $ do
+          i <- newTChanRW
+          -- initial input
+          writeTChanRW i input
+          pure i
+        let threadInput = readOnly threadInput'
+        threadOutput' <- liftIO (atomically newTChanRW)
+        let threadOutput = writeOnly threadOutput'
+
+
+        -- relays the process's output to the whole output - can't be done in lock-step, must let one finish
+        -- before writing
+        outputRelay <- liftIO $ async $ forever $ do
+          output <- atomically (readTChanRW threadOutput')
+          atomically (writeTChanRW (allowWriting outgoing) (k, output))
+
+        -- main thread
+        thread <- liftBaseWith $ \runInBase' -> fmap (fmap runSingleton) $ async $ runInBase' $ do
+          process threadInput threadOutput
+          -- thread finished processing
+          mThread' <- liftIO (atomically (tryLookup threads k))
+          case mThread' of
+            Nothing -> error ("Thread's facilities don't exist: " ++ show k)
+            Just ThreadedInternal{thread = thread'} -> liftIO $ do
+              -- kill the thread's supervisors
+              cancel outputRelay
+              cancel thread'
+
+        -- store threads and channels
+        liftIO $ atomically $
+          insertForce threads k ThreadedInternal{thread,outputRelay,threadInput,threadOutput}
+
+
+      -- thread is still processing - relay input to its channel
+      Just ThreadedInternal{threadInput} ->
+        liftIO (atomically (writeTChanRW (allowWriting threadInput) input))
+
+  pure (threadRunner, outgoing)
diff --git a/src/Control/Concurrent/Threaded/Hash.hs b/src/Control/Concurrent/Threaded/Hash.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Concurrent/Threaded/Hash.hs
@@ -0,0 +1,91 @@
+{-# LANGUAGE
+    DataKinds
+  , RankNTypes
+  , NamedFieldPuns
+  , FlexibleContexts
+  , ScopedTypeVariables
+  #-}
+
+module Control.Concurrent.Threaded.Hash where
+
+import Control.Concurrent.Threaded (ThreadedInternal (..))
+import Control.Concurrent.Async (Async, async, cancel)
+import Control.Concurrent.Chan.Scope (Scope (Read, Write))
+import Control.Concurrent.Chan.Extra (ChanScoped (readOnly, allowReading, writeOnly, allowWriting))
+import Control.Concurrent.STM (atomically)
+import Control.Concurrent.STM.TChan.Typed (TChanRW, newTChanRW, readTChanRW, writeTChanRW)
+import Control.Concurrent.STM.TMapMVar.Hash (TMapMVar, newTMapMVar, tryObserve, insertForce, tryLookup)
+import Control.Monad (forever)
+import Control.Monad.IO.Class (MonadIO (liftIO))
+import Control.Monad.Trans.Control.Aligned (MonadBaseControl (liftBaseWith))
+import Data.Singleton.Class (Extractable (runSingleton))
+import Data.Hashable (Hashable)
+
+
+-- | Segregates concurrently operating threads by some key type @k@. Returns the
+-- thread that processes all other threads (this function is non-blocking), and the
+-- channel that dispenses the outputs from each thread.
+threaded :: forall m stM k input output
+          . Hashable k
+         => Eq k
+         => Show k
+         => MonadIO m
+         => MonadBaseControl IO m stM
+         => Extractable stM
+         => -- | Incoming messages, identified by thread @k@
+            TChanRW 'Write (k, input)
+         -> -- | Process to spark in a new thread. When @m ()@ returns, the thread is considered \"dead\",
+            -- and is internally cleaned up.
+            (TChanRW 'Read input -> TChanRW 'Write output -> m ())
+         -> m (Async (), TChanRW 'Read (k, output))
+threaded incoming process = do
+  ( threads :: TMapMVar k (ThreadedInternal incoming outgoing)
+    ) <- liftIO (atomically newTMapMVar)
+  outgoing <- liftIO (atomically (readOnly <$> newTChanRW))
+
+  -- the main function that organizes the execution and plumbing of the threads
+  threadRunner <- liftBaseWith $ \runInBase -> fmap (fmap runSingleton) $ async $ runInBase $ forever $ do
+    (k, input) <- liftIO (atomically (readTChanRW (allowReading incoming)))
+
+    mThread <- liftIO (atomically (tryObserve threads k))
+    case mThread of
+      Nothing -> do
+        -- thread-specific channels
+        threadInput' <- liftIO $ atomically $ do
+          i <- newTChanRW
+          -- initial input
+          writeTChanRW i input
+          pure i
+        let threadInput = readOnly threadInput'
+        threadOutput' <- liftIO (atomically newTChanRW)
+        let threadOutput = writeOnly threadOutput'
+
+
+        -- relays the process's output to the whole output - can't be done in lock-step, must let one finish
+        -- before writing
+        outputRelay <- liftIO $ async $ forever $ do
+          output <- atomically (readTChanRW threadOutput')
+          atomically (writeTChanRW (allowWriting outgoing) (k, output))
+
+        -- main thread
+        thread <- liftBaseWith $ \runInBase' -> fmap (fmap runSingleton) $ async $ runInBase' $ do
+          process threadInput threadOutput
+          -- thread finished processing
+          mThread' <- liftIO (atomically (tryLookup threads k))
+          case mThread' of
+            Nothing -> error ("Thread's facilities don't exist: " ++ show k)
+            Just ThreadedInternal{thread = thread'} -> liftIO $ do
+              -- kill the thread's supervisors
+              cancel outputRelay
+              cancel thread'
+
+        -- store threads and channels
+        liftIO $ atomically $
+          insertForce threads k ThreadedInternal{thread,outputRelay,threadInput,threadOutput}
+
+
+      -- thread is still processing - relay input to its channel
+      Just ThreadedInternal{threadInput} ->
+        liftIO (atomically (writeTChanRW (allowWriting threadInput) input))
+
+  pure (threadRunner, outgoing)
diff --git a/test/Spec.hs b/test/Spec.hs
new file mode 100644
--- /dev/null
+++ b/test/Spec.hs
@@ -0,0 +1,39 @@
+import Control.Concurrent.Threaded (threaded)
+import Control.Concurrent.Async (async, cancel)
+import Control.Concurrent.Chan.Extra (writeOnly)
+import Control.Concurrent.STM (atomically)
+import Control.Concurrent.STM.TChan.Typed (newTChanRW, readTChanRW, writeTChanRW)
+import Control.Monad (forever)
+import Control.Concurrent (threadDelay)
+
+
+main :: IO ()
+main = do
+  let mult inputs outputs = do
+        x <- atomically (readTChanRW inputs)
+        putStrLn $ "Got x: " ++ show x
+        y <- atomically (readTChanRW inputs)
+        putStrLn $ "Got y: " ++ show y
+        let o :: Integer
+            o = (x :: Integer) * (y :: Integer)
+        atomically (writeTChanRW outputs o)
+        putStrLn $ "Sent o: " ++ show o
+  incoming <- writeOnly <$> atomically newTChanRW
+
+  (mainThread, outgoing) <- threaded incoming mult
+
+  echoingThread <- async $ forever $ do
+    (k,o) <- atomically (readTChanRW outgoing)
+    putStrLn $ show k ++ ": " ++ show o
+
+  atomically $ writeTChanRW incoming (1,1)
+  atomically $ writeTChanRW incoming (2,2)
+  atomically $ writeTChanRW incoming (3,3)
+  atomically $ writeTChanRW incoming (1,1)
+  atomically $ writeTChanRW incoming (2,2)
+  atomically $ writeTChanRW incoming (3,3)
+
+  threadDelay 1000000
+
+  cancel echoingThread
+  cancel mainThread
diff --git a/threaded.cabal b/threaded.cabal
new file mode 100644
--- /dev/null
+++ b/threaded.cabal
@@ -0,0 +1,69 @@
+cabal-version: 1.12
+
+-- This file has been generated from package.yaml by hpack version 0.31.2.
+--
+-- see: https://github.com/sol/hpack
+--
+-- hash: 563aa3ae542c712e727ae4fedfd4c05457e0e45535ae5b2c7952e318f45d7d21
+
+name:           threaded
+version:        0.0.0
+synopsis:       Manage concurrently operating threads without having to spark them
+description:    Please see the README on GitHub at <https://github.com/athanclark/threaded#readme>
+category:       Concurrent
+homepage:       https://github.com/athanclark/threaded#readme
+bug-reports:    https://github.com/athanclark/threaded/issues
+author:         Athan Clark
+maintainer:     athan.clark@gmail.com
+copyright:      2020 Athan Clark
+license:        BSD3
+license-file:   LICENSE
+build-type:     Simple
+extra-source-files:
+    README.md
+    ChangeLog.md
+
+source-repository head
+  type: git
+  location: https://github.com/athanclark/threaded
+
+library
+  exposed-modules:
+      Control.Concurrent.Threaded
+      Control.Concurrent.Threaded.Hash
+  other-modules:
+      Paths_threaded
+  hs-source-dirs:
+      src
+  build-depends:
+      async
+    , base >=4.7 && <5
+    , chan
+    , extractable-singleton
+    , hashable
+    , monad-control-aligned
+    , mtl
+    , stm
+    , tmapmvar
+  default-language: Haskell2010
+
+test-suite threaded-test
+  type: exitcode-stdio-1.0
+  main-is: Spec.hs
+  other-modules:
+      Paths_threaded
+  hs-source-dirs:
+      test
+  ghc-options: -threaded -rtsopts -with-rtsopts=-N
+  build-depends:
+      async
+    , base >=4.7 && <5
+    , chan
+    , extractable-singleton
+    , hashable
+    , monad-control-aligned
+    , mtl
+    , stm
+    , threaded
+    , tmapmvar
+  default-language: Haskell2010