From f3d5f23cf9b0e299f52f1bac287fa4cb35d3dc8e Mon Sep 17 00:00:00 2001 From: Kleissner Date: Mon, 8 Nov 2021 04:40:53 +0100 Subject: [PATCH] Transfer: Record close reason of underlying library (UDT). This helps for debugging issues. Close #54 --- Transfer UDT.go | 21 ++++++++------- Transfer Virtual Connection.go | 48 ++++++++++++++++++++++------------ udt/listener.go | 5 ++-- udt/multiplexer.go | 5 ++-- udt/udt.go | 22 ++++++++++++++-- udt/udtsocket.go | 14 +++++----- udt/udtsocket_recv.go | 2 +- udt/udtsocket_send.go | 20 +++++++------- webapi/File IO.go | 7 ++--- 9 files changed, 89 insertions(+), 55 deletions(-) diff --git a/Transfer UDT.go b/Transfer UDT.go index 530165a..107b8ea 100644 --- a/Transfer UDT.go +++ b/Transfer 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. diff --git a/Transfer Virtual Connection.go b/Transfer Virtual Connection.go index 9d3999e..5090b6e 100644 --- a/Transfer Virtual Connection.go +++ b/Transfer Virtual 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 } diff --git a/udt/listener.go b/udt/listener.go index fe5ec40..882b9ea 100644 --- 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{ diff --git a/udt/multiplexer.go b/udt/multiplexer.go index a7fcea2..694a359 100644 --- 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, diff --git a/udt/udt.go b/udt/udt.go index 548c750..05a44aa 100644 --- 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) diff --git a/udt/udtsocket.go b/udt/udtsocket.go index aeaefbc..c7924fc 100644 --- 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: diff --git a/udt/udtsocket_recv.go b/udt/udtsocket_recv.go index 4045817..a7173f8 100644 --- 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 { diff --git a/udt/udtsocket_send.go b/udt/udtsocket_send.go index 3482bf4..66a7a15 100644 --- 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 } diff --git a/webapi/File IO.go b/webapi/File IO.go index c61fa27..e6de3b7 100644 --- a/webapi/File IO.go +++ b/webapi/File IO.go @@ -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 }