Finishing touches on ObjectDiffusion protocol internals

Author: tbagrel1

cabal.project

@@ -59,8 +59,8 @@ allow-newer:
 source-repository-package
    type: git
    location: https://github.com/IntersectMBO/ouroboros-network
-   tag: 9f94642533c2fbfae0852741f5c9ecd7f3776505
-   --sha256: 1pb36hlazh0b94x0kp8s2gk2sbknqk2q86yz28d4jsqvg36bfhk9
+   tag: c2e936f454a0026b9a854e5f230714de81b9965c
+   --sha256: sha256-139VtT1VJkBqIcqf+vak7h4Fh+Z748dHoHwaCCpKOy4=
    subdir:
      ouroboros-network
      ouroboros-network-protocols

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

@@ -309,7 +309,7 @@ mkHandlers
       , hPerasCertDiffusionInbound = \version peer ->
           objectDiffusionInbound
             (contramap (TraceLabelPeer peer) (Node.certDiffusionInboundTracer tracers))
-            ( perasCertDiffusionMaxFifoLength  miniProtocolParameters
+            ( perasCertDiffusionMaxFifoLength miniProtocolParameters
             , 10 -- TODO: change this to a sensible value
             , 10 -- TODO: change this to a sensible value
             )

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

@@ -3,8 +3,8 @@ module Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.ObjectPool.API
   , ObjectPoolWriter (..)
   ) where
 
-import Data.Word (Word64)
 import Control.Concurrent.Class.MonadSTM.Strict (STM)
+import Data.Word (Word64)
 
 data ObjectPoolReader objectId object ticketNo m
   = ObjectPoolReader

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/MiniProtocol/ObjectDiffusion/ObjectPool/API.hs

@@ -27,10 +27,11 @@ makePerasCertPoolReader getCertSnapshot =
     , oprZeroTicketNo = PerasCertDB.zeroPerasCertTicketNo
     , oprObjectsAfter = \lastKnown limit -> do
         certSnapshot <- getCertSnapshot
-        pure $ take (fromIntegral limit) $
-                [ (ticketNo, perasCertRound cert, pure cert)
-                | (cert, ticketNo) <- PerasCertDB.getCertsAfter certSnapshot lastKnown
-                ]
+        pure $
+          take (fromIntegral limit) $
+            [ (ticketNo, perasCertRound cert, pure cert)
+            | (cert, ticketNo) <- PerasCertDB.getCertsAfter certSnapshot lastKnown
+            ]
     }
 
 makePerasCertPoolReaderFromCertDB ::

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/MiniProtocol/ObjectDiffusion/ObjectPool/PerasCert.hs

@@ -12,13 +12,14 @@ module Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.Outbound
   ) where
 
 import Control.Exception (assert)
-import Control.Monad (unless, when, forM)
+import Control.Monad (forM, unless, when)
 import Control.Monad.Class.MonadSTM
 import Control.Monad.Class.MonadThrow
 import Control.Tracer (Tracer, traceWith)
 import Data.List.NonEmpty qualified as NonEmpty
 import Data.Sequence.Strict (StrictSeq)
 import Data.Sequence.Strict qualified as Seq
