WIP #731

IntersectMBO · Jul 11, 2019 · 6634f76 · 6634f76
1 parent 6d31410
commit 6634f76
Show file tree

Hide file tree

Showing 5 changed files with 167 additions and 55 deletions.
diff --git a/ouroboros-consensus/src/Ouroboros/Consensus/Node.hs b/ouroboros-consensus/src/Ouroboros/Consensus/Node.hs
@@ -1,13 +1,12 @@
-{-# LANGUAGE ConstraintKinds            #-}
-{-# LANGUAGE DataKinds                  #-}
-{-# LANGUAGE DuplicateRecordFields      #-}
-{-# LANGUAGE FlexibleContexts           #-}
-{-# LANGUAGE GeneralizedNewtypeDeriving #-}
-{-# LANGUAGE NamedFieldPuns             #-}
-{-# LANGUAGE RankNTypes                 #-}
-{-# LANGUAGE RecordWildCards            #-}
-{-# LANGUAGE ScopedTypeVariables        #-}
-{-# LANGUAGE TypeFamilies               #-}
+{-# LANGUAGE ConstraintKinds       #-}
+{-# LANGUAGE DataKinds             #-}
+{-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE FlexibleContexts      #-}
+{-# LANGUAGE NamedFieldPuns        #-}
+{-# LANGUAGE RankNTypes            #-}
+{-# LANGUAGE RecordWildCards       #-}
+{-# LANGUAGE ScopedTypeVariables   #-}
+{-# LANGUAGE TypeFamilies          #-}
 
 {-# OPTIONS_GHC -Wredundant-constraints -Werror=missing-fields #-}
 
@@ -18,6 +17,8 @@ module Ouroboros.Consensus.Node (
   , NodeParams (..)
   , TraceConstraints
   , nodeKernel
+  , getMempoolReader
+  , getMempoolWriter
     -- * Auxiliary functions
   , tracePrefix
   ) where
@@ -26,6 +27,8 @@ import           Control.Monad (void)
 import           Crypto.Random (ChaChaDRG)
 import           Data.Functor.Contravariant (contramap)
 import           Data.Map.Strict (Map)
+import           Data.Maybe (isNothing)
+import           Data.Word (Word16)
 
 import           Control.Monad.Class.MonadAsync
 import           Control.Monad.Class.MonadFork (MonadFork)
@@ -39,6 +42,10 @@ import           Ouroboros.Network.Block
 import           Ouroboros.Network.BlockFetch
 import           Ouroboros.Network.BlockFetch.State (FetchMode (..))
 import qualified Ouroboros.Network.Chain as Chain
+import           Ouroboros.Network.TxSubmission.Inbound
+import           Ouroboros.Network.TxSubmission.Outbound hiding
+                     (MempoolSnapshot)
+import qualified Ouroboros.Network.TxSubmission.Outbound as Outbound
 
 import           Ouroboros.Consensus.Block
 import           Ouroboros.Consensus.BlockchainTime
@@ -119,6 +126,8 @@ data NodeParams m peer blk = NodeParams {
     , mempoolTracer      :: Tracer m (TraceEventMempool blk)
     , decisionTracer     :: Tracer m [TraceLabelPeer peer (FetchDecision [Point (Header blk)])]
     , fetchClientTracer  :: Tracer m (TraceLabelPeer peer (TraceFetchClientState (Header blk)))
+    , txInboundTracer    :: Tracer m (TraceTxSubmissionInbound  (GenTxId blk) (GenTx blk))
+    , txOutboundTracer   :: Tracer m (TraceTxSubmissionOutbound (GenTxId blk) (GenTx blk))
     , threadRegistry     :: ThreadRegistry m
     , maxClockSkew       :: ClockSkew
     , cfg                :: NodeConfig (BlockProtocol blk)
@@ -128,6 +137,7 @@ data NodeParams m peer blk = NodeParams {
     , callbacks          :: NodeCallbacks m blk
     , blockFetchSize     :: Header blk -> SizeInBytes
     , blockMatchesHeader :: Header blk -> blk -> Bool
+    , maxUnackTxs        :: Word16
     }
 
 nodeKernel
@@ -376,3 +386,38 @@ forkBlockProduction IS{..} =
                        $ simChaChaT varDRG
                        $ id
 
+
+{-------------------------------------------------------------------------------
+  TxSubmission integration
+-------------------------------------------------------------------------------}
+
+getMempoolReader
+  :: forall m blk.
+     (MonadSTM m)
+  => Mempool m blk TicketNo
+  -> TxSubmissionMempoolReader (GenTxId blk) (GenTx blk) TicketNo m
+getMempoolReader mempool = TxSubmissionMempoolReader
+    { mempoolZeroIdx     = zeroIdx mempool
+    , mempoolGetSnapshot = convertSnapshot <$> getSnapshot mempool
+    }
+  where
+    convertSnapshot
+      :: MempoolSnapshot          (GenTxId blk) (GenTx blk) TicketNo
+      -> Outbound.MempoolSnapshot (GenTxId blk) (GenTx blk) TicketNo
+    convertSnapshot MempoolSnapshot{getTxsAfter, getTx} = Outbound.MempoolSnapshot
+      { mempoolTxIdsAfter = \idx ->
+          -- TODO we need #700
+          [(txid, idx', undefined) | (txid, _tx, idx') <- getTxsAfter idx]
+      , mempoolLookupTx   = fmap snd . getTx
+       }
+
+getMempoolWriter
+  :: (Monad m, ApplyTx blk)
+  => Mempool m blk TicketNo
+  -> TxSubmissionMempoolWriter (GenTxId blk) (GenTx blk) TicketNo m
+getMempoolWriter mempool = TxSubmissionMempoolWriter
+    { txId          = computeGenTxId
+    , mempoolAddTxs = \txs ->
+        map (computeGenTxId . fst) . filter (isNothing . snd) <$>
+        addTxs mempool txs
+    }