diff --git a/Commands.go b/Commands.go index 916bfe3..e12e17c 100644 --- a/Commands.go +++ b/Commands.go @@ -266,14 +266,14 @@ func (peer *PeerInfo) cmdTransfer(msg *protocol.MessageTransfer, connection *Con case protocol.TransferControlNotAvailable: if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok { - v.Terminate(false, 404) + v.Terminate(404) return } case protocol.TransferControlTerminate: if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok { // Since an incoming terminate notice means the remote peer already terminated the connection, set sendNotice to false. - v.Terminate(false, 2) + v.Terminate(2) return } diff --git a/Transfer UDT.go b/Transfer UDT.go index cb9dcad..2ed0fef 100644 --- a/Transfer UDT.go +++ b/Transfer UDT.go @@ -32,7 +32,7 @@ func (peer *PeerInfo) startFileTransferUDT(hash []byte, offset, limit uint64, se networks.Sequences.RegisterSequenceBi(peer.PublicKey, sequenceNumber, virtualConnection, transferSequenceTimeout, virtualConnection.sequenceTerminate) // start UDT sender - udtConn, err := udt.DialUDT(udt.DefaultConfig(), virtualConnection, true) + udtConn, err := udt.DialUDT(udt.DefaultConfig(), virtualConnection, virtualConnection.incomingData, virtualConnection.outgoingData, virtualConnection.terminateChan, true) if err != nil { return err } @@ -40,25 +40,24 @@ func (peer *PeerInfo) startFileTransferUDT(hash []byte, offset, limit uint64, se _, err = UserWarehouse.ReadFile(hash, int64(offset), int64(limit), udtConn) // close the UDT client and virtual connection in any case - udtConn.Close() // warning: This is currently blocking. - //virtualConnection.Terminate(false, 1) + udtConn.Close() // warning: This is currently blocking in case the other side does not call Close(). return err } // RequestFileTransferUDT creates a UDT server listening for incoming data transfer and requests a file transfer from a remote peer. -func (peer *PeerInfo) RequestFileTransferUDT(hash []byte, offset, limit uint64) (udtConn net.Conn, udtListener net.Listener, err error) { +func (peer *PeerInfo) RequestFileTransferUDT(hash []byte, offset, limit uint64) (udtConn net.Conn, err error) { virtualConnection := newVirtualPacketConn(peer, 0, hash, offset, limit, true) // new sequence sequence := networks.Sequences.NewSequenceBi(peer.PublicKey, &peer.messageSequence, virtualConnection, transferSequenceTimeout, virtualConnection.sequenceTerminate) if sequence == nil { - return nil, nil, errors.New("cannot acquire sequence") + return nil, errors.New("cannot acquire sequence") } virtualConnection.sequenceNumber = sequence.SequenceNumber // start UDT receiver - udtListener = udt.ListenUDT(udt.DefaultConfig(), virtualConnection) + udtListener := udt.ListenUDT(udt.DefaultConfig(), virtualConnection, virtualConnection.incomingData, virtualConnection.outgoingData, virtualConnection.terminateChan) // request file transfer peer.sendTransfer(nil, protocol.TransferControlRequestStart, virtualConnection.transferProtocol, hash, offset, limit, virtualConnection.sequenceNumber) @@ -67,8 +66,10 @@ func (peer *PeerInfo) RequestFileTransferUDT(hash []byte, offset, limit uint64) udtConn, err = udtListener.Accept() if err != nil { udtListener.Close() - return nil, nil, err + return nil, err } - return udtConn, udtListener, nil + // We do not close the UDT listener here. It should automatically close after udtConn is closed. + + return udtConn, nil } diff --git a/Transfer Virtual Connection.go b/Transfer Virtual Connection.go index 25c8aff..be2d0bf 100644 --- a/Transfer Virtual Connection.go +++ b/Transfer Virtual Connection.go @@ -3,23 +3,19 @@ File Name: Transfer Virtual Connection.go Copyright: 2021 Peernet s.r.o. Author: Peter Kleissner -This file defines a virtual net.PacketConn which sends transfer messages and can be used to embed other transfer protocols. +This file defines a virtual connection between a transfer protocol and Peernet messages. +If either the downstream transfer protocol or upstream Peernet messages indicate termination, the virtual connection ceases to exist. */ package core import ( - "errors" - "io" - "net" "sync" - "time" "github.com/PeernetOfficial/core/protocol" ) // virtualPacketConn is a virtual connection. -// The required functions are ReadFrom, WriteTo, Close, LocalAddr, SetDeadline, SetReadDeadline, SetWriteDeadline. type virtualPacketConn struct { peer *PeerInfo @@ -37,6 +33,7 @@ type virtualPacketConn struct { // data channel incomingData chan []byte + outgoingData chan []byte // internal data closed bool @@ -47,7 +44,7 @@ type virtualPacketConn struct { // newVirtualPacketConn creates a new virtual connection (both incomign and outgoing). func newVirtualPacketConn(peer *PeerInfo, protocol uint8, hash []byte, offset, limit uint64, isListen bool) (v *virtualPacketConn) { - return &virtualPacketConn{ + v = &virtualPacketConn{ transferProtocol: protocol, peer: peer, hash: hash, @@ -55,64 +52,44 @@ func newVirtualPacketConn(peer *PeerInfo, protocol uint8, hash []byte, offset, l limit: limit, isListenMode: isListen, incomingData: make(chan []byte), + outgoingData: make(chan []byte), terminateChan: make(chan struct{}), } -} -// ReadFrom reads a packet from the connection, copying the payload into p. It returns the number of bytes copied into p and the return address that was on the packet. -func (v *virtualPacketConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) { - if v.closed { - return 0, nil, errors.New("connection closed") - } - - // The underlying transfer messages feature a message sequence timeout (via Sequences.RegisterSequenceBi and Sequences.NewSequenceBi). - // The sequence timeout triggers the virtualPacketConn.Terminate function which closes terminateChan. - // Therefore an additional timeout here would be redundant. Instead, a transfer-wide timeout, that closes terminateChan upon expiration, shall be used. - - select { - case data := <-v.incomingData: - n = copy(p, data) - case <-v.terminateChan: - return n, nil, io.EOF - } + go v.writeForward() return } -// Write sends the data to the remote peer. It makes it possible to use virtualPacketConn as io.Writer. -// Note: If the packet size exceed the limit from protocol.EncodeTransfer, this function fails. -func (v *virtualPacketConn) Write(p []byte) (n int, err error) { - if v.IsTerminated() { - return 0, errors.New("connection closed") - } +// writeForward forwards outgoing messages +func (v *virtualPacketConn) writeForward() { + for { + select { + case data := <-v.outgoingData: + v.peer.sendTransfer(data, protocol.TransferControlActive, v.transferProtocol, v.hash, v.offset, v.limit, v.sequenceNumber) - // create a new packet - err = v.peer.sendTransfer(p, protocol.TransferControlActive, v.transferProtocol, v.hash, v.offset, v.limit, v.sequenceNumber) - if err == nil { - n = len(p) + case <-v.terminateChan: + return + } } - - return } // receiveData receives incoming data via an external message. Blocking until a read occurs or the connection is terminated! -func (v *virtualPacketConn) receiveData(data []byte) (err error) { +func (v *virtualPacketConn) receiveData(data []byte) { if v.IsTerminated() { - return errors.New("connection closed") + return } - // pass on the data + // pass the data on select { case v.incomingData <- data: case <-v.terminateChan: } - - return } -// Terminate closes the connection and optionally sends a termination message to the remote peer. Multiple termination signals have no effect. -// Reason: 404 = Remote peer does not store file, 1 = Local termination signal, 2 = Remote termination signal, 3 = Sequence invalidation or expiration -func (v *virtualPacketConn) Terminate(sendSignal bool, reason int) (err error) { +// 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) +func (v *virtualPacketConn) Terminate(reason int) (err error) { v.Lock() defer v.Unlock() @@ -124,10 +101,6 @@ func (v *virtualPacketConn) Terminate(sendSignal bool, reason int) (err error) { v.reason = reason close(v.terminateChan) - if sendSignal { - err = v.peer.sendTransfer(nil, protocol.TransferControlTerminate, v.transferProtocol, v.hash, 0, 0, v.sequenceNumber) - } - return } @@ -138,35 +111,12 @@ func (v *virtualPacketConn) IsTerminated() bool { // sequenceTerminate is a wrapper for sequenece termination (invalidation or expiration) func (v *virtualPacketConn) sequenceTerminate() { - v.Terminate(false, 3) + v.Terminate(3) } -// Close closes the connection. +// 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) { - return v.Terminate(false, 1) -} - -// WriteTo writes a packet with payload p to addr. WriteTo can be made to time out and return an Error after a fixed time limit. -func (v *virtualPacketConn) WriteTo(p []byte, addr net.Addr) (n int, err error) { - return v.Write(p) -} - -// LocalAddr returns the local network address -func (v *virtualPacketConn) LocalAddr() (addr net.Addr) { - return -} - -// SetDeadline sets the read and write deadlines associated with the connection. It is equivalent to calling both SetReadDeadline and SetWriteDeadline. -func (v *virtualPacketConn) SetDeadline(t time.Time) (err error) { - return nil -} - -// SetReadDeadline sets the deadline for future ReadFrom calls and any currently-blocked ReadFrom call. -func (v *virtualPacketConn) SetReadDeadline(t time.Time) (err error) { - return nil -} - -// SetWriteDeadline sets the deadline for future WriteTo calls and any currently-blocked WriteTo call. -func (v *virtualPacketConn) SetWriteDeadline(t time.Time) (err error) { - return nil + networks.Sequences.InvalidateSequence(v.peer.PublicKey, v.sequenceNumber, true) + return v.Terminate(1) } diff --git a/protocol/Sequence.go b/protocol/Sequence.go index 80448a9..b5b8410 100644 --- a/protocol/Sequence.go +++ b/protocol/Sequence.go @@ -165,9 +165,9 @@ func (manager *SequenceManager) ValidateSequence(publicKey *btcec.PublicKey, seq return sequence, sequence.expires.After(time.Now()), rtt } -// InvalidateSequence invalidates the sequence number. -func (manager *SequenceManager) InvalidateSequence(publicKey *btcec.PublicKey, sequenceNumber uint32) { - key := sequence2Key(false, publicKey, sequenceNumber) +// InvalidateSequence invalidates the sequence number. It does not call invalidateFunc. +func (manager *SequenceManager) InvalidateSequence(publicKey *btcec.PublicKey, sequenceNumber uint32, bidirectional bool) { + key := sequence2Key(bidirectional, publicKey, sequenceNumber) manager.Lock() delete(manager.sequences, key) diff --git a/udt/listener.go b/udt/listener.go index f247418..76e042d 100644 --- a/udt/listener.go +++ b/udt/listener.go @@ -5,6 +5,7 @@ import ( "encoding/binary" "errors" "fmt" + "io" "net" "sync" "time" @@ -162,8 +163,8 @@ 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, packetConn net.PacketConn) net.Listener { - m := newMultiplexer(packetConn, config.MaxPacketSize) +func ListenUDT(config *Config, closer io.Closer, incomingData <-chan []byte, outgoingData chan<- []byte, terminationSignal <-chan struct{}) net.Listener { + m := newMultiplexer(closer, config.MaxPacketSize, incomingData, outgoingData, terminationSignal) l := &listener{ m: m, diff --git a/udt/multiplexer.go b/udt/multiplexer.go index 2678293..598b54e 100644 --- a/udt/multiplexer.go +++ b/udt/multiplexer.go @@ -4,8 +4,8 @@ package udt import ( "fmt" + "io" "math/rand" - "net" "sync" "github.com/PeernetOfficial/core/udt/packet" @@ -13,20 +13,27 @@ import ( // A multiplexer is a single UDT socket over a single PacketConn. type multiplexer struct { - conn net.PacketConn // the UDPConn from which we read/write - socket *udtSocket // Socket - socketID uint32 // Socket ID - listenSock *listener // the server socket listening to incoming connections, if there is one. Set by caller. - maxPacketSize uint // the Maximum Transmission Unit of packets sent from this address - pktOut chan packet.Packet // packets queued for immediate sending - sync.Mutex // Synchronized access to socket/listenSock + socket *udtSocket // Socket + socketID uint32 // Socket ID + listenSock *listener // the server socket listening to incoming connections, if there is one. Set by caller. + maxPacketSize uint // the Maximum Transmission Unit of packets sent from this address + pktOut chan packet.Packet // packets queued for immediate sending + sync.Mutex // Synchronized access to socket/listenSock + 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 } -func newMultiplexer(conn net.PacketConn, maxPacketSize uint) (m *multiplexer) { +// 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) { m = &multiplexer{ - conn: conn, - maxPacketSize: maxPacketSize, // to be verified?! - pktOut: make(chan packet.Packet, 100), // todo: figure out how to size this + maxPacketSize: maxPacketSize, + pktOut: make(chan packet.Packet, 100), + closer: closer, + incomingData: incomingData, + outgoingData: outgoingData, + terminationSignal: terminationSignal, } go m.goRead() @@ -46,7 +53,7 @@ func (m *multiplexer) unlistenUDT(l *listener) { m.listenSock = nil - m.conn.Close() + m.closer.Close() close(m.pktOut) } @@ -66,22 +73,23 @@ func (m *multiplexer) closeSocket(sockID uint32) { m.socket = nil - m.conn.Close() + m.closer.Close() close(m.pktOut) } // read runs in a goroutine and reads packets from conn using a buffer from the readBufferPool, or a new buffer. func (m *multiplexer) goRead() { - buf := make([]byte, m.maxPacketSize) for { - numBytes, _, err := m.conn.ReadFrom(buf) - if err != nil { + var buf []byte + select { + case buf = <-m.incomingData: + case <-m.terminationSignal: return } - p, err := packet.DecodePacket(buf[0:numBytes]) + p, err := packet.DecodePacket(buf) if err != nil { - fmt.Printf("Unable to decode packet: %s\n", err) + fmt.Printf("Error decoding UDT packet: %s\n", err) return } @@ -111,16 +119,16 @@ func (m *multiplexer) goRead() { func (m *multiplexer) goWrite() { buf := make([]byte, m.maxPacketSize) for pkt := range m.pktOut { - plen, err := pkt.WriteTo(buf) + plen, err := pkt.WriteTo(buf) // encode if err != nil { // TODO: handle write error - fmt.Printf("Unable to buffer out: %s\n", err.Error()) + fmt.Printf("Error encoding UDT packet: %s\n", err.Error()) return } - if _, err = m.conn.WriteTo(buf[0:plen], nil); err != nil { - // TODO: handle write error - fmt.Printf("Unable to write out: %s\n", err.Error()) + select { + case m.outgoingData <- buf[0:plen]: + case <-m.terminationSignal: return } } diff --git a/udt/udt.go b/udt/udt.go index edb6ece..548c750 100644 --- a/udt/udt.go +++ b/udt/udt.go @@ -12,12 +12,13 @@ implemented: */ import ( + "io" "net" ) // DialUDT establishes an outbound UDT connection using the existing provided packet connection. It creates a UDT client. -func DialUDT(config *Config, packetConn net.PacketConn, isStream bool) (net.Conn, error) { - m := newMultiplexer(packetConn, config.MaxPacketSize) +func DialUDT(config *Config, closer io.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) err := s.startConnect()