diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/concurrent-machines.cabal b/concurrent-machines.cabal --- a/concurrent-machines.cabal +++ b/concurrent-machines.cabal @@ -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" diff --git a/src/Data/Machine/Concurrent.hs b/src/Data/Machine/Concurrent.hs --- a/src/Data/Machine/Concurrent.hs +++ b/src/Data/Machine/Concurrent.hs @@ -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 diff --git a/src/Data/Machine/Concurrent/Buffer.hs b/src/Data/Machine/Concurrent/Buffer.hs --- a/src/Data/Machine/Concurrent/Buffer.hs +++ b/src/Data/Machine/Concurrent/Buffer.hs @@ -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 diff --git a/src/Data/Machine/Concurrent/Fanout.hs b/src/Data/Machine/Concurrent/Fanout.hs --- a/src/Data/Machine/Concurrent/Fanout.hs +++ b/src/Data/Machine/Concurrent/Fanout.hs @@ -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' diff --git a/src/Data/Machine/Concurrent/Scatter.hs b/src/Data/Machine/Concurrent/Scatter.hs --- a/src/Data/Machine/Concurrent/Scatter.hs +++ b/src/Data/Machine/Concurrent/Scatter.hs @@ -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 diff --git a/src/Data/Machine/Regulated.hs b/src/Data/Machine/Regulated.hs --- a/src/Data/Machine/Regulated.hs +++ b/src/Data/Machine/Regulated.hs @@ -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 diff --git a/tests/AllTests.hs b/tests/AllTests.hs --- a/tests/AllTests.hs +++ b/tests/AllTests.hs @@ -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 ] diff --git a/tests/SPlotTests.hs b/tests/SPlotTests.hs new file mode 100644 --- /dev/null +++ b/tests/SPlotTests.hs @@ -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