+import Data.Set qualified as Set
 import Ouroboros.Consensus.MiniProtocol.ObjectDiffusion.ObjectPool.API
 import Ouroboros.Network.ControlMessage
   ( ControlMessage
@@ -28,16 +29,14 @@ import Ouroboros.Network.ControlMessage
 import Ouroboros.Network.NodeToNode.Version (NodeToNodeVersion)
 import Ouroboros.Network.Protocol.ObjectDiffusion.Outbound
 import Ouroboros.Network.Protocol.ObjectDiffusion.Type
-import qualified Data.Set as Set
+
+-- Note: This module is inspired from TxSubmission outbound side.
 
 data TraceObjectDiffusionOutbound objectId object
-  = 
-    TraceObjectDiffusionOutboundRecvMsgRequestObjectIds NumObjectIdsReq
-  | 
-    -- | The IDs to be sent in the response
+  = TraceObjectDiffusionOutboundRecvMsgRequestObjectIds NumObjectIdsReq
+  | -- | The IDs to be sent in the response
     TraceObjectDiffusionOutboundSendMsgReplyObjectIds [objectId]
-  |
-    -- | The IDs of the objects requested.
+  | -- | The IDs of the objects requested.
     TraceObjectDiffusionOutboundRecvMsgRequestObjects
       [objectId]
   | -- | The objects to be sent in the response.
@@ -100,20 +99,27 @@ objectDiffusionOutbound tracer maxFifoLength ObjectPoolReader{..} _version contr
   makeBundle :: OutboundSt objectId object ticketNo -> OutboundStIdle objectId object m ()
   makeBundle !st =
     OutboundStIdle
-    { recvMsgRequestObjectIds = recvMsgRequestObjectIds st
-    , recvMsgRequestObjects = recvMsgRequestObjects st}
+      { recvMsgRequestObjectIds = recvMsgRequestObjectIds st
+      , recvMsgRequestObjects = recvMsgRequestObjects st
+      }
 
-  updateStNewObjects :: OutboundSt objectId object ticketNo -> [(object, ticketNo)] -> OutboundSt objectId object ticketNo
+  updateStNewObjects ::
+    OutboundSt objectId object ticketNo ->
+    [(object, ticketNo)] ->
+    OutboundSt objectId object ticketNo
   updateStNewObjects !OutboundSt{..} newObjectsWithTicketNos =
     -- These objects should all be fresh
-      assert (all (\(_, ticketNo) -> ticketNo > lastTicketNo) newObjectsWithTicketNos) $
-        let !outstandingFifo' =
-                  outstandingFifo
-                    <> (Seq.fromList $ fst <$> newObjectsWithTicketNos)
-            !lastTicketNo'
-                  | null newObjectsWithTicketNos = lastTicketNo
-                  | otherwise = snd $ last newObjectsWithTicketNos
-        in OutboundSt{ outstandingFifo = outstandingFifo', lastTicketNo = lastTicketNo' }
+    assert (all (\(_, ticketNo) -> ticketNo > lastTicketNo) newObjectsWithTicketNos) $
+      let !outstandingFifo' =
+            outstandingFifo
+              <> (Seq.fromList $ fst <$> newObjectsWithTicketNos)
+          !lastTicketNo'
+            | null newObjectsWithTicketNos = lastTicketNo
+            | otherwise = snd $ last newObjectsWithTicketNos
+       in OutboundSt
+            { outstandingFifo = outstandingFifo'
+            , lastTicketNo = lastTicketNo'
+            }
 
   recvMsgRequestObjectIds ::
     forall blocking.
@@ -136,14 +142,15 @@ objectDiffusionOutbound tracer maxFifoLength ObjectPoolReader{..} _version contr
       )
       $ throwIO (ProtocolErrorRequestedTooManyObjectIds numIdsToReq maxFifoLength)
 
-    -- First we update our FIFO to remove the number of objectIds
-    -- that the inbound peer has acknowledged.
+    -- First we update our FIFO to remove the number of objectIds that the
+    -- inbound peer has acknowledged.
     let !outstandingFifo' = Seq.drop (fromIntegral numIdsToAck) outstandingFifo
-        st' :: OutboundSt objectId object ticketNo  -- must specify the type here otherwise GHC complains about mismatch objectId types
-        !st' = st{ outstandingFifo = outstandingFifo' }
+        -- must specify the type here otherwise GHC complains about mismatch objectId types
+        st' :: OutboundSt objectId object ticketNo
+        !st' = st{outstandingFifo = outstandingFifo'}
 
-    -- Grab info about any new objects after the last object ticketNo we've seen,
-    -- up to the number that the peer has requested.
+    -- Grab info about any new objects after the last object ticketNo we've
+    -- seen, up to the number that the peer has requested.
     case blocking of
       -----------------------------------------------------------------------
       SingBlocking -> do
@@ -154,28 +161,34 @@ objectDiffusionOutbound tracer maxFifoLength ObjectPoolReader{..} _version contr
 
         mbNewContent <- timeoutWithControlMessage controlMessageSTM $
           do
-            newObjectsWithTicketNos <- oprObjectsAfter lastTicketNo (fromIntegral numIdsToReq)
+            newObjectsWithTicketNos <-
+              oprObjectsAfter
+                lastTicketNo
+                (fromIntegral numIdsToReq)
             check (not $ null newObjectsWithTicketNos)
             pure newObjectsWithTicketNos
 
         case mbNewContent of
           Nothing -> pure (SendMsgDone ())
           Just newContent -> do
-            newObjectsWithTicketNos <- forM newContent $ \(ticketNo, _, getObject) -> do
-              object <- getObject
-              pure (object, ticketNo)
-            
-          
+            newObjectsWithTicketNos <- forM newContent $
+              \(ticketNo, _, getObject) -> do
+                object <- getObject
+                pure (object, ticketNo)
+
             let !newIds = oprObjectId . fst <$> newObjectsWithTicketNos
                 st'' = updateStNewObjects st' newObjectsWithTicketNos
-            
+
             traceWith tracer (TraceObjectDiffusionOutboundSendMsgReplyObjectIds newIds)
 
-            -- Assert objects is non-empty: we blocked until objects was non-null,
-            -- and we know reqNo > 0, hence `take reqNo objects` is non-null.
+            -- Assert objects is non-empty: we blocked until objects was
+            -- non-null, and we know numIdsToReq > 0, hence
+            -- `take numIdsToReq objects` is non-null.
             assert (not $ null newObjectsWithTicketNos) $
