Peernet-Archive/core
1
0
Fork 0
mirror of https://github.com/PeernetOfficial/core.git synced 2026-07-17 02:47:51 +01:00
Code Issues Packages Projects Releases Wiki Activity

Lite Packet algorithm for data transfer. It bypasses the CPU expensive Peernet Protocol packet encoding (which uses public key signing).

Browse Source
This commit is contained in:
Kleissner
2022-01-17 03:48:40 +01:00
parent 6a5312d834
commit f305e5b158
14 changed files with 403 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
Commands.go
View File

@@ -14,6 +14,7 @@ import (
"github.com/PeernetOfficial/core/dht"
"github.com/PeernetOfficial/core/protocol"
"github.com/PeernetOfficial/core/warehouse"
"github.com/google/uuid"
)
// respondClosesContactsCount is the number of closest contact to respond.
@@ -251,7 +252,7 @@ func (peer *PeerInfo) cmdTransfer(msg *protocol.MessageTransfer, connection *Con
_, fileInfo, status, _ := peer.Backend.UserWarehouse.FileExists(msg.Hash)
if status != warehouse.StatusOK {
// File not available.
peer.sendTransfer(nil, protocol.TransferControlNotAvailable, msg.TransferProtocol, msg.Hash, 0, 0, msg.Sequence)
peer.sendTransfer(nil, protocol.TransferControlNotAvailable, msg.TransferProtocol, msg.Hash, 0, 0, msg.Sequence, uuid.UUID{}, false)
return
} else if msg.Limit > 0 && fileInfo.Size() < int64(msg.Offset)+int64(msg.Limit) {
// If the read limit is out of bounds, this request is considered invalid and silently discarded.
@@ -259,7 +260,7 @@ func (peer *PeerInfo) cmdTransfer(msg *protocol.MessageTransfer, connection *Con
}
// Create a local UDT client to connect to the remote UDT server and serve the file!
go peer.startFileTransferUDT(msg.Hash, uint64(fileInfo.Size()), msg.Offset, msg.Limit, msg.Sequence)
go peer.startFileTransferUDT(msg.Hash, uint64(fileInfo.Size()), msg.Offset, msg.Limit, msg.Sequence, msg.TransferID, msg.TransferProtocol)
case protocol.TransferControlActive:
if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok {
@@ -288,18 +289,18 @@ func (peer *PeerInfo) cmdGetBlock(msg *protocol.MessageGetBlock, connection *Con
case protocol.GetBlockControlRequestStart:
// Currently only support the local blockchain.
if !msg.BlockchainPublicKey.IsEqual(peer.Backend.peerPublicKey) {
peer.sendGetBlock(nil, protocol.GetBlockControlNotAvailable, msg.BlockchainPublicKey, 0, 0, nil, msg.Sequence)
peer.sendGetBlock(nil, protocol.GetBlockControlNotAvailable, msg.BlockchainPublicKey, 0, 0, nil, msg.Sequence, uuid.UUID{}, false)
return
} else if _, height, _ := peer.Backend.UserBlockchain.Header(); height == 0 {
peer.sendGetBlock(nil, protocol.GetBlockControlEmpty, msg.BlockchainPublicKey, 0, 0, nil, msg.Sequence)
peer.sendGetBlock(nil, protocol.GetBlockControlEmpty, msg.BlockchainPublicKey, 0, 0, nil, msg.Sequence, uuid.UUID{}, false)
return
} else if msg.LimitBlockCount == 0 {
peer.sendGetBlock(nil, protocol.GetBlockControlTerminate, msg.BlockchainPublicKey, 0, 0, nil, msg.Sequence)
peer.sendGetBlock(nil, protocol.GetBlockControlTerminate, msg.BlockchainPublicKey, 0, 0, nil, msg.Sequence, uuid.UUID{}, false)
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)
go peer.startBlockTransfer(msg.BlockchainPublicKey, msg.LimitBlockCount, msg.MaxBlockSize, msg.TargetBlocks, msg.Sequence, msg.TransferID)
case protocol.GetBlockControlActive:
if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok {

3
Config Default.yaml
View File

@@ -13,6 +13,9 @@ Listen: []
# Count of workers to process incoming raw packets. Default 2.
ListenWorkers: 0
# Count of workers to process incoming lite packets. Default 2.
ListenWorkersLite: 0
# AutoUpdateSeedList enables auto update of the seed list.
AutoUpdateSeedList: true

5
Config.go
View File

@@ -34,8 +34,9 @@ type Config struct {
LogTarget int `yaml:"LogTarget"`
// Listen settings
Listen []string `yaml:"Listen"` // IP:Port combinations
ListenWorkers int `yaml:"ListenWorkers"` // Count of workers to process incoming raw packets. Default 2.
Listen []string `yaml:"Listen"` // IP:Port combinations
ListenWorkers int `yaml:"ListenWorkers"` // Count of workers to process incoming raw packets. Default 2.
ListenWorkersLite int `yaml:"ListenWorkersLite"` // Count of workers to process incoming lite packets. Default 2.
// User specific settings
PrivateKey string `yaml:"PrivateKey"` // The Private Key, hex encoded so it can be copied manually

39
Connection.go
View File

@@ -458,3 +458,42 @@ func (nets *Networks) sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet
return nil
}
// send sends a raw packet to the peer. Only uses active connections.
func (peer *PeerInfo) sendLite(raw []byte) (err error) {
if peer.isVirtual { // special case for peers that were not contacted before
return errors.New("cannot send lite packet to virtual peer")
} else if len(peer.connectionActive) == 0 {
return errors.New("no valid connection to peer")
} else if atomic.LoadUint64(&peer.StatsPacketSent) == 0 && atomic.LoadUint64(&peer.StatsPacketReceived) == 0 {
return errors.New("uncontacted peer") // A valid connection must have been established.
}
// always count as one sent packet even if sent via broadcast
atomic.AddUint64(&peer.StatsPacketSent, 1)
// Send out the wire. Use connectionLatest if available.
cLatest := peer.connectionLatest
if cLatest != nil {
if err := cLatest.Network.send(cLatest.Address.IP, cLatest.Address.Port, raw); err == nil {
return nil
} else if IsNetworkErrorFatal(err) {
// Invalid connection, immediately invalidate. Fallback to broadcast to all other active ones.
// Windows: A common error when the network adapter is disabled is "wsasendto: The requested address is not valid in its context".
peer.invalidateActiveConnection(cLatest)
}
}
// If no latest connection available, broadcast on all other available connections.
for _, c := range peer.GetConnections(true) {
if c == cLatest {
continue
}
if err := c.Network.send(c.Address.IP, c.Address.Port, raw); err != nil && IsNetworkErrorFatal(err) {
peer.invalidateActiveConnection(c)
}
}
return nil // on broadcast no error is known and returned
}

27
Message Send.go
View File

@@ -11,6 +11,7 @@ import (
"github.com/PeernetOfficial/core/btcec"
"github.com/PeernetOfficial/core/protocol"
"github.com/google/uuid"
)
// pingConnection sends a ping to the target peer via the specified connection
@@ -107,8 +108,17 @@ func (peer *PeerInfo) sendTraverse(packet *protocol.PacketRaw, receiverEnd *btce
}
// sendTransfer sends a transfer message
func (peer *PeerInfo) sendTransfer(data []byte, control, transferProtocol uint8, hash []byte, offset, limit uint64, sequenceNumber uint32) (err error) {
packetRaw, err := protocol.EncodeTransfer(peer.Backend.peerPrivateKey, data, control, transferProtocol, hash, offset, limit)
func (peer *PeerInfo) sendTransfer(data []byte, control, transferProtocol uint8, hash []byte, offset, limit uint64, sequenceNumber uint32, transferID uuid.UUID, isLite bool) (err error) {
// Send optionally as lite packet. This bypasses the signing overhead of regular Peernet packets which is CPU intensive and a bottleneck.
if control == protocol.TransferControlActive && isLite {
raw, err := protocol.PacketLiteEncode(transferID, data)
if err != nil {
return err
}
return peer.sendLite(raw)
}
packetRaw, err := protocol.EncodeTransfer(peer.Backend.peerPrivateKey, data, control, transferProtocol, hash, offset, limit, transferID)
if err != nil {
return err
}
@@ -121,8 +131,17 @@ func (peer *PeerInfo) sendTransfer(data []byte, control, transferProtocol uint8,
}
// 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(peer.Backend.peerPrivateKey, data, control, blockchainPublicKey, limitBlockCount, maxBlockSize, targetBlocks)
func (peer *PeerInfo) sendGetBlock(data []byte, control uint8, blockchainPublicKey *btcec.PublicKey, limitBlockCount, maxBlockSize uint64, targetBlocks []protocol.BlockRange, sequenceNumber uint32, transferID uuid.UUID, isLite bool) (err error) {
// Send optionally as lite packet. This bypasses the signing overhead of regular Peernet packets which is CPU intensive and a bottleneck.
if control == protocol.GetBlockControlActive && isLite {
raw, err := protocol.PacketLiteEncode(transferID, data)
if err != nil {
return err
}
return peer.sendLite(raw)
}
packetRaw, err := protocol.EncodeGetBlock(peer.Backend.peerPrivateKey, data, control, blockchainPublicKey, limitBlockCount, maxBlockSize, targetBlocks, transferID)
if err != nil {
return err
}

6
Network Init.go
View File

@@ -38,9 +38,15 @@ func (backend *Backend) initNetwork() {
if backend.Config.ListenWorkers == 0 {
backend.Config.ListenWorkers = 2
}
if backend.Config.ListenWorkersLite == 0 {
backend.Config.ListenWorkersLite = 2
}
for n := 0; n < backend.Config.ListenWorkers; n++ {
go backend.networks.packetWorker()
}
for n := 0; n < backend.Config.ListenWorkersLite; n++ {
go backend.networks.packetWorkerLite()
}
// check if user specified where to listen
if len(backend.Config.Listen) > 0 {

28
Network.go
View File

@@ -121,6 +121,14 @@ func (network *Network) Listen() {
continue
}
// handle lite packets before regular ones
if isLite, err := network.networkGroup.LiteRouter.IsPacketLite(buffer[:length]); isLite && err != nil {
continue
} else if isLite {
network.networkGroup.litePacketsIncoming <- networkWire{network: network, sender: sender, raw: buffer[:length], receiverPublicKey: network.backend.peerPublicKey, unicast: true}
continue
}
if length < protocol.PacketLengthMin {
// Discard packets that do not meet the minimum length.
continue
@@ -404,3 +412,23 @@ func (backend *Backend) FeatureSupport() (feature byte) {
}
return feature
}
// Handles incoming lite packets. It will decrypt them as needed.
func (nets *Networks) packetWorkerLite() {
for wire := range nets.litePacketsIncoming {
packet, err := nets.LiteRouter.PacketLiteDecode(wire.raw)
if err != nil {
continue
}
// Handle the received data. Note this is called in the same Go routine.
// The underlying data receiver must not stall.
if v, ok := packet.Session.Data.(*virtualPacketConn); ok {
// update stats TODO
//atomic.AddUint64(&packet.Session.Data.(*virtualPacketConn).peer.StatsPacketReceived, 1)
//connection.LastPacketIn = time.Now()
v.receiveData(packet.Payload)
}
}
}

10
Networks.go
View File

@@ -26,11 +26,15 @@ type Networks struct {
countListen4, countListen6 int64
// channel for processing incoming decoded packets by workers, across all networks
rawPacketsIncoming chan networkWire
rawPacketsIncoming chan networkWire
litePacketsIncoming chan networkWire
// Sequences keeps track of all message sequence number, regardless of the network connection.
Sequences *protocol.SequenceManager
// Keep track of valid IDs for lite packets.
LiteRouter *protocol.LiteRouter
// ipListen keeps a simple list of IPs listened to. This allows quickly identifying if an IP matches with a listened one.
ipListen *ipList
@@ -51,9 +55,11 @@ const ReplyTimeout = 20
func (backend *Backend) initMessageSequence() {
backend.networks = &Networks{backend: backend}
backend.networks.rawPacketsIncoming = make(chan networkWire, 1000) // buffer up to 1000 UDP packets before they get buffered by the OS network stack and eventually dropped
backend.networks.rawPacketsIncoming = make(chan networkWire, 1000) // buffer up to 1000 UDP packets before they get buffered by the OS network stack and eventually dropped
backend.networks.litePacketsIncoming = make(chan networkWire, 1000) // buffer up to 1000 UDP packets before they get buffered by the OS network stack and eventually dropped
backend.networks.Sequences = protocol.NewSequenceManager(ReplyTimeout)
backend.networks.LiteRouter = protocol.NewLiteRouter()
backend.networks.ipListen = NewIPList()

33
Transfer Block.go
View File

@@ -15,23 +15,32 @@ import (
"github.com/PeernetOfficial/core/btcec"
"github.com/PeernetOfficial/core/protocol"
"github.com/PeernetOfficial/core/udt"
"github.com/google/uuid"
)
// blockSequenceTimeout is the timeout for a follow-up message to appear, otherwise the transfer will be terminated.
var blockSequenceTimeout = time.Second * 10
// Whether to use the lite protocol for transfer of data.
const blockTransferLite = true
// 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)
func (peer *PeerInfo) startBlockTransfer(BlockchainPublicKey *btcec.PublicKey, LimitBlockCount uint64, MaxBlockSize uint64, TargetBlocks []protocol.BlockRange, sequenceNumber uint32, transferID uuid.UUID) (err error) {
virtualConn := newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32, transferID uuid.UUID) {
peer.sendGetBlock(data, protocol.GetBlockControlActive, BlockchainPublicKey, 0, 0, nil, sequenceNumber, transferID, blockTransferLite)
})
// use the transfer ID indicated by the remote peer
// 17.01.2021: Due to using lite IDs, the sequence termination function in RegisterSequenceBi is no longer used, as data packets are only sent via lite packets.
virtualConn.transferID = transferID
peer.Backend.networks.LiteRouter.RegisterLiteID(transferID, virtualConn, blockSequenceTimeout, virtualConn.sequenceTerminate)
// register the sequence since packets are sent bi-directional
virtualConn.sequenceNumber = sequenceNumber
peer.Backend.networks.Sequences.RegisterSequenceBi(peer.PublicKey, sequenceNumber, virtualConn, blockSequenceTimeout, virtualConn.sequenceTerminate)
peer.Backend.networks.Sequences.RegisterSequenceBi(peer.PublicKey, sequenceNumber, virtualConn, blockSequenceTimeout, nil)
udtConfig := udt.DefaultConfig()
udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSize
udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSizeLite
udtConfig.MaxFlowWinSize = maxFlowWinSize
// start UDT sender
@@ -79,26 +88,30 @@ func (peer *PeerInfo) startBlockTransfer(BlockchainPublicKey *btcec.PublicKey, L
// 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)
virtualConn = newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32, transferID uuid.UUID) {
peer.sendGetBlock(data, protocol.GetBlockControlActive, BlockchainPublicKey, 0, 0, nil, sequenceNumber, transferID, blockTransferLite)
})
// new lite ID
liteID := peer.Backend.networks.LiteRouter.NewLiteID(virtualConn, blockSequenceTimeout, virtualConn.sequenceTerminate)
virtualConn.transferID = liteID.ID
// new sequence
sequence := peer.Backend.networks.Sequences.NewSequenceBi(peer.PublicKey, &peer.messageSequence, virtualConn, blockSequenceTimeout, virtualConn.sequenceTerminate)
sequence := peer.Backend.networks.Sequences.NewSequenceBi(peer.PublicKey, &peer.messageSequence, virtualConn, blockSequenceTimeout, nil)
if sequence == nil {
return nil, nil, errors.New("cannot acquire sequence")
}
virtualConn.sequenceNumber = sequence.SequenceNumber
udtConfig := udt.DefaultConfig()
udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSize
udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSizeLite
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)
err = peer.sendGetBlock(nil, protocol.GetBlockControlRequestStart, BlockchainPublicKey, LimitBlockCount, MaxBlockSize, TargetBlocks, virtualConn.sequenceNumber, virtualConn.transferID, false)
if err != nil {
udtListener.Close()
return nil, nil, err

35
Transfer UDT.go
View File

@@ -15,6 +15,7 @@ import (
"github.com/PeernetOfficial/core/protocol"
"github.com/PeernetOfficial/core/udt"
"github.com/google/uuid"
)
// transferSequenceTimeout is the timeout for a follow-up message to appear, otherwise the transfer will be terminated.
@@ -25,9 +26,12 @@ var transferSequenceTimeout = time.Minute * 1
// The actual used number will be negotiated through the UDT handshake and must be a minimum of 32.
const maxFlowWinSize = 64
// Whether to use the lite protocol for transfer of data.
const transferLite = true
// startFileTransferUDT starts a file transfer from the local warehouse to the remote peer.
// It creates a virtual UDT client to transfer data to a remote peer. Counterintuitively, this will be the "file server" peer.
func (peer *PeerInfo) startFileTransferUDT(hash []byte, fileSize uint64, offset, limit uint64, sequenceNumber uint32) (err error) {
func (peer *PeerInfo) startFileTransferUDT(hash []byte, fileSize uint64, offset, limit uint64, sequenceNumber uint32, transferID uuid.UUID, transferProtocol uint8) (err error) {
if limit > 0 && offset+limit > fileSize {
return errors.New("invalid limit")
} else if offset > fileSize {
@@ -36,16 +40,21 @@ func (peer *PeerInfo) startFileTransferUDT(hash []byte, fileSize uint64, offset,
limit = fileSize - offset
}
virtualConnection := newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32) {
peer.sendTransfer(data, protocol.TransferControlActive, 0, hash, offset, limit, sequenceNumber)
virtualConnection := newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32, transferID uuid.UUID) {
peer.sendTransfer(data, protocol.TransferControlActive, 0, hash, offset, limit, sequenceNumber, transferID, transferLite)
})
// use the transfer ID indicated by the remote peer
// 17.01.2021: Due to using lite IDs, the sequence termination function in RegisterSequenceBi is no longer used, as data packets are only sent via lite packets.
virtualConnection.transferID = transferID
peer.Backend.networks.LiteRouter.RegisterLiteID(transferID, virtualConnection, transferSequenceTimeout, virtualConnection.sequenceTerminate)
// register the sequence since packets are sent bi-directional
virtualConnection.sequenceNumber = sequenceNumber
peer.Backend.networks.Sequences.RegisterSequenceBi(peer.PublicKey, sequenceNumber, virtualConnection, transferSequenceTimeout, virtualConnection.sequenceTerminate)
peer.Backend.networks.Sequences.RegisterSequenceBi(peer.PublicKey, sequenceNumber, virtualConnection, transferSequenceTimeout, nil)
udtConfig := udt.DefaultConfig()
udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSize
udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSizeLite
udtConfig.MaxFlowWinSize = maxFlowWinSize
// start UDT sender
@@ -65,30 +74,34 @@ func (peer *PeerInfo) startFileTransferUDT(hash []byte, fileSize uint64, offset,
return err
}
// FileTransferRequestUDT creates a UDT server listening for incoming data transfer and requests a file transfer from a remote peer.
// FileTransferRequestUDT creates a UDT server listening for incoming data transfer via the lite protocol and requests a file transfer from a remote peer.
// The caller must call udtConn.Close() when done. Do not use any of the closing functions of virtualConn.
// Limit is optional. 0 means the entire file.
func (peer *PeerInfo) FileTransferRequestUDT(hash []byte, offset, limit uint64) (udtConn net.Conn, virtualConn *virtualPacketConn, err error) {
virtualConn = newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32) {
peer.sendTransfer(data, protocol.TransferControlActive, protocol.TransferProtocolUDT, hash, offset, limit, sequenceNumber)
virtualConn = newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32, transferID uuid.UUID) {
peer.sendTransfer(data, protocol.TransferControlActive, protocol.TransferProtocolUDT, hash, offset, limit, sequenceNumber, transferID, transferLite)
})
// new lite ID
liteID := peer.Backend.networks.LiteRouter.NewLiteID(virtualConn, transferSequenceTimeout, virtualConn.sequenceTerminate)
virtualConn.transferID = liteID.ID
// new sequence
sequence := peer.Backend.networks.Sequences.NewSequenceBi(peer.PublicKey, &peer.messageSequence, virtualConn, transferSequenceTimeout, virtualConn.sequenceTerminate)
sequence := peer.Backend.networks.Sequences.NewSequenceBi(peer.PublicKey, &peer.messageSequence, virtualConn, transferSequenceTimeout, nil)
if sequence == nil {
return nil, nil, errors.New("cannot acquire sequence")
}
virtualConn.sequenceNumber = sequence.SequenceNumber
udtConfig := udt.DefaultConfig()
udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSize
udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSizeLite
udtConfig.MaxFlowWinSize = maxFlowWinSize
// start UDT receiver
udtListener := udt.ListenUDT(udtConfig, virtualConn, virtualConn.incomingData, virtualConn.outgoingData, virtualConn.terminationSignal)
// request file transfer
peer.sendTransfer(nil, protocol.TransferControlRequestStart, protocol.TransferProtocolUDT, hash, offset, limit, virtualConn.sequenceNumber)
peer.sendTransfer(nil, protocol.TransferControlRequestStart, protocol.TransferProtocolUDT, hash, offset, limit, virtualConn.sequenceNumber, virtualConn.transferID, false)
// accept the connection
udtConn, err = udtListener.Accept()

11
Transfer Virtual Connection.go
View File

@@ -11,6 +11,8 @@ package core
import (
"sync"
"github.com/google/uuid"
)
// virtualPacketConn is a virtual connection.
@@ -18,11 +20,14 @@ type virtualPacketConn struct {
peer *PeerInfo
// function to send data to the remote peer
sendData func(data []byte, sequenceNumber uint32)
sendData func(data []byte, sequenceNumber uint32, transferID uuid.UUID)
// Sequence number from the first outgoing or incoming packet.
sequenceNumber uint32
// Transfer ID represents a session ID valid only for the duration of the transfer.
transferID uuid.UUID
// data channel
incomingData chan []byte
outgoingData chan []byte
@@ -35,7 +40,7 @@ type virtualPacketConn struct {
}
// newVirtualPacketConn creates a new virtual connection (both incomign and outgoing).
func newVirtualPacketConn(peer *PeerInfo, sendData func(data []byte, sequenceNumber uint32)) (v *virtualPacketConn) {
func newVirtualPacketConn(peer *PeerInfo, sendData func(data []byte, sequenceNumber uint32, transferID uuid.UUID)) (v *virtualPacketConn) {
v = &virtualPacketConn{
peer: peer,
sendData: sendData,
@@ -54,7 +59,7 @@ func (v *virtualPacketConn) writeForward() {
for {
select {
case data := <-v.outgoingData:
v.sendData(data, v.sequenceNumber)
v.sendData(data, v.sequenceNumber, v.transferID)
case <-v.terminationSignal:
return

30
protocol/Message Encoding Get Block.go
View File

@@ -11,8 +11,9 @@ Offset Size Info
Control = 0: Request Blocks
34 8 Limit total count of blocks to transfer. The transfer will be terminated if the limit is reached.
42 8 Limit of bytes per block to transfer max. Blocks exceeding this limit will not be transferred.
50 2 Count of block ranges
52 16 * ? List of block ranges
50 16 Transfer ID. This will identify lite packets.
66 2 Count of block ranges
68 16 * ? List of block ranges
Block range:
0 8 Block number
@@ -41,6 +42,7 @@ import (
"io"
"github.com/PeernetOfficial/core/btcec"
"github.com/google/uuid"
)
const (
@@ -57,6 +59,9 @@ const (
GetBlockStatusSizeExceed = 2
)
// Min size of header for Get Block control 0 message.
const getBlockRequestHeaderSize = 68
// MessageGetBlock is the decoded Get Block message.
type MessageGetBlock struct {
*MessageRaw // Underlying raw message.
@@ -64,6 +69,7 @@ type MessageGetBlock struct {
BlockchainPublicKey *btcec.PublicKey // Peer ID of blockchain to transfer.
// fields valid only for GetBlockControlRequestStart
TransferID uuid.UUID // Transfer ID to identify lite packets.
LimitBlockCount uint64 // Limit total count of blocks to transfer
MaxBlockSize uint64 // Limit of bytes per block to transfer max. Blocks exceeding this limit will not be transferred.
TargetBlocks []BlockRange // Target list of block ranges to transfer.
@@ -96,21 +102,22 @@ func DecodeGetBlock(msg *MessageRaw) (result *MessageGetBlock, err error) {
}
if result.Control == GetBlockControlRequestStart {
if len(msg.Payload) < 52 {
if len(msg.Payload) < getBlockRequestHeaderSize {
return nil, errors.New("get block: invalid minimum length")
}
result.LimitBlockCount = binary.LittleEndian.Uint64(msg.Payload[34 : 34+8])
result.MaxBlockSize = binary.LittleEndian.Uint64(msg.Payload[42 : 42+8])
copy(result.TransferID[:], msg.Payload[50:50+16])
countBlockRanges := int(binary.LittleEndian.Uint16(msg.Payload[50:52]))
countBlockRanges := int(binary.LittleEndian.Uint16(msg.Payload[66 : 66+2]))
if countBlockRanges == 0 {
return nil, errors.New("get block: empty block range")
} else if len(msg.Payload) < 52+16*countBlockRanges {
} else if len(msg.Payload) < getBlockRequestHeaderSize+16*countBlockRanges {
return nil, errors.New("get block: cound block ranges exceeds length")
}
index := 52
index := getBlockRequestHeaderSize
for n := 0; n < countBlockRanges; n++ {
var target BlockRange
@@ -128,18 +135,18 @@ func DecodeGetBlock(msg *MessageRaw) (result *MessageGetBlock, err error) {
}
// EncodeGetBlock encodes a Get Block message. The embedded packet size must be smaller than TransferMaxEmbedSize.
func EncodeGetBlock(senderPrivateKey *btcec.PrivateKey, data []byte, control uint8, blockchainPublicKey *btcec.PublicKey, limitBlockCount, maxBlockSize uint64, targetBlocks []BlockRange) (packetRaw []byte, err error) {
func EncodeGetBlock(senderPrivateKey *btcec.PrivateKey, data []byte, control uint8, blockchainPublicKey *btcec.PublicKey, limitBlockCount, maxBlockSize uint64, targetBlocks []BlockRange, transferID uuid.UUID) (packetRaw []byte, err error) {
if control == GetBlockControlRequestStart && len(data) != 0 {
return nil, errors.New("get block encode: payload not allowed in start")
} else if isPacketSizeExceed(transferPayloadHeaderSize, len(data)) {
return nil, errors.New("get block encode: embedded packet too big")
} else if control == GetBlockControlRequestStart && isPacketSizeExceed(52, len(targetBlocks)*16) {
} else if control == GetBlockControlRequestStart && isPacketSizeExceed(getBlockRequestHeaderSize, len(targetBlocks)*16) {
return nil, errors.New("get block encode: too many target block ranges")
}
packetSize := transferPayloadHeaderSize
if control == GetBlockControlRequestStart {
packetSize = 52 + len(targetBlocks)*16
packetSize = getBlockRequestHeaderSize + len(targetBlocks)*16
} else if control == GetBlockControlActive {
packetSize += len(data)
}
@@ -153,9 +160,10 @@ func EncodeGetBlock(senderPrivateKey *btcec.PrivateKey, data []byte, control uin
if control == GetBlockControlRequestStart {
binary.LittleEndian.PutUint64(raw[34:34+8], limitBlockCount)
binary.LittleEndian.PutUint64(raw[42:42+8], maxBlockSize)
binary.LittleEndian.PutUint16(raw[50:50+2], uint16(len(targetBlocks)))
copy(raw[50:50+16], transferID[:])
binary.LittleEndian.PutUint16(raw[66:66+2], uint16(len(targetBlocks)))
index := 52
index := getBlockRequestHeaderSize
for _, target := range targetBlocks {
binary.LittleEndian.PutUint64(raw[index:index+8], target.Offset)
binary.LittleEndian.PutUint64(raw[index+8:index+16], target.Limit)

39
protocol/Message Encoding Transfer.go
View File

@@ -12,11 +12,10 @@ Offset Size Info
Control = 0: Request Start
34 8 Offset to start reading in the file
42 8 Limit of bytes to read at the offset
50 16 Transfer ID. This will identify lite packets.
Offset + limit must not exceed the file size.
Control = 3: Active
34 ? Embedded protocol data
Offset + limit must not exceed the file size. Actual data transfer should be sent via lite packets.
The regular Peernet packets would be too CPU expensive and slow due to public key signing.
*/
@@ -27,18 +26,20 @@ import (
"errors"
"github.com/PeernetOfficial/core/btcec"
"github.com/google/uuid"
)
// MessageTransfer is the decoded transfer message.
// It is sent to initiate a file transfer, and to send data as part of a file transfer. The actual file data is encapsulated via UDT.
type MessageTransfer struct {
*MessageRaw // Underlying raw message.
Control uint8 // Control. See TransferControlX.
TransferProtocol uint8 // Embedded transfer protocol: 0 = UDT
Hash []byte // Hash of the file to transfer.
Offset uint64 // Offset to start reading at. Only TransferControlRequestStart.
Limit uint64 // Limit (count of bytes) to read starting at the offset. Only TransferControlRequestStart.
Data []byte // Embedded protocol data. Only TransferControlActive.
*MessageRaw // Underlying raw message.
Control uint8 // Control. See TransferControlX.
TransferProtocol uint8 // Embedded transfer protocol: 0 = UDT
Hash []byte // Hash of the file to transfer.
Offset uint64 // Offset to start reading at. Only TransferControlRequestStart.
Limit uint64 // Limit (count of bytes) to read starting at the offset. Only TransferControlRequestStart.
TransferID uuid.UUID // Transfer ID to identify lite packets.
Data []byte // Embedded protocol data. Only TransferControlActive.
}
const (
@@ -48,7 +49,9 @@ const (
TransferControlTerminate = 3 // Terminate
)
const TransferProtocolUDT = 0 // Indicates that UDT is used as embedded transfer protocol.
const (
TransferProtocolUDT = 0 // UDT via lite packets. No encryption.
)
const transferPayloadHeaderSize = 34
@@ -76,9 +79,11 @@ func DecodeTransfer(msg *MessageRaw) (result *MessageTransfer, err error) {
result.Offset = binary.LittleEndian.Uint64(msg.Payload[34 : 34+8])
result.Limit = binary.LittleEndian.Uint64(msg.Payload[42 : 42+8])
copy(result.TransferID[:], msg.Payload[50:50+16])
case TransferControlActive:
result.Data = msg.Payload[34:]
// Data should be transferred via lite packets for performance reasons, but it is allowed to be encapsulated in Peernet packets.
result.Data = msg.Payload[transferPayloadHeaderSize:]
}
@@ -90,8 +95,11 @@ func DecodeTransfer(msg *MessageRaw) (result *MessageTransfer, err error) {
// The caller may send bigger payloads but may risk that data packets are simply dropped and never arrive. A MTU negotiation or detection could pimp that.
const TransferMaxEmbedSize = internetSafeMTU - PacketLengthMin - transferPayloadHeaderSize
// Same as TransferMaxEmbedSize but for encoding via lite packets.
const TransferMaxEmbedSizeLite = internetSafeMTU - PacketLiteSizeMin
// EncodeTransfer encodes a transfer message. The embedded packet size must be smaller than TransferMaxEmbedSize.
func EncodeTransfer(senderPrivateKey *btcec.PrivateKey, data []byte, control, transferProtocol uint8, hash []byte, offset, limit uint64) (packetRaw []byte, err error) {
func EncodeTransfer(senderPrivateKey *btcec.PrivateKey, data []byte, control, transferProtocol uint8, hash []byte, offset, limit uint64, transferID uuid.UUID) (packetRaw []byte, err error) {
if control == TransferControlRequestStart && len(data) != 0 {
return nil, errors.New("transfer encode: payload not allowed in start")
} else if isPacketSizeExceed(transferPayloadHeaderSize, len(data)) {
@@ -100,7 +108,7 @@ func EncodeTransfer(senderPrivateKey *btcec.PrivateKey, data []byte, control, tr
packetSize := transferPayloadHeaderSize
if control == TransferControlRequestStart {
packetSize += 16
packetSize += 32
} else if control == TransferControlActive {
packetSize += len(data)
}
@@ -114,6 +122,7 @@ func EncodeTransfer(senderPrivateKey *btcec.PrivateKey, data []byte, control, tr
if control == TransferControlRequestStart {
binary.LittleEndian.PutUint64(raw[34:34+8], offset)
binary.LittleEndian.PutUint64(raw[42:42+8], limit)
copy(raw[50:50+16], transferID[:])
} else if control == TransferControlActive {
copy(raw[34:34+len(data)], data)
}

188
protocol/Packet Lite.go Normal file
View File

@@ -0,0 +1,188 @@
/*
File Name: Packet Lite.go
Copyright: 2021 Peernet s.r.o.
Author: Peter Kleissner
The lite packet header is used for encoding data transfer packets. The regular header is too expensive in terms of CPU consumption due to public key signing.
Instead, a simple session ID will identify lite packets. The ID is randomized and only valid during the session.
Unsolicited lite packets are therefore impossible; the receiver must have the ID already whitelisted for the packet to be recognized.
Offset Size Info
0 16 ID
16 2 Size of data to follow
*/
package protocol
import (
"encoding/binary"
"errors"
"sync"
"time"
"github.com/google/uuid"
)
// PacketLiteRaw is a decrypted P2P lite packet
type PacketLiteRaw struct {
ID uuid.UUID // ID
Payload []byte // Payload
Session *LiteID // Session info
}
// Minimum packet size of lite packets.
const PacketLiteSizeMin = 16 + 2
// IsPacketLite identifies a lite packet based on its ID. If the ID is not recognized, it fails.
func (router *LiteRouter) IsPacketLite(raw []byte) (isLite bool, err error) {
if len(raw) < PacketLiteSizeMin {
return false, errors.New("invalid packet size")
}
// Parse the ID and then look it up.
var id uuid.UUID
copy(id[:], raw[0:16])
return router.LookupLiteID(id) != nil, nil
}
// PacketLiteDecode a lite packet. It will identify the lite packet based on its ID. If the ID is not recognized (which is the case for regular Peernet packets), the function fails.
// It does not perform any decryption.
func (router *LiteRouter) PacketLiteDecode(raw []byte) (packet *PacketLiteRaw, err error) {
if len(raw) < PacketLiteSizeMin {
return nil, errors.New("invalid packet size")
}
// Parse the ID and look it up. It will contain information about the decryption algorithm to use.
var id uuid.UUID
copy(id[:], raw[0:16])
session := router.LookupLiteID(id)
if session == nil {
return nil, errors.New("packet ID not found")
}
// TODO: Decrypt the data if indicated by the session.
sizePayload := binary.LittleEndian.Uint16(raw[16 : 16+2])
if int(sizePayload) > len(raw)-PacketLiteSizeMin { // invalid size field?
return nil, errors.New("invalid packet size field")
}
// Valid packet received, extend expiration.
session.expires = time.Now().Add(session.timeout)
return &PacketLiteRaw{Payload: raw[PacketLiteSizeMin:], ID: id, Session: session}, nil
}
// Encodes a lite packet.
func PacketLiteEncode(id uuid.UUID, data []byte) (raw []byte, err error) {
raw = make([]byte, PacketLiteSizeMin+len(data))
copy(raw[0:16], id[:])
binary.LittleEndian.PutUint16(raw[16:16+2], uint16(len(data)))
copy(raw[PacketLiteSizeMin:], data)
return raw, nil
}
// ---- Lite packet ID management. This is similar to packet sequences. ----
// LiteRouter keeps track of accepted (expected) packet IDs.
type LiteRouter struct {
// list of recognized IDs
ids map[uuid.UUID]*LiteID
sync.Mutex // synchronized access to the IDs
}
// LiteID contains session information for a bidirectional transfer of data
type LiteID struct {
ID uuid.UUID // ID
created time.Time // When the ID was created.
expires time.Time // When the ID expires. This can be extended on the fly!
Data interface{} // Optional high-level data associated with the ID
timeout time.Duration // Timeout for receiving the next message
invalidateFunc func() // Called on expiration.
}
// Creates a new manager to keep track of accepted IDs.
func NewLiteRouter() (router *LiteRouter) {
router = &LiteRouter{
ids: make(map[uuid.UUID]*LiteID),
}
go router.autoDeleteExpired()
return
}
// autoDeleteExpired deletes all IDs that are expired.
func (router *LiteRouter) autoDeleteExpired() {
for {
time.Sleep(4 * time.Second)
now := time.Now()
router.Lock()
for id, info := range router.ids {
if info.expires.Before(now) {
delete(router.ids, id)
if info.invalidateFunc != nil {
go info.invalidateFunc()
}
}
}
router.Unlock()
}
}
func (router *LiteRouter) LookupLiteID(id uuid.UUID) (info *LiteID) {
router.Lock()
info = router.ids[id]
router.Unlock()
return info
}
// Returns a new lite ID to be used.
func (router *LiteRouter) NewLiteID(data interface{}, timeout time.Duration, invalidateFunc func()) (info *LiteID) {
info = &LiteID{
created: time.Now(),
expires: time.Now().Add(timeout),
timeout: timeout,
invalidateFunc: invalidateFunc,
Data: data,
ID: uuid.New(),
}
router.Lock()
router.ids[info.ID] = info
router.Unlock()
return
}
func (router *LiteRouter) RegisterLiteID(id uuid.UUID, data interface{}, timeout time.Duration, invalidateFunc func()) (info *LiteID) {
info = &LiteID{
ID: id,
created: time.Now(),
expires: time.Now().Add(timeout),
timeout: timeout,
invalidateFunc: invalidateFunc,
Data: data,
}
router.Lock()
existingInfo := router.ids[info.ID]
router.ids[info.ID] = info
router.Unlock()
// Call the invalidate function if there is a collision. This should never happen.
if existingInfo != nil && existingInfo.invalidateFunc != nil {
go existingInfo.invalidateFunc()
}
return
}