Added Light Peer Sharing

Added a Control Channel between the Inbound Governor and the
OutboundGovernor. Reading from this channel is abstracted as a
PeerSelectionAction. When the InboundGovernor receives an Inbound
NewConnection it will write the address and PeerSharing Willingness
information to the Channel.

ouroboros-network-framework/demo/connection-manager.hs

@@ -64,6 +64,8 @@ import qualified Ouroboros.Network.InboundGovernor.ControlChannel as Server
 import           Ouroboros.Network.IOManager
 import           Ouroboros.Network.Mux
 import           Ouroboros.Network.MuxMode
+import           Ouroboros.Network.PeerSelection.PeerSharing.Type
+                     (PeerSharing (..))
 import           Ouroboros.Network.Protocol.Handshake
 import           Ouroboros.Network.Protocol.Handshake.Codec
                      (timeLimitsHandshake)
@@ -209,6 +211,7 @@ withBidirectionalConnectionManager snocket makeBearer socket
                                    k = do
     mainThreadId <- myThreadId
     inbgovControlChannel      <- Server.newControlChannel
+    outgovControlChannel      <- Server.newControlChannel
     -- as in the 'withInitiatorOnlyConnectionManager' we use a `StrictTVar` to
     -- pass list of requests, but since we are also interested in the results we
     -- need to have multable cells to pass the accumulators around.
@@ -275,6 +278,8 @@ withBidirectionalConnectionManager snocket makeBearer socket
                     serverConnectionManager = connectionManager,
                     serverInboundIdleTimeout = Just protocolIdleTimeout,
                     serverControlChannel = inbgovControlChannel,
+                    governorControlChannel = outgovControlChannel,
+                    getPeerSharing = \_ -> NoPeerSharing,
                     serverObservableStateVar = observableStateVar
                   }
               )

ouroboros-network-framework/src/Ouroboros/Network/InboundGovernor.hs

@@ -60,11 +60,12 @@ import           Ouroboros.Network.ConnectionId (ConnectionId (..))
 import           Ouroboros.Network.ConnectionManager.Types hiding
                      (TrUnexpectedlyFalseAssertion)
 import           Ouroboros.Network.InboundGovernor.ControlChannel
-                     (ServerControlChannel)
+                     (GovernorControlChannel, ServerControlChannel)
 import qualified Ouroboros.Network.InboundGovernor.ControlChannel as ControlChannel
 import           Ouroboros.Network.InboundGovernor.Event
 import           Ouroboros.Network.InboundGovernor.State
 import           Ouroboros.Network.Mux
+import           Ouroboros.Network.PeerSelection.PeerSharing.Type (PeerSharing)
 import           Ouroboros.Network.Server.RateLimiting
 
 
@@ -99,13 +100,16 @@ inboundGovernor :: forall (muxMode :: MuxMode) socket peerAddr versionData versi
                 => Tracer m (RemoteTransitionTrace peerAddr)
                 -> Tracer m (InboundGovernorTrace peerAddr)
                 -> ServerControlChannel muxMode peerAddr versionData ByteString m a b
+                -> GovernorControlChannel peerAddr m
+                -> (versionData -> PeerSharing)
                 -> Maybe DiffTime -- protocol idle timeout
                 -> MuxConnectionManager muxMode socket peerAddr versionData
                                         versionNumber ByteString m a b
                 -> StrictTVar m InboundGovernorObservableState
                 -> m Void
-inboundGovernor trTracer tracer serverControlChannel inboundIdleTimeout
-                connectionManager observableStateVar = do
+inboundGovernor trTracer tracer serverControlChannel governorControlChannel
+                getPeerSharing inboundIdleTimeout connectionManager
+                observableStateVar = do
     -- State needs to be a TVar, otherwise, when catching the exception inside
     -- the loop we do not have access to the most recent version of the state
     -- and might be truncating transitions.
@@ -176,10 +180,18 @@ inboundGovernor trTracer tracer serverControlChannel inboundIdleTimeout
             provenance
             connId
             csDataFlow
-            (Handle csMux muxBundle _ _)) -> do
+            (Handle csMux muxBundle _ versionData)) -> do
 
               traceWith tracer (TrNewConnection provenance connId)
 
