concurrent-machines 0.2.3.3 → 0.3.1.5
raw patch · 9 files changed
Files
- CHANGELOG.md +12/−0
- concurrent-machines.cabal +37/−14
- src/Data/Machine/Concurrent.hs +1/−2
- src/Data/Machine/Concurrent/Buffer.hs +9/−22
- src/Data/Machine/Concurrent/Fanout.hs +3/−1
- src/Data/Machine/Concurrent/Scatter.hs +2/−2
- src/Data/Machine/Regulated.hs +1/−1
- tests/AllTests.hs +26/−2
- tests/SPlotTests.hs +172/−0
CHANGELOG.md view
@@ -1,3 +1,15 @@+# 0.3.0++- Removed `buffer` and `rolling` functions++These functions did not properly use the buffers intended to mediate+the connection to a downstream consumer. This functionality is+provided by `bufferConnect` and `rollingConnect`. Thanks to Ben+Sinclair for identifying the problem.++The issue is that these are necessarily connectors between machines,+and can not be treated as modifiers of a single machine in isolation.+ # 0.2.3 - GHC-8.0.1 compatibility
concurrent-machines.cabal view
@@ -1,5 +1,5 @@ name: concurrent-machines-version: 0.2.3.3+version: 0.3.1.5 synopsis: Concurrent networked stream transducers description: A simple use-case for this library is to run the stages@@ -31,12 +31,17 @@ build-type: Simple extra-source-files: README.md, CHANGELOG.md cabal-version: >=1.10-tested-with: GHC == 7.10.3, GHC == 8.0.1+tested-with: GHC == 8.6.5 || == 8.8.4 || == 8.10.7 || == 9.0.2 || == 9.2.1 source-repository head type: git location: http://github.com/acowley/concurrent-machines.git +flag splot+ description: Build test with splot visual output+ default: False+ manual: True+ library exposed-modules: Data.Machine.Concurrent, Data.Machine.Regulated,@@ -49,16 +54,16 @@ -- other-modules: other-extensions: GADTs, FlexibleContexts, RankNTypes, TupleSections, ScopedTypeVariables- build-depends: base >= 4.8 && < 4.10,+ build-depends: base >= 4.8 && < 5, monad-control >= 1.0 && < 1.1,- transformers >= 0.4 && < 0.6,- time >= 1.4 && < 1.8,- containers >= 0.5 && < 0.6,- transformers-base >= 0.4 && < 0.6,- machines >= 0.5 && < 0.7,- async >= 2.0.1 && < 2.2,- lifted-async >= 0.1 && < 0.10,- semigroups >= 0.8 && < 0.19+ transformers >= 0.4 && < 0.7,+ time >= 1.4 && < 1.14,+ containers >= 0.5 && < 0.7,+ transformers-base >= 0.4 && < 0.7,+ machines >= 0.5 && < 0.8,+ async >= 2.0.1 && < 2.3,+ lifted-async >= 0.1 && < 0.11,+ semigroups >= 0.8 && < 0.21 hs-source-dirs: src default-language: Haskell2010 ghc-options: -Wall@@ -67,15 +72,33 @@ type: exitcode-stdio-1.0 hs-source-dirs: tests main-is: AllTests.hs- ghc-options: -Wall -O0+ ghc-options: -Wall -threaded default-language: Haskell2010- build-depends: base >= 4.6 && < 5, concurrent-machines, machines,+ build-depends: base, concurrent-machines, machines, tasty, tasty-hunit, transformers, time +-- This is a nice modification of the tests that generates plots for+-- visualizing the execution of parallel processes using the splot+-- tool. That program requires cairo, so the test is disabled by+-- default.+test-suite splotime+ type: exitcode-stdio-1.0+ if flag(splot)+ buildable: True+ else+ buildable: False+ hs-source-dirs: tests+ main-is: SPlotTests.hs+ ghc-options: -Wall -O0+ default-language: Haskell2010+ if flag(splot)+ build-depends: base, concurrent-machines, machines,+ process, 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+ build-depends: base, time, machines, concurrent-machines default-language: Haskell2010 ghc-options: -threaded "-with-rtsopts=-N2"
src/Data/Machine/Concurrent.hs view
@@ -16,7 +16,6 @@ -- * Concurrent connection (>~>), (<~<), -- * Buffered machines- buffer, rolling, bufferConnect, rollingConnect, -- * Concurrent processing of shared inputs fanout, fanoutSteps,@@ -61,7 +60,7 @@ infixl 7 >~> -- | We want the first available response.-waitEither' :: MonadBaseControl IO m +waitEither' :: MonadBaseControl IO m => Maybe (Async (StM m a)) -> Async (StM m b) -> m (Either a b) waitEither' Nothing y = Right <$> wait y
src/Data/Machine/Concurrent/Buffer.hs view
@@ -3,9 +3,9 @@ -- irregular production rates. module Data.Machine.Concurrent.Buffer ( -- * Blocking buffers- bufferConnect, buffer,+ bufferConnect, -- * Non-blocking (rolling) buffers- rollingConnect, rolling,+ rollingConnect, -- * Internal helpers mediatedConnect, BufferRoom(..) ) where@@ -71,23 +71,10 @@ EmptyL -> Nothing x :< acc' -> Just (x, acc') --- | Eagerly request values from the wrapped machine. Values are--- placed in a buffer of the given size. When the buffer is full--- (i.e. downstream is running behind), we stop pumping the wrapped--- machine.-buffer :: MonadBaseControl IO m => Int -> MachineT m k o -> MachineT m k o-buffer n src = bufferConnect n src echo---- | Eagerly request values from the wrapped machine. Values are--- placed in a rolling buffer of the given size. If downstream can not--- catch up, values yielded by the wrapped machine will be dropped.-rolling :: MonadBaseControl IO m => Int -> MachineT m k o -> MachineT m k o-rolling n src = rollingConnect n src echo- -- | Indication if the payload value is "full" or not. data BufferRoom a = NoVacancy a | Vacancy a deriving (Eq, Ord, Show) --- | Mediate a 'MachineT' and a 'ProcessT' with a buffer. +-- | Mediate a 'MachineT' and a 'ProcessT' with a buffer. -- -- @mediatedConnect z snoc view source sink@ pipes @source@ into -- @sink@ through a buffer initialized to @z@ and updated with@@ -98,7 +85,7 @@ mediatedConnect :: forall m t b k c. MonadBaseControl IO m => t -> (t -> b -> BufferRoom t) -> (t -> Maybe (b,t)) -> MachineT m k b -> ProcessT m b c -> MachineT m k c-mediatedConnect z snoc view src0 snk0 = +mediatedConnect z snoc view src0 snk0 = MachineT $ do srcFuture <- asyncRun src0 snkFuture <- asyncRun snk0 go z (Just srcFuture) snkFuture@@ -115,7 +102,7 @@ -> Maybe (MachineT m k b) -> ProcessT m b c -> m (MachineStep m k c)- goAsync acc src snk = + goAsync acc src snk = join $ go acc <$> traverse asyncRun src <*> asyncRun snk -- Handle whichever step is ready first@@ -125,22 +112,22 @@ goStep acc step = case step of -- @src@ stepped first Left (Stop, snk) -> go acc Nothing snk- Left (Await g kg fg, snk) -> + Left (Await g kg fg, snk) -> asyncAwait g kg fg (MachineT . flip (go acc) snk . Just) Left (Yield o k, snk) -> case snoc acc o of -- add it to the right end of the buffer Vacancy acc' -> asyncRun k >>= flip (go acc') snk . Just -- buffer was full- NoVacancy acc' -> + NoVacancy acc' -> let go' snk' = do src' <- asyncRun k goStep acc' (Right (snk', Just src')) in wait snk >>= flip drain go' -- @snk@ stepped first Right (Stop, _) -> return Stop- Right (Yield o k, src) -> + Right (Yield o k, src) -> do return $ Yield o (MachineT $ asyncRun k >>= go acc src)- Right (Await f Refl ff, src) -> + Right (Await f Refl ff, src) -> case view acc of Nothing -> maybe (goAsync acc Nothing ff) (wait >=> demandSrc) src Just (x, acc') -> asyncRun (f x) >>= go acc' src
src/Data/Machine/Concurrent/Fanout.hs view
@@ -6,7 +6,7 @@ import Control.Concurrent.Async.Lifted (Async, async, wait) import Control.Monad (foldM) import Control.Monad.Trans.Control (MonadBaseControl, StM)-import Data.Machine (Step(..), MachineT(..), encased, ProcessT, Is(..))+import Data.Machine (Step(..), MachineT(..), encased, stopped, ProcessT, Is(..)) import Data.Machine.Concurrent.AsyncStep (MachineStep) import Data.Maybe (catMaybes) #if defined(__GLASGOW_HASKELL__) && __GLASGOW_HASKELL__ < 710@@ -53,6 +53,7 @@ -- a single yield from the composite process. fanout :: (MonadBaseControl IO m, Semigroup r) => [ProcessT m a r] -> ProcessT m a r+fanout [] = stopped 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'@@ -72,6 +73,7 @@ -- followed by a 'taking' process. fanoutSteps :: (MonadBaseControl IO m, Monoid r) => [ProcessT m a r] -> ProcessT m a r+fanoutSteps [] = stopped 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'
src/Data/Machine/Concurrent/Scatter.hs view
@@ -186,13 +186,13 @@ -- keep running it as long as upstream does not yield a 'Left' which -- we can not handle. When upstream yields a 'Left', we 'stop'. rightOnly :: Monad m => ProcessT m b r -> ProcessT m (Either a b) r-rightOnly snk = repeatedly (await >>= either (const stop) yield) ~> snk+rightOnly snk = repeatedly (await >>= either (const stop) (\x -> yield x)) ~> snk -- | We have a sink for the Left output of a source, so we want to -- keep running it as long as upstream does not yield a 'Right' which -- we can not handle. When upstream yields a 'Right', we 'stop'. leftOnly :: Monad m => ProcessT m a r -> ProcessT m (Either a b) r-leftOnly snk = repeatedly (await >>= either yield (const stop)) ~> snk+leftOnly snk = repeatedly (await >>= either (\x -> yield x) (const stop)) ~> snk -- | Connect two processes to the downstream tails of a 'Machine' that -- produces tuples. The two downstream consumers are run
src/Data/Machine/Regulated.hs view
@@ -14,7 +14,7 @@ regulated :: MonadIO m => Double -> ProcessT m a a regulated target = construct $ liftIO getCurrentTime >>= go 0 where go dt prevT =- do await >>= yield+ do await >>= \x -> yield x t <- liftIO getCurrentTime let e = target - realToFrac (diffUTCTime t prevT) dt' = dt + 0.5 * e
tests/AllTests.hs view
@@ -1,4 +1,6 @@+{-# LANGUAGE RankNTypes #-} import Data.Time.Clock (getCurrentTime, diffUTCTime)+import Control.Applicative ((<|>)) import Control.Concurrent (threadDelay) import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Machine.Concurrent@@ -17,6 +19,28 @@ t2 <- liftIO getCurrentTime return (r, realToFrac $ t2 `diffUTCTime` t1) ++-- Based on GitHub Issue 7+-- https://github.com/acowley/concurrent-machines/issues/7+alternativeWorks :: TestTree+alternativeWorks = testCase "alternative" $ do+ xs <- runT (replicated 5 "Step" ~> construct aux)+ assertEqual "Results" (replicate 5 "Step" ++ ["Done"]) xs+ where aux = do x <- await <|> yield "Done" *> stop+ yield x+ aux++alternativeWorksDelay :: TestTree+alternativeWorksDelay = testCase "alternative with delay" $ do+ xs <- runT (construct (gen 5) ~> construct aux)+ assertEqual "Results" (replicate 5 "Step" ++ ["Done"]) xs+ where aux = do x <- await <|> yield "Done" *> stop+ yield x+ aux+ gen :: MonadIO m => Int -> PlanT k String m a+ gen 0 = stop+ gen n = yield "Step" >> liftIO (threadDelay 100000) >> gen (n-1)+ pipeline :: TestTree pipeline = testCaseSteps "pipeline" $ \step -> do (r,dt) <- timed . runT . supply (repeat ()) $@@ -46,6 +70,6 @@ yield (x * 2) main :: IO ()-main = defaultMain $ +main = defaultMain $ testGroup "concurrent-machines"- [ pipeline, workStealing ]+ [ pipeline, workStealing, alternativeWorks, alternativeWorksDelay ]
+ tests/SPlotTests.hs view
@@ -0,0 +1,172 @@+module Main (main) where++import Control.Concurrent (threadDelay)+import Control.Exception (catch, throwIO)+import Control.Monad (when, forM_)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Trans.Writer+import Control.Monad.Trans.Class (lift)+import Data.Machine.Concurrent+import Data.Time.Clock (UTCTime, addUTCTime, diffUTCTime, getCurrentTime)+import Text.Printf (printf)+import Data.Time.Format (defaultTimeLocale, formatTime, readPTime)+import System.Exit (ExitCode (ExitSuccess), exitSuccess)+import System.IO (writeFile)+import qualified System.Process as Proc+import Text.ParserCombinators.ReadP (readP_to_S)++import qualified Test.Tasty as T+import qualified Test.Tasty.HUnit as TH++writeSPlot :: Bool+writeSPlot = True++showTime :: UTCTime -> String+showTime = formatTime defaultTimeLocale "%Y-%m-%d %H:%M:%S%Q"++worker :: (Show a, MonadIO m)+ => (a -> b) -> Int -> Double -> ProcessT (WriterT [String] m) a b+worker f i dt = repeatedly $ do+ x <- await+ t1 <- liftIO getCurrentTime+ lift $ tell [ printf "%s >%d colour%d" (showTime t1) i i+ , printf "%s !%d black %s" (showTime t1) i (show x) ]+ liftIO $ threadDelay (floor (dt * 10000))+ t2 <- liftIO getCurrentTime+ lift $ tell [printf "%s <%d" (showTime t2) i]+ yield (f x)++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)++pipeline :: T.TestTree+pipeline = TH.testCaseSteps "pipeline" $ \step -> do+ let xs = [(0::Int)..]++ ((r,dt), ls) <- runWriterT . timed . runT $+ source xs ~> worker id 1 3 ~> worker id 2 5 ~> worker id 3 10 ~> taking 10++ ((r',dt'), ls') <- runWriterT . timed . runT $+ source xs ~> worker id 1 2 >~> worker id 2 4 >~> worker id 3 8 ~> taking 10++ when writeSPlot $ do+ writeFile "pipeline-seq.splot" (unlines ls)+ writeFile "pipeline-par.splot" (unlines ls')++ step "Consistent results"+ TH.assertEqual "Results" r r'++ step "Parallelism"+ TH.assertBool ("Pipeline faster than sequential" ++ show (dt',dt))+ (dt' * 1.5 < dt)++buffering1 :: T.TestTree+buffering1 = TH.testCaseSteps "buffering1" $ \step -> do+ let xs = [1..32::Int]++ ((r, dt), ls) <- runWriterT . timed . runT $+ source xs ~> worker (*2) 1 2 ~> worker (+1) 2 4++ ((r', dt'), ls') <- runWriterT . timed . runT $+ source xs ~> bufferConnect 5 (worker (*2) 1 2) (worker (+1) 2 4)++ when writeSPlot $ do+ writeFile "buffering1-seq.splot" (unlines ls)+ writeFile "buffering1-par.splot" (unlines ls')++ step "Consistent results"+ TH.assertEqual "Results" r r'++ step "Parallelism"+ TH.assertBool ("Buffered pipeline faster than sequential" ++ show (dt', dt))+ (dt' * 1.4 < dt)++rolling1 :: T.TestTree+rolling1 = TH.testCaseSteps "rolling1" $ \step -> do+ let xs = [1..32::Int]++ ((r, dt), ls) <- runWriterT . timed . runT $+ source xs ~> worker (*2) 1 2 ~> worker (+1) 2 4++ ((r', dt'), ls') <- runWriterT . timed . runT $+ source xs ~> rollingConnect 5 (worker (*2) 1 2) (worker (+1) 2 4)++ when writeSPlot $ do+ writeFile "rolling1-seq.splot" (unlines ls)+ writeFile "rolling1-par.splot" (unlines ls')++ step "Consistent results"+ TH.assertBool "Results" (all (`elem` r) r')++ step "Parallelism"+ TH.assertBool ("Rolling pipeline faster than sequential" ++ show (dt', dt))+ (dt' * 1.5 < dt)++workStealing :: T.TestTree+workStealing = TH.testCaseSteps "work stealing" $ \step -> do+ let xs = [1..32::Int]++ ((r,dt), ls) <- runWriterT . timed . runT $+ source xs ~> (worker (*2) 0 4)++ ((r',dt'), ls') <- runWriterT . timed . runT $+ source xs ~> scatter (map (\i -> worker (*2) i 4) [1..4])++ when writeSPlot $ do+ writeFile "work-stealing-seq.splot" (unlines ls)+ writeFile "work-stealing-par.splot" (unlines ls')++ step "Consistent results"+ TH.assertBool "Predicted Serial Length" (length r == length xs)+ TH.assertBool "Predicted Parallel Length" (length r' == length xs)+ TH.assertBool "Predicted Results" (all (`elem` r') (map (*2) xs))+ TH.assertBool "Results" (all (`elem` r') r)++ step "Parallelism"+ TH.assertBool ("Work Stealing faster than sequential" ++ show (dt,dt'))+ (dt' * 1.5 < dt)++main :: IO ()+main = do+ catch+ (T.defaultMain+ (T.testGroup "concurrent-machines"+ [ pipeline, buffering1, rolling1, workStealing ]))+ (\e -> if e == ExitSuccess+ then return ()+ else throwIO e)+ when writeSPlot $ do+ let splots = [ "pipeline-seq.splot"+ , "pipeline-par.splot"+ , "buffering1-seq.splot"+ , "buffering1-par.splot"+ , "rolling1-seq.splot"+ , "rolling1-par.splot"+ , "work-stealing-seq.splot"+ , "work-stealing-par.splot"+ ]+ forM_ splots $ \path -> do+ l:_ <- lines <$> readFile path+ case readP_to_S (readPTime False defaultTimeLocale "%Y-%m-%d %H:%M:%S%Q") l of+ [] ->+ fail "Shit no."+ (fromTime,_):_ -> do+ let toTime = addUTCTime 2.5 fromTime+ let args = [ "-if", path+ , "-o", path ++ ".png"+ , "-w", "2048"+ , "-h", "200"+ , "-bh", "1"+ , "-tickInterval", "100"+ , "-legendWidth", "20"+ , "-numTracks", "4"+ , "-fromTime", showTime fromTime+ , "-toTime", showTime toTime+ ]+ Proc.readProcess "splot" args "" >>= putStr+ putStrLn $ "OK: " ++ show (fromTime :: UTCTime) ++ " " ++ path+ exitSuccess