Transfer: Record close reason of underlying library (UDT). This helps for debugging issues. Close #54

2026-07-17 02:47:51 +01:00 · 2021-11-08 04:40:53 +01:00
parent 2924f0074d
commit f3d5f23cf9
9 changed files with 89 additions and 55 deletions
--- a/UDT.go
+++ b/UDT.go
@@ -49,7 +49,7 @@ 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.terminateChan, true)
+	udtConn, err := udt.DialUDT(udtConfig, virtualConnection, virtualConnection.incomingData, virtualConnection.outgoingData, virtualConnection.terminationSignal, true)
 	if err != nil {
 		return err
 	}
@@ -72,35 +72,36 @@ func (peer *PeerInfo) startFileTransferUDT(hash []byte, fileSize uint64, offset,
 }

 // FileTransferRequestUDT creates a UDT server listening for incoming data transfer and requests a file transfer from a remote peer.
-func (peer *PeerInfo) FileTransferRequestUDT(hash []byte, offset, limit uint64) (udtConn net.Conn, err error) {
-	virtualConnection := newVirtualPacketConn(peer, 0, hash, offset, limit)
+// The caller must call udtConn.Close() when done. Do not use any of the closing functions of virtualConn.
+func (peer *PeerInfo) FileTransferRequestUDT(hash []byte, offset, limit uint64) (udtConn net.Conn, virtualConn *virtualPacketConn, err error) {
+	virtualConn = newVirtualPacketConn(peer, 0, hash, offset, limit)

 	// new sequence
-	sequence := networks.Sequences.NewSequenceBi(peer.PublicKey, &peer.messageSequence, virtualConnection, transferSequenceTimeout, virtualConnection.sequenceTerminate)
+	sequence := networks.Sequences.NewSequenceBi(peer.PublicKey, &peer.messageSequence, virtualConn, transferSequenceTimeout, virtualConn.sequenceTerminate)
 	if sequence == nil {
-		return nil, errors.New("cannot acquire sequence")
+		return nil, nil, errors.New("cannot acquire sequence")
 	}
-	virtualConnection.sequenceNumber = sequence.SequenceNumber
+	virtualConn.sequenceNumber = sequence.SequenceNumber

 	udtConfig := udt.DefaultConfig()
 	udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSize

 	// start UDT receiver
-	udtListener := udt.ListenUDT(udtConfig, virtualConnection, virtualConnection.incomingData, virtualConnection.outgoingData, virtualConnection.terminateChan)
+	udtListener := udt.ListenUDT(udtConfig, virtualConn, virtualConn.incomingData, virtualConn.outgoingData, virtualConn.terminationSignal)

 	// request file transfer
-	peer.sendTransfer(nil, protocol.TransferControlRequestStart, virtualConnection.transferProtocol, hash, offset, limit, virtualConnection.sequenceNumber)
+	peer.sendTransfer(nil, protocol.TransferControlRequestStart, virtualConn.transferProtocol, hash, offset, limit, virtualConn.sequenceNumber)

 	// accept the connection
 	udtConn, err = udtListener.Accept()
 	if err != nil {
 		udtListener.Close()
-		return nil, err
+		return nil, nil, err
 	}

 	// We do not close the UDT listener here. It should automatically close after udtConn is closed.

-	return udtConn, nil
+	return udtConn, virtualConn, nil
 }

 // FileTransferReadHeaderUDT starts reading a file via UDT. It will only read the header and keeps the connection open.
--- a/Connection.go
+++ b/Connection.go
@@ -33,23 +33,23 @@ type virtualPacketConn struct {
 	outgoingData chan []byte

 	// internal data
-	closed        bool
-	terminateChan chan struct{}
-	reason        int // Reason why it was closed
+	closed            bool
+	terminationSignal chan struct{} // The termination signal shall be used by the underlying protocol to detect upstream termination.
+	reason            int           // Reason why it was closed
 	sync.Mutex
 }

 // newVirtualPacketConn creates a new virtual connection (both incomign and outgoing).
 func newVirtualPacketConn(peer *PeerInfo, protocol uint8, hash []byte, offset, limit uint64) (v *virtualPacketConn) {
 	v = &virtualPacketConn{
-		peer:             peer,
-		transferProtocol: protocol,
-		hash:             hash,
-		offset:           offset,
-		limit:            limit,
-		incomingData:     make(chan []byte, 100),
-		outgoingData:     make(chan []byte),
-		terminateChan:    make(chan struct{}),
+		peer:              peer,
+		transferProtocol:  protocol,
+		hash:              hash,
+		offset:            offset,
+		limit:             limit,
+		incomingData:      make(chan []byte, 100),
+		outgoingData:      make(chan []byte),
+		terminationSignal: make(chan struct{}),
 	}

 	go v.writeForward()
@@ -64,7 +64,7 @@ func (v *virtualPacketConn) writeForward() {
 		case data := <-v.outgoingData:
 			v.peer.sendTransfer(data, protocol.TransferControlActive, v.transferProtocol, v.hash, v.offset, v.limit, v.sequenceNumber)

-		case <-v.terminateChan:
+		case <-v.terminationSignal:
 			return
 		}
 	}
@@ -79,12 +79,13 @@ func (v *virtualPacketConn) receiveData(data []byte) {
 	// pass the data on
 	select {
 	case v.incomingData <- data:
-	case <-v.terminateChan:
+	case <-v.terminationSignal:
 	}
+	// TODO: Add read timeout
 }

 // 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), 1 = Transfer Protocol indicated closing (downstream), 2 = Remote termination signal (upstream), 3 = Sequence invalidation or expiration (upstream)
+// 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) {
 	v.Lock()
 	defer v.Unlock()
@@ -95,7 +96,7 @@ func (v *virtualPacketConn) Terminate(reason int) (err error) {

 	v.closed = true
 	v.reason = reason
-	close(v.terminateChan)
+	close(v.terminationSignal)

 	return
 }
@@ -112,7 +113,20 @@ func (v *virtualPacketConn) sequenceTerminate() {

 // 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).
-func (v *virtualPacketConn) Close() (err error) {
+// Rather, use the underlying transfer protocol's close function.
+func (v *virtualPacketConn) Close(reason int) (err error) {
 	networks.Sequences.InvalidateSequence(v.peer.PublicKey, v.sequenceNumber, true)
-	return v.Terminate(1)
+	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) {
+	v.reason = reason
+	return nil
+}
+
+// GetTerminateReason returns the termination reason. 0 = Not yet terminated.
+func (v *virtualPacketConn) GetTerminateReason() int {
+	return v.reason
 }
--- a/udt/listener.go
+++ b/udt/listener.go
@@ -5,7 +5,6 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
-	"io"
 	"net"
 	"sync"
 	"time"
@@ -48,7 +47,7 @@ func (l *listener) Close() (err error) {
 	close(a)
 	close(c)

-	l.m.closer.Close()
+	l.m.closer.Close(TerminateReasonListenerClosed)
 	return nil
 }

@@ -163,7 +162,7 @@ func (l *listener) readHandshake(m *multiplexer, hsPacket *packet.HandshakePacke
 }

 // ListenUDT listens for incoming UDT connections using the existing provided packet connection. It creates a UDT server.
-func ListenUDT(config *Config, closer io.Closer, incomingData <-chan []byte, outgoingData chan<- []byte, terminationSignal <-chan struct{}) net.Listener {
+func ListenUDT(config *Config, closer Closer, incomingData <-chan []byte, outgoingData chan<- []byte, terminationSignal <-chan struct{}) net.Listener {
 	m := newMultiplexer(closer, config.MaxPacketSize, incomingData, outgoingData, terminationSignal)

 	l := &listener{
--- a/udt/multiplexer.go
+++ b/udt/multiplexer.go
@@ -4,7 +4,6 @@ package udt

 import (
 	"fmt"
-	"io"
 	"math/rand"

 	"github.com/PeernetOfficial/core/udt/packet"
@@ -19,11 +18,11 @@ type multiplexer struct {
 	incomingData      <-chan []byte   // source to read packets from
 	outgoingData      chan<- []byte   // destination to send packets to
 	terminationSignal <-chan struct{} // external termination signal to watch
-	closer            io.Closer       // external closer to call in case the local socket/listener closes
+	closer            Closer          // external closer to call in case the local socket/listener closes
 }

 // The closer is called when the socket/listener closes. The terminationSignal is an external (upstream) signal to watch for.
-func newMultiplexer(closer io.Closer, maxPacketSize uint, incomingData <-chan []byte, outgoingData chan<- []byte, terminationSignal <-chan struct{}) (m *multiplexer) {
+func newMultiplexer(closer Closer, maxPacketSize uint, incomingData <-chan []byte, outgoingData chan<- []byte, terminationSignal <-chan struct{}) (m *multiplexer) {
 	m = &multiplexer{
 		maxPacketSize:     maxPacketSize,
 		closer:            closer,
--- a/udt/udt.go
+++ b/udt/udt.go
@@ -12,12 +12,30 @@ implemented:
 */

 import (
-	"io"
 	"net"
 )

+// Closer provides a status code indicating why the closing happens.
+type Closer interface {
+	Close(reason int) error       // Close is called when the socket is actually closed.
+	CloseLinger(reason int) error // CloseLinger is called when the socket indicates to be closed soon, after the linger time.
+}
+
+// The termination reason is passed on to the close function
+const (
+	TerminateReasonListenerClosed        = 1000 // Listener: The listener.Close function was called.
+	TerminateReasonLingerTimerExpired    = 1001 // Socket: The linger timer expired. Use CloseLinger to know the actual closing reason.
+	TerminateReasonConnectTimeout        = 1002 // Socket: The connection timed out when sending the initial handshake.
+	TerminateReasonRemoteSentShutdown    = 1003 // Remote peer sent a shutdown message.
+	TerminateReasonCannotProcessOutgoing = 1004 // Send: Cannot process outgoing messages.
+	TerminateReasonInvalidPacketIDAck    = 1005 // Send: Invalid packet ID received in ACK message.
+	TerminateReasonInvalidPacketIDNak    = 1006 // Send: Invalid packet ID received in NAK message.
+	TerminateReasonCorruptPacketNak      = 1007 // Send: Invalid NAK packet received.
+	TerminateReasonExpireTimer           = 1008 // Send: EXP timer expired.
+)
+
 // DialUDT establishes an outbound UDT connection using the existing provided packet connection. It creates a UDT client.
-func DialUDT(config *Config, closer io.Closer, incomingData <-chan []byte, outgoingData chan<- []byte, terminationSignal <-chan struct{}, isStream bool) (net.Conn, error) {
+func DialUDT(config *Config, closer Closer, incomingData <-chan []byte, outgoingData chan<- []byte, terminationSignal <-chan struct{}, isStream bool) (net.Conn, error) {
 	m := newMultiplexer(closer, config.MaxPacketSize, incomingData, outgoingData, terminationSignal)

 	s := m.newSocket(config, false, !isStream)
--- a/udt/udtsocket.go
+++ b/udt/udtsocket.go
@@ -38,6 +38,7 @@ type shutdownMessage struct {
 	sockState    sockState
 	permitLinger bool
 	err          error
+	reason       int
 }

 /*
@@ -463,7 +464,7 @@ func (s *udtSocket) goManageConnection() {
 	for {
 		select {
 		case <-s.lingerTimer: // linger timer expired, shut everything down
-			s.shutdown(sockStateClosed, false, nil)
+			s.shutdown(sockStateClosed, false, nil, TerminateReasonLingerTimerExpired)
 			return
 		case <-s.sockClosed:
 			return
@@ -473,9 +474,9 @@ func (s *udtSocket) goManageConnection() {
 			//fmt.Printf("(id=%d) sending %s  (id=%d)\n", s.sockID, packet.PacketTypeName(p.PacketType()), s.farSockID)
 			s.m.sendPacket(s.farSockID, ts, p)
 		case sd := <-s.shutdownEvent: // connection shut down
-			s.shutdown(sd.sockState, sd.permitLinger, sd.err)
+			s.shutdown(sd.sockState, sd.permitLinger, sd.err, sd.reason)
 		case <-s.connTimeout: // connection timed out
-			s.shutdown(sockStateTimeout, true, nil)
+			s.shutdown(sockStateTimeout, true, nil, TerminateReasonConnectTimeout)
 		case <-s.connRetry: // resend connection attempt
 			s.connRetry = nil
 			switch s.sockState {
@@ -596,7 +597,7 @@ func (s *udtSocket) readHandshake(m *multiplexer, p *packet.HandshakePacket) boo
 	return false
 }

-func (s *udtSocket) shutdown(sockState sockState, permitLinger bool, err error) {
+func (s *udtSocket) shutdown(sockState sockState, permitLinger bool, err error, reason int) {
 	if !s.isOpen() {
 		return // already closed
 	}
@@ -612,6 +613,7 @@ func (s *udtSocket) shutdown(sockState sockState, permitLinger bool, err error)
 			linger = DefaultConfig().LingerTime
 		}
 		s.lingerTimer = time.After(linger)
+		s.m.closer.CloseLinger(reason)
 		return
 	}

@@ -627,7 +629,7 @@ func (s *udtSocket) shutdown(sockState sockState, permitLinger bool, err error)
 	close(s.sockClosed)
 	close(s.recvEvent)

-	s.m.closer.Close()
+	s.m.closer.Close(reason)

 	s.messageIn <- nil
 }
@@ -688,7 +690,7 @@ func (s *udtSocket) readPacket(m *multiplexer, p packet.Packet) {
 	case *packet.HandshakePacket: // sent by both peers
 		s.readHandshake(m, sp)
 	case *packet.ShutdownPacket: // sent by either peer
-		s.shutdownEvent <- shutdownMessage{sockState: sockStateClosed, permitLinger: s.isServer} // if client tells us done, it is done.
+		s.shutdownEvent <- shutdownMessage{sockState: sockStateClosed, permitLinger: s.isServer, reason: TerminateReasonRemoteSentShutdown} // if client tells us done, it is done.
 	case *packet.AckPacket, *packet.NakPacket: // receiver -> sender
 		s.sendEvent <- recvPktEvent{pkt: p, now: now}
 	case *packet.UserDefControlPacket:
--- a/udt/udtsocket_recv.go
+++ b/udt/udtsocket_recv.go
@@ -301,7 +301,7 @@ func (s *udtSocketRecv) reassemblePacketPiecesDatagram(p *packet.DataPacket) (pi
 			nextBoundary, _, nextMsg := nextPiece.pkt.GetMessageData()
 			if nextMsg != msgID {
 				// ...oops? previous piece isn't in the same message
-				break
+				return nil, false
 			}
 			pieces = append(pieces, nextPiece.pkt)
 			if nextBoundary == packet.MbLast {
--- a/udt/udtsocket_send.go
+++ b/udt/udtsocket_send.go
@@ -116,7 +116,7 @@ func (s *udtSocketSend) goSendEvent() {
 		case msg, ok := <-thisMsgChan: // nil if we can't process outgoing messages right now
 			if !ok {
 				s.sendPacket <- &packet.ShutdownPacket{}
-				s.shutdownEvent <- shutdownMessage{sockState: sockStateClosed, permitLinger: !s.socket.isServer}
+				s.shutdownEvent <- shutdownMessage{sockState: sockStateClosed, permitLinger: !s.socket.isServer, reason: TerminateReasonCannotProcessOutgoing}
 				return
 			}
 			s.msgPartialSend = &msg
@@ -334,10 +334,10 @@ func (s *udtSocketSend) sendDataPacket(dp sendPacketEntry, isResend bool) {
 	}
 }

-func (s *udtSocketSend) assertValidSentPktID(pktType string, pktSeq packet.PacketID) bool {
+func (s *udtSocketSend) assertValidSentPktID(pktType string, pktSeq packet.PacketID, reason int) bool {
 	if s.sendPktSeq.BlindDiff(pktSeq) < 0 {
 		s.shutdownEvent <- shutdownMessage{sockState: sockStateCorrupted, permitLinger: false,
-			err: fmt.Errorf("FAULT: Received an %s for packet %d, but the largest packet we've sent has been %d", pktType, pktSeq.Seq, s.sendPktSeq.Seq)}
+			err: fmt.Errorf("FAULT: Received an %s for packet %d, but the largest packet we've sent has been %d", pktType, pktSeq.Seq, s.sendPktSeq.Seq), reason: reason}
 		return false
 	}
 	return true
@@ -350,7 +350,7 @@ func (s *udtSocketSend) ingestAck(p *packet.AckPacket, now time.Time) {
 	// Send back an ACK2 with the same ACK sequence number in this ACK.
 	s.sendPacket <- &packet.Ack2Packet{AckSeqNo: p.AckSeqNo}

-	if !s.assertValidSentPktID("ACK", p.PktSeqHi) || p.PktSeqHi.BlindDiff(s.recvAckSeq) <= 0 {
+	if !s.assertValidSentPktID("ACK", p.PktSeqHi, TerminateReasonInvalidPacketIDAck) || p.PktSeqHi.BlindDiff(s.recvAckSeq) <= 0 {
 		return
 	}

@@ -388,20 +388,20 @@ func (s *udtSocketSend) ingestNak(p *packet.NakPacket, now time.Time) {
 			thisPktID := packet.PacketID{Seq: thisEntry & 0x7FFFFFFF}
 			if idx+1 == clen {
 				s.shutdownEvent <- shutdownMessage{sockState: sockStateCorrupted, permitLinger: false,
-					err: fmt.Errorf("FAULT: While unpacking a NAK, the last entry (%x) was describing a start-of-range", thisEntry)}
+					err: fmt.Errorf("FAULT: While unpacking a NAK, the last entry (%x) was describing a start-of-range", thisEntry), reason: TerminateReasonCorruptPacketNak}
 				return
 			}
-			if !s.assertValidSentPktID("NAK", thisPktID) {
+			if !s.assertValidSentPktID("NAK", thisPktID, TerminateReasonInvalidPacketIDNak) {
 				return
 			}
 			lastEntry := p.CmpLossInfo[idx+1]
 			if lastEntry&0x80000000 != 0 {
 				s.shutdownEvent <- shutdownMessage{sockState: sockStateCorrupted, permitLinger: false,
-					err: fmt.Errorf("FAULT: While unpacking a NAK, a start-of-range (%x) was followed by another start-of-range (%x)", thisEntry, lastEntry)}
+					err: fmt.Errorf("FAULT: While unpacking a NAK, a start-of-range (%x) was followed by another start-of-range (%x)", thisEntry, lastEntry), reason: TerminateReasonCorruptPacketNak}
 				return
 			}
 			lastPktID := packet.PacketID{Seq: lastEntry}
-			if !s.assertValidSentPktID("NAK", lastPktID) {
+			if !s.assertValidSentPktID("NAK", lastPktID, TerminateReasonInvalidPacketIDNak) {
 				return
 			}
 			idx++
@@ -411,7 +411,7 @@ func (s *udtSocketSend) ingestNak(p *packet.NakPacket, now time.Time) {
 			}
 		} else {
 			thisPktID := packet.PacketID{Seq: thisEntry}
-			if !s.assertValidSentPktID("NAK", thisPktID) {
+			if !s.assertValidSentPktID("NAK", thisPktID, TerminateReasonInvalidPacketIDNak) {
 				return
 			}
 			s.sendLossList.Add(recvLossEntry{packetID: thisPktID})
@@ -457,7 +457,7 @@ func (s *udtSocketSend) expEvent(currTime time.Time) {
 	// timeout: at least 16 expirations and must be greater than 10 seconds
 	if (s.expCount > 16) && (currTime.Sub(s.lastRecvTime) > 5*time.Second) {
 		// Connection is broken.
-		s.shutdownEvent <- shutdownMessage{sockState: sockStateTimeout, permitLinger: true}
+		s.shutdownEvent <- shutdownMessage{sockState: sockStateTimeout, permitLinger: true, reason: TerminateReasonExpireTimer}
 		return
 	}

--- a/webapi/File
+++ b/webapi/File
@@ -204,7 +204,7 @@ func PeerConnectPublicKey(publicKey *btcec.PublicKey, timeout time.Duration) (pe
 	}

 	// otherwise not found :(
-	return nil, errors.New("peer not found :(")
+	return nil, errors.New("peer not found")
 }

 // PeerConnectNode tries to connect via the node ID
@@ -218,7 +218,8 @@ func PeerConnectNode(nodeID []byte, timeout time.Duration) (peer *core.PeerInfo,
 		return peer, nil
 	}

-	return nil, nil
+	// otherwise not found :(
+	return nil, errors.New("peer not found")
 }

 // FileStartReader providers a reader to a remote file. The reader must be closed by the caller.
@@ -232,7 +233,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, _, err := peer.FileTransferRequestUDT(hash, offset, limit)
 	if err != nil {
 		return nil, 0, 0, err
 	}