+              -- Comunicate this new inbound connection to the Outbound Governor
+              when (provenance == Inbound) $
+                atomically $
+                  ControlChannel.writeMessage governorControlChannel
+                                              ( remoteAddress connId
+                                              , getPeerSharing versionData
+                                              )
+
               igsConnections <- Map.alterF
                       (\case
                         -- connection

ouroboros-network-framework/src/Ouroboros/Network/Protocol/Handshake/Unversioned.hs

@@ -26,6 +26,8 @@ import           Network.TypedProtocol.Codec
 
 import           Ouroboros.Network.CodecCBORTerm
 import           Ouroboros.Network.ConnectionManager.Types (DataFlow (..))
+import           Ouroboros.Network.PeerSelection.PeerSharing.Type
+                     (PeerSharing (..))
 import           Ouroboros.Network.Protocol.Handshake.Codec
 import           Ouroboros.Network.Protocol.Handshake.Type
 import           Ouroboros.Network.Protocol.Handshake.Version
@@ -70,24 +72,32 @@ unversionedProtocol = simpleSingletonVersions UnversionedProtocol UnversionedPro
 
 -- | Alternative for 'UnversionedProtocolData' which contains 'DataFlow'.
 --
-newtype DataFlowProtocolData =
-    DataFlowProtocolData { getProtocolDataFlow :: DataFlow }
+data DataFlowProtocolData =
+    DataFlowProtocolData { getProtocolDataFlow :: DataFlow, getProtocolPeerSharing :: PeerSharing }
   deriving (Eq, Show)
 
 instance Acceptable DataFlowProtocolData where
-  acceptableVersion (DataFlowProtocolData local) (DataFlowProtocolData remote) =
-    Accept (DataFlowProtocolData $ local `min` remote)
+  acceptableVersion (DataFlowProtocolData local _) (DataFlowProtocolData remote ps) =
+    Accept (DataFlowProtocolData (local `min` remote) ps)
 
 dataFlowProtocolDataCodec :: UnversionedProtocol -> CodecCBORTerm Text DataFlowProtocolData
 dataFlowProtocolDataCodec _ = CodecCBORTerm {encodeTerm, decodeTerm}
     where
       encodeTerm :: DataFlowProtocolData -> CBOR.Term
-      encodeTerm (DataFlowProtocolData Unidirectional) = CBOR.TBool False
-      encodeTerm (DataFlowProtocolData Duplex)         = CBOR.TBool True
+      encodeTerm (DataFlowProtocolData Unidirectional NoPeerSharing)      = CBOR.TList [CBOR.TBool False, CBOR.TInt 0]
+      encodeTerm (DataFlowProtocolData Unidirectional PeerSharingPrivate) = CBOR.TList [CBOR.TBool False, CBOR.TInt 1]
+      encodeTerm (DataFlowProtocolData Unidirectional PeerSharingPublic)  = CBOR.TList [CBOR.TBool False, CBOR.TInt 2]
+      encodeTerm (DataFlowProtocolData Duplex NoPeerSharing)      = CBOR.TList [CBOR.TBool True, CBOR.TInt 0]
+      encodeTerm (DataFlowProtocolData Duplex PeerSharingPrivate) = CBOR.TList [CBOR.TBool True, CBOR.TInt 1]
+      encodeTerm (DataFlowProtocolData Duplex PeerSharingPublic)  = CBOR.TList [CBOR.TBool True, CBOR.TInt 2]
 
       decodeTerm :: CBOR.Term -> Either Text DataFlowProtocolData
-      decodeTerm (CBOR.TBool False) = Right (DataFlowProtocolData Unidirectional)
-      decodeTerm (CBOR.TBool True)  = Right (DataFlowProtocolData Duplex)
+      decodeTerm (CBOR.TList [CBOR.TBool False, CBOR.TInt 0]) = Right (DataFlowProtocolData Unidirectional NoPeerSharing)
+      decodeTerm (CBOR.TList [CBOR.TBool False, CBOR.TInt 1]) = Right (DataFlowProtocolData Unidirectional PeerSharingPrivate)
+      decodeTerm (CBOR.TList [CBOR.TBool False, CBOR.TInt 2]) = Right (DataFlowProtocolData Unidirectional PeerSharingPublic)
+      decodeTerm (CBOR.TList [CBOR.TBool True, CBOR.TInt 0])  = Right (DataFlowProtocolData Duplex NoPeerSharing)
+      decodeTerm (CBOR.TList [CBOR.TBool True, CBOR.TInt 1])  = Right (DataFlowProtocolData Duplex PeerSharingPrivate)
+      decodeTerm (CBOR.TList [CBOR.TBool True, CBOR.TInt 2])  = Right (DataFlowProtocolData Duplex PeerSharingPublic)
       decodeTerm t                  = Left $ T.pack $ "unexpected term: " ++ show t
 
 dataFlowProtocol :: DataFlow
@@ -96,7 +106,7 @@ dataFlowProtocol :: DataFlow
                              DataFlowProtocolData
                              app
 dataFlowProtocol dataFlow =
-    simpleSingletonVersions UnversionedProtocol (DataFlowProtocolData dataFlow)
+    simpleSingletonVersions UnversionedProtocol (DataFlowProtocolData dataFlow NoPeerSharing)
 
 -- | 'Handshake' codec used in various tests.
 --

ouroboros-network-framework/src/Ouroboros/Network/Server2.hs

@@ -55,6 +55,7 @@ import           Ouroboros.Network.InboundGovernor
 import           Ouroboros.Network.InboundGovernor.ControlChannel
 import qualified Ouroboros.Network.InboundGovernor.ControlChannel as ControlChannel
 import           Ouroboros.Network.Mux
+import           Ouroboros.Network.PeerSelection.PeerSharing.Type (PeerSharing)
 import           Ouroboros.Network.Server.RateLimiting
 import           Ouroboros.Network.Snocket
 
@@ -90,6 +91,16 @@ data ServerArguments (muxMode  :: MuxMode) socket peerAddr versionData versionNu
       serverControlChannel        :: ServerControlChannel muxMode peerAddr versionData
                                                           bytes m a b,
 
+      -- | Governor control channel var is passed as an argument; this allows the Server
+      -- to communicate with the Outbound Governor telling it about new inbound
+      -- connections to add to the Known Peers Set.
+      --
+      governorControlChannel        :: GovernorControlChannel peerAddr m,
+
+      -- | Extract 'PeerSharing' value from 'versionData'
+      --
+      getPeerSharing :: versionData -> PeerSharing,
+
       -- | Observable mutable state.
       --
       serverObservableStateVar    :: StrictTVar m InboundGovernorObservableState
@@ -141,6 +152,8 @@ run ServerArguments {
       serverInboundIdleTimeout,
       serverConnectionManager,
       serverControlChannel,
+      governorControlChannel,
+      getPeerSharing,
       serverObservableStateVar
     } = do
       let sockets = NonEmpty.toList serverSockets
@@ -156,6 +169,8 @@ run ServerArguments {
                         inboundGovernor serverTrTracer
                                         inboundGovernorTracer
                                         serverControlChannel
+                                        governorControlChannel
+                                        getPeerSharing
                                         serverInboundIdleTimeout
                                         serverConnectionManager
                                         serverObservableStateVar)

ouroboros-network-framework/test/Test/Ouroboros/Network/Server2.hs

@@ -360,7 +360,7 @@ withInitiatorOnlyConnectionManager name timeouts trTracer cmTracer snocket makeB
           cmSnocket = snocket,
           cmMakeBearer = makeBearer,
           cmConfigureSocket = \_ _ -> return (),
-          connectionDataFlow = \_ (DataFlowProtocolData df) -> df,
+          connectionDataFlow = \_ (DataFlowProtocolData df _) -> df,
           cmPrunePolicy = simplePrunePolicy,
           cmConnectionsLimits = acceptedConnLimit,
           cmTimeWaitTimeout = tTimeWaitTimeout timeouts,
@@ -520,6 +520,7 @@ withBidirectionalConnectionManager name timeouts
                                    acceptedConnLimit k = do
     mainThreadId <- myThreadId
     inbgovControlChannel      <- Server.newControlChannel
+    outbgovControlChannel      <- Server.newControlChannel
     -- we are not using the randomness
     observableStateVar        <- Server.newObservableStateVarFromSeed 0
     let muxTracer = WithName name `contramap` nullTracer -- mux tracer
@@ -541,7 +542,7 @@ withBidirectionalConnectionManager name timeouts
           cmConfigureSocket = \sock _ -> confSock sock,
           cmTimeWaitTimeout = tTimeWaitTimeout timeouts,
           cmOutboundIdleTimeout = tOutboundIdleTimeout timeouts,
-          connectionDataFlow = \_ (DataFlowProtocolData df) -> df,
+          connectionDataFlow = \_ (DataFlowProtocolData df _) -> df,
           cmPrunePolicy = simplePrunePolicy,
           cmConnectionsLimits = acceptedConnLimit
         }
