tx-submission: TX peer ranking

Use hashWithSalt to breaks ties when ranking peers with equal score.
Track rejections in a bucket that slowly drains over time.

Author: karknu

ouroboros-network/sim-tests-lib/Test/Ouroboros/Network/Diffusion/Node/Kernel.hs

@@ -52,7 +52,8 @@ import Data.Monoid.Synchronisation
 import Data.Void (Void)
 import GHC.Generics (Generic)
 import Numeric.Natural (Natural)
-import System.Random (RandomGen, StdGen, randomR, split)
+
+import System.Random (RandomGen, StdGen, mkStdGen, random, randomR, split)
 
 import Network.Socket (PortNumber)
 
@@ -302,23 +303,24 @@ newNodeKernel :: ( MonadSTM m
                  , Eq txid
                  )
               => s
+              -> Int
               -> [Tx txid]
               -> m (NodeKernel header block s txid m)
-newNodeKernel rng txs = do
+newNodeKernel psRng txSeed txs = do
     publicStateVar <- makePublicPeerSelectionStateVar
     NodeKernel
       <$> newTVarIO Map.empty
       <*> newTVarIO (ChainProducerState Chain.Genesis Map.empty 0)
       <*> newFetchClientRegistry
       <*> newPeerSharingRegistry
       <*> ChainDB.newChainDB
-      <*> newPeerSharingAPI publicStateVar rng
+      <*> newPeerSharingAPI publicStateVar psRng
                             ps_POLICY_PEER_SHARE_STICKY_TIME
                             ps_POLICY_PEER_SHARE_MAX_PEERS
       <*> pure publicStateVar
       <*> newMempool txs
       <*> Strict.newMVar (TxChannels Map.empty)
-      <*> newSharedTxStateVar
+      <*> newSharedTxStateVar (mkStdGen txSeed)
 
 -- | Register a new upstream chain-sync client.
 --
@@ -407,11 +409,12 @@ withNodeKernelThread
 withNodeKernelThread BlockGeneratorArgs { bgaSlotDuration, bgaBlockGenerator, bgaSeed }
                      txs
                      k = do
-    kernel <- newNodeKernel psSeed txs
+    kernel <- newNodeKernel psSeed txSeed txs
     withSlotTime bgaSlotDuration $ \waitForSlot ->
       withAsync (blockProducerThread kernel waitForSlot) (k kernel)
   where
-    (bpSeed, psSeed) = split bgaSeed
+    (bpSeed, rng) = split bgaSeed
+    (txSeed, psSeed) = random rng
 
     blockProducerThread :: NodeKernel header block seed txid m
                         -> (SlotNo -> STM m SlotNo)

ouroboros-network/sim-tests-lib/Test/Ouroboros/Network/TxSubmission/AppV2.hs

@@ -34,11 +34,13 @@ import Data.ByteString.Lazy (ByteString)
 import Data.ByteString.Lazy qualified as BSL
 import Data.Foldable (traverse_)
 import Data.Function (on)
+import Data.Hashable
 import Data.List (nubBy)
 import Data.Map.Strict (Map)
 import Data.Map.Strict qualified as Map
 import Data.Maybe (fromMaybe)
 import Data.Void (Void)
+import System.Random (mkStdGen)
 
 import Ouroboros.Network.Channel
 import Ouroboros.Network.ControlMessage (ControlMessage (..), ControlMessageSTM)
@@ -129,6 +131,7 @@ runTxSubmission
      , NoThunks (Tx txid)
      , Show peeraddr
      , Ord peeraddr
+     , Hashable peeraddr
 
      , txid ~ Int
      )
@@ -150,9 +153,10 @@ runTxSubmission tracer tracerTxLogic state txDecisionPolicy = do
         ) state
 
     inboundMempool <- emptyMempool
+    let txRng = mkStdGen 42 -- TODO
 
     txChannelsMVar <- newMVar (TxChannels Map.empty)
-    sharedTxStateVar <- newSharedTxStateVar
+    sharedTxStateVar <- newSharedTxStateVar txRng
     labelTVarIO sharedTxStateVar "shared-tx-state"
     gsvVar <- newTVarIO Map.empty
     labelTVarIO gsvVar "gsv"

