From cb1441c97ac418a160673980e9a0932e6fcded8e Mon Sep 17 00:00:00 2001 From: Kleissner Date: Wed, 16 Mar 2022 23:52:55 +0100 Subject: [PATCH] VirtualPacketConn: New Stats field maintained by caller. This allows insight into active transfers. New FileTransferStats and BlockTransferStats structures. --- Commands.go | 14 +++++++------- Network.go | 4 ++-- Networks.go | 5 +++++ Transfer Block.go | 16 ++++++++++++++-- Transfer UDT.go | 34 ++++++++++++++++++++++++++------- Transfer Virtual Connection.go | 35 ++++++++++++++++++---------------- protocol/Packet Lite.go | 11 +++++++++++ webapi/File IO.go | 4 +++- 8 files changed, 88 insertions(+), 35 deletions(-) diff --git a/Commands.go b/Commands.go index abf5120..8b6c1a1 100644 --- a/Commands.go +++ b/Commands.go @@ -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 } diff --git a/Network.go b/Network.go index 5a7f2e3..727b131 100644 --- a/Network.go +++ b/Network.go @@ -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) diff --git a/Networks.go b/Networks.go index 962cc9c..de8929c 100644 --- a/Networks.go +++ b/Networks.go @@ -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() +} diff --git a/Transfer Block.go b/Transfer Block.go index 90b8361..a56485b 100644 --- a/Transfer Block.go +++ b/Transfer Block.go @@ -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 +} diff --git a/Transfer UDT.go b/Transfer UDT.go index aca5ade..3802fcf 100644 --- a/Transfer UDT.go +++ b/Transfer UDT.go @@ -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 +) diff --git a/Transfer Virtual Connection.go b/Transfer Virtual Connection.go index 0d8f20a..45a497d 100644 --- a/Transfer Virtual Connection.go +++ b/Transfer Virtual Connection.go @@ -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 } diff --git a/protocol/Packet Lite.go b/protocol/Packet Lite.go index fb0e07b..5996d69 100644 --- a/protocol/Packet Lite.go +++ b/protocol/Packet Lite.go @@ -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 +} diff --git a/webapi/File IO.go b/webapi/File IO.go index 5d8641c..434d382 100644 --- a/webapi/File IO.go +++ b/webapi/File IO.go @@ -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 }