<commit_before>{-# LANGUAGE CPP                 #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE RecordWildCards     #-}
{-# LANGUAGE RankNTypes          #-}

module Pos.Diffusion.Full
    ( diffusionLayerFull
    ) where

import           Universum hiding (bracket)

import           Control.Monad.Fix (MonadFix)
import qualified Data.Map as M
import           Data.Reflection (give)
import           Formatting (sformat, shown, (%))
import           Mockable (Mockable, Bracket, bracket, Catch, Throw, throw, withAsync)
import           Mockable.Production (Production)
import qualified Network.Broadcast.OutboundQueue as OQ
import           Network.Broadcast.OutboundQueue.Types (MsgType (..), Origin (..))
import           Network.QDisc.Fair (fairQDisc)
import qualified Network.Transport as NT (closeTransport)
import           Network.Transport.Abstract (Transport)
import           Network.Transport.Concrete (concrete)
import qualified Network.Transport.TCP as TCP
import           Node (Node, NodeAction (..), simpleNodeEndPoint, NodeEnvironment (..), defaultNodeEnvironment, node)
import           Node.Conversation (Converse, converseWith, Conversation)
import qualified System.Metrics as Monitoring
import           System.Random (newStdGen)
import           System.Wlog (WithLogger, CanLog, logError, getLoggerName, usingLoggerName)

import           Pos.Block.Network.Types (MsgGetHeaders, MsgHeaders, MsgGetBlocks, MsgBlock)
import           Pos.Communication (NodeId, VerInfo (..), PeerData, PackingType, EnqueueMsg, makeEnqueueMsg, bipPacking, Listener, MkListeners (..), HandlerSpecs, InSpecs (..), OutSpecs (..), createOutSpecs, toOutSpecs, convH, InvOrDataTK, MsgSubscribe, makeSendActions, SendActions, Msg)
import           Pos.Communication.Limits (SscLimits (..), UpdateLimits (..), TxpLimits (..), BlockLimits (..))
import           Pos.Communication.Relay.Logic (invReqDataFlowTK)
import           Pos.Communication.Util (wrapListener)
import           Pos.Configuration (HasNodeConfiguration, conversationEstablishTimeout, networkConnectionTimeout)
import           Pos.Core.Block (Block, MainBlockHeader, BlockHeader)
import           Pos.Core.Coin (coinPortionToDouble)
import           Pos.Core.Configuration (protocolMagic)
import           Pos.Core.Types (HeaderHash, BlockVersionData (..), StakeholderId)
import           Pos.Core.Txp (TxAux)
import           Pos.Core.Update (UpId, UpdateVote, UpdateProposal)
import           Pos.Crypto.Configuration (ProtocolMagic (..))
import           Pos.DHT.Real (KademliaDHTInstance, KademliaParams (..), startDHTInstance, stopDHTInstance)
import qualified Pos.Diffusion.Full.Block      as Diffusion.Block
import qualified Pos.Diffusion.Full.Delegation as Diffusion.Delegation
import qualified Pos.Diffusion.Full.Ssc        as Diffusion.Ssc
import qualified Pos.Diffusion.Full.Txp        as Diffusion.Txp
import qualified Pos.Diffusion.Full.Update     as Diffusion.Update
import           Pos.Diffusion.Full.Types  (DiffusionWorkMode)
import           Pos.Diffusion.Types (Diffusion (..), DiffusionLayer (..), GetBlocksError (..))
import           Pos.Logic.Types (Logic (..))
import           Pos.Network.CLI (NetworkConfigOpts (..), intNetworkConfigOpts)
import           Pos.Network.Types (NetworkConfig (..), Topology (..), Bucket, initQueue,
                                    topologySubscribers, SubscriptionWorker (..),
                                    topologySubscriptionWorker)
import           Pos.Ssc.Message (MCOpening, MCShares, MCCommitment, MCVssCertificate)
import           Pos.Diffusion.Subscription.Common (subscriptionListeners)
import           Pos.Diffusion.Subscription.Dns (dnsSubscriptionWorker)
import           Pos.Diffusion.Subscription.Dht (dhtSubscriptionWorker)
import           Pos.Update.Configuration (lastKnownBlockVersion)
import           Pos.Util.OutboundQueue (EnqueuedConversation (..))

-- | The full diffusion layer.
--
-- NB: we could do the whole logic/diffusion layer interface using typeclasses
-- against a monad, but we'd end up with a bunch of reader constraints over
-- the values that would otherwise appear in the Logic and Diffusion records.
-- That's to say, we'd have to do the same work anyway, but then even more
-- work to juggle the instances.
diffusionLayerFull
    :: forall d x .
       ( DiffusionWorkMode d
       , MonadFix d
       )
    => NetworkConfigOpts
    -> Maybe Monitoring.Store
    -> ((Logic d -> Production (DiffusionLayer d)) -> Production x)
    -> Production x
diffusionLayerFull networkConfigOpts mEkgStore expectLogic =
    bracket acquire release $ \_ -> expectLogic $ \logic -> do

        -- Read in network topology / configuration / policies.
        networkConfig <- intNetworkConfigOpts networkConfigOpts

        -- Make the outbound queue using network policies.
        oq :: OQ.OutboundQ (EnqueuedConversation d) NodeId Bucket <-
            initQueue networkConfig mEkgStore 

        let -- VerInfo is a diffusion-layer-specific thing. It's only used for
            -- negotiating with peers.
            ourVerInfo :: VerInfo
            ourVerInfo = VerInfo (getProtocolMagic protocolMagic) lastKnownBlockVersion ins (outs <> workerOuts)

            ins :: HandlerSpecs
            InSpecs ins = inSpecs mkL

            -- The out specs come not just from listeners but also from workers.
            -- Workers in the existing implementation were bundled up in
            --   allWorkers :: ([WorkerSpec m], OutSpecs)
            -- and they performed logic layer tasks, so having out specs defined
            -- by them doesn't make sense.
            -- For the first iteration, we just dump those out specs here, since
            -- we know in the diffusion layer the set of all requests that might
            -- be made.
            --
            -- Find below a definition of each of the worker out specs,
            -- copied from Pos.Worker (allWorkers). Each one was manually
            -- inspected to determine the out specs.
            --
            -- FIXME this system must change. Perhaps replace it with a
            -- version number?
            outs :: HandlerSpecs
            OutSpecs outs = outSpecs mkL

            workerOuts :: HandlerSpecs
            OutSpecs workerOuts = mconcat
                [ sscWorkerOutSpecs
                , securityWorkerOutSpecs
                , usWorkerOutSpecs
                , blockWorkerOutSpecs
                , delegationWorkerOutSpecs
                , slottingWorkerOutSpecs
                , subscriptionWorkerOutSpecs
                , dhtWorkerOutSpecs
                ]

            -- An onNewSlotWorker and a localWorker. Latter is mempty. Former
            -- actually does the ssc stuff.
            sscWorkerOutSpecs = mconcat
                [ createOutSpecs (Proxy @(InvOrDataTK StakeholderId MCCommitment))
                , createOutSpecs (Proxy @(InvOrDataTK StakeholderId MCOpening))
                , createOutSpecs (Proxy @(InvOrDataTK StakeholderId MCShares))
                , createOutSpecs (Proxy @(InvOrDataTK StakeholderId MCVssCertificate))
                ]

            -- A single worker checkForReceivedBlocksWorker with
            -- requestTipOuts from Pos.Block.Network.
            securityWorkerOutSpecs = toOutSpecs
                [ convH (Proxy :: Proxy MsgGetHeaders)
                        (Proxy :: Proxy MsgHeaders)
                ]

            -- Definition of usWorkers plainly shows the out specs = mempty.
            usWorkerOutSpecs = mempty

            -- announceBlockOuts from blkCreatorWorker
            -- announceBlockOuts from blkMetricCheckerWorker
            -- along with the retrieval worker outs which also include
            -- announceBlockouts.
            blockWorkerOutSpecs = mconcat
                [ announceBlockOuts
                , announceBlockOuts
                , announceBlockOuts <> toOutSpecs [ convH (Proxy :: Proxy MsgGetBlocks)
                                                          (Proxy :: Proxy MsgBlock)
                                                  ]
                ]

            announceBlockOuts = toOutSpecs [ convH (Proxy :: Proxy MsgHeaders)
                                                   (Proxy :: Proxy MsgGetHeaders)
                                           ]

            -- It's a local worker, no out specs.
            delegationWorkerOutSpecs = mempty

            -- Plainly mempty from the definition of allWorkers.
            slottingWorkerOutSpecs = mempty

            -- Copied from existing implementation but
            -- FIXME it will be wrong when the patch to include a keepalive
            -- is merged. That shall be the first test of this inspec/outspec
            -- system I suppose.
            subscriptionWorkerOutSpecs = case topologySubscriptionWorker (ncTopology networkConfig) of
                Just (SubscriptionWorkerBehindNAT _) -> specs
                Just (SubscriptionWorkerKademlia __ _ _ _) -> specs
                _ -> mempty
              where
                specs = toOutSpecs [ convH (Proxy @MsgSubscribe) (Proxy @Void) ]

            -- It's a localOnNewSlotWorker, so mempty.
            dhtWorkerOutSpecs = mempty

            mkL :: MkListeners d
            --mkL = error "listeners" -- allListeners oq (ncTopology networkConfig) enqueue
            mkL = mconcat $
                [ lmodifier "block"       $ give blockLimits $ Diffusion.Block.blockListeners logic oq
                , lmodifier "tx"          $ give txpLimits $ Diffusion.Txp.txListeners logic oq enqueue
                , lmodifier "update"      $ give updateLimits $ Diffusion.Update.updateListeners logic oq enqueue
                , lmodifier "delegation"  $ Diffusion.Delegation.delegationListeners logic oq enqueue
                , lmodifier "ssc"         $ give sscLimits $ Diffusion.Ssc.sscListeners logic oq enqueue
                ] ++ [
                  lmodifier "subscription" $ subscriptionListeners oq subscriberNodeType
                | Just (subscriberNodeType, _) <- [topologySubscribers (ncTopology networkConfig)]
                ]

            lmodifier lname mkLs = mkLs { mkListeners = mkListeners' }
              where
                mkListeners' v p =
                    let ls = mkListeners mkL v p
                        f = wrapListener ("server" <> lname)
                    in  map f ls

            listeners :: VerInfo -> [Listener d]
            listeners = mkListeners mkL ourVerInfo

            -- Bracket kademlia and network-transport, create a node. This
            -- will be very involved. Should make it top-level I think.
            runDiffusionLayer :: forall y . d y -> d y
            runDiffusionLayer = runDiffusionLayerFull networkConfig ourVerInfo oq listeners

            enqueue :: EnqueueMsg d
            enqueue = makeEnqueueMsg ourVerInfo $ \msgType k -> do
                itList <- OQ.enqueue oq msgType (EnqueuedConversation (msgType, k))<commit_msg>[CSL-2031] Fix infinite loop in mkL.<commit_after>{-# LANGUAGE CPP                 #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE RecordWildCards     #-}
{-# LANGUAGE RankNTypes          #-}

module Pos.Diffusion.Full
    ( diffusionLayerFull
    ) where

import           Universum hiding (bracket)

import           Control.Monad.Fix (MonadFix)
import qualified Data.Map as M
import           Data.Reflection (give)
import           Formatting (sformat, shown, (%))
import           Mockable (Mockable, Bracket, bracket, Catch, Throw, throw, withAsync)
import           Mockable.Production (Production)
import qualified Network.Broadcast.OutboundQueue as OQ
import           Network.Broadcast.OutboundQueue.Types (MsgType (..), Origin (..))
import           Network.QDisc.Fair (fairQDisc)
import qualified Network.Transport as NT (closeTransport)
import           Network.Transport.Abstract (Transport)
import           Network.Transport.Concrete (concrete)
import qualified Network.Transport.TCP as TCP
import           Node (Node, NodeAction (..), simpleNodeEndPoint, NodeEnvironment (..), defaultNodeEnvironment, node)
import           Node.Conversation (Converse, converseWith, Conversation)
import qualified System.Metrics as Monitoring
import           System.Random (newStdGen)
import           System.Wlog (WithLogger, CanLog, logError, getLoggerName, usingLoggerName)

import           Pos.Block.Network.Types (MsgGetHeaders, MsgHeaders, MsgGetBlocks, MsgBlock)
import           Pos.Communication (NodeId, VerInfo (..), PeerData, PackingType, EnqueueMsg, makeEnqueueMsg, bipPacking, Listener, MkListeners (..), HandlerSpecs, InSpecs (..), OutSpecs (..), createOutSpecs, toOutSpecs, convH, InvOrDataTK, MsgSubscribe, makeSendActions, SendActions, Msg)
import           Pos.Communication.Limits (SscLimits (..), UpdateLimits (..), TxpLimits (..), BlockLimits (..))
import           Pos.Communication.Relay.Logic (invReqDataFlowTK)
import           Pos.Communication.Util (wrapListener)
import           Pos.Configuration (HasNodeConfiguration, conversationEstablishTimeout, networkConnectionTimeout)
import           Pos.Core.Block (Block, MainBlockHeader, BlockHeader)
import           Pos.Core.Coin (coinPortionToDouble)
import           Pos.Core.Configuration (protocolMagic)
import           Pos.Core.Types (HeaderHash, BlockVersionData (..), StakeholderId)
import           Pos.Core.Txp (TxAux)
import           Pos.Core.Update (UpId, UpdateVote, UpdateProposal)
import           Pos.Crypto.Configuration (ProtocolMagic (..))
import           Pos.DHT.Real (KademliaDHTInstance, KademliaParams (..), startDHTInstance, stopDHTInstance)
import qualified Pos.Diffusion.Full.Block      as Diffusion.Block
import qualified Pos.Diffusion.Full.Delegation as Diffusion.Delegation
import qualified Pos.Diffusion.Full.Ssc        as Diffusion.Ssc
import qualified Pos.Diffusion.Full.Txp        as Diffusion.Txp
import qualified Pos.Diffusion.Full.Update     as Diffusion.Update
import           Pos.Diffusion.Full.Types  (DiffusionWorkMode)
import           Pos.Diffusion.Types (Diffusion (..), DiffusionLayer (..), GetBlocksError (..))
import           Pos.Logic.Types (Logic (..))
import           Pos.Network.CLI (NetworkConfigOpts (..), intNetworkConfigOpts)
import           Pos.Network.Types (NetworkConfig (..), Topology (..), Bucket, initQueue,
                                    topologySubscribers, SubscriptionWorker (..),
                                    topologySubscriptionWorker)
import           Pos.Ssc.Message (MCOpening, MCShares, MCCommitment, MCVssCertificate)
import           Pos.Diffusion.Subscription.Common (subscriptionListeners)
import           Pos.Diffusion.Subscription.Dns (dnsSubscriptionWorker)
import           Pos.Diffusion.Subscription.Dht (dhtSubscriptionWorker)
import           Pos.Update.Configuration (lastKnownBlockVersion)
import           Pos.Util.OutboundQueue (EnqueuedConversation (..))

-- | The full diffusion layer.
--
-- NB: we could do the whole logic/diffusion layer interface using typeclasses
-- against a monad, but we'd end up with a bunch of reader constraints over
-- the values that would otherwise appear in the Logic and Diffusion records.
-- That's to say, we'd have to do the same work anyway, but then even more
-- work to juggle the instances.
diffusionLayerFull
    :: forall d x .
       ( DiffusionWorkMode d
       , MonadFix d
       )
    => NetworkConfigOpts
    -> Maybe Monitoring.Store
    -> ((Logic d -> Production (DiffusionLayer d)) -> Production x)
    -> Production x
diffusionLayerFull networkConfigOpts mEkgStore expectLogic =
    bracket acquire release $ \_ -> expectLogic $ \logic -> do

        -- Read in network topology / configuration / policies.
        networkConfig <- intNetworkConfigOpts networkConfigOpts

        -- Make the outbound queue using network policies.
        oq :: OQ.OutboundQ (EnqueuedConversation d) NodeId Bucket <-
            initQueue networkConfig mEkgStore 

        let -- VerInfo is a diffusion-layer-specific thing. It's only used for
            -- negotiating with peers.
            ourVerInfo :: VerInfo
            ourVerInfo = VerInfo (getProtocolMagic protocolMagic) lastKnownBlockVersion ins (outs <> workerOuts)

            ins :: HandlerSpecs
            InSpecs ins = inSpecs mkL

            -- The out specs come not just from listeners but also from workers.
            -- Workers in the existing implementation were bundled up in
            --   allWorkers :: ([WorkerSpec m], OutSpecs)
            -- and they performed logic layer tasks, so having out specs defined
            -- by them doesn't make sense.
            -- For the first iteration, we just dump those out specs here, since
            -- we know in the diffusion layer the set of all requests that might
            -- be made.
            --
            -- Find below a definition of each of the worker out specs,
            -- copied from Pos.Worker (allWorkers). Each one was manually
            -- inspected to determine the out specs.
            --
            -- FIXME this system must change. Perhaps replace it with a
            -- version number?
            outs :: HandlerSpecs
            OutSpecs outs = outSpecs mkL

            workerOuts :: HandlerSpecs
            OutSpecs workerOuts = mconcat
                [ sscWorkerOutSpecs
                , securityWorkerOutSpecs
                , usWorkerOutSpecs
                , blockWorkerOutSpecs
                , delegationWorkerOutSpecs
                , slottingWorkerOutSpecs
                , subscriptionWorkerOutSpecs
                , dhtWorkerOutSpecs
                ]

            -- An onNewSlotWorker and a localWorker. Latter is mempty. Former
            -- actually does the ssc stuff.
            sscWorkerOutSpecs = mconcat
                [ createOutSpecs (Proxy @(InvOrDataTK StakeholderId MCCommitment))
                , createOutSpecs (Proxy @(InvOrDataTK StakeholderId MCOpening))
                , createOutSpecs (Proxy @(InvOrDataTK StakeholderId MCShares))
                , createOutSpecs (Proxy @(InvOrDataTK StakeholderId MCVssCertificate))
                ]

            -- A single worker checkForReceivedBlocksWorker with
            -- requestTipOuts from Pos.Block.Network.
            securityWorkerOutSpecs = toOutSpecs
                [ convH (Proxy :: Proxy MsgGetHeaders)
                        (Proxy :: Proxy MsgHeaders)
                ]

            -- Definition of usWorkers plainly shows the out specs = mempty.
            usWorkerOutSpecs = mempty

            -- announceBlockOuts from blkCreatorWorker
            -- announceBlockOuts from blkMetricCheckerWorker
            -- along with the retrieval worker outs which also include
            -- announceBlockouts.
            blockWorkerOutSpecs = mconcat
                [ announceBlockOuts
                , announceBlockOuts
                , announceBlockOuts <> toOutSpecs [ convH (Proxy :: Proxy MsgGetBlocks)
                                                          (Proxy :: Proxy MsgBlock)
                                                  ]
                ]

            announceBlockOuts = toOutSpecs [ convH (Proxy :: Proxy MsgHeaders)
                                                   (Proxy :: Proxy MsgGetHeaders)
                                           ]

            -- It's a local worker, no out specs.
            delegationWorkerOutSpecs = mempty

            -- Plainly mempty from the definition of allWorkers.
            slottingWorkerOutSpecs = mempty

            -- Copied from existing implementation but
            -- FIXME it will be wrong when the patch to include a keepalive
            -- is merged. That shall be the first test of this inspec/outspec
            -- system I suppose.
            subscriptionWorkerOutSpecs = case topologySubscriptionWorker (ncTopology networkConfig) of
                Just (SubscriptionWorkerBehindNAT _) -> specs
                Just (SubscriptionWorkerKademlia __ _ _ _) -> specs
                _ -> mempty
              where
                specs = toOutSpecs [ convH (Proxy @MsgSubscribe) (Proxy @Void) ]

            -- It's a localOnNewSlotWorker, so mempty.
            dhtWorkerOutSpecs = mempty

            mkL :: MkListeners d
            --mkL = error "listeners" -- allListeners oq (ncTopology networkConfig) enqueue
            mkL = mconcat $
                [ lmodifier "block"       $ give blockLimits $ Diffusion.Block.blockListeners logic oq
                , lmodifier "tx"          $ give txpLimits $ Diffusion.Txp.txListeners logic oq enqueue
                , lmodifier "update"      $ give updateLimits $ Diffusion.Update.updateListeners logic oq enqueue
                , lmodifier "delegation"  $ Diffusion.Delegation.delegationListeners logic oq enqueue
                , lmodifier "ssc"         $ give sscLimits $ Diffusion.Ssc.sscListeners logic oq enqueue
                ] ++ [
                  lmodifier "subscription" $ subscriptionListeners oq subscriberNodeType
                | Just (subscriberNodeType, _) <- [topologySubscribers (ncTopology networkConfig)]
                ]

            lmodifier lname mkLs = mkLs { mkListeners = mkListeners' }
              where
                mkListeners' v p =
                    let ls = mkListeners mkLs v p
                        f = wrapListener ("server" <> lname)
                    in  map f ls

            listeners :: VerInfo -> [Listener d]
            listeners = mkListeners mkL ourVerInfo

            -- Bracket kademlia and network-transport, create a node. This
            -- will be very involved. Should make it top-level I think.
            runDiffusionLayer :: forall y . d y -> d y
            runDiffusionLayer = runDiffusionLayerFull networkConfig ourVerInfo oq listeners

            enqueue :: EnqueueMsg d
            enqueue = makeEnqueueMsg ourVerInfo $ \msgType k -> do
                itList <- OQ.enqueue oq msgType (EnqueuedConversation (msgType, k))
{'fraction_non_alphanumeric': 0.072658090233738, 'fraction_numerical': 0.0001811922449719152, 'mean_word_length': 2.015983606557377, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6933681', 'n_tokens_mistral': 5738, 'n_tokens_neox': 5562, 'n_words': 1676}