ouroboros-network/sim-tests-lib/Test/Ouroboros/Network/TxSubmission/TxLogic.hs

@@ -1,13 +1,16 @@
 {-# LANGUAGE BangPatterns        #-}
 {-# LANGUAGE BlockArguments      #-}
 {-# LANGUAGE CPP                 #-}
+{-# LANGUAGE DataKinds           #-}
 {-# LANGUAGE DeriveGeneric       #-}
+{-# LANGUAGE DerivingVia         #-}
 {-# LANGUAGE FlexibleContexts    #-}
 {-# LANGUAGE FlexibleInstances   #-}
 {-# LANGUAGE GADTs               #-}
 {-# LANGUAGE LambdaCase          #-}
 {-# LANGUAGE NamedFieldPuns      #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE StandaloneDeriving  #-}
 {-# LANGUAGE TupleSections       #-}
 {-# LANGUAGE TypeOperators       #-}
 
@@ -17,6 +20,7 @@ module Test.Ouroboros.Network.TxSubmission.TxLogic where
 
 import Prelude hiding (seq)
 
+import Control.Monad.Class.MonadTime.SI (Time (..))
 import Control.Exception (assert)
 
 import Data.Foldable (
@@ -36,6 +40,7 @@ import Data.Sequence.Strict qualified as StrictSeq
 import Data.Set (Set)
 import Data.Set qualified as Set
 import Data.Typeable
+import System.Random (mkStdGen, StdGen)
 
 import NoThunks.Class
 
@@ -304,6 +309,7 @@ mkArbPeerTxState mempoolHasTxFun txIdsInflight unacked txMaskMap =
                     requestedTxsInflightSize,
                     unknownTxs,
                     rejectedTxs = 0,
+                    rejectedTxsTs = Time 0,
                     fetchedTxs = Set.empty }
       (Set.fromList $ Map.elems inflightMap)
       bufferedMap
@@ -384,6 +390,7 @@ genSharedTxState maxTxIdsInflight = do
     _mempoolHasTxFun@(Fun (_, _, x) _) <- arbitrary :: Gen (Fun Bool Bool)
     let mempoolHasTxFun = Fun (function (const False), False, x) (const False)
     pss <- listOf1 (genArbPeerTxState mempoolHasTxFun maxTxIdsInflight)
+    seed <- arbitrary
 
     let pss' :: [(PeerAddr, ArbPeerTxState txid (Tx txid))]
         pss' = [0..] `zip` pss
@@ -410,7 +417,8 @@ genSharedTxState maxTxIdsInflight = do
                                  | ArbPeerTxState { arbBufferedMap }
                                    <- pss
                                  ],
-                 referenceCounts = Map.empty
+                 referenceCounts = Map.empty,
+                 peerRng         = mkStdGen seed
                }
 
     return ( mempoolHasTxFun
@@ -637,7 +645,7 @@ prop_acknowledgeTxIds :: ArbDecisionContextWithReceivedTxIds
                       -> Property
 prop_acknowledgeTxIds (ArbDecisionContextWithReceivedTxIds policy SharedDecisionContext { sdcSharedTxState = st } ps _ _ _) =
     case TXS.acknowledgeTxIds policy st ps of
-      (numTxIdsToAck, txIdsToRequest, txs, TXS.RefCountDiff { TXS.txIdsToAck }, ps') | txIdsToRequest > 0 ->
+      (numTxIdsToAck, txIdsToRequest, txIdsTxs, TXS.RefCountDiff { TXS.txIdsToAck }, ps') | txIdsToRequest > 0 ->
              counterexample "number of tx ids to ack must agree with RefCountDiff"
              ( fromIntegral numTxIdsToAck
                ===
@@ -660,7 +668,7 @@ prop_acknowledgeTxIds (ArbDecisionContextWithReceivedTxIds policy SharedDecision
                          | txid <- take (fromIntegral numTxIdsToAck) (toList $ unacknowledgedTxIds ps)
                          , Just _ <- maybeToList $ txid `Map.lookup` bufferedTxs st
                          ]
-             in getTxId `map` txs === acked)
+             in map (getTxId . snd) txIdsTxs === acked)
       _otherwise -> property True
   where
     stripSuffix :: Eq a => [a] -> [a] -> Maybe [a]
@@ -913,6 +921,9 @@ prop_collectTxsImpl (ArbCollectTxs _mempoolHasTxFun txidsRequested txsReceived p
       Nothing -> error "impossible happened! Is the test still using `TxId` for `txid`?"
 
 
+
+deriving via OnlyCheckWhnfNamed "StdGen" StdGen instance NoThunks StdGen
+
 -- | Verify that `SharedTxState` returned by `collectTxsImpl` if evaluated to
 -- WHNF, it doesn't contain any thunks.
 --

ouroboros-network/src/Ouroboros/Network/TxSubmission/Inbound/Decision.hs

@@ -20,13 +20,15 @@ import Control.Arrow ((>>>))
 import Control.Exception (assert)
 
 import Data.Bifunctor (second)
+import Data.Hashable
 import Data.List (mapAccumR, sortOn)
 import Data.Map.Merge.Strict qualified as Map
 import Data.Map.Strict (Map)
 import Data.Map.Strict qualified as Map
 import Data.Maybe (mapMaybe)
 import Data.Set (Set)
 import Data.Set qualified as Set
+import System.Random (random)
 
 import Data.Sequence.Strict qualified as StrictSeq
 import Ouroboros.Network.DeltaQ (PeerGSV (..), defaultGSV,
@@ -43,6 +45,7 @@ makeDecisions
     :: forall peeraddr txid tx.
        ( Ord peeraddr
        , Ord txid
+       , Hashable peeraddr
        )
     => TxDecisionPolicy
     -- ^ decision policy
@@ -62,9 +65,11 @@ makeDecisions policy SharedDecisionContext {
       sdcPeerGSV = _peerGSV,
       sdcSharedTxState = st
     }
-    = fn
-    . pickTxsToDownload policy st
-    . orderByRejections
+    = let (salt, rng') = random (peerRng st)
+          st' = st { peerRng = rng' } in
+    fn
+    . pickTxsToDownload policy st'
+    . orderByRejections salt
   where
     fn :: forall a.
           (a, [(peeraddr, TxDecision txid tx)])
@@ -76,13 +81,16 @@ makeDecisions policy SharedDecisionContext {
 --
 -- TXs delivered late will fail to apply because they where included in
 -- a recently adopted block. Peers can race against each other by setting
--- `txInflightMultiplicity` to > 1.
+-- `txInflightMultiplicity` to > 1. In case of a tie a hash of the peeraddr
+-- is used as a tie breaker. Since every invocation use a new salt a given
+-- peeraddr does not have an advantage over time.
 --
--- TODO: Should not depend on plain `peeraddr` as a tie breaker.
-orderByRejections :: Map peeraddr (PeerTxState txid tx)
+orderByRejections :: Hashable peeraddr
+                  => Int
+                  -> Map peeraddr (PeerTxState txid tx)
                   -> [ (peeraddr, PeerTxState txid tx)]
-orderByRejections =
-        sortOn (\(_peeraddr, ps) -> rejectedTxs ps)
+orderByRejections salt =
+        sortOn (\(peeraddr, ps) -> (rejectedTxs ps, hashWithSalt salt peeraddr))
       . Map.toList
 
 -- | Order peers by `DeltaQ`.

ouroboros-network/src/Ouroboros/Network/TxSubmission/Inbound/Registry.hs

@@ -13,6 +13,7 @@ module Ouroboros.Network.TxSubmission.Inbound.Registry
   , newTxChannelsVar
   , PeerTxAPI (..)
   , decisionLogicThread
+  , drainRejectionThread
   , withPeer
   ) where
 
@@ -21,20 +22,22 @@ import Control.Concurrent.Class.MonadSTM.Strict
 import Control.Monad.Class.MonadFork
 import Control.Monad.Class.MonadThrow
 import Control.Monad.Class.MonadTimer.SI
+import Control.Monad.Class.MonadTime.SI
 
 import Data.Foldable (traverse_
 #if !MIN_VERSION_base(4,20,0)
                      , foldl'
 #endif
                      )
-import Data.Typeable (Typeable)
+import Data.Hashable
 import Data.Map.Strict (Map)
 import Data.Map.Strict qualified as Map
 import Data.Maybe (fromMaybe)
 import Data.Sequence.Strict (StrictSeq)
 import Data.Sequence.Strict qualified as StrictSeq
 import Data.Set (Set)
 import Data.Set qualified as Set
+import Data.Typeable (Typeable)
 import Data.Void (Void)
 
 import Control.Tracer (Tracer, traceWith)
@@ -79,8 +82,11 @@ data PeerTxAPI m txid tx = PeerTxAPI {
                         -> m (Maybe TxSubmissionProtocolError),
     -- ^ handle received txs
 
-    countRejectedTxs    :: Int
-                        -> m Int,
+    countRejectedTxs    :: Time
+                        -> Double
+                        -> m Double,
+    -- ^ updated score.  The `Double` is difference between accepted and
+    -- rejected transactions.
 
     consumeFetchedTxs   :: Set txid
                         -> m (Set txid)
@@ -166,6 +172,7 @@ withPeer tracer
                  unacknowledgedTxIds      = StrictSeq.empty,
                  unknownTxs               = Set.empty,
                  rejectedTxs              = 0,
+                 rejectedTxsTs            = Time 0,
                  fetchedTxs               = Set.empty }
                peerTxStates
          }
@@ -238,9 +245,11 @@ withPeer tracer
                   peeraddr peerTxStates
           in st {peerTxStates = peerTxStates' }
 
-    countRejectedTxs :: Int
-                     -> m Int
-    countRejectedTxs n = atomically $ do
+
+    countRejectedTxs :: Time
+                     -> Double
+                     -> m Double
+    countRejectedTxs now n = atomically $ do
       modifyTVar sharedStateVar cntRejects
       st <- readTVar sharedStateVar
       case Map.lookup peeraddr (peerTxStates st)  of
@@ -251,10 +260,10 @@ withPeer tracer
                    -> SharedTxState peeraddr txid tx
         cntRejects st@SharedTxState { peerTxStates } =
           let peerTxStates' =
-                Map.update
-                  (\ps -> Just $! ps { rejectedTxs = min 42 (max (-42) (rejectedTxs ps + n)) })
-                  peeraddr peerTxStates
-          in st { peerTxStates = peerTxStates' }
+                Map.update (\ps -> Just $! (updateRejects now n ps))
+                peeraddr peerTxStates
+          in st {peerTxStates = peerTxStates'}
+
 
     consumeFetchedTxs :: Set txid
                       -> m (Set txid)
@@ -274,6 +283,48 @@ withPeer tracer
              return o
 
 
+updateRejects
+  :: Time
+  -> Double
+  -> PeerTxState txid tx
+  -> PeerTxState txid tx
+updateRejects now 0 pts | rejectedTxs pts == 0
+                        = pts {rejectedTxsTs = now}
+updateRejects now n pts@PeerTxState { rejectedTxs, rejectedTxsTs } =
+    let duration = diffTime now rejectedTxsTs
+        rate = 0.1                 -- 0.1 rejected tx/s
+        maxTokens = 15 * 60 * rate -- 15 minutes worth of rejections
+        !drain = realToFrac duration * rate
+        !drained = max 0 $ rejectedTxs - drain in
+    pts { rejectedTxs = max 0 $ min maxTokens $ drained + n
+        , rejectedTxsTs = now }
+
+
+drainRejectionThread
+  :: forall m peeraddr txid tx.
+     ( MonadDelay m
+     , MonadSTM m
+     , MonadThread m
+     )
+  => SharedTxStateVar m peeraddr txid tx
+  -> m Void
+drainRejectionThread sharedStateVar = do
+    labelThisThread "tx-rejection-drain"
+    go
+  where
+    go :: m Void
+    go = do
+      threadDelay 7
+
+      !now <- getMonotonicTime
+      atomically $ do
+        st <- readTVar sharedStateVar
+        let ptss = Map.map (\pts -> updateRejects now 0 pts) (peerTxStates st)
+        writeTVar sharedStateVar (st { peerTxStates = ptss })
+
+      go
+
+
 decisionLogicThread
     :: forall m peeraddr txid tx.
        ( MonadDelay m
@@ -283,6 +334,7 @@ decisionLogicThread
        , MonadFork  m
        , Ord peeraddr
        , Ord txid
+       , Hashable peeraddr
        )
     => Tracer m (TraceTxLogic peeraddr txid tx)
     -> TxDecisionPolicy