[WIP] Experimental ObjectDiffusion v2 (just inbound side changes) for efficient vote retrieval

cabal.project

@@ -59,8 +59,8 @@ allow-newer:
 source-repository-package
    type: git
    location: https://github.com/IntersectMBO/ouroboros-network
-   tag: peras-staging/pr-5202
-   --sha256: sha256-nTbjunQaqt6/syzSKw24Lne50083dI2SZFirG2/1T9U=
+   tag: 0db8669b67982cba755e80bf2e413527def41244
+   --sha256: sha256-vEO721Xab0RTVKFQFKal5VCV5y+OUzELo8+7Z8TETJQ=
    subdir:
      ouroboros-network
      ouroboros-network-protocols

ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Network/NodeToNode.hs

@@ -70,8 +70,8 @@ import Ouroboros.Consensus.MiniProtocol.ChainSync.Client
   )
 import qualified Ouroboros.Consensus.MiniProtocol.ChainSync.Client as CsClient
 import Ouroboros.Consensus.MiniProtocol.ChainSync.Server
-import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound (objectDiffusionInbound)
-import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.State
+import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V1 (objectDiffusionInbound)
+import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V1.State
   ( ObjectDiffusionInboundStateView
   , bracketObjectDiffusionInbound
   )

ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Node/GSM/PeerState.hs

@@ -17,7 +17,7 @@ import Ouroboros.Consensus.MiniProtocol.ChainSync.Client.State
   , ChainSyncClientHandleCollection (..)
   , ChainSyncState (..)
   )
