Tests for concurrent-machines

AllTests.hs

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)

buffering2 :: T.TestTree
buffering2 = TH.testCaseSteps "buffering2" $ \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 ~> buffer 5 (worker (*2) 1 2) ~> worker (+1) 2 4

  when writeSPlot $ do
    writeFile "buffering2-seq.splot" (unlines ls)
    writeFile "buffering2-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.1 < 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)

rolling2 :: T.TestTree
rolling2 = TH.testCaseSteps "rolling2" $ \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 ~> rolling 5 (worker (*2) 1 2) ~> worker (+1) 2 4

  when writeSPlot $ do
    writeFile "rolling2-seq.splot" (unlines ls)
    writeFile "rolling2-par.splot" (unlines ls')

  step "Consistent results"
  TH.assertEqual "Results" r r'

  step "Parallelism"
  TH.assertBool ("Rolling pipeline faster than sequential" ++ show (dt', dt))
                (dt' * 1.1 < 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, buffering2, rolling1, rolling2, 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"
                 , "buffering2-seq.splot"
                 , "buffering2-par.splot"
                 , "rolling1-seq.splot"
                 , "rolling1-par.splot"
                 , "rolling2-seq.splot"
                 , "rolling2-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