@@ -581,6 +582,8 @@ withBidirectionalConnectionManager name timeouts
                     serverConnectionManager = connectionManager,
                     serverInboundIdleTimeout = Just (tProtocolIdleTimeout timeouts),
                     serverControlChannel = inbgovControlChannel,
+                    governorControlChannel = outbgovControlChannel,
+                    getPeerSharing = \(DataFlowProtocolData _ ps) -> ps,
                     serverObservableStateVar = observableStateVar
                   }
               )

ouroboros-network/src/Ouroboros/Network/Diffusion/P2P.hs

@@ -86,6 +86,8 @@ import           Ouroboros.Network.Diffusion.Utils
 import           Ouroboros.Network.ExitPolicy
 import           Ouroboros.Network.InboundGovernor (InboundGovernorTrace (..),
                      RemoteTransitionTrace)
+import           Ouroboros.Network.InboundGovernor.ControlChannel
+                     (ControlChannel (..), GovernorControlChannel)
 import           Ouroboros.Network.IOManager
 import           Ouroboros.Network.Mux hiding (MiniProtocol (..))
 import           Ouroboros.Network.MuxMode
@@ -346,6 +348,7 @@ data ConnectionManagerDataInMode peerAddr versionData m a (mode :: MuxMode) wher
 
     CMDInInitiatorResponderMode
       :: ServerControlChannel InitiatorResponderMode peerAddr versionData ByteString  m a ()
