Minimize message decodes

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)

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

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 <laurent.decotret@outlook.com> 0.7.10
 
 * Improved the performance of a process death, as it were, to not scale with the

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.

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

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 ()