concurrent-machines 0.1.0.2 → 0.2.0

raw patch · 6 files changed

+122/−89 lines, 6 filesdep ~basedep ~machinesdep ~semigroupsPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: base, machines, semigroups, time

API changes (from Hackage documentation)

Files

• CHANGELOG.md +4/−0
• README.md +53/−0
• concurrent-machines.cabal +12/−5
• examples/ExampleFanout.hs +38/−0
• src/Data/Machine/Concurrent/Fanout.hs +15/−12
• src/Data/Machine/Fanout.hs +0/−72

@@ -0,0 +1,4 @@+# 0.2.0++- Fix fanout behavior (Ben SInclair)+- Make ExampleFanout a benchmark that cabal can run

@@ -0,0 +1,53 @@+A simple example of a pipelined computation whose throughput is improved by concurrently running distinct processing stages is given in [`examples/Pipeline.hs`](http://github.com/acowley/concurrent-machines/blob/master/examples/Pipeline.hs).++```haskell+import Data.Time.Clock (getCurrentTime, diffUTCTime)+import Control.Concurrent (threadDelay)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Data.Machine.Concurrent+```++Suppose we have a worker that performs a computation on its input before producing output. This operation may take some time due to, say, network IO or just CPU load. Here we simulate an operation that takes some time with `threadDelay`.++```haskell+worker :: String -> Double -> ProcessT IO () ()+worker name dt = repeatedly $ do _ <- await+                                 liftIO $ do+                                   putStrLn $ name ++ " working on its input"+                                   threadDelay dt'+                                 yield ()+  where dt' = floor $ dt * 1000000+```++We will use a little helper to time two variations of our test program.++```haskell+timed :: MonadIO m => m a -> m (a, Double)+timed m = do t1 <- liftIO getCurrentTime+             r <- m+             t2 <- liftIO getCurrentTime+             return (r, realToFrac $ t2 `diffUTCTime` t1)+```++Now we will run a three-stage pipeline where each stage takes one second to process its input before yielding some output. A sequential execution strategy will thus take three seconds to pass an input through this pipeline. At the top-level, we will request three outputs, each of which will take three seconds to produce, resulting in a total execution time of approximately nine seconds.++```haskell+main :: IO ()+main = do (r,dt) <- timed . runT . supply (repeat ()) $+            worker "A" 1 ~> worker "B" 1 ~> worker "C" 1 ~> taking 3+          putStrLn $ "Sequentially produced "++show r+          putStrLn $ "Sequential processing took "++show dt++"s"+```++If we instead run the same arrangement as a pipelined computation, we allow the independent stages to run concurrently, much like the stages in a pipelined CPU.++```haskell+          (r',dt') <- timed . runT . supply (repeat ()) $+            worker "A" 1 >~> worker "B" 1 >~> worker "C" 1 >~> taking 3+          putStrLn $ "Pipeline produced "++show r'+          putStrLn $ "Pipeline processing took "++show dt'++"s"+```++With this arrangement, the first output we request takes three seconds to produce as we must wait for an input to pass through the entire length of the pipeline. However, successive outputs are following that first output through the pipeline so that we will produce more output at one second intervals. Therefore it takes is `3 + 2 = 5` seconds to produce three outputs: three seconds for the first output, and one second for each of the following two.++[![Build Status](https://travis-ci.org/acowley/concurrent-machines.png)](https://travis-ci.org/acowley/concurrent-machines)

@@ -1,5 +1,5 @@ name:                concurrent-machines-version:             0.1.0.2+version:             0.2.0 synopsis:            Concurrent networked stream transducers  description: A simple use-case for this library is to run the stages@@ -29,7 +29,7 @@ copyright:           Copyright (C) 2014 Anthony Cowley category:            Concurrency, Control build-type:          Simple--- extra-source-files:  +extra-source-files:  README.md, CHANGELOG.md cabal-version:       >=1.10  source-repository head@@ -38,7 +38,6 @@  library   exposed-modules:     Data.Machine.Concurrent,-                       Data.Machine.Fanout,                        Data.Machine.Regulated,                        Data.Machine.Concurrent.AsyncStep,                        Data.Machine.Concurrent.Buffer,@@ -49,7 +48,7 @@   -- other-modules:          other-extensions:    GADTs, FlexibleContexts, RankNTypes, TupleSections,                         ScopedTypeVariables-  build-depends:       base >= 4.6 && < 5, +  build-depends:       base >= 4.6 && < 5,                        monad-control >= 1.0 && < 1.1,                        transformers >= 0.4 && < 0.5,                        time >= 1.4 && < 1.6,@@ -58,7 +57,7 @@                        machines >= 0.5 && < 0.6,                        async >= 2.0.1 && < 2.1,                        lifted-async >= 0.1 && < 0.8,-                       semigroups >= 0.8 && < 0.18+                       semigroups >= 0.8 && < 0.19   hs-source-dirs:      src   default-language:    Haskell2010 @@ -70,3 +69,11 @@   default-language: Haskell2010   build-depends: base >= 4.6 && < 5, concurrent-machines, machines,                  tasty, tasty-hunit, transformers, time++benchmark fanout+  type: exitcode-stdio-1.0+  hs-source-dirs: examples+  main-is: ExampleFanout.hs+  build-depends: base >= 4.8 && < 5, time, machines, concurrent-machines+  default-language: Haskell2010+  ghc-options: -threaded "-with-rtsopts=-N2"

@@ -0,0 +1,38 @@+-- | A demonstration of concurrent vs serialized fanout. A single+-- input is fed to two workers who both perform a CPU intensive+-- task. The concurrent version is approximately twice as fast when+-- two CPU cores are available.+import Control.Monad (when)+import Data.List (sort)+import Data.Machine.Concurrent+import qualified Data.Machine.Fanout as M+import Data.Time.Clock+import Text.Printf++-- A slow Fibonacci sequence can occupy the CPU+fib :: Int -> Int+fib 0 = 0+fib 1 = 1+fib n = fib (n-1) + fib (n-2)++-- Build a worker process with the given name.+worker :: String -> ProcessT IO Int [String]+worker name = repeatedly $ do+                seed <- await+                let x = fib seed+                x `seq` yield $ [name++": "++show seed++" => "++show x]++main :: IO ()+main = do t1 <- getCurrentTime+          [responses] <- runT $ supply [42] (fanout workers) ~> taking 1+          t2 <- getCurrentTime+          [responses'] <- runT $ supply [42] (M.fanout workers) ~> taking 1+          t3 <- getCurrentTime+          when (map sort responses /= map sort responses') $+            putStrLn "Concurrent and serial workers gave different results!"+          let dt = realToFrac (diffUTCTime t2 t1) :: Double+              dt' = realToFrac (diffUTCTime t3 t2) :: Double+          putStrLn $ "Concurrent vs. serial ("++show responses++"): "+          putStrLn $ "  " ++ printf "%0.1fs vs %0.1fs" dt dt'+  where workers = [ worker "alpha"+                  , worker "beta" ]

@@ -2,7 +2,7 @@ -- | Provide a notion of fanout wherein a single input is passed to -- several consumers. The consumers are run concurrently. module Data.Machine.Concurrent.Fanout (fanout, fanoutSteps) where-import Control.Arrow (second)+import Control.Arrow (first, second) import Control.Concurrent.Async.Lifted (Async, async, wait) import Control.Monad (foldM) import Control.Monad.Trans.Control (MonadBaseControl, StM)@@ -12,17 +12,18 @@ import Data.Monoid (Monoid, mempty, mconcat) import Data.Semigroup (Semigroup(sconcat)) import Data.List.NonEmpty (NonEmpty((:|)))+import Data.Coerce (coerce)  -- | Feed a value to a 'ProcessT' at an 'Await' 'Step'. If the -- 'ProcessT' is awaiting a value, then its next step is -- returned. Otherwise, the original process is returned. feed :: forall m a b. MonadBaseControl IO m      => a -> ProcessT m a b-     -> m (Async (StM m ([b], Maybe (MachineStep m (Is a) b))))+     -> m (Async (StM m (MachineStep m (Is a) b))) feed x m = async $ runMachineT m >>= \(v :: MachineStep m (Is a) b) ->              case v of-               Await f Refl _ -> runMachineT (f x) >>= flushYields-               s -> return ([]::[b], Just s)+               Await f Refl _ -> runMachineT (f x)+               s -> return s  -- | Like 'Data.List.mapAccumL' but with a monadic accumulating -- function.@@ -37,11 +38,11 @@ -- 'MachineT'. Returns the list of yielded values and the next -- (non-Yield) step of the machine. flushYields :: Monad m-            => Step k o (MachineT m k o) -> m ([o], Maybe (MachineStep m k o))+            => Step k o (MachineT m k o) -> m ([o], Maybe (MachineT m k o)) flushYields = go id   where go rs (Yield o s) = runMachineT s >>= go ((o:) . rs)         go rs Stop = return (rs [], Nothing)-        go rs s = return (rs [], Just s)+        go rs s = return (rs [], Just $ encased s)  -- | Share inputs with each of a list of processes in lockstep. Any -- values yielded by the processes for a given input are combined into@@ -50,12 +51,13 @@        => [ProcessT m a r] -> ProcessT m a r fanout xs = encased $ Await (MachineT . aux) Refl (fanout xs)   where aux y = do (rs,xs') <- mapM (feed y) xs >>= mapAccumLM yields []-                   let nxt = fanout . map encased $ catMaybes xs'+                   let nxt = fanout $ catMaybes xs'                    case rs of                      [] -> runMachineT nxt                      (r:rs') -> return $ Yield (sconcat $ r :| rs') nxt-        yields rs (rs', Nothing) = return (rs' ++ rs, Nothing)-        yields rs (rs', Just s) = return (rs' ++ rs, Just s)+        yields rs Stop = return (rs,Nothing)+        yields rs y@Yield{} = first (++ rs) <$> flushYields y+        yields rs a@Await{} = return (rs, Just $ encased a)  -- | Share inputs with each of a list of processes in lockstep. If -- none of the processes yields a value, the composite process will@@ -68,9 +70,10 @@             => [ProcessT m a r] -> ProcessT m a r fanoutSteps xs = encased $ Await (MachineT . aux) Refl (fanoutSteps xs)   where aux y = do (rs,xs') <- mapM (feed y) xs >>= mapAccumLM yields []-                   let nxt = fanoutSteps . map encased $ catMaybes xs'+                   let nxt = fanoutSteps $ catMaybes xs'                    if null rs                    then return $ Yield mempty nxt                    else return $ Yield (mconcat rs) nxt-        yields rs (rs', Nothing) = return (rs' ++ rs,Nothing)-        yields rs (rs', Just s) = return (rs'++rs, Just s)+        yields rs Stop = return (rs,Nothing)+        yields rs y@Yield{} = first (++rs) <$> flushYields y+        yields rs a@Await{} = return (rs, Just $ encased a)

@@ -1,72 +0,0 @@-{-# LANGUAGE GADTs #-}--- | Provide a notion of fanout wherein a single input is passed to--- several consumers. The consumers are run sequentially.-module Data.Machine.Fanout (fanout, fanoutSteps) where-import Control.Applicative-import Control.Arrow-import Control.Monad (foldM)-import Data.Machine-import Data.Maybe (catMaybes)-import Data.Monoid-import Data.Semigroup (Semigroup(sconcat))-import Data.List.NonEmpty (NonEmpty((:|)))---- | Feed a value to a 'ProcessT' at an 'Await' 'Step'. If the--- 'ProcessT' is awaiting a value, then its next step is--- returned. Otherwise, the original process is returned.-feed :: Monad m => a -> ProcessT m a b -> m (Step (Is a) b (ProcessT m a b))-feed x m = runMachineT m >>= \v ->-            case v of-              Await f Refl _ -> runMachineT (f x)-              s -> return s---- | Like 'Data.List.mapAccumL' but with a monadic accumulating--- function.-mapAccumLM :: (Functor m, Monad m)-           => (acc -> x -> m (acc, y)) -> acc -> [x] -> m (acc, [y])-mapAccumLM f z = fmap (second ($ [])) . foldM aux (z,id)-  where aux (acc,ys) x = second ((. ys) . (:)) <$> f acc x---- | Exhaust a sequence of all successive 'Yield' steps taken by a--- 'MachineT'. Returns the list of yielded values and the next--- (non-Yield) step of the machine.-flushYields :: Monad m-            => Step k o (MachineT m k o) -> m ([o], Maybe (MachineT m k o))-flushYields = go id-  where go rs (Yield o s) = runMachineT s >>= go ((o:) . rs)-        go rs Stop = return (rs [], Nothing)-        go rs s = return (rs [], Just $ encased s)---- | Share inputs with each of a list of processes in lockstep. Any--- values yielded by the processes for a given input are combined into--- a single yield from the composite process.-fanout :: (Functor m, Monad m, Semigroup r)-       => [ProcessT m a r] -> ProcessT m a r-fanout xs = encased $ Await (MachineT . aux) Refl (fanout xs)-  where aux y = do (rs,xs') <- mapM (feed y) xs >>= mapAccumLM yields []-                   let nxt = fanout $ catMaybes xs'-                   case rs of-                     [] -> runMachineT nxt-                     (r:rs') -> return $ Yield (sconcat $ r :| rs') nxt-        yields rs Stop = return (rs,Nothing)-        yields rs y@(Yield _ _) = first (++ rs) <$> flushYields y-        yields rs a@(Await _ _ _) = return (rs, Just $ encased a)---- | Share inputs with each of a list of processes in lockstep. If--- none of the processes yields a value, the composite process will--- itself yield 'mempty'. The idea is to provide a handle on steps--- only executed for their side effects. For instance, if you want to--- run a collection of 'ProcessT's that await but don't yield some--- number of times, you can use 'fanOutSteps . map (fmap (const ()))'--- followed by a 'taking' process.-fanoutSteps :: (Functor m, Monad m, Monoid r)-            => [ProcessT m a r] -> ProcessT m a r-fanoutSteps xs = encased $ Await (MachineT . aux) Refl (fanoutSteps xs)-  where aux y = do (rs,xs') <- mapM (feed y) xs >>= mapAccumLM yields []-                   let nxt = fanoutSteps $ catMaybes xs'-                   if null rs-                   then return $ Yield mempty nxt-                   else return $ Yield (mconcat rs) nxt-        yields rs Stop = return (rs,Nothing)-        yields rs y@(Yield _ _) = first (++rs) <$> flushYields y-        yields rs a@(Await _ _ _) = return (rs, Just $ encased a)