concurrent-machines 0.2.3.3 → 0.3.0

raw patch · 5 files changed

+219/−26 lines, 5 filesdep +processdep ~basedep ~machinesdep ~timePVP ok

version bump matches the API change (PVP)

Dependencies added: process

Dependency ranges changed: base, machines, time, transformers

API changes (from Hackage documentation)

Files

• CHANGELOG.md +12/−0
• concurrent-machines.cabal +25/−2
• src/Data/Machine/Concurrent.hs +1/−2
• src/Data/Machine/Concurrent/Buffer.hs +9/−22
• tests/SPlotTests.hs +172/−0

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

@@ -1,5 +1,5 @@ name:                concurrent-machines-version:             0.2.3.3+version:             0.3.0 synopsis:            Concurrent networked stream transducers  description: A simple use-case for this library is to run the stages@@ -37,6 +37,11 @@   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,@@ -52,7 +57,7 @@   build-depends:       base >= 4.8 && < 4.10,                        monad-control >= 1.0 && < 1.1,                        transformers >= 0.4 && < 0.6,-                       time >= 1.4 && < 1.8,+                       time >= 1.4 && < 1.9,                        containers >= 0.5 && < 0.6,                        transformers-base >= 0.4 && < 0.6,                        machines >= 0.5 && < 0.7,@@ -71,6 +76,24 @@   default-language: Haskell2010   build-depends: base >= 4.6 && < 5, 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 >= 4.6 && < 5, concurrent-machines, machines,+                   process, tasty, tasty-hunit, transformers, time  benchmark fanout   type: exitcode-stdio-1.0

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

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

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