-              pure (SendMsgReplyObjectIds (BlockingReply (NonEmpty.fromList $ newIds)) (makeBundle st''))
-                
+              pure $
+                SendMsgReplyObjectIds
+                  (BlockingReply (NonEmpty.fromList $ newIds))
+                  (makeBundle st'')
 
       -----------------------------------------------------------------------
       SingNonBlocking -> do
@@ -184,14 +197,17 @@ objectDiffusionOutbound tracer maxFifoLength ObjectPoolReader{..} _version contr
         when (Seq.null outstandingFifo') $
           throwIO ProtocolErrorRequestNonBlocking
 
-        newContent <- atomically $ oprObjectsAfter lastTicketNo (fromIntegral numIdsToReq)
-        newObjectsWithTicketNos <- forM newContent $ \(ticketNo, _, getObject) -> do
-              object <- getObject
-              pure (object, ticketNo)
+        newContent <-
+          atomically $
+            oprObjectsAfter lastTicketNo (fromIntegral numIdsToReq)
+        newObjectsWithTicketNos <- forM newContent $
+          \(ticketNo, _, getObject) -> do
+            object <- getObject
+            pure (object, ticketNo)
 
         let !newIds = oprObjectId . fst <$> newObjectsWithTicketNos
             st'' = updateStNewObjects st' newObjectsWithTicketNos
-        
+
         traceWith tracer (TraceObjectDiffusionOutboundSendMsgReplyObjectIds newIds)
 
         pure (SendMsgReplyObjectIds (NonBlockingReply newIds) (makeBundle st''))
@@ -203,22 +219,31 @@ objectDiffusionOutbound tracer maxFifoLength ObjectPoolReader{..} _version contr
   recvMsgRequestObjects !st@OutboundSt{..} requestedIds = do
     traceWith tracer (TraceObjectDiffusionOutboundRecvMsgRequestObjects requestedIds)
 
-    -- All the objects correspond to advertised objectIds are already in the outstandingFifo.
-    -- So we don't need to read from the object pool here.
+    -- All the objects correspond to advertised objectIds are already in the
+    -- outstandingFifo. So we don't need to read from the object pool here.
 
-    -- TODO: I've improved the search to do only one traversal of 'outstandingFifo'.
+    -- I've optimized the search to do only one traversal of 'outstandingFifo'.
     -- When the 'requestedIds' is exactly the whole 'outstandingFifo', then this
     -- should take O(n * log n) time.
-    -- We will need to revisit the underlying outstandingFifo data structure and search if
-    -- performance isn't sufficient when we'll use ObjectDiffusion for votes (and not just cert diffusion).
+    --
+    -- TODO: We might need to revisit the underlying 'outstandingFifo' data
+    -- structure and the search if performance isn't sufficient when we'll use
+    -- ObjectDiffusion for votes diffusion (and not just cert diffusion).
+
     let requestedIdsSet = Set.fromList requestedIds
 
     when (Set.size requestedIdsSet /= length requestedIds) $
       throwIO ProtocolErrorRequestedDuplicateObject
 
-    let requestedObjects = foldr (\obj acc -> if Set.member (oprObjectId obj) requestedIdsSet
-                                                  then obj : acc
-                                                  else acc) [] outstandingFifo
+    let requestedObjects =
+          foldr
+            ( \obj acc ->
+                if Set.member (oprObjectId obj) requestedIdsSet
+                  then obj : acc
+                  else acc
+            )
+            []
+            outstandingFifo
 
     when (Set.size requestedIdsSet /= length requestedObjects) $
       throwIO ProtocolErrorRequestedUnavailableObject

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/MiniProtocol/ObjectDiffusion/Outbound.hs

@@ -119,15 +119,18 @@ prop_smoke (ListWithUniqueIds certs) =
     IOSim
       s
       ( ObjectPoolReader PerasRoundNo (PerasCert TestBlock) PerasCertTicketNo (IOSim s)
-      , ObjectPoolReader PerasRoundNo (PerasCert TestBlock) PerasCertTicketNo (IOSim s)
       , ObjectPoolWriter PerasRoundNo (PerasCert TestBlock) (IOSim s)
+      , (IOSim s) [PerasCert TestBlock]
       )
   mkPoolInterfaces = do
     outboundPool <- newCertDB certs
     inboundPool <- newCertDB []
 
     let outboundPoolReader = makePerasCertPoolReaderFromCertDB outboundPool
-        inboundPoolReader = makePerasCertPoolReaderFromCertDB inboundPool
         inboundPoolWriter = makePerasCertPoolWriterFromCertDB inboundPool
+        getAllInboundPoolContent = do
+          snap <- atomically $ PerasCertDB.getCertSnapshot inboundPool
+          let rawContent = PerasCertDB.getCertsAfter snap (PerasCertDB.zeroPerasCertTicketNo)
+          pure $ fst <$> rawContent
 
-    return (outboundPoolReader, inboundPoolReader, inboundPoolWriter)
+    return (outboundPoolReader, inboundPoolWriter, getAllInboundPoolContent)