From c63f5543f680104e3b9bc1aa9692e31c118f7fd9 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Sat, 4 Dec 2021 03:17:28 +0100 Subject: [PATCH] Implement protocol.CommandGetBlock. Close #33 --- Commands.go | 47 +++++++++++++++++- Message Send.go | 14 ++++++ Network.go | 16 +++++++ Transfer Blocks.go | 117 +++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 193 insertions(+), 1 deletion(-) create mode 100644 Transfer Blocks.go diff --git a/Commands.go b/Commands.go index 5b785b3..837e21a 100644 --- a/Commands.go +++ b/Commands.go @@ -274,7 +274,52 @@ func (peer *PeerInfo) cmdTransfer(msg *protocol.MessageTransfer, connection *Con case protocol.TransferControlTerminate: if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok { - // Since an incoming terminate notice means the remote peer already terminated the connection, set sendNotice to false. + v.Terminate(2) + return + } + + } +} + +// cmdGetBlock handles an incoming block message +func (peer *PeerInfo) cmdGetBlock(msg *protocol.MessageGetBlock, connection *Connection) { + switch msg.Control { + case protocol.GetBlockControlRequestStart: + // Currently only support the local blockchain. + if !msg.BlockchainPublicKey.IsEqual(peerPublicKey) { + peer.sendGetBlock(nil, protocol.GetBlockControlNotAvailable, msg.BlockchainPublicKey, 0, 0, nil, msg.Sequence) + return + } else if _, height, _ := UserBlockchain.Header(); height == 0 { + peer.sendGetBlock(nil, protocol.GetBlockControlEmpty, msg.BlockchainPublicKey, 0, 0, nil, msg.Sequence) + return + } else if msg.LimitBlockCount == 0 { + peer.sendGetBlock(nil, protocol.GetBlockControlTerminate, msg.BlockchainPublicKey, 0, 0, nil, msg.Sequence) + return + } + + // Create a local UDT client to connect to the remote UDT server and serve the blocks! + go peer.startBlockTransfer(msg.BlockchainPublicKey, msg.LimitBlockCount, msg.MaxBlockSize, msg.TargetBlocks, msg.Sequence) + + case protocol.GetBlockControlActive: + if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok { + go v.receiveData(msg.Data) + return + } + + case protocol.GetBlockControlNotAvailable: + if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok { + v.Terminate(404) + return + } + + case protocol.GetBlockControlEmpty: + if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok { + v.Terminate(410) + return + } + + case protocol.GetBlockControlTerminate: + if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok { v.Terminate(2) return } diff --git a/Message Send.go b/Message Send.go index 8e0e3fc..a71ac15 100644 --- a/Message Send.go +++ b/Message Send.go @@ -99,3 +99,17 @@ func (peer *PeerInfo) sendTransfer(data []byte, control, transferProtocol uint8, return peer.send(raw) } + +// sendGetBlock sends a get block message +func (peer *PeerInfo) sendGetBlock(data []byte, control uint8, blockchainPublicKey *btcec.PublicKey, limitBlockCount, maxBlockSize uint64, targetBlocks []protocol.BlockRange, sequenceNumber uint32) (err error) { + packetRaw, err := protocol.EncodeGetBlock(peerPrivateKey, data, control, blockchainPublicKey, limitBlockCount, maxBlockSize, targetBlocks) + if err != nil { + return err + } + + raw := &protocol.PacketRaw{Command: protocol.CommandGetBlock, Payload: packetRaw, Sequence: sequenceNumber} + + //Filters.MessageOutGetBlock(peer, raw, control, ) + + return peer.send(raw) +} diff --git a/Network.go b/Network.go index da1755c..acf2a63 100644 --- a/Network.go +++ b/Network.go @@ -276,6 +276,22 @@ func (nets *Networks) packetWorker() { peer.cmdTransfer(msg, connection) } + case protocol.CommandGetBlock: + if msg, _ := protocol.DecodeGetBlock(raw); msg != nil { + // Validate sequence number which prevents unsolicited responses. + isLast := msg.IsLast() + sequenceInfo, valid, rtt := nets.Sequences.ValidateSequenceBi(raw.SenderPublicKey, raw.Sequence, isLast) + if msg.Control != protocol.GetBlockControlRequestStart && !valid { + //Filters.LogError("packetWorker", "message with invalid sequence %d command %d from %s\n", raw.Sequence, raw.Command, raw.connection.Address.String()) // Only log for debug purposes. + continue + } else if rtt > 0 { + connection.RoundTripTime = rtt + } + raw.SequenceInfo = sequenceInfo + + peer.cmdGetBlock(msg, connection) + } + default: // Unknown command Filters.MessageIn(peer, raw, nil) diff --git a/Transfer Blocks.go b/Transfer Blocks.go new file mode 100644 index 0000000..fb776c4 --- /dev/null +++ b/Transfer Blocks.go @@ -0,0 +1,117 @@ +/* +File Name: Transfer Blocks.go +Copyright: 2021 Peernet s.r.o. +Author: Peter Kleissner +*/ + +package core + +import ( + "errors" + "net" + "time" + + "github.com/PeernetOfficial/core/blockchain" + "github.com/PeernetOfficial/core/btcec" + "github.com/PeernetOfficial/core/protocol" + "github.com/PeernetOfficial/core/udt" +) + +// blockSequenceTimeout is the timeout for a follow-up message to appear, otherwise the transfer will be terminated. +var blockSequenceTimeout = time.Second * 10 + +// startBlockTransfer starts the transfer of blocks. Currently it only serves the user's blockchain. +func (peer *PeerInfo) startBlockTransfer(BlockchainPublicKey *btcec.PublicKey, LimitBlockCount uint64, MaxBlockSize uint64, TargetBlocks []protocol.BlockRange, sequenceNumber uint32) (err error) { + virtualConn := newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32) { + peer.sendGetBlock(data, protocol.GetBlockControlActive, BlockchainPublicKey, 0, 0, nil, sequenceNumber) + }) + + // register the sequence since packets are sent bi-directional + virtualConn.sequenceNumber = sequenceNumber + networks.Sequences.RegisterSequenceBi(peer.PublicKey, sequenceNumber, virtualConn, blockSequenceTimeout, virtualConn.sequenceTerminate) + + udtConfig := udt.DefaultConfig() + udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSize + udtConfig.MaxFlowWinSize = maxFlowWinSize + + // start UDT sender + // Set streaming to true, otherwise udtSocket.Read returns the error "Message truncated" in case the reader has a smaller buffer. + udtConn, err := udt.DialUDT(udtConfig, virtualConn, virtualConn.incomingData, virtualConn.outgoingData, virtualConn.terminationSignal, true) + if err != nil { + return err + } + + defer udtConn.Close() + + // loop through the requested TargetBlocks range. + sentBlocks := uint64(0) + + for _, target := range TargetBlocks { + for blockN := target.Offset; blockN < target.Offset+target.Limit; blockN++ { + blockData, status, err := UserBlockchain.GetBlockRaw(blockN) + if err != nil { + protocol.BlockTransferWriteHeader(udtConn, protocol.GetBlockStatusNotAvailable, protocol.BlockRange{Offset: blockN, Limit: 1}, 0) + continue + } + blockSize := uint64(len(blockData)) + + if status != blockchain.StatusOK { + protocol.BlockTransferWriteHeader(udtConn, protocol.GetBlockStatusNotAvailable, protocol.BlockRange{Offset: blockN, Limit: 1}, 0) + continue + } else if blockSize > MaxBlockSize { + protocol.BlockTransferWriteHeader(udtConn, protocol.GetBlockStatusSizeExceed, protocol.BlockRange{Offset: blockN, Limit: 1}, blockSize) + continue + } + + protocol.BlockTransferWriteHeader(udtConn, protocol.GetBlockStatusAvailable, protocol.BlockRange{Offset: blockN, Limit: 1}, blockSize) + udtConn.Write(blockData) + + sentBlocks++ + if sentBlocks >= MaxBlockSize { + break + } + } + } + + return err +} + +// BlockTransferRequest requests blocks from the peer. +// The caller must call udtConn.Close() when done. Do not use any of the closing functions of virtualConn. +func (peer *PeerInfo) BlockTransferRequest(BlockchainPublicKey *btcec.PublicKey, LimitBlockCount uint64, MaxBlockSize uint64, TargetBlocks []protocol.BlockRange) (udtConn net.Conn, virtualConn *virtualPacketConn, err error) { + virtualConn = newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32) { + peer.sendGetBlock(data, protocol.GetBlockControlActive, BlockchainPublicKey, 0, 0, nil, sequenceNumber) + }) + + // new sequence + sequence := networks.Sequences.NewSequenceBi(peer.PublicKey, &peer.messageSequence, virtualConn, blockSequenceTimeout, virtualConn.sequenceTerminate) + if sequence == nil { + return nil, nil, errors.New("cannot acquire sequence") + } + virtualConn.sequenceNumber = sequence.SequenceNumber + + udtConfig := udt.DefaultConfig() + udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSize + udtConfig.MaxFlowWinSize = maxFlowWinSize + + // start UDT receiver + udtListener := udt.ListenUDT(udtConfig, virtualConn, virtualConn.incomingData, virtualConn.outgoingData, virtualConn.terminationSignal) + + // request block transfer + err = peer.sendGetBlock(nil, protocol.GetBlockControlRequestStart, BlockchainPublicKey, LimitBlockCount, MaxBlockSize, TargetBlocks, virtualConn.sequenceNumber) + if err != nil { + udtListener.Close() + return nil, nil, err + } + + // accept the connection + udtConn, err = udtListener.Accept() // TODO: Add timeout! + if err != nil { + udtListener.Close() + return nil, nil, err + } + + // We do not close the UDT listener here. It should automatically close after udtConn is closed. + + return udtConn, virtualConn, nil +}