+      -> GovernorControlChannel peerAddr m
       -> StrictTVar m Server.InboundGovernorObservableState
       -> ConnectionManagerDataInMode peerAddr versionData m a InitiatorResponderMode
 
@@ -728,6 +731,7 @@ runM Interfaces
               Just $ withLocalSocket tracer diNtcGetFileDescriptor diNtcSnocket localAddr
                        $ \localSocket -> do
                 localControlChannel <- Server.newControlChannel
+                localGovControlChannel <- Server.newControlChannel
                 localServerStateVar <- Server.newObservableStateVar ntcInbgovRng
 
                 let localConnectionLimits = AcceptedConnectionsLimit maxBound maxBound 0
@@ -798,6 +802,11 @@ runM Interfaces
                             serverConnectionLimits      = localConnectionLimits,
                             serverConnectionManager     = localConnectionManager,
                             serverControlChannel        = localControlChannel,
+                            governorControlChannel      = localGovControlChannel,
+                            -- local thread does not start a Outbound Governor
+                            -- so it doesn't matter what we put here.
+                            -- 'NoPeerSharing' is set for all connections.
+                            getPeerSharing              = \_ -> NoPeerSharing,
                             serverObservableStateVar    = localServerStateVar
                           }) Async.wait
 
@@ -824,6 +833,7 @@ runM Interfaces
                   HasInitiatorResponder <$>
                     (CMDInInitiatorResponderMode
                       <$> Server.newControlChannel
+                      <*> Server.newControlChannel
                       <*> Server.newObservableStateVar ntnInbgovRng)
 
           -- RNGs used for picking random peers from the ledger and for
@@ -941,6 +951,7 @@ runM Interfaces
                       daOwnPeerSharing
                       (pchPeerSharing diNtnPeerSharing)
                       (readTVar (getPeerSharingRegistry daPeerSharingRegistry))
+                      retry -- Will never receive inbound connections
                       peerStateActions
                       requestLedgerPeers
                       $ \localPeerSelectionActionsThread
@@ -985,7 +996,7 @@ runM Interfaces
               -- Run peer selection and the server.
               --
               HasInitiatorResponder
