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

VirtualPacketConn: New Stats field maintained by caller. This allows insight into active transfers.

Browse Source 
New FileTransferStats and BlockTransferStats structures.
This commit is contained in:
Kleissner
2022-03-16 23:52:55 +01:00
parent 51336263f5
commit cb1441c97a
8 changed files with 88 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
Commands.go
View File

@@ -263,19 +263,19 @@ func (peer *PeerInfo) cmdTransfer(msg *protocol.MessageTransfer, connection *Con
go peer.startFileTransferUDT(msg.Hash, fileSize, msg.Offset, msg.Limit, msg.Sequence, msg.TransferID, msg.TransferProtocol)
case protocol.TransferControlActive:
if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok {
if v, ok := msg.SequenceInfo.Data.(*VirtualPacketConn); ok {
go v.receiveData(msg.Data)
return
}
case protocol.TransferControlNotAvailable:
if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok {
if v, ok := msg.SequenceInfo.Data.(*VirtualPacketConn); ok {
v.Terminate(404)
return
}
case protocol.TransferControlTerminate:
if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok {
if v, ok := msg.SequenceInfo.Data.(*VirtualPacketConn); ok {
v.Terminate(2)
return
}
@@ -303,25 +303,25 @@ func (peer *PeerInfo) cmdGetBlock(msg *protocol.MessageGetBlock, connection *Con
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 {
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 {
if v, ok := msg.SequenceInfo.Data.(*VirtualPacketConn); ok {
v.Terminate(404)
return
}
case protocol.GetBlockControlEmpty:
if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok {
if v, ok := msg.SequenceInfo.Data.(*VirtualPacketConn); ok {
v.Terminate(410)
return
}
case protocol.GetBlockControlTerminate:
if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok {
if v, ok := msg.SequenceInfo.Data.(*VirtualPacketConn); ok {
v.Terminate(2)
return
}

4
Network.go
View File

@@ -423,9 +423,9 @@ func (nets *Networks) packetWorkerLite() {
// 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 {
if v, ok := packet.Session.Data.(*VirtualPacketConn); ok {
// update stats TODO
//atomic.AddUint64(&packet.Session.Data.(*virtualPacketConn).peer.StatsPacketReceived, 1)
//atomic.AddUint64(&packet.Session.Data.(*VirtualPacketConn).peer.StatsPacketReceived, 1)
//connection.LastPacketIn = time.Now()
v.receiveData(packet.Payload)

5
Networks.go
View File

@@ -87,3 +87,8 @@ func (backend *Backend) firewallDetectIndicatorFile() {
os.Remove(firewallIndicatorFile)
}
}
// List of all lite sessions
func (backend *Backend) LiteSessions() (sessions []*protocol.LiteID) {
return backend.networks.LiteRouter.All()
}

16
Transfer Block.go
View File

@@ -8,7 +8,6 @@ package core
import (
"errors"
"net"
"time"
"github.com/PeernetOfficial/core/blockchain"
@@ -29,6 +28,7 @@ func (peer *PeerInfo) startBlockTransfer(BlockchainPublicKey *btcec.PublicKey, L
virtualConn := newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32, transferID uuid.UUID) {
peer.sendGetBlock(data, protocol.GetBlockControlActive, BlockchainPublicKey, 0, 0, nil, sequenceNumber, transferID, blockTransferLite)
})
virtualConn.Stats = &BlockTransferStats{BlockchainPublicKey: BlockchainPublicKey, Direction: DirectionOut, LimitBlockCount: LimitBlockCount, MaxBlockSize: MaxBlockSize, TargetBlocks: TargetBlocks}
// 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.
@@ -51,6 +51,7 @@ func (peer *PeerInfo) startBlockTransfer(BlockchainPublicKey *btcec.PublicKey, L
}
defer udtConn.Close()
virtualConn.Stats.(*BlockTransferStats).UDTConn = udtConn
// loop through the requested TargetBlocks range.
sentBlocks := uint64(0)
@@ -87,10 +88,11 @@ 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) {
func (peer *PeerInfo) BlockTransferRequest(BlockchainPublicKey *btcec.PublicKey, LimitBlockCount uint64, MaxBlockSize uint64, TargetBlocks []protocol.BlockRange) (udtConn *udt.UDTSocket, virtualConn *VirtualPacketConn, 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)
})
virtualConn.Stats = &BlockTransferStats{BlockchainPublicKey: BlockchainPublicKey, Direction: DirectionIn, LimitBlockCount: LimitBlockCount, MaxBlockSize: MaxBlockSize, TargetBlocks: TargetBlocks}
// new lite ID
liteID := peer.Backend.networks.LiteRouter.NewLiteID(virtualConn, blockSequenceTimeout, virtualConn.sequenceTerminate)
@@ -123,6 +125,7 @@ func (peer *PeerInfo) BlockTransferRequest(BlockchainPublicKey *btcec.PublicKey,
udtListener.Close()
return nil, nil, err
}
virtualConn.Stats.(*BlockTransferStats).UDTConn = udtConn
// We do not close the UDT listener here. It should automatically close after udtConn is closed.
@@ -169,3 +172,12 @@ func isTargetInRange(targets []protocol.BlockRange, offset, limit uint64) (valid
return false
}
type BlockTransferStats struct {
BlockchainPublicKey *btcec.PublicKey // Target blockchain
Direction int // Direction of the data transfer
LimitBlockCount uint64 // Max count of blocks to be transferred
MaxBlockSize uint64 // Max single block size to transfer
TargetBlocks []protocol.BlockRange // List of blocks to transfer
UDTConn *udt.UDTSocket // Underlying UDT connection
}

34
Transfer UDT.go
View File

@@ -39,18 +39,19 @@ func (peer *PeerInfo) startFileTransferUDT(hash []byte, fileSize uint64, offset,
limit = fileSize - offset
}
virtualConnection := newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32, transferID uuid.UUID) {
virtualConn := newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32, transferID uuid.UUID) {
peer.sendTransfer(data, protocol.TransferControlActive, 0, hash, offset, limit, sequenceNumber, transferID, transferLite)
})
virtualConn.Stats = &FileTransferStats{Hash: hash, Direction: DirectionOut, FileSize: fileSize, Offset: offset, Limit: limit}
// 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)
virtualConn.transferID = transferID
peer.Backend.networks.LiteRouter.RegisterLiteID(transferID, virtualConn, transferSequenceTimeout, virtualConn.sequenceTerminate)
// register the sequence since packets are sent bi-directional
virtualConnection.sequenceNumber = sequenceNumber
peer.Backend.networks.Sequences.RegisterSequenceBi(peer.PublicKey, sequenceNumber, virtualConnection, transferSequenceTimeout, nil)
virtualConn.sequenceNumber = sequenceNumber
peer.Backend.networks.Sequences.RegisterSequenceBi(peer.PublicKey, sequenceNumber, virtualConn, transferSequenceTimeout, nil)
udtConfig := udt.DefaultConfig()
udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSizeLite
@@ -58,12 +59,13 @@ func (peer *PeerInfo) startFileTransferUDT(hash []byte, fileSize uint64, offset,
// 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, virtualConnection, virtualConnection.incomingData, virtualConnection.outgoingData, virtualConnection.terminationSignal, true)
udtConn, err := udt.DialUDT(udtConfig, virtualConn, virtualConn.incomingData, virtualConn.outgoingData, virtualConn.terminationSignal, true)
if err != nil {
return err
}
defer udtConn.Close()
virtualConn.Stats.(*FileTransferStats).UDTConn = udtConn
// First send the header (Total File Size, Transfer Size) and then the file data.
protocol.FileTransferWriteHeader(udtConn, fileSize, limit)
@@ -76,7 +78,7 @@ func (peer *PeerInfo) startFileTransferUDT(hash []byte, fileSize uint64, offset,
// 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 *udt.UDTSocket, virtualConn *virtualPacketConn, err error) {
func (peer *PeerInfo) FileTransferRequestUDT(hash []byte, offset, limit uint64) (udtConn *udt.UDTSocket, virtualConn *VirtualPacketConn, err error) {
virtualConn = newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32, transferID uuid.UUID) {
peer.sendTransfer(data, protocol.TransferControlActive, protocol.TransferProtocolUDT, hash, offset, limit, sequenceNumber, transferID, transferLite)
})
@@ -84,6 +86,7 @@ func (peer *PeerInfo) FileTransferRequestUDT(hash []byte, offset, limit uint64)
// new lite ID
liteID := peer.Backend.networks.LiteRouter.NewLiteID(virtualConn, transferSequenceTimeout, virtualConn.sequenceTerminate)
virtualConn.transferID = liteID.ID
virtualConn.Stats = &FileTransferStats{Hash: hash, Direction: DirectionIn, Offset: offset, Limit: limit}
// new sequence
sequence := peer.Backend.networks.Sequences.NewSequenceBi(peer.PublicKey, &peer.messageSequence, virtualConn, transferSequenceTimeout, nil)
@@ -108,8 +111,25 @@ func (peer *PeerInfo) FileTransferRequestUDT(hash []byte, offset, limit uint64)
udtListener.Close()
return nil, nil, err
}
virtualConn.Stats.(*FileTransferStats).UDTConn = udtConn
// We do not close the UDT listener here. It should automatically close after udtConn is closed.
return udtConn, virtualConn, nil
}
type FileTransferStats struct {
Hash []byte // Hash of the file to transfer
Direction int // Direction of the data transfer
FileSize uint64 // File size if known
Offset uint64 // Offset to start the transfer
Limit uint64 // Limit in bytes to transfer
UDTConn *udt.UDTSocket // Underlying UDT connection
}
// Transfer directions
const (
DirectionIn = 0
DirectionOut = 1
DirectionBi = 2
)

35
Transfer Virtual Connection.go
View File

@@ -15,9 +15,12 @@ import (
"github.com/google/uuid"
)
// virtualPacketConn is a virtual connection.
type virtualPacketConn struct {
peer *PeerInfo
// VirtualPacketConn is a virtual connection.
type VirtualPacketConn struct {
Peer *PeerInfo
// Stats are maintained by the caller
Stats interface{}
// function to send data to the remote peer
sendData func(data []byte, sequenceNumber uint32, transferID uuid.UUID)
@@ -39,10 +42,10 @@ type virtualPacketConn struct {
sync.Mutex
}
// newVirtualPacketConn creates a new virtual connection (both incomign and outgoing).
func newVirtualPacketConn(peer *PeerInfo, sendData func(data []byte, sequenceNumber uint32, transferID uuid.UUID)) (v *virtualPacketConn) {
v = &virtualPacketConn{
peer: peer,
// newVirtualPacketConn creates a new virtual connection (both incoming and outgoing).
func newVirtualPacketConn(peer *PeerInfo, sendData func(data []byte, sequenceNumber uint32, transferID uuid.UUID)) (v *VirtualPacketConn) {
v = &VirtualPacketConn{
Peer: peer,
sendData: sendData,
incomingData: make(chan []byte, 512),
outgoingData: make(chan []byte),
@@ -55,7 +58,7 @@ func newVirtualPacketConn(peer *PeerInfo, sendData func(data []byte, sequenceNum
}
// writeForward forwards outgoing messages
func (v *virtualPacketConn) writeForward() {
func (v *VirtualPacketConn) writeForward() {
for {
select {
case data := <-v.outgoingData:
@@ -68,7 +71,7 @@ func (v *virtualPacketConn) writeForward() {
}
// receiveData receives incoming data via an external message. Non-blocking.
func (v *virtualPacketConn) receiveData(data []byte) {
func (v *VirtualPacketConn) receiveData(data []byte) {
if v.IsTerminated() {
return
}
@@ -84,7 +87,7 @@ func (v *virtualPacketConn) receiveData(data []byte) {
// Terminate closes the connection. Do not call this function manually. Use the underlying protocol's function to close the connection.
// Reason: 404 = Remote peer does not store file (upstream), 2 = Remote termination signal (upstream), 3 = Sequence invalidation or expiration (upstream), 1000+ = Transfer protocol indicated closing (downstream)
func (v *virtualPacketConn) Terminate(reason int) (err error) {
func (v *VirtualPacketConn) Terminate(reason int) (err error) {
v.Lock()
defer v.Unlock()
@@ -100,31 +103,31 @@ func (v *virtualPacketConn) Terminate(reason int) (err error) {
}
// IsTerminated checks if the connection is terminated
func (v *virtualPacketConn) IsTerminated() bool {
func (v *VirtualPacketConn) IsTerminated() bool {
return v.closed
}
// sequenceTerminate is a wrapper for sequenece termination (invalidation or expiration)
func (v *virtualPacketConn) sequenceTerminate() {
func (v *VirtualPacketConn) sequenceTerminate() {
v.Terminate(3)
}
// Close provides a Close function to be called by the underlying transfer protocol.
// Do not call the function manually; otherwise the underlying transfer protocol may not have time to send a termination message (and the remote peer would subsequently try to reconnect).
// Rather, use the underlying transfer protocol's close function.
func (v *virtualPacketConn) Close(reason int) (err error) {
v.peer.Backend.networks.Sequences.InvalidateSequence(v.peer.PublicKey, v.sequenceNumber, true)
func (v *VirtualPacketConn) Close(reason int) (err error) {
v.Peer.Backend.networks.Sequences.InvalidateSequence(v.Peer.PublicKey, v.sequenceNumber, true)
return v.Terminate(reason)
}
// CloseLinger is to be called by the underlying transfer protocol when it will close the socket soon after lingering around.
// Lingering happens to resend packets at the end of transfer, when it is not immediately known whether the remote peer received all packets.
func (v *virtualPacketConn) CloseLinger(reason int) (err error) {
func (v *VirtualPacketConn) CloseLinger(reason int) (err error) {
v.reason = reason
return nil
}
// GetTerminateReason returns the termination reason. 0 = Not yet terminated.
func (v *virtualPacketConn) GetTerminateReason() int {
func (v *VirtualPacketConn) GetTerminateReason() int {
return v.reason
}

11
protocol/Packet Lite.go
View File

@@ -186,3 +186,14 @@ func (router *LiteRouter) RegisterLiteID(id uuid.UUID, data interface{}, timeout
return
}
// Returns all lite sessions
func (router *LiteRouter) All() (sessions []*LiteID) {
router.Lock()
for _, info := range router.ids {
sessions = append(sessions, info)
}
router.Unlock()
return sessions
}

4
webapi/File IO.go
View File

@@ -275,7 +275,7 @@ func FileStartReader(peer *core.PeerInfo, hash []byte, offset, limit uint64, can
return nil, 0, 0, errors.New("no valid connection to peer")
}
udtConn, _, err := peer.FileTransferRequestUDT(hash, offset, limit)
udtConn, virtualConn, err := peer.FileTransferRequestUDT(hash, offset, limit)
if err != nil {
return nil, 0, 0, err
}
@@ -293,6 +293,8 @@ func FileStartReader(peer *core.PeerInfo, hash []byte, offset, limit uint64, can
return nil, 0, 0, err
}
virtualConn.Stats.(*core.FileTransferStats).FileSize = fileSize
return udtConn, fileSize, transferSize, nil
}