-import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.State
+import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V1.State
   ( ObjectDiffusionInboundHandle (..)
   , ObjectDiffusionInboundHandleCollection (..)
   , ObjectDiffusionInboundState (..)

ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/NodeKernel.hs

@@ -82,7 +82,7 @@ import Ouroboros.Consensus.MiniProtocol.ChainSync.Client.HistoricityCheck
 import Ouroboros.Consensus.MiniProtocol.ChainSync.Client.InFutureCheck
   ( SomeHeaderInFutureCheck
   )
-import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.State
+import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V1.State
   ( ObjectDiffusionInboundHandleCollection (..)
   , newObjectDiffusionInboundHandleCollection
   )

ouroboros-consensus/ouroboros-consensus.cabal

@@ -191,8 +191,14 @@ library
     Ouroboros.Consensus.MiniProtocol.LocalStateQuery.Server
     Ouroboros.Consensus.MiniProtocol.LocalTxMonitor.Server
     Ouroboros.Consensus.MiniProtocol.LocalTxSubmission.Server
-    Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound
-    Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.State
+    Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V1
+    Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V1.State
+    Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V2
+    Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V2.Decision
+    Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V2.Policy
+    Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V2.Registry
+    Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V2.State
+    Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V2.Types
     Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.ObjectPool.API
     Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.ObjectPool.PerasCert
     Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Outbound
@@ -354,6 +360,7 @@ library
     primitive,
     psqueues ^>=0.2.3,
     quiet ^>=0.2,
+    random,
     rawlock ^>=0.1.1,
     resource-registry ^>=0.1,
     semialign >=1.1,

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/MiniProtocol/ObjectDiffusion/Inbound.hs → ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/MiniProtocol/ObjectDiffusion/Inbound/V1.hs

@@ -11,7 +11,7 @@
 {-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 
-module Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound
+module Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V1
   ( objectDiffusionInbound
   , TraceObjectDiffusionInbound (..)
   , ObjectDiffusionInboundError (..)
@@ -38,7 +38,7 @@ import Data.Word (Word64)
 import GHC.Generics (Generic)
 import Network.TypedProtocol.Core (N (Z), Nat (..), natToInt)
 import NoThunks.Class (NoThunks (..), unsafeNoThunks)
-import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.State
+import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V1.State
   ( ObjectDiffusionInboundStateView (..)
   )
 import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.ObjectPool.API
@@ -146,7 +146,7 @@ objectDiffusionInbound
   _version
   controlMessageSTM
   state =
-    ObjectDiffusionInboundPipelined $ do
+    ObjectDiffusionInboundPipelined $
       continueWithStateM (go Zero) initialInboundSt
    where
     canRequestMoreObjects :: InboundSt k object -> Bool
@@ -320,9 +320,10 @@ objectDiffusionInbound
         -- request.
         let st' = st{numIdsInFlight = numIdsInFlight st - numIdsRequested}
         poolHasObject <- atomically $ opwHasObject
-        continueWithStateM
-          (go n)
-          (preAcknowledge st' poolHasObject collectedIds)
+        pure $
+          continueWithStateM
+            (go n)
+            (preAcknowledge st' poolHasObject collectedIds)
       CollectObjects requestedIds collectedObjects -> do
         let requestedIdsSet = Set.fromList requestedIds
             obtainedIdsSet = Set.fromList (opwObjectId <$> collectedObjects)
@@ -368,15 +369,16 @@ objectDiffusionInbound
         traceWith tracer $
           TraceObjectDiffusionProcessed
             (NumObjectsProcessed (fromIntegral $ length objectsToAck))
-        continueWithStateM
-          (go n)
-          st
-            { pendingObjects = pendingObjects''
-            , outstandingFifo = outstandingFifo'
-            , numToAckOnNextReq =
-                numToAckOnNextReq st
-                  + fromIntegral (Seq.length objectIdsToAck)
-            }
+        pure $
+          continueWithStateM
+            (go n)
+            st
+              { pendingObjects = pendingObjects''
+              , outstandingFifo = outstandingFifo'
+              , numToAckOnNextReq =
+                  numToAckOnNextReq st
+                    + fromIntegral (Seq.length objectIdsToAck)
+              }
 
     goReqObjectIdsBlocking :: Stateful (InboundSt objectId object) 'Z objectId object m
     goReqObjectIdsBlocking = Stateful $ \st -> do
@@ -392,20 +394,21 @@ objectDiffusionInbound
         $ SendMsgRequestObjectIdsBlocking
           (numToAckOnNextReq st)
           numIdsToRequest
-          ( \neCollectedIds -> do
+          ( \neCollectedIds -> WithEffect $ do
               -- We just got some new object id's, so we are no longer idling
               --
               -- NOTE this change of state should be made explicit:
               -- https://github.com/tweag/cardano-peras/issues/144
               Idling.idlingStop (odisvIdling state)
               traceWith tracer TraceObjectInboundStoppedIdling
-              collectAndContinueWithState
-                (goCollect Zero)
-                st
-                  { numToAckOnNextReq = 0
-                  , numIdsInFlight = numIdsToRequest
-                  }
-                (CollectObjectIds numIdsToRequest (NonEmpty.toList neCollectedIds))
+              pure $
+                collectAndContinueWithState
+                  (goCollect Zero)
+                  st
+                    { numToAckOnNextReq = 0
+                    , numIdsInFlight = numIdsToRequest
+                    }
+                  (CollectObjectIds numIdsToRequest (NonEmpty.toList neCollectedIds))
           )
 
     goReqObjectsAndObjectIdsPipelined ::
@@ -433,7 +436,7 @@ objectDiffusionInbound
       let numIdsToRequest = numIdsToReq st
 
       if numIdsToRequest <= 0
-        then continueWithStateM (go n) st
+        then pure $ continueWithStateM (go n) st
         else
           pure $
             SendMsgRequestObjectIdsPipelined
@@ -454,8 +457,8 @@ objectDiffusionInbound
     terminateAfterDrain ::
       Nat n -> InboundStIdle n objectId object m ()
     terminateAfterDrain = \case
-      Zero -> SendMsgDone (pure ())
-      Succ n -> CollectPipelined Nothing $ \_ignoredMsg -> pure $ terminateAfterDrain n
+      Zero -> SendMsgDone ()
+      Succ n -> CollectPipelined Nothing $ \_ignoredMsg -> terminateAfterDrain n
 
 -------------------------------------------------------------------------------
 -- Utilities to deal with stateful continuations (copied from TX-submission)
@@ -487,9 +490,9 @@ continueWithStateM ::
   NoThunks s =>
   StatefulM s n objectId object m ->
   s ->
-  m (InboundStIdle n objectId object m ())
+  InboundStIdle n objectId object m ()
 continueWithStateM (StatefulM f) !st =
-  checkInvariant (show <$> unsafeNoThunks st) (f st)
+  checkInvariant (show <$> unsafeNoThunks st) (WithEffect $! f st)
 {-# NOINLINE continueWithStateM #-}
 
 -- | A variant of 'continueWithState' to be more easily utilized with
@@ -499,7 +502,7 @@ collectAndContinueWithState ::
   StatefulCollect s n objectId object m ->
   s ->
   Collect objectId object ->
-  m (InboundStIdle n objectId object m ())
+  InboundStIdle n objectId object m ()
 collectAndContinueWithState (StatefulCollect f) !st c =
-  checkInvariant (show <$> unsafeNoThunks st) (f st c)
+  checkInvariant (show <$> unsafeNoThunks st) (WithEffect $! f st c)
 {-# NOINLINE collectAndContinueWithState #-}

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/MiniProtocol/ObjectDiffusion/Inbound/State.hs → ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/MiniProtocol/ObjectDiffusion/Inbound/V1/State.hs

@@ -6,7 +6,7 @@
 {-# LANGUAGE StandaloneDeriving #-}
 {-# LANGUAGE UndecidableInstances #-}
 
-module Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.State
+module Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V1.State
   ( ObjectDiffusionInboundState (..)
   , initObjectDiffusionInboundState
   , ObjectDiffusionInboundHandle (..)

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/MiniProtocol/ObjectDiffusion/Inbound/V2.hs

@@ -0,0 +1,175 @@
+{-# LANGUAGE BlockArguments #-}
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE ImportQualifiedPost #-}
+{-# LANGUAGE KindSignatures #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V2
+  ( -- * ObjectDiffusion Inbound client
+    objectDiffusionInbound
+
+    -- * PeerStateAPI
+  , withPeer
+  , PeerStateAPI
+
+    -- * Supporting types
+  , module V2
+  , PeerDecisionChannelsVar
+  , newPeerDecisionChannelsVar
+  , DecisionPolicy (..)
+  , defaultDecisionPolicy
+  ) where
+
+import Control.Concurrent.Class.MonadSTM (MonadSTM, atomically)
+import Control.Monad.Class.MonadThrow
+import Control.Tracer (Tracer, traceWith)
+import Data.List.NonEmpty qualified as NonEmpty
+import Data.Set qualified as Set
+import Network.TypedProtocol
+import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V2.Policy
+import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V2.Registry
+import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Inbound.V2.Types as V2
+import Ouroboros.Network.ControlMessage (ControlMessage (..), ControlMessageSTM)
+import Ouroboros.Network.Protocol.ObjectDiffusion.Inbound
+
+-- TODO: Add checks and validation
+
+-- | A object-diffusion inbound side (client).
+--
+-- The steps are as follow
+-- 1. Block on next decision from the decision logic
+-- 2. Handle any available reply (`goCollect`)
+-- 3. Request new objects if possible (`goReqObjects`)
+-- 4. Request new ids (also responsible for ack) (`goReqIds`)
+-- 5. signal psaOnDecisionCompleted (as part of `goReqIds{Blocking,NonBlocking}`)
+-- And loop again
+-- We need to make sure we don't go again into `goIdle` until `psaOnDecisionCompleted` has been called
+objectDiffusionInbound ::
+  forall objectId object m.
+  ( MonadThrow m
+  , MonadSTM m
+  ) =>
+  Tracer m (TraceObjectDiffusionInbound objectId object) ->
+  ControlMessageSTM m ->
+  PeerStateAPI m objectId object ->
+  ObjectDiffusionInboundPipelined objectId object m ()
+objectDiffusionInbound
+  tracer
+  controlMessageSTM
+  PeerStateAPI
+    { psaReadDecision
+    , psaOnDecisionCompleted
+    , psaOnRequestIds
+    , psaOnRequestObjects
+    , psaOnReceiveIds
+    , psaOnReceiveObjects
+    } =
+    ObjectDiffusionInboundPipelined $ goIdle Zero
+   where
+    goIdle :: forall (n :: N). Nat n -> InboundStIdle n objectId object m ()
+    goIdle n = WithEffect $ do
+      ctrlMsg <- atomically controlMessageSTM
+      traceWith tracer $ TraceObjectDiffusionInboundReceivedControlMessage ctrlMsg
+      case ctrlMsg of
+        -- The peer selection governor is asking us to terminate the connection.
+        Terminate ->
+          pure $ terminateAfterDrain n
+        -- Otherwise, we can continue the protocol normally.
+        _continue -> do
+          -- Block on next decision.
+          decision <- psaReadDecision
+          traceWith tracer (TraceObjectDiffusionInboundReceivedDecision decision)
+          pure $ goCollect n decision
+
+    terminateAfterDrain ::
+      Nat n -> InboundStIdle n objectId object m ()
+    terminateAfterDrain = \case
+      Zero -> SendMsgDone ()
+      Succ n -> CollectPipelined Nothing $ \_ignoredMsg -> terminateAfterDrain n
+
+    goCollect :: Nat n -> PeerDecision objectId object -> InboundStIdle n objectId object m ()
+    goCollect Zero decision =
+      goReqObjects Zero decision
+    goCollect (Succ n) decision =
+      CollectPipelined
+        (Just $ goReqObjects (Succ n) decision)
+        ( \case
+            CollectObjectIds numIdsRequested ids -> WithEffect $ do
+              -- TODO: Add checks and validation
+              psaOnReceiveIds numIdsRequested ids
+              pure $ goCollect n decision
+            CollectObjects _objectIds objects -> WithEffect $ do
+              -- TODO: Add checks and validation
+              psaOnReceiveObjects objects
+              pure $ goCollect n decision
+        )
+
+    goReqObjects ::
+      Nat n ->
+      PeerDecision objectId object ->
+      InboundStIdle n objectId object m ()
+    goReqObjects n object@PeerDecision{pdObjectsToReqIds} =
+      if Set.null pdObjectsToReqIds
+        then
+          goReqIds n object
+        else WithEffect $ do
+          psaOnRequestObjects pdObjectsToReqIds
+          pure $
+            SendMsgRequestObjectsPipelined
+              (Set.toList pdObjectsToReqIds)
+              (goReqIds (Succ n) object)
+
+    goReqIds ::
+      forall (n :: N).
+      Nat n ->
+      PeerDecision objectId object ->
+      InboundStIdle n objectId object m ()
+    goReqIds n pd@PeerDecision{pdCanPipelineIdsRequests} =
+      if pdCanPipelineIdsRequests
+        then goReqIdsPipelined n pd
+        else case n of
+          Zero -> goReqIdsBlocking pd
+          Succ{} -> error "Impossible to have pipelined requests when we have no known unacknowledged objectIds"
+
+    goReqIdsBlocking ::
+      PeerDecision objectId object ->
+      InboundStIdle Z objectId object m ()
+    goReqIdsBlocking PeerDecision{pdNumIdsToAck, pdNumIdsToReq} = WithEffect $ do
+      if pdNumIdsToReq == 0
+        then do
+          psaOnDecisionCompleted
+          pure $ goIdle Zero
+        else do
+          psaOnRequestIds pdNumIdsToAck pdNumIdsToReq
+          psaOnDecisionCompleted
+          pure $
+            SendMsgRequestObjectIdsBlocking
+              pdNumIdsToAck
+              pdNumIdsToReq
+              ( \objectIds -> WithEffect $ do
+                  -- TODO: Add checks and validation
+                  psaOnReceiveIds pdNumIdsToReq (NonEmpty.toList objectIds)
+                  pure $ goIdle Zero
+              )
+
+    goReqIdsPipelined ::
+      forall (n :: N).
+      Nat n ->
+      PeerDecision objectId object ->
+      InboundStIdle n objectId object m ()
+    goReqIdsPipelined n PeerDecision{pdNumIdsToAck, pdNumIdsToReq} = WithEffect $ do
+      if pdNumIdsToReq == 0
+        then do
+          psaOnDecisionCompleted
+          pure $ goIdle n
+        else do
+          psaOnRequestIds pdNumIdsToAck pdNumIdsToReq
+          psaOnDecisionCompleted
+          pure $
+            SendMsgRequestObjectIdsPipelined
+              pdNumIdsToAck
+              pdNumIdsToReq
+              (goIdle (Succ n))