Skip to content

netsync: Rework external request from peer. #3500

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
davecgh merged 1 commit into from
Aug 29, 2025
Merged

netsync: Rework external request from peer. #3500

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
117 changes: 31 additions & 86 deletions internal/netsync/manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,13 +8,11 @@ package netsync
import (
"context"
"errors"
"fmt"
"math"
"sync"
"sync/atomic"
"time"

"github.com/decred/dcrd/blockchain/stake/v5"
"github.com/decred/dcrd/chaincfg/chainhash"
"github.com/decred/dcrd/chaincfg/v3"
"github.com/decred/dcrd/container/apbf"
Expand Down Expand Up @@ -2062,116 +2060,63 @@ func (m *SyncManager) SyncPeerID() int32 {
// the peer is not banned for sending unrequested data when it responds.
//
// This function is safe for concurrent access.
func (m *SyncManager) RequestFromPeer(peer *Peer, blocks, voteHashes,
tSpendHashes []chainhash.Hash) error {

func (m *SyncManager) RequestFromPeer(peer *Peer, blocks, voteHashes, tSpendHashes []chainhash.Hash) {
if m.shutdownRequested() {
return nil
return
}

defer m.requestMtx.Unlock()
m.requestMtx.Lock()

// Add the blocks to the request.
// Request as many needed blocks as possible at once.
var numRequested uint32
gdMsg := wire.NewMsgGetData()
for i := range blocks {
// Skip the block when it has already been requested.
bh := &blocks[i]
if m.isRequestedBlock(bh) {
continue
}

// Skip the block when it is already known.
if m.cfg.Chain.HaveBlock(bh) {
// Skip the block when it has already been requested or is already
// known.
blockHash := &blocks[i]
if m.isRequestedBlock(blockHash) || m.cfg.Chain.HaveBlock(blockHash) {
continue
}

err := gdMsg.AddInvVect(wire.NewInvVect(wire.InvTypeBlock, bh))
if err != nil {
return fmt.Errorf("unexpected error encountered building request "+
"for block %v: %w", bh, err)
gdMsg.AddInvVect(wire.NewInvVect(wire.InvTypeBlock, blockHash))
m.requestedBlocks[*blockHash] = peer
numRequested++
if numRequested == wire.MaxInvPerMsg {
// Send full getdata message and reset.
peer.QueueMessage(gdMsg, nil)
gdMsg = wire.NewMsgGetData()
numRequested = 0
}

m.requestedBlocks[*bh] = peer
}

addTxsToRequest := func(hashes []chainhash.Hash, txType stake.TxType) error {
// Return immediately if txs is nil.
if hashes == nil {
return nil
}

// Request as many needed votes and treasury spend transactions as possible
// at once.
for _, hashes := range [][]chainhash.Hash{voteHashes, tSpendHashes} {
for i := range hashes {
// Skip the transaction when it has already been requested.
// Skip the transaction when it has already been requested or is
// otherwise not needed.
txHash := &hashes[i]
if _, ok := m.requestedTxns[*txHash]; ok {
continue
}

// Ask the transaction memory pool if the transaction is known
// to it in any form (main pool or orphan).
if m.cfg.TxMemPool.HaveTransaction(txHash) {
continue
}

// Check if the transaction exists from the point of view of the main
// chain tip. Note that this is only a best effort since it is expensive
// to check existence of every output and the only purpose of this check
// is to avoid requesting already known transactions.
//
// Check for a specific outpoint based on the tx type.
outpoint := wire.OutPoint{Hash: *txHash}
switch txType {
case stake.TxTypeSSGen:
// The first two outputs of vote transactions are OP_RETURN <data>, and
// therefore never exist as an unspent txo. Use the third output, as
// the third output (and subsequent outputs) are OP_SSGEN outputs.
outpoint.Index = 2
outpoint.Tree = wire.TxTreeStake
case stake.TxTypeTSpend:
// The first output of a tSpend transaction is OP_RETURN <data>, and
// therefore never exists as an unspent txo. Use the second output, as
// the second output (and subsequent outputs) are OP_TGEN outputs.
outpoint.Index = 1
outpoint.Tree = wire.TxTreeStake
}
entry, err := m.cfg.Chain.FetchUtxoEntry(outpoint)
if err != nil {
return err
}
if entry != nil {
_, alreadyRequested := m.requestedTxns[*txHash]
if alreadyRequested || !m.needTx(txHash) {
continue
}

err = gdMsg.AddInvVect(wire.NewInvVect(wire.InvTypeTx, txHash))
if err != nil {
return fmt.Errorf("unexpected error encountered building request "+
"for tx %v: %w", txHash, err)
}

gdMsg.AddInvVect(wire.NewInvVect(wire.InvTypeTx, txHash))
m.requestedTxns[*txHash] = peer
numRequested++
if numRequested == wire.MaxInvPerMsg {
// Send full getdata message and reset.
peer.QueueMessage(gdMsg, nil)
gdMsg = wire.NewMsgGetData()
numRequested = 0
}
}

return nil
}

// Add the vote transactions to the request.
err := addTxsToRequest(voteHashes, stake.TxTypeSSGen)
if err != nil {
return err
}

// Add the tspend transactions to the request.
err = addTxsToRequest(tSpendHashes, stake.TxTypeTSpend)
if err != nil {
return err
}

if len(gdMsg.InvList) > 0 {
peer.QueueMessage(gdMsg, nil)
}

return nil
}

// RequestMixMsgFromPeer requests the specified mix message from the given peer.
Expand Down
12 changes: 3 additions & 9 deletions server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1343,11 +1343,8 @@ func (sp *serverPeer) OnMiningState(_ *peer.Peer, msg *wire.MsgMiningState) {
}
}

err := sp.server.syncManager.RequestFromPeer(sp.syncMgrPeer, blockHashes,
sp.server.syncManager.RequestFromPeer(sp.syncMgrPeer, blockHashes,
voteHashes, nil)
if err != nil {
peerLog.Warnf("couldn't handle mining state message: %v", err)
}
}

// OnGetInitState is invoked when a peer receives a getinitstate wire message.
Expand Down Expand Up @@ -1429,11 +1426,8 @@ func (sp *serverPeer) OnGetInitState(_ *peer.Peer, msg *wire.MsgGetInitState) {
// OnInitState is invoked when a peer receives a initstate wire message. It
// requests the data advertised in the message from the peer.
func (sp *serverPeer) OnInitState(_ *peer.Peer, msg *wire.MsgInitState) {
err := sp.server.syncManager.RequestFromPeer(sp.syncMgrPeer,
msg.BlockHashes, msg.VoteHashes, msg.TSpendHashes)
if err != nil {
peerLog.Warnf("couldn't handle init state message: %v", err)
}
sp.server.syncManager.RequestFromPeer(sp.syncMgrPeer, msg.BlockHashes,
msg.VoteHashes, msg.TSpendHashes)
}

// OnTx is invoked when a peer receives a tx wire message. It blocks until the
Expand Down