-                (CMDInInitiatorResponderMode controlChannel observableStateVar) -> do
+                (CMDInInitiatorResponderMode serverControlChannel governorControlChannel observableStateVar) -> do
                 let connectionManagerArguments
                       :: NodeToNodeConnectionManagerArguments
                           InitiatorResponderMode
@@ -1046,7 +1057,7 @@ runM Interfaces
                   connectionManagerArguments
                   connectionHandler
                   classifyHandleError
-                  (InResponderMode controlChannel)
+                  (InResponderMode serverControlChannel)
                   $ \(connectionManager
                         :: NodeToNodeConnectionManager
                              InitiatorResponderMode ntnFd ntnAddr ntnVersionData ntnVersion m a ()
@@ -1088,6 +1099,7 @@ runM Interfaces
                       daOwnPeerSharing
                       (pchPeerSharing diNtnPeerSharing)
                       (readTVar (getPeerSharingRegistry daPeerSharingRegistry))
+                      (readMessage governorControlChannel)
                       peerStateActions
                       requestLedgerPeers
                       $ \localPeerRootProviderThread
@@ -1131,7 +1143,9 @@ runM Interfaces
                                   serverConnectionLimits      = daAcceptedConnectionsLimit,
                                   serverConnectionManager     = connectionManager,
                                   serverInboundIdleTimeout    = Just daProtocolIdleTimeout,
-                                  serverControlChannel        = controlChannel,
+                                  serverControlChannel        = serverControlChannel,
+                                  governorControlChannel      = governorControlChannel,
+                                  getPeerSharing              = diNtnPeerSharing,
                                   serverObservableStateVar    = observableStateVar
                                 })
                                 $ \serverThread ->

ouroboros-network/src/Ouroboros/Network/PeerSelection/Governor.hs

@@ -552,10 +552,11 @@ peerSelectionGovernorLoop tracer
                      -> Guarded (STM m) (TimedDecision m peeraddr peerconn)
     guardedDecisions blockedAt peerSharing st =
       -- All the alternative potentially-blocking decisions.
-         Monitor.connections          actions st
-      <> Monitor.jobs                 jobPool st
-      <> Monitor.targetPeers          actions st
-      <> Monitor.localRoots           actions policy st
+         Monitor.connections           actions st
+      <> Monitor.jobs                  jobPool st
+      <> Monitor.targetPeers           actions st
+      <> Monitor.localRoots            actions policy st
+      <> Monitor.newInboundConnections actions st
 
       -- All the alternative non-blocking internal decisions.
       <> RootPeers.belowTarget        actions blockedAt         st

ouroboros-network/src/Ouroboros/Network/PeerSelection/Governor/Monitor.hs

@@ -13,6 +13,7 @@ module Ouroboros.Network.PeerSelection.Governor.Monitor
   , jobs
   , connections
   , localRoots
+  , newInboundConnections
   ) where
 
 import           Data.Map.Strict (Map)
@@ -91,6 +92,34 @@ jobs jobPool st =
       Completion completion <- JobPool.waitForJob jobPool
       return (completion st)
 
+-- | Monitor new inbound connections
+--
+newInboundConnections :: forall m peeraddr peerconn.
+                          (MonadSTM m, Ord peeraddr)
+                      => PeerSelectionActions peeraddr peerconn m
+                      -> PeerSelectionState peeraddr peerconn
+                      -> Guarded (STM m) (TimedDecision m peeraddr peerconn)
+newInboundConnections PeerSelectionActions{
+                        readNewInboundConnection
+                      }
+                      st@PeerSelectionState {
+                        knownPeers
+                      } =
+    Guarded Nothing $ do
+      (addr, ps) <- readNewInboundConnection
+      return $ \_ ->
+        let -- If peer happens to already be present in the Known Peer set
+            -- 'insert' is going to do its due diligence before adding.
+            newEntry    = Map.singleton addr (ps, DoAdvertisePeer, IsNotLedgerPeer)
+            knownPeers' = KnownPeers.insert newEntry knownPeers
+         in Decision {
+              decisionTrace = [TraceNewInboundConnection addr ps],
+              decisionJobs = [],
+              decisionState =
+                st {
+                  knownPeers = knownPeers'
+                }
+            }
 
 -- | Monitor connections.
 --