threaded (empty) → 0.0.0

raw patch · 8 files changed

+378/−0 lines, 8 filesdep +asyncdep +basedep +chansetup-changed

Dependencies added: async, base, chan, extractable-singleton, hashable, monad-control-aligned, mtl, stm, threaded, tmapmvar

Files

ChangeLog.md +3/−0
LICENSE +30/−0
README.md +47/−0
Setup.hs +2/−0
src/Control/Concurrent/Threaded.hs +97/−0
src/Control/Concurrent/Threaded/Hash.hs +91/−0
test/Spec.hs +39/−0
threaded.cabal +69/−0

+ ChangeLog.md view

@@ -0,0 +1,3 @@+# Changelog for threaded++## Unreleased changes

+ LICENSE view

@@ -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.

+ README.md view

@@ -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.

+ Setup.hs view

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

+ src/Control/Concurrent/Threaded.hs view

@@ -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)

+ src/Control/Concurrent/Threaded/Hash.hs view

@@ -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)

+ test/Spec.hs view

@@ -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

+ threaded.cabal view

@@ -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