From 691e84650b7f530622480d69886104471b9d6276 Mon Sep 17 00:00:00 2001 From: Laurent Date: Fri, 18 Sep 2026 12:20:27 -0400 Subject: [PATCH 1/2] Re-work distributed-process benchmarks --- cabal.project | 1 + packages/distributed-process/ChangeLog | 1 + .../benchmarks/Channels.hs | 43 -- .../distributed-process/benchmarks/Latency.hs | 39 - .../distributed-process/benchmarks/Main.hs | 711 ++++++++++++++++++ .../benchmarks/ProcessRing.hs | 117 --- .../distributed-process/benchmarks/Spawns.hs | 49 -- .../benchmarks/Throughput.hs | 74 -- .../distributed-process.cabal | 57 +- 9 files changed, 722 insertions(+), 370 deletions(-) delete mode 100644 packages/distributed-process/benchmarks/Channels.hs delete mode 100644 packages/distributed-process/benchmarks/Latency.hs create mode 100644 packages/distributed-process/benchmarks/Main.hs delete mode 100644 packages/distributed-process/benchmarks/ProcessRing.hs delete mode 100644 packages/distributed-process/benchmarks/Spawns.hs delete mode 100644 packages/distributed-process/benchmarks/Throughput.hs diff --git a/cabal.project b/cabal.project index 2ce3cec64..d612efaf7 100644 --- a/cabal.project +++ b/cabal.project @@ -1,5 +1,6 @@ packages: packages/*/**.cabal tests: true +benchmarks: true package distributed-process-tests flags: +tcp diff --git a/packages/distributed-process/ChangeLog b/packages/distributed-process/ChangeLog index a65812d1f..7055e4aa4 100644 --- a/packages/distributed-process/ChangeLog +++ b/packages/distributed-process/ChangeLog @@ -1,6 +1,7 @@ Unreleased * Added support for `containers-0.8`. +* Reworked benchmarks, which can now be run using `cabal bench distributed-process`. 2025-02-04 Laurent P. René de Cotret 0.7.8 diff --git a/packages/distributed-process/benchmarks/Channels.hs b/packages/distributed-process/benchmarks/Channels.hs deleted file mode 100644 index 7ec82b64b..000000000 --- a/packages/distributed-process/benchmarks/Channels.hs +++ /dev/null @@ -1,43 +0,0 @@ --- | Like Latency, but creating lots of channels -import System.Environment -import Control.Monad -import Control.Applicative -import Control.Distributed.Process -import Control.Distributed.Process.Node -import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters) -import Data.Binary (encode, decode) -import qualified Data.ByteString.Lazy as BSL - -pingServer :: Process () -pingServer = forever $ do - them <- expect - sendChan them () - -- TODO: should this be automatic? - reconnectPort them - -pingClient :: Int -> ProcessId -> Process () -pingClient n them = do - replicateM_ n $ do - (sc, rc) <- newChan :: Process (SendPort (), ReceivePort ()) - send them sc - receiveChan rc - liftIO . putStrLn $ "Did " ++ show n ++ " pings" - -initialProcess :: String -> Process () -initialProcess "SERVER" = do - us <- getSelfPid - liftIO $ BSL.writeFile "pingServer.pid" (encode us) - pingServer -initialProcess "CLIENT" = do - n <- liftIO $ getLine - them <- liftIO $ decode <$> BSL.readFile "pingServer.pid" - pingClient (read n) them - -main :: IO () -main = do - [role, host, port] <- getArgs - trans <- createTransport (defaultTCPAddr host port) defaultTCPParameters - case trans of - Right transport -> do node <- newLocalNode transport initRemoteTable - runProcess node $ initialProcess role - Left other -> error $ show other diff --git a/packages/distributed-process/benchmarks/Latency.hs b/packages/distributed-process/benchmarks/Latency.hs deleted file mode 100644 index 6d9a6cf48..000000000 --- a/packages/distributed-process/benchmarks/Latency.hs +++ /dev/null @@ -1,39 +0,0 @@ -import System.Environment -import Control.Monad -import Control.Applicative -import Control.Distributed.Process -import Control.Distributed.Process.Node -import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters) -import Data.Binary (encode, decode) -import qualified Data.ByteString.Lazy as BSL - -pingServer :: Process () -pingServer = forever $ do - them <- expect - send them () - -pingClient :: Int -> ProcessId -> Process () -pingClient n them = do - us <- getSelfPid - replicateM_ n $ send them us >> (expect :: Process ()) - liftIO . putStrLn $ "Did " ++ show n ++ " pings" - -initialProcess :: String -> Process () -initialProcess "SERVER" = do - us <- getSelfPid - liftIO $ BSL.writeFile "pingServer.pid" (encode us) - pingServer -initialProcess "CLIENT" = do - n <- liftIO $ getLine - them <- liftIO $ decode <$> BSL.readFile "pingServer.pid" - pingClient (read n) them - -main :: IO () -main = do - [role, host, port] <- getArgs - trans <- createTransport - (defaultTCPAddr host port) defaultTCPParameters - case trans of - Right transport -> do node <- newLocalNode transport initRemoteTable - runProcess node $ initialProcess role - Left other -> error $ show other diff --git a/packages/distributed-process/benchmarks/Main.hs b/packages/distributed-process/benchmarks/Main.hs new file mode 100644 index 000000000..26f100f0a --- /dev/null +++ b/packages/distributed-process/benchmarks/Main.hs @@ -0,0 +1,711 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} +{-# OPTIONS_GHC -Wno-unused-top-binds #-} + +module Main (main) where + +import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) +import Control.Concurrent.STM + ( TQueue, + atomically, + newTQueueIO, + readTQueue, + writeTQueue, + ) +import Control.Distributed.Process + ( Handler (Handler), + MonitorRef, + NodeId, + Process, + ProcessId, + ProcessMonitorNotification (ProcessMonitorNotification), + ReceivePort, + SendPort, + WhereIsReply (WhereIsReply), + call, + callLocal, + catchExit, + catches, + catchesExit, + delegate, + die, + exit, + expect, + expectTimeout, + forward, + getLocalNodeStats, + getNodeStats, + getProcessInfo, + getSelfNode, + getSelfPid, + handleMessage, + kill, + liftIO, + link, + match, + matchAny, + matchChan, + matchIf, + matchMessage, + matchSTM, + matchUnknown, + mergePortsBiased, + mergePortsRR, + monitor, + monitorNode, + monitorPort, + newChan, + nsend, + nsendRemote, + proxy, + receiveChan, + receiveChanTimeout, + receiveTimeout, + receiveWait, + register, + relay, + reregister, + send, + sendChan, + spawn, + spawnChannel, + spawnChannelLocal, + spawnLocal, + spawnMonitor, + uforward, + unlink, + unmonitor, + unregister, + unsafeSend, + unwrapMessage, + usend, + whereis, + whereisRemoteAsync, + withMonitor_, + wrapMessage, + ) +import Control.Distributed.Process.Closure + ( functionTDict, + mkClosure, + remotable, + sdictUnit, + ) +import Control.Distributed.Process.Node + ( LocalNode (..), + closeLocalNode, + forkProcess, + initRemoteTable, + newLocalNode, + runProcess, + ) +import Control.Distributed.Process.Serializable (Serializable) +import qualified Control.Exception as E +import Control.Monad (forever, replicateM, replicateM_, void, when) +import qualified Control.Monad.Catch as Catch +import Data.Binary (Binary) +import qualified Data.ByteString.Char8 as BS +import GHC.Generics (Generic) +import qualified Network.Transport as NT +import Network.Transport.TCP + ( createTransport, + defaultTCPAddr, + defaultTCPParameters, + ) +import Test.Tasty.Bench + ( Benchmark, + bench, + bgroup, + defaultMain, + whnfIO, + ) + +-- A top-level splice only brings names into scope for later declaration +-- groups, so these must precede 'main' and every use of 'mkClosure'. + +remoteSignal :: ProcessId -> Process () +remoteSignal them = send them () + +remoteChanEcho :: ProcessId -> ReceivePort () -> Process () +remoteChanEcho them rp = receiveChan rp >> send them () + +remoteAnswer :: () -> Process Int +remoteAnswer () = return 42 + +remotable ['remoteSignal, 'remoteChanEcho, 'remoteAnswer] + +main :: IO () +main = do + let rtable = __remoteTable initRemoteTable + transport <- + either E.throwIO return + =<< createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters + ( E.bracket (newLocalNode transport rtable) closeLocalNode $ \node1 -> + E.bracket (newLocalNode transport rtable) closeLocalNode $ \node2 -> do + fx <- setup node1 node2 + defaultMain (benchmarks fx) + ) + `E.finally` NT.closeTransport transport + +benchmarks :: Fixture -> [Benchmark] +benchmarks fx = + [ bgroup + "local" + [ baseline fx, + messaging fx, + channels fx, + receiving fx, + messages fx, + processes fx, + monitoring fx, + registry fx, + exceptions fx, + introspection fx, + ring fx + ], + remote fx + ] + +-- | Cost of the harness alone, and so the floor below which the other numbers +-- say nothing. +baseline :: Fixture -> Benchmark +baseline fx = + bgroup + "baseline" + [ oneBench fx "runner round trip (no work)" (return ()), + repsBench fx "empty loop" 1000 (return ()) + ] + +messaging :: Fixture -> Benchmark +messaging fx = + bgroup + "messaging" + [ repsBench fx "send/expect" 1000 $ do + us <- getSelfPid + send echo us + expect :: Process (), + repsBench fx "usend/expect" 1000 $ do + us <- getSelfPid + usend echo us + expect :: Process (), + repsBench fx "unsafeSend/expect" 1000 $ do + us <- getSelfPid + unsafeSend echo us + expect :: Process (), + repsBench fx "nsend/expect" 100 $ do + us <- getSelfPid + nsend echoName us + expect :: Process (), + bgroup + "throughput/bytestring" + [ oneBench fx (show sz ++ "B") $ + sendThrough (fxCounter fx) 1000 (BS.replicate sz 'x') + | sz <- [8, 1024, 65536] + ], + bgroup + "throughput/list-of-int" + [ oneBench fx (show n ++ "elems") $ + sendThrough (fxCounter fx) 1000 [1 .. n] + | n <- [1, 100 :: Int] + ] + ] + where + echo = fxEcho fx + +channels :: Fixture -> Benchmark +channels fx = + bgroup + "channels" + [ repsBench fx "newChan" 100 $ + void (newChan :: Process (SendPort (), ReceivePort ())), + repsBench fx "sendChan/receiveChan" 1000 $ do + sendChan sp () + receiveChan rp, + repsBench fx "receiveChanTimeout (empty)" 1000 $ + void (receiveChanTimeout 0 rp), + repsBench fx "newChan + roundtrip via echo server" 100 $ do + (sp', rp') <- newChan + send (fxEcho fx) sp' + receiveChan rp', + repsBench fx "spawnChannelLocal" 100 $ do + us <- getSelfPid + sp' <- spawnChannelLocal $ \rp' -> + (receiveChan rp' :: Process ()) >> send us () + sendChan sp' () + expect :: Process (), + repsBench fx "mergePortsBiased" 100 (mergeBench mergePortsBiased), + repsBench fx "mergePortsRR" 100 (mergeBench mergePortsRR) + ] + where + (sp, rp) = fxChan fx + + mergeBench merge = do + (sps, rps) <- + unzip + <$> replicateM 4 (newChan :: Process (SendPort (), ReceivePort ())) + merged <- merge rps + mapM_ (`sendChan` ()) sps + replicateM_ 4 (receiveChan merged) + +-- | Each of these puts one message in the runner's own mailbox and takes it +-- out again, so the spread between them is the cost of the 'Match'. +receiving :: Fixture -> Benchmark +receiving fx = + bgroup + "receiving" + [ repsBench fx "expect" 1000 $ do + selfSend () + expect :: Process (), + repsBench fx "receiveWait (first of 1 match)" 1000 $ do + selfSend () + receiveWait [match (\() -> return ())], + repsBench fx "receiveWait (last of 6 matches)" 1000 $ do + selfSend () + receiveWait + [ match (\(_ :: Int) -> return ()), + match (\(_ :: Bool) -> return ()), + match (\(_ :: Char) -> return ()), + match (\(_ :: BS.ByteString) -> return ()), + match (\(_ :: Ping) -> return ()), + match (\() -> return ()) + ], + repsBench fx "matchIf" 1000 $ do + selfSend (1 :: Int) + receiveWait [matchIf (> (0 :: Int)) (\_ -> return ())], + repsBench fx "matchAny" 1000 $ do + selfSend () + receiveWait [matchAny (\_ -> return ())], + repsBench fx "matchUnknown" 1000 $ do + selfSend () + receiveWait [match (\(_ :: Int) -> return ()), matchUnknown (return ())], + repsBench fx "matchMessage" 1000 $ do + selfSend () + void (receiveWait [matchMessage return]), + repsBench fx "matchChan" 1000 $ do + sendChan sp () + receiveWait [matchChan rp return], + repsBench fx "matchSTM" 1000 $ do + liftIO (atomically (writeTQueue q ())) + receiveWait [matchSTM (readTQueue q) return], + repsBench fx "receiveTimeout (empty mailbox)" 1000 $ + void (receiveTimeout 0 [match (\() -> return ())]), + repsBench fx "expectTimeout (hit)" 1000 $ do + selfSend () + void (expectTimeout 0 :: Process (Maybe ())) + ] + where + (sp, rp) = fxChan fx + q = fxQueue fx + + selfSend :: (Serializable a) => a -> Process () + selfSend x = getSelfPid >>= \us -> unsafeSend us x + +messages :: Fixture -> Benchmark +messages fx = + bgroup + "messages" + [ repsBench fx "unwrapMessage (hit)" 1000 $ + void (unwrapMessage intMessage :: Process (Maybe Int)), + repsBench fx "unwrapMessage (miss)" 1000 $ + void (unwrapMessage intMessage :: Process (Maybe Bool)), + repsBench fx "handleMessage (hit)" 1000 $ + void (handleMessage intMessage (\(_ :: Int) -> return ())), + repsBench fx "handleMessage (miss)" 1000 $ + void (handleMessage intMessage (\(_ :: Bool) -> return ())), + repsBench fx "wrapMessage + unwrapMessage" 1000 $ + void (unwrapMessage (wrapMessage (42 :: Int)) :: Process (Maybe Int)), + -- A 'ProcessId' is sent rather than @()@ so that 'echoServer' recognises + -- the forwarded message and replies. + repsBench fx "forward" 1000 $ do + us <- getSelfPid + unsafeSend us us + receiveWait [matchAny (`forward` fxEcho fx)] + expect :: Process (), + repsBench fx "uforward" 1000 $ do + us <- getSelfPid + unsafeSend us us + receiveWait [matchAny (`uforward` fxEcho fx)] + expect :: Process (), + repsBench fx "relay" 1000 $ do + send (fxRelay fx) () + expect :: Process (), + repsBench fx "delegate" 1000 $ do + send (fxDelegate fx) () + expect :: Process (), + repsBench fx "proxy" 1000 $ do + send (fxProxy fx) () + expect :: Process () + ] + where + intMessage = wrapMessage (42 :: Int) + +processes :: Fixture -> Benchmark +processes fx = + bgroup + "processes" + [ repsBench fx "spawnLocal (sequential)" 100 $ do + us <- getSelfPid + _ <- spawnLocal (send us ()) + expect :: Process (), + oneBench fx "spawnLocal (pipelined)" $ do + us <- getSelfPid + replicateM_ 100 (spawnLocal (send us ())) + replicateM_ 100 (expect :: Process ()), + repsBench fx "callLocal" 100 $ + callLocal (return ()), + repsBench fx "getSelfPid" 1000 $ + void getSelfPid, + repsBench fx "getSelfNode" 1000 $ + void getSelfNode + ] + +monitoring :: Fixture -> Benchmark +monitoring fx = + bgroup + "monitoring" + [ repsBench fx "monitor/unmonitor" 100 $ + monitor echo >>= unmonitor, + repsBench fx "withMonitor_" 100 $ + withMonitor_ echo (return ()), + repsBench fx "link/unlink" 100 $ + link echo >> unlink echo, + repsBench fx "monitorNode/unmonitor" 100 $ + (getSelfNode >>= monitorNode) >>= unmonitor, + repsBench fx "monitorPort/unmonitor" 100 $ + monitorPort (fst (fxChan fx)) >>= unmonitor, + repsBench fx "notification (normal exit)" 100 $ do + pid <- spawnLocal (expect :: Process ()) + ref <- monitor pid + send pid () + awaitDown ref, + repsBench fx "notification (kill)" 100 $ do + pid <- spawnLocal (expect :: Process ()) + ref <- monitor pid + kill pid "benchmark" + awaitDown ref, + repsBench fx "notification (die)" 100 $ do + pid <- spawnLocal (die "benchmark") + ref <- monitor pid + awaitDown ref, + repsBench fx "exit caught by catchExit" 100 $ do + pid <- + spawnLocal $ + catchExit (expect :: Process ()) (\_ (_ :: String) -> return ()) + ref <- monitor pid + exit pid "benchmark" + awaitDown ref, + repsBench fx "exit caught by catchesExit" 100 $ do + pid <- + spawnLocal $ + catchesExit + (expect :: Process ()) + [\_ m -> handleMessage m (\(_ :: String) -> return ())] + ref <- monitor pid + exit pid "benchmark" + awaitDown ref + ] + where + echo = fxEcho fx + +registry :: Fixture -> Benchmark +registry fx = + bgroup + "registry" + [ repsBench fx "whereis (hit)" 100 $ + void (whereis echoName), + repsBench fx "whereis (miss)" 100 $ + void (whereis "benchmarks.absent"), + repsBench fx "register/unregister" 100 $ do + register "benchmarks.tmp" (fxEcho fx) + unregister "benchmarks.tmp", + repsBench fx "reregister" 100 $ + reregister echoName (fxEcho fx) + ] + +exceptions :: Fixture -> Benchmark +exceptions fx = + bgroup + "exceptions" + [ repsBench fx "catch (not thrown)" 1000 $ + Catch.catch (return ()) (\(_ :: E.SomeException) -> return ()), + repsBench fx "catch (thrown)" 1000 $ + Catch.catch (Catch.throwM Boom) (\Boom -> return ()), + repsBench fx "try" 1000 $ + void (Catch.try (return ()) :: Process (Either E.SomeException ())), + repsBench fx "catches (distributed-process Handler)" 1000 $ + catches + (return ()) + [ Handler (\(_ :: E.ArithException) -> return ()), + Handler (\(_ :: E.SomeException) -> return ()) + ], + repsBench fx "bracket" 1000 $ + Catch.bracket (return ()) (\_ -> return ()) (\_ -> return ()), + repsBench fx "finally" 1000 $ + Catch.finally (return ()) (return ()), + repsBench fx "onException" 1000 $ + Catch.onException (return ()) (return ()), + repsBench fx "mask_" 1000 $ + Catch.mask_ (return ()) + ] + +introspection :: Fixture -> Benchmark +introspection fx = + bgroup + "introspection" + [ repsBench fx "getProcessInfo" 100 $ + void (getProcessInfo (fxEcho fx)), + repsBench fx "getLocalNodeStats" 100 $ + void getLocalNodeStats, + repsBench fx "getNodeStats" 100 $ + void (getSelfNode >>= getNodeStats) + ] + +-- | 100 laps around each of the rings built by 'setup'. +ring :: Fixture -> Benchmark +ring fx = + bgroup + "ring" + [ oneBench fx nm $ do + replicateM_ 100 (send entry (Ping 0)) + replicateM_ 100 (void (expect :: Process Ping)) + | (nm, entry) <- fxRings fx + ] + +remote :: Fixture -> Benchmark +remote fx = + bgroup + "remote" + [ repsBench fx "send/expect" 100 $ do + us <- getSelfPid + send echo us + expect :: Process (), + repsBench fx "usend/expect" 100 $ do + us <- getSelfPid + usend echo us + expect :: Process (), + repsBench fx "newChan + sendChan/receiveChan" 100 $ do + (sp, rp) <- newChan + send echo sp + receiveChan rp, + bgroup + "throughput/bytestring" + [ oneBench fx (show sz ++ "B") $ + sendThrough (fxRemoteCounter fx) 100 (BS.replicate sz 'x') + | sz <- [8, 1024, 65536] + ], + repsBench fx "nsendRemote/expect" 100 $ do + us <- getSelfPid + nsendRemote nid echoName us + expect :: Process (), + repsBench fx "whereisRemoteAsync" 100 $ do + whereisRemoteAsync nid echoName + receiveWait + [ matchIf + (\(WhereIsReply n _) -> n == echoName) + (\_ -> return ()) + ], + repsBench fx "spawn" 100 $ do + us <- getSelfPid + _ <- spawn nid ($(mkClosure 'remoteSignal) us) + expect :: Process (), + repsBench fx "spawnMonitor + notification" 100 $ do + us <- getSelfPid + (_, ref) <- spawnMonitor nid ($(mkClosure 'remoteSignal) us) + expect :: Process () + awaitDown ref, + repsBench fx "spawnChannel" 100 $ do + us <- getSelfPid + sp <- spawnChannel sdictUnit nid ($(mkClosure 'remoteChanEcho) us) + sendChan sp () + expect :: Process (), + repsBench fx "call" 100 $ + void + ( call + $(functionTDict 'remoteAnswer) + nid + ($(mkClosure 'remoteAnswer) ()) + ), + repsBench fx "getNodeStats" 100 $ + void (getNodeStats nid), + repsBench fx "getProcessInfo" 100 $ + void (getProcessInfo echo) + ] + where + echo = fxRemoteEcho fx + nid = fxRemoteNodeId fx + +-- | tasty-bench already repeats the body of the benchmark, but the benchmark +-- fixture adds a baseline amount of time which drowns some of the faster benchmarks. +-- +-- Therefore, we amortize the fixture overhead by looping. +repsBench :: Fixture -> String -> Int -> Process () -> Benchmark +repsBench fx name reps act = + bench (name ++ " (x" ++ show reps ++ ")") $ + whnfIO (fxRun fx (replicateM_ reps act)) + +oneBench :: Fixture -> String -> Process () -> Benchmark +oneBench fx name act = bench name $ whnfIO (fxRun fx act) + +data Fixture = Fixture + { fxRun :: Process () -> IO (), + fxRemoteNodeId :: NodeId, + fxEcho :: ProcessId, + fxCounter :: ProcessId, + fxRemoteEcho :: ProcessId, + fxRemoteCounter :: ProcessId, + fxChan :: (SendPort (), ReceivePort ()), + fxQueue :: TQueue (), + fxRelay :: ProcessId, + fxDelegate :: ProcessId, + fxProxy :: ProcessId, + fxRings :: [(String, ProcessId)] + } + +echoName :: String +echoName = "benchmarks.echo" + +setup :: LocalNode -> LocalNode -> IO Fixture +setup node1 node2 = do + run <- newRunner node1 + queue <- newTQueueIO + echo <- forkProcess node1 echoServer + counter <- forkProcess node1 counterServer + remoteEcho <- forkProcess node2 echoServer + remoteCount <- forkProcess node2 counterServer + -- 'register' acts on the caller's node. + runProcess node1 (register echoName echo) + runProcess node2 (register echoName remoteEcho) + -- 'relay', 'delegate' and 'proxy' never return, so they cannot be spawned + -- per iteration. They, the rings and the shared channel all have to be + -- rooted at the runner, since that is the process each iteration runs on. + var <- newEmptyMVar + run $ do + self <- getSelfPid + chan <- newChan + rly <- spawnLocal (relay self) + dlg <- spawnLocal (delegate self (const True)) + prx <- spawnLocal (proxy self (\() -> return True)) + rings <- + mapM + (\(nm, mode) -> (,) nm <$> makeRing mode 10 self) + [ ("send", RelaySend), + ("unsafeSend", RelayUnsafeSend), + ("forward", RelayForward) + ] + liftIO $ putMVar var (chan, rly, dlg, prx, rings) + (chan, rly, dlg, prx, rings) <- takeMVar var + return + Fixture + { fxRun = run, + fxRemoteNodeId = localNodeId node2, + fxEcho = echo, + fxCounter = counter, + fxRemoteEcho = remoteEcho, + fxRemoteCounter = remoteCount, + fxChan = chan, + fxQueue = queue, + fxRelay = rly, + fxDelegate = dlg, + fxProxy = prx, + fxRings = rings + } + +-- | Runs actions on one long-lived process. Using 'runProcess' instead would +-- fold a 'forkProcess' into every measurement and give each iteration a fresh +-- 'ProcessId', defeating the connection caching real applications rely on. +newRunner :: LocalNode -> IO (Process () -> IO ()) +newRunner node = do + reqVar <- newEmptyMVar + respVar <- newEmptyMVar + _ <- forkProcess node $ forever $ do + act <- liftIO (takeMVar reqVar) + r <- Catch.try act + drainMailbox + liftIO $ putMVar respVar (r :: Either E.SomeException ()) + return $ \act -> do + putMVar reqVar act + takeMVar respVar >>= either E.throwIO return + +-- | Keeps a benchmark from perturbing later ones through the runner's mailbox. +drainMailbox :: Process () +drainMailbox = do + r <- receiveTimeout 0 [matchAny (\_ -> return ())] + case r of + Nothing -> return () + Just () -> drainMailbox + +awaitDown :: MonitorRef -> Process () +awaitDown ref = + receiveWait + [ matchIf + (\(ProcessMonitorNotification ref' _ _) -> ref' == ref) + (\_ -> return ()) + ] + +-- | Pipelined throughput: @n@ one-way sends, then one round trip to confirm +-- they all arrived. +sendThrough :: (Serializable a) => ProcessId -> Int -> a -> Process () +sendThrough srv n payload = do + us <- getSelfPid + replicateM_ n (send srv payload) + send srv (Report us) + n' <- expect + when (n' /= n) $ + die ("expected " ++ show n ++ " messages, server saw " ++ show n') + +-- | The trailing 'matchAny' stops the mailbox growing if a benchmark sends +-- something unexpected; a growing mailbox is rescanned on every 'receiveWait' +-- and would skew every benchmark that follows. +echoServer :: Process () +echoServer = + forever $ + receiveWait + [ match $ \(them :: ProcessId) -> send them (), + match $ \(them, n :: Int) -> send them n, + match $ \(them, bs :: BS.ByteString) -> send them bs, + match $ \(sp :: SendPort ()) -> sendChan sp (), + matchAny $ \_ -> return () + ] + +-- | Counts one-way messages, and on 'Report' replies with the number seen +-- since the last report. +counterServer :: Process () +counterServer = go 0 + where + go :: Int -> Process () + go !n = + receiveWait + [ match $ \(Report them) -> send them n >> go 0, + matchAny $ \_ -> go (n + 1) + ] + +data RelayMode = RelaySend | RelayUnsafeSend | RelayForward + +relayLoop :: RelayMode -> ProcessId -> Process () +relayLoop mode next = forever $ case mode of + RelaySend -> expect >>= \m -> send next (m :: Ping) + RelayUnsafeSend -> expect >>= \m -> unsafeSend next (m :: Ping) + RelayForward -> receiveWait [matchAny (`forward` next)] + +-- | Ring of @n@ relays whose last member relays to @target@; returns the entry +-- point. +makeRing :: RelayMode -> Int -> ProcessId -> Process ProcessId +makeRing mode n target + | n <= 0 = return target + | otherwise = makeRing mode (n - 1) =<< spawnLocal (relayLoop mode target) + +newtype Ping = Ping Int + deriving (Generic) + +instance Binary Ping + +newtype Report = Report ProcessId + deriving (Generic) + +instance Binary Report + +data Boom = Boom + deriving (Show) + +instance E.Exception Boom diff --git a/packages/distributed-process/benchmarks/ProcessRing.hs b/packages/distributed-process/benchmarks/ProcessRing.hs deleted file mode 100644 index 55f3c7433..000000000 --- a/packages/distributed-process/benchmarks/ProcessRing.hs +++ /dev/null @@ -1,117 +0,0 @@ -{- ProcessRing benchmarks. - -To run the benchmarks, select a value for the ring size (sz) and -the number of times to send a message around the ring - --} - -{-# LANGUAGE BangPatterns #-} -{-# LANGUAGE ScopedTypeVariables #-} - -import Control.Monad -import Control.Distributed.Process hiding (catch) -import Control.Distributed.Process.Node -import Control.Exception (catch, SomeException) -import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters) -import System.Environment -import System.Console.GetOpt - -data Options = Options - { optRingSize :: Int - , optIterations :: Int - , optForward :: Bool - , optParallel :: Bool - , optUnsafe :: Bool - } deriving Show - -initialProcess :: Options -> Process () -initialProcess op = - let ringSz = optRingSize op - msgCnt = optIterations op - fwd = optForward op - unsafe = optUnsafe op - msg = ("foobar", "baz") - in do - self <- getSelfPid - ring <- makeRing fwd unsafe ringSz self - forM_ [1..msgCnt] (\_ -> send ring msg) - collect msgCnt - where relay fsend pid = do - msg <- expect :: Process (String, String) - fsend pid msg - relay fsend pid - - forward' pid = - receiveWait [ matchAny (\m -> forward m pid) ] >> forward' pid - - makeRing :: Bool -> Bool -> Int -> ProcessId -> Process ProcessId - makeRing !f !u !n !pid - | n == 0 = go f u pid - | otherwise = go f u pid >>= makeRing f u (n - 1) - - go :: Bool -> Bool -> ProcessId -> Process ProcessId - go False False next = spawnLocal $ relay send next - go False True next = spawnLocal $ relay unsafeSend next - go True _ next = spawnLocal $ forward' next - - collect :: Int -> Process () - collect !n - | n == 0 = return () - | otherwise = do - receiveWait [ - matchIf (\(a, b) -> a == "foobar" && b == "baz") - (\_ -> return ()) - , matchAny (\_ -> error "unexpected input!") - ] - collect (n - 1) - -defaultOptions :: Options -defaultOptions = Options - { optRingSize = 10 - , optIterations = 100 - , optForward = False - , optParallel = False - , optUnsafe = False - } - -options :: [OptDescr (Options -> Options)] -options = - [ Option ['s'] ["ring-size"] (OptArg optSz "SIZE") "# of processes in ring" - , Option ['i'] ["iterations"] (OptArg optMsgCnt "ITER") "# of times to send" - , Option ['f'] ["forward"] - (NoArg (\opts -> opts { optForward = True })) - "use `forward' instead of send - default = False" - , Option ['u'] ["unsafe-send"] - (NoArg (\opts -> opts { optUnsafe = True })) - "use 'unsafeSend' (ignored with -f) - default = False" - , Option ['p'] ["parallel"] - (NoArg (\opts -> opts { optParallel = True })) - "send in parallel and consume sequentially - default = False" - ] - -optMsgCnt :: Maybe String -> Options -> Options -optMsgCnt Nothing opts = opts -optMsgCnt (Just c) opts = opts { optIterations = ((read c) :: Int) } - -optSz :: Maybe String -> Options -> Options -optSz Nothing opts = opts -optSz (Just s) opts = opts { optRingSize = ((read s) :: Int) } - -parseArgv :: [String] -> IO (Options, [String]) -parseArgv argv = do - pn <- getProgName - case getOpt Permute options argv of - (o,n,[] ) -> return (foldl (flip id) defaultOptions o, n) - (_,_,errs) -> ioError (userError (concat errs ++ usageInfo (header pn) options)) - where header pn' = "Usage: " ++ pn' ++ " [OPTION...]" - -main :: IO () -main = do - argv <- getArgs - (opt, _) <- parseArgv argv - putStrLn $ "options: " ++ (show opt) - Right transport <- createTransport - (defaultTCPAddr "127.0.0.1" "8090" ) defaultTCPParameters - node <- newLocalNode transport initRemoteTable - catch (void $ runProcess node $ initialProcess opt) - (\(e :: SomeException) -> putStrLn $ "ERROR: " ++ (show e)) diff --git a/packages/distributed-process/benchmarks/Spawns.hs b/packages/distributed-process/benchmarks/Spawns.hs deleted file mode 100644 index 6ae56a2ac..000000000 --- a/packages/distributed-process/benchmarks/Spawns.hs +++ /dev/null @@ -1,49 +0,0 @@ -{-# LANGUAGE BangPatterns #-} - --- | Like Throughput, but send every ping from a different process --- (i.e., require a lightweight connection per ping) -import System.Environment -import Control.Monad -import Control.Applicative -import Control.Distributed.Process -import Control.Distributed.Process.Node -import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters) -import Data.Binary (encode, decode) -import qualified Data.ByteString.Lazy as BSL - -counter :: Process () -counter = go 0 - where - go :: Int -> Process () - go !n = do - b <- expect - case b of - Nothing -> go (n + 1) - Just them -> send them n >> go 0 - -count :: Int -> ProcessId -> Process () -count n them = do - us <- getSelfPid - replicateM_ n . spawnLocal $ send them (Nothing :: Maybe ProcessId) - send them (Just us) - n' <- expect - liftIO $ print (n == n') - -initialProcess :: String -> Process () -initialProcess "SERVER" = do - us <- getSelfPid - liftIO $ BSL.writeFile "counter.pid" (encode us) - counter -initialProcess "CLIENT" = do - n <- liftIO $ getLine - them <- liftIO $ decode <$> BSL.readFile "counter.pid" - count (read n) them - -main :: IO () -main = do - [role, host, port] <- getArgs - trans <- createTransport (defaultTCPAddr host port) defaultTCPParameters - case trans of - Right transport -> do node <- newLocalNode transport initRemoteTable - runProcess node $ initialProcess role - Left other -> error $ show other diff --git a/packages/distributed-process/benchmarks/Throughput.hs b/packages/distributed-process/benchmarks/Throughput.hs deleted file mode 100644 index 3f80a140f..000000000 --- a/packages/distributed-process/benchmarks/Throughput.hs +++ /dev/null @@ -1,74 +0,0 @@ -{-# LANGUAGE BangPatterns #-} -{-# LANGUAGE DeriveDataTypeable #-} - -import System.Environment -import Control.Monad -import Control.Applicative -import Control.Distributed.Process -import Control.Distributed.Process.Node -import Network.Transport.TCP (createTransport, defaultTCPParameters, defaultTCPAddr) -import Data.Binary -import qualified Data.ByteString.Lazy as BSL -import Data.Typeable - -data SizedList a = SizedList { size :: Int , elems :: [a] } - deriving (Typeable) - -instance Binary a => Binary (SizedList a) where - put (SizedList sz xs) = put sz >> mapM_ put xs - get = do - sz <- get - xs <- getMany sz - return (SizedList sz xs) - --- Copied from Data.Binary -getMany :: Binary a => Int -> Get [a] -getMany = go [] - where - go xs 0 = return $! reverse xs - go xs i = do x <- get - x `seq` go (x:xs) (i-1) -{-# INLINE getMany #-} - -nats :: Int -> SizedList Int -nats = \n -> SizedList n (aux n) - where - aux 0 = [] - aux n = n : aux (n - 1) - -counter :: Process () -counter = go 0 - where - go :: Int -> Process () - go !n = - receiveWait - [ match $ \xs -> go (n + size (xs :: SizedList Int)) - , match $ \them -> send them n >> go 0 - ] - -count :: (Int, Int) -> ProcessId -> Process () -count (packets, sz) them = do - us <- getSelfPid - replicateM_ packets $ send them (nats sz) - send them us - n' <- expect - liftIO $ print (packets * sz, n' == packets * sz) - -initialProcess :: String -> Process () -initialProcess "SERVER" = do - us <- getSelfPid - liftIO $ BSL.writeFile "counter.pid" (encode us) - counter -initialProcess "CLIENT" = do - n <- liftIO getLine - them <- liftIO $ decode <$> BSL.readFile "counter.pid" - count (read n) them - -main :: IO () -main = do - [role, host, port] <- getArgs - trans <- createTransport (defaultTCPAddr host port) defaultTCPParameters - case trans of - Right transport -> do node <- newLocalNode transport initRemoteTable - runProcess node $ initialProcess role - Left other -> error $ show other diff --git a/packages/distributed-process/distributed-process.cabal b/packages/distributed-process/distributed-process.cabal index bad340211..2e0aee8b3 100644 --- a/packages/distributed-process/distributed-process.cabal +++ b/packages/distributed-process/distributed-process.cabal @@ -21,7 +21,7 @@ Description: This is an implementation of Cloud Haskell, as described in You will probably also want to install a Cloud Haskell backend such as distributed-process-simplelocalnet. -tested-with: GHC==8.10.7 GHC==9.0.2 GHC==9.2.8 GHC==9.4.8 GHC==9.6.7 GHC==9.8.4 GHC==9.10.3 GHC==9.12.2 GHC==9.14.1 GHC==9.14.1 +tested-with: GHC==8.10.7 GHC==9.0.2 GHC==9.2.8 GHC==9.4.8 GHC==9.6.7 GHC==9.8.4 GHC==9.10.3 GHC==9.12.2 GHC==9.14.1 Category: Control extra-doc-files: ChangeLog @@ -119,58 +119,19 @@ Library -- Tests are in distributed-process-test package, for convenience. -benchmark distributed-process-throughput +benchmark distributed-process-benchmarks import: warnings Type: exitcode-stdio-1.0 + Main-Is: Main.hs + HS-Source-Dirs: benchmarks Build-Depends: base >= 4.14 && < 5, - distributed-process, - network-transport-tcp >= 0.3 && <= 0.9, - bytestring >= 0.10 && < 0.13, - binary >= 0.8 && < 0.10 - Main-Is: benchmarks/Throughput.hs - default-language: Haskell2010 - -benchmark distributed-process-latency - import: warnings - Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.14 && < 5, - distributed-process, - network-transport-tcp >= 0.3 && <= 0.9, - bytestring >= 0.10 && < 0.13, - binary >= 0.8 && < 0.10 - Main-Is: benchmarks/Latency.hs - default-language: Haskell2010 - -benchmark distributed-process-channels - import: warnings - Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.14 && < 5, - distributed-process, - network-transport-tcp >= 0.3 && <= 0.9, - bytestring >= 0.10 && < 0.13, - binary >= 0.8 && < 0.10 - Main-Is: benchmarks/Channels.hs - default-language: Haskell2010 - -benchmark distributed-process-spawns - import: warnings - Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.14 && < 5, - distributed-process, - network-transport-tcp >= 0.3 && <= 0.9, + binary >= 0.8 && < 0.10, bytestring >= 0.10 && < 0.13, - binary >= 0.8 && < 0.10 - Main-Is: benchmarks/Spawns.hs - default-language: Haskell2010 - -benchmark distributed-process-ring - import: warnings - Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.14 && < 5, distributed-process, + exceptions >= 0.10, + network-transport >= 0.4.1.0 && < 0.6, network-transport-tcp >= 0.3 && <= 0.9, - bytestring >= 0.10 && < 0.13, - binary >= 0.8 && < 0.10 - Main-Is: benchmarks/ProcessRing.hs + stm >= 2.4 && < 2.6, + tasty-bench >= 0.3.4 && < 0.6 default-language: Haskell2010 ghc-options: -threaded -O2 -rtsopts From 4b7c37ae5ed0834173ed4f2aa5940291798e6d59 Mon Sep 17 00:00:00 2001 From: Laurent Date: Fri, 18 Sep 2026 12:20:38 -0400 Subject: [PATCH 2/2] Run benchmarks in CI --- .github/workflows/cabal.yml | 44 ++++++++++++++++++++++++++++++++++++- cabal.optimized.project | 8 +++++++ 2 files changed, 51 insertions(+), 1 deletion(-) create mode 100644 cabal.optimized.project diff --git a/.github/workflows/cabal.yml b/.github/workflows/cabal.yml index 500ba498c..20207e383 100644 --- a/.github/workflows/cabal.yml +++ b/.github/workflows/cabal.yml @@ -10,7 +10,7 @@ on: branches: ['master'] jobs: - continuous-integration: + tests: # You can skip continuous integration by writing '[ci skip]' or '[skip ci]' in a commit message, # which is useful to preserve computing resources # @@ -83,3 +83,45 @@ jobs: timeout-minutes: 10 # We run each test suite one-by-one to better observe problems. run: cabal test all -j1 + + benchmark: + # You can skip continuous integration by writing '[ci skip]' or '[skip ci]' in a commit message, + # which is useful to preserve computing resources + # + # For example: + # > git commit -am "[skip ci] fixed x y z" + if: contains(toJson(github.event.commits), '[ci skip]') == false && contains(toJson(github.event.commits), '[skip ci]') == false + runs-on: ubuntu-latest + env: + # See issue 479 for when we can move up to GHC 9.14 + GHC_VERSION: '9.12' + + steps: + - uses: actions/checkout@v4 + + - uses: haskell-actions/setup@v2 + id: setup + with: + ghc-version: ${{ env.GHC_VERSION }} + + - name: Configure build + run: | + cabal configure --disable-tests --semaphore -j + # Generate a plan.json + cabal build all --project-file=cabal.optimized.project --dry-run + + - name: Cache cabal store + uses: actions/cache@v4 + with: + path: ${{ steps.setup.outputs.cabal-store }} + key: ${{ runner.os }}-ghc-${{ env.GHC_VERSION }}-benchmark-${{ hashFiles('**/plan.json') }} + restore-keys: | + ${{ runner.os }}-ghc-${{ env.GHC_VERSION }}-benchmark + + - name: Build + run: cabal build all --project-file=cabal.optimized.project + + # Run benchmarks with j1 so that we don't get interleaved output + - name: Run benchmarks + run: | + cabal bench all -j1 --project-file=cabal.optimized.project \ No newline at end of file diff --git a/cabal.optimized.project b/cabal.optimized.project new file mode 100644 index 000000000..5bbc5248d --- /dev/null +++ b/cabal.optimized.project @@ -0,0 +1,8 @@ +import: cabal.project + +-- Turns on optimizations for local packages +-- and also for third-party packages +optimization: 2 + +package * + optimization: 2 \ No newline at end of file