From 258fd4dccd03a30145c34e75443358c398b9e59b Mon Sep 17 00:00:00 2001 From: Laurent Date: Wed, 14 Oct 2026 13:07:04 -0400 Subject: [PATCH] Minimize message decodes --- .../Control/Distributed/Process/Tests/CH.hs | 27 ++++++++- .../Process/Tests/Internal/Utils.hs | 6 +- packages/distributed-process/ChangeLog | 6 ++ .../distributed-process/benchmarks/Main.hs | 43 ++++++++++++++ .../Distributed/Process/Internal/CQueue.hs | 45 ++++++++++----- .../Process/Internal/Primitives.hs | 57 +++++++++++-------- .../Distributed/Process/Internal/Types.hs | 15 +++-- .../src/Control/Distributed/Process/Node.hs | 2 +- 8 files changed, 154 insertions(+), 47 deletions(-) diff --git a/packages/distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs b/packages/distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs index 469eb9ae..7fe23f9e 100644 --- a/packages/distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs +++ b/packages/distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs @@ -37,15 +37,17 @@ import Control.Distributed.Process.Internal.Types ( LocalNode(localEndPoint) , ProcessExitException(..) , nullProcessId + , createMessage , createUnencodedMessage ) import Control.Distributed.Process.Node -import Control.Distributed.Process.Tests.Internal.Utils (pause) +import Control.Distributed.Process.Tests.Internal.Utils (pause, (@^?=)) import Control.Distributed.Process.Serializable (Serializable) import Data.Maybe (isNothing, isJust) import System.Timeout (timeout) import Test.Tasty (TestTree, testGroup) import Test.Tasty.HUnit (Assertion, assertBool, assertEqual, testCase, assertFailure) +import Data.Functor ((<&>)) newtype Ping = Ping ProcessId deriving (Typeable, Binary, Show) @@ -1070,6 +1072,28 @@ testMatchAnyNoHandle TestTransport{..} = do takeMVar clientDone takeMVar serverDone +-- | Check that a decoded message that fails to match any of the predicates isn't lost +testRejectedMessagesKeepPlace :: TestTransport -> Assertion +testRejectedMessagesKeepPlace TestTransport{..} = do + node <- newLocalNode testTransport initRemoteTable + + runProcess node $ do + us <- getSelfPid + forM_ [1, 2, 3 :: Int] $ \i -> uforward (createMessage i) us + + -- Wait until [1,2,3] messages are received by sending and receiving a sentinel value, () + send us () + receiveWait [match (\() -> pure ())] + + receiveTimeout 0 [matchIf (> (5 :: Int)) pure] @^?= Nothing + receiveTimeout 0 [matchIf (> (5 :: Int)) pure] @^?= Nothing + receiveTimeout 0 [matchAnyIf (> (5 :: Int)) (\_ -> pure ())] @^?= Nothing + + receiveTimeout 0 [matchIf (== (2 :: Int)) pure] @^?= Just 2 + receiveWait [matchAnyIf (== (3 :: Int)) (\m -> handleMessage m pure)] @^?= Just (3 :: Int) + expect @^?= (1 :: Int) + (receiveTimeout 0 [matchAny pure] <&> isNothing) @^?= True + -- | Test 'matchAnyIf'. We provide an /echo/ server, but it ignores requests -- unless the text body @/= "bar"@ - this case should time out rather than -- removing the message from the process mailbox. @@ -1806,6 +1830,7 @@ tests testtrans = return $ testGroup "CH" [ , testCase "MatchAnyHandle" (testMatchAnyHandle testtrans) , testCase "MatchAnyNoHandle" (testMatchAnyNoHandle testtrans) , testCase "MatchAnyIf" (testMatchAnyIf testtrans) + , testCase "RejectedMessagesKeepPlace" (testRejectedMessagesKeepPlace testtrans) , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans) , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans) , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans) diff --git a/packages/distributed-process-tests/src/Control/Distributed/Process/Tests/Internal/Utils.hs b/packages/distributed-process-tests/src/Control/Distributed/Process/Tests/Internal/Utils.hs index f5d534b8..eb6a5620 100644 --- a/packages/distributed-process-tests/src/Control/Distributed/Process/Tests/Internal/Utils.hs +++ b/packages/distributed-process-tests/src/Control/Distributed/Process/Tests/Internal/Utils.hs @@ -33,6 +33,7 @@ module Control.Distributed.Process.Tests.Internal.Utils , testProcessReport , delayedAssertion , assertComplete + , (@^?=) -- logging , Logger() , newLogger @@ -79,7 +80,7 @@ import Control.Monad.STM (atomically) import Data.Binary import Data.Typeable (Typeable) -import Test.Tasty.HUnit (Assertion, assertBool) +import Test.Tasty.HUnit (Assertion, assertBool, assertEqual) import GHC.Generics import System.Timeout (timeout) @@ -222,3 +223,6 @@ tryForkProcess :: LocalNode -> Process () -> IO ProcessId tryForkProcess node p = do tid <- liftIO myThreadId forkProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException)) + +(@^?=) :: (Eq a, Show a) => Process a -> a -> Process () +test @^?= expected = test >>= \t -> liftIO $ assertEqual "" expected t diff --git a/packages/distributed-process/ChangeLog b/packages/distributed-process/ChangeLog index 4c88a52c..19b39e6a 100644 --- a/packages/distributed-process/ChangeLog +++ b/packages/distributed-process/ChangeLog @@ -1,3 +1,9 @@ +Unreleased + +* Messages that arrive from other nodes are now decoded at most once. + This can substantially lower the overhead of matching messages, + especially with messages queueing up and with a large number of `Match`es. + 2026-10-13 Laurent P. René de Cotret 0.7.10 * Improved the performance of a process death, as it were, to not scale with the diff --git a/packages/distributed-process/benchmarks/Main.hs b/packages/distributed-process/benchmarks/Main.hs index 179c6152..d9381abd 100644 --- a/packages/distributed-process/benchmarks/Main.hs +++ b/packages/distributed-process/benchmarks/Main.hs @@ -16,6 +16,7 @@ import Control.Concurrent.STM ) import Control.Distributed.Process ( Handler (Handler), + Message, MonitorRef, NodeId, Process, @@ -86,6 +87,7 @@ import Control.Distributed.Process withMonitor_, wrapMessage, ) +import Control.Distributed.Process.Internal.Types (createMessage) import Control.Distributed.Process.Closure ( functionTDict, mkClosure, @@ -276,6 +278,11 @@ receiving fx = repsBench fx "matchIf" 1000 $ do selfSend (1 :: Int) receiveWait [matchIf (> (0 :: Int)) (\_ -> return ())], + bgroup + "rescan mailbox of 100 encoded [Int] (1000 elems)" + [ oneBench fx (show scans ++ " scans") (rescanEncoded 100 scans) + | scans <- [0, 1, 10] + ], repsBench fx "matchAny" 1000 $ do selfSend () receiveWait [matchAny (\_ -> return ())], @@ -371,6 +378,17 @@ messages fx = unsafeSend us us receiveWait [matchAny (`uforward` fxEcho fx)] expect :: Process (), + -- A message wrapped once and then serialized on every forward: the + -- encoding should be shared rather than recomputed. + bgroup + "uforward (wrapped once) to remote" + [ oneBench fx name (uforwardTo (fxRemoteEcho fx) 1000 msg) + | (name, msg) <- + [ ("100 Ints", wrapMessage [1 .. 100 :: Int]), + ("10000 Ints", wrapMessage [1 .. 10000 :: Int]), + ("64KiB ByteString", wrapMessage (BS.replicate 65536 'x')) + ] + ], repsBench fx "relay" 1000 $ do send (fxRelay fx) () expect :: Process (), @@ -759,6 +777,31 @@ sendThrough srv n payload = do when (n' /= n) $ die ("expected " ++ show n ++ " messages, server saw " ++ show n') +-- | Fills our mailbox with @k@ still-encoded messages, scans it @scans@ times +-- with a predicate that rejects all of them, then empties it. +rescanEncoded :: Int -> Int -> Process () +rescanEncoded k scans = do + us <- getSelfPid + let msg = createMessage [1 .. 1000 :: Int] + replicateM_ k (uforward msg us) + -- Local delivery is asynchronous (via the node controller, which is FIFO), so + -- wait for a marker sent afterwards: all the messages are then in the mailbox. + -- Matching on () skips them by fingerprint, without decoding. + send us () + receiveWait [match (\() -> return ())] + replicateM_ scans $ + receiveTimeout 0 [matchIf (\(xs :: [Int]) -> null xs) (\_ -> return ())] + replicateM_ k (expect :: Process [Int]) + +-- | @n@ forwards of the same message, then one round trip to confirm the +-- receiver has drained them (messages between a pair of processes are ordered). +uforwardTo :: ProcessId -> Int -> Message -> Process () +uforwardTo dest n msg = do + us <- getSelfPid + replicateM_ n (uforward msg dest) + send dest us + expect :: Process () + -- | 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. diff --git a/packages/distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs b/packages/distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs index e6646b1a..2277a106 100644 --- a/packages/distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs +++ b/packages/distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs @@ -7,6 +7,7 @@ module Control.Distributed.Process.Internal.CQueue ( CQueue , BlockSpec(..) , MatchOn(..) + , MatchResult(..) , newCQueue , enqueue , enqueueSTM @@ -74,11 +75,26 @@ data BlockSpec = | Blocking | Timeout Int -- ^ Timeout in microseconds +-- | The outcome of matching a message. +-- +-- For performance reasons, a decoded message that has not matched +-- returned 'Refined' instead of 'NoMatch'. It is expected that messages +-- are decoded at most once. +data MatchResult m a + = NoMatch + -- ^ The message does not match, and is left in the mailbox as it is. + | Matched a + -- ^ The message matches, and is removed from the mailbox. + | Refined m + -- ^ The message does not match, but matching it required decoding the message. + -- The refined message should be placed back in the mailbox. + deriving (Functor) + -- Match operations -- -- They can be either a message match or a channel match. data MatchOn m a - = MatchMsg (m -> Maybe a) + = MatchMsg (m -> MatchResult m a) | MatchChan (STM a) deriving (Functor) @@ -87,7 +103,7 @@ data MatchOn m a -- Two consecutive chunks never have the same kind of matches. i.e. if one chunk -- contains message matches then the next one must contain channel matches and -- viceversa. -type MatchChunks m a = [Either [m -> Maybe a] [STM a]] +type MatchChunks m a = [Either [m -> MatchResult m a] [STM a]] -- Splits a list of matches into chunks. -- @@ -101,7 +117,7 @@ chunkMatches (MatchChan r : ms) = Right (r : chk) : chunkMatches rest where (chk, rest) = spanMatchChan ms -- | @spanMatchMsg = first (map (\(MatchMsg x) -> x)) . span isMatchMsg@ -spanMatchMsg :: [MatchOn m a] -> ([m -> Maybe a], [MatchOn m a]) +spanMatchMsg :: [MatchOn m a] -> ([m -> MatchResult m a], [MatchOn m a]) spanMatchMsg [] = ([],[]) spanMatchMsg (m : ms) | MatchMsg msg <- m = (msg:msgs, rest) @@ -264,15 +280,16 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus goCheck1 (Left matches : rest) m old = do case checkMatches matches m of - Nothing -> goCheck1 rest m old - Just p -> returnOld old (Just (Right p)) + (m', Nothing) -> goCheck1 rest m' old -- m' may be a refinement of an encoded message, leading to at most one decode per message + (_, Just p) -> returnOld old (Just (Right p)) -- a common pattern for putting back the arrived queue at the end returnOld :: StrictList m -> Maybe (Either a a) -> IO (Maybe (Either a a)) returnOld old r = do putMVar arrived old; return r - -- as a side-effect, this left-biases the list - checkArrived :: [m -> Maybe a] -> StrictList m -> (StrictList m, Maybe a) + -- as a side-effect, this left-biases the list, and puts refined messages + -- in the place of the ones they refine + checkArrived :: [m -> MatchResult m a] -> StrictList m -> (StrictList m, Maybe a) checkArrived matches list = go list Nil where -- @go xs ys@ searches for a message match in @append xs ys@ @@ -281,13 +298,15 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus go (Append xs ys) tl = go xs (append ys tl) go (Snoc xs x) tl = go xs (Cons x tl) go (Cons x xs) tl - | Just y <- checkMatches matches x = (append xs tl, Just y) - | otherwise = let !(rest,r) = go xs tl in (Cons x rest, r) + = case checkMatches matches x of + (_, Just y) -> (append xs tl, Just y) + (x', Nothing) -> let !(rest,r) = go xs tl in (Cons x' rest, r) - checkMatches :: [m -> Maybe a] -> m -> Maybe a - checkMatches [] _ = Nothing - checkMatches (m:ms) a = case m a of Nothing -> checkMatches ms a - Just b -> Just b + checkMatches :: [m -> MatchResult m a] -> m -> (m, Maybe a) + checkMatches [] a = (a, Nothing) + checkMatches (m:ms) a = case m a of NoMatch -> checkMatches ms a + Refined a' -> checkMatches ms a' + Matched b -> (a, Just b) -- | Weak reference to a CQueue mkWeakCQueue :: CQueue a -> IO () -> IO (Weak (CQueue a)) diff --git a/packages/distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs b/packages/distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs index be07075a..9ccc01af 100644 --- a/packages/distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs +++ b/packages/distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs @@ -159,6 +159,7 @@ import Control.Distributed.Process.Internal.CQueue ( dequeue , BlockSpec(..) , MatchOn(..) + , MatchResult(..) ) import Control.Distributed.Process.Serializable (Serializable, fingerprint) import Data.Accessor ((^.), (^:), (^=)) @@ -458,32 +459,35 @@ match = matchIf (const True) matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b matchIf c p = Match $ MatchMsg $ \msg -> case messageFingerprint msg == fingerprint (undefined :: a) of - False -> Nothing + False -> NoMatch True -> case msg of - (UnencodedMessage _ m) -> + (UnencodedMessage _ m _) -> let m' = unsafeCoerce m :: a in case (c m') of - True -> Just (p m') - False -> Nothing - (EncodedMessage _ _) -> - if (c decoded) then Just (p decoded) else Nothing + True -> Matched (p m') + False -> NoMatch + (EncodedMessage fp enc) -> + if (c decoded) + then Matched (p decoded) + -- the mailbox keeps the decoded value, so as not to decode it again on the next receive (#275) + else Refined (UnencodedMessage { messageFingerprint = fp, messagePayload = decoded, messageEncoded = enc }) where decoded :: a -- Make sure the value is fully decoded so that we don't hang to -- bytestrings when the process calling 'matchIf' doesn't process -- the values immediately - !decoded = decode (messageEncoding msg) + !decoded = decode enc -- | Match against any message, regardless of the underlying (contained) type matchMessage :: (Message -> Process Message) -> Match Message -matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg) +matchMessage p = Match $ MatchMsg $ \msg -> Matched (p msg) -- | Match against any message (regardless of underlying type) that satisfies a predicate matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message matchMessageIf c p = Match $ MatchMsg $ \msg -> case (c msg) of - True -> Just (p msg) - False -> Nothing + True -> Matched (p msg) + False -> NoMatch -- | Forward a raw 'Message' to the given 'ProcessId'. forward :: Message -> ProcessId -> Process () @@ -562,7 +566,7 @@ unwrapMessage msg = case messageFingerprint msg == fingerprint (undefined :: a) of False -> return Nothing :: m (Maybe a) True -> case msg of - (UnencodedMessage _ ms) -> + (UnencodedMessage _ ms _) -> let ms' = unsafeCoerce ms :: a in return (Just ms') (EncodedMessage _ _) -> @@ -598,7 +602,7 @@ handleMessageIf msg c proc = do case messageFingerprint msg == fingerprint (undefined :: a) of False -> return Nothing :: m (Maybe b) True -> case msg of - (UnencodedMessage _ ms) -> + (UnencodedMessage _ ms _) -> let ms' = unsafeCoerce ms :: a in case (c ms') of True -> do { r <- proc ms'; return (Just r) } @@ -632,7 +636,7 @@ handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return () -- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'. -- matchAny :: forall b. (Message -> Process b) -> Match b -matchAny p = Match $ MatchMsg $ \msg -> Just (p msg) +matchAny p = Match $ MatchMsg $ \msg -> Matched (p msg) -- | Match against an arbitrary message. Intended for use with 'handleMessage' -- and 'unwrapMessage', this function /only/ removes a message from the process @@ -649,17 +653,20 @@ matchAnyIf :: forall a b. (Serializable a) -> Match b matchAnyIf c p = Match $ MatchMsg $ \msg -> case messageFingerprint msg == fingerprint (undefined :: a) of - True | check -> Just (p msg) - where - check :: Bool - !check = - case msg of - (EncodedMessage _ _) -> c decoded - (UnencodedMessage _ m') -> c (unsafeCoerce m') - - decoded :: a -- note [decoding] - !decoded = decode (messageEncoding msg) - _ -> Nothing + False -> NoMatch + True -> case msg of + (UnencodedMessage _ m' _) + | c (unsafeCoerce m') -> Matched (p msg) + | otherwise -> NoMatch + (EncodedMessage fp enc) + | c decoded -> Matched (p decodedMsg) + | otherwise -> Refined decodedMsg + where + decoded :: a -- note [decoding] + !decoded = decode enc + + decodedMsg :: Message + decodedMsg = UnencodedMessage { messageFingerprint = fp, messagePayload = decoded, messageEncoded = enc } {- note [decoding] For an EncodedMessage, we need to ensure the value is fully decoded so that @@ -670,7 +677,7 @@ succeeds) that unsafeCoerce will not fail. -- | Remove any message from the queue matchUnknown :: Process b -> Match b -matchUnknown p = Match $ MatchMsg (const (Just p)) +matchUnknown p = Match $ MatchMsg (const (Matched p)) -- | Receives messages and forwards them to @pid@ if @p msg == True@. delegate :: ProcessId -> (Message -> Bool) -> Process () diff --git a/packages/distributed-process/src/Control/Distributed/Process/Internal/Types.hs b/packages/distributed-process/src/Control/Distributed/Process/Internal/Types.hs index 21fcc706..036f72ab 100644 --- a/packages/distributed-process/src/Control/Distributed/Process/Internal/Types.hs +++ b/packages/distributed-process/src/Control/Distributed/Process/Internal/Types.hs @@ -475,17 +475,19 @@ data Message = { messageFingerprint :: !Fingerprint , messagePayload :: !a + -- | Contrary to other fields, we keep the encoded message payload deliberately lazy, so + -- that it is computed at most once + , messageEncoded :: BSL.ByteString } deriving (Typeable) instance NFData Message where rnf (EncodedMessage _ e) = rnf e `seq` () - rnf (UnencodedMessage _ a) = e `seq` () - where e = BSL.length (encode a) + rnf (UnencodedMessage _ a e) = a `seq` BSL.length e `seq` () instance Show Message where show (EncodedMessage fp enc) = show enc ++ " :: " ++ showFingerprint fp [] - show (UnencodedMessage _ uenc) = "[unencoded message] :: " ++ (show $ typeOf uenc) + show (UnencodedMessage _ uenc _) = "[unencoded message] :: " ++ (show $ typeOf uenc) -- | /internal use only/. isEncoded :: Message -> Bool @@ -506,17 +508,18 @@ createMessage a = EncodedMessage (fingerprint a) (encode a) -- | Turn any serializable term into an unencoded/local message createUnencodedMessage :: Serializable a => a -> Message createUnencodedMessage a = - let encoded = encode a in BSL.length encoded `seq` UnencodedMessage (fingerprint a) a + let encoded = encode a + in a `seq` BSL.length encoded `seq` UnencodedMessage { messageFingerprint = fingerprint a, messagePayload = a, messageEncoded = encoded } -- | Turn any serializable term into an unencodede/local message, without -- evalutaing it! This is a dangerous business. unsafeCreateUnencodedMessage :: Serializable a => a -> Message -unsafeCreateUnencodedMessage a = UnencodedMessage (fingerprint a) a +unsafeCreateUnencodedMessage a = UnencodedMessage { messageFingerprint = fingerprint a, messagePayload = a, messageEncoded = (encode a) } -- | Serialize a message messageToPayload :: Message -> [BSS.ByteString] messageToPayload (EncodedMessage fp enc) = encodeFingerprint fp : BSL.toChunks enc -messageToPayload (UnencodedMessage fp m) = messageToPayload ((EncodedMessage fp (encode m))) +messageToPayload (UnencodedMessage fp _ enc) = encodeFingerprint fp : BSL.toChunks enc -- | Deserialize a message payloadToMessage :: [BSS.ByteString] -> Message diff --git a/packages/distributed-process/src/Control/Distributed/Process/Node.hs b/packages/distributed-process/src/Control/Distributed/Process/Node.hs index c527e47d..6bf09f29 100644 --- a/packages/distributed-process/src/Control/Distributed/Process/Node.hs +++ b/packages/distributed-process/src/Control/Distributed/Process/Node.hs @@ -1047,7 +1047,7 @@ ncEffectLocalPortSend from msg = do ch <- deRefWeak chan' forM_ ch $ \chan -> deliverChan node from msg chan where deliverChan :: forall a . LocalNode -> SendPortId -> Message -> TQueue a -> IO () - deliverChan n p (UnencodedMessage _ raw) chan' = do + deliverChan n p (UnencodedMessage _ raw _) chan' = do atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a) trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw) deliverChan _ _ (EncodedMessage _ _) _ =