diff --git a/udt/config.go b/udt/config.go index 8aac7d3..bf30e5a 100644 --- a/udt/config.go +++ b/udt/config.go @@ -15,7 +15,6 @@ type Config struct { MaxBandwidth uint64 // Maximum bandwidth to take with this connection (in bytes/sec, 0 = unlimited) LingerTime time.Duration // time to wait for retransmit requests after connection shutdown MaxFlowWinSize uint // maximum number of unacknowledged packets to permit (minimum 32) - MTU uint // MTU is the maximum UDT packet size SynTime time.Duration // SynTime CanAccept func(hsPacket *packet.HandshakePacket) error // can this listener accept this connection? @@ -30,7 +29,7 @@ func DefaultConfig() *Config { ListenReplayWindow: 5 * time.Minute, LingerTime: 10 * time.Second, MaxFlowWinSize: 64, - MTU: 65535, + MaxPacketSize: 65535, SynTime: 10000 * time.Microsecond, CongestionForSocket: func(sock *udtSocket) CongestionControl { return &NativeCongestionControl{} diff --git a/udt/listener.go b/udt/listener.go index f3d6310..f247418 100644 --- a/udt/listener.go +++ b/udt/listener.go @@ -163,7 +163,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, packetConn net.PacketConn) net.Listener { - m := newMultiplexer(packetConn, config.MTU) + m := newMultiplexer(packetConn, config.MaxPacketSize) l := &listener{ m: m, diff --git a/udt/multiplexer.go b/udt/multiplexer.go index c431c06..2678293 100644 --- a/udt/multiplexer.go +++ b/udt/multiplexer.go @@ -13,20 +13,20 @@ 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. - mtu 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 + 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 } -func newMultiplexer(conn net.PacketConn, mtu uint) (m *multiplexer) { +func newMultiplexer(conn net.PacketConn, maxPacketSize uint) (m *multiplexer) { m = &multiplexer{ - conn: conn, - mtu: mtu, // to be verified?! - pktOut: make(chan packet.Packet, 100), // todo: figure out how to size this + conn: conn, + maxPacketSize: maxPacketSize, // to be verified?! + pktOut: make(chan packet.Packet, 100), // todo: figure out how to size this } go m.goRead() @@ -72,7 +72,7 @@ func (m *multiplexer) closeSocket(sockID uint32) { // 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.mtu) + buf := make([]byte, m.maxPacketSize) for { numBytes, _, err := m.conn.ReadFrom(buf) if err != nil { @@ -109,7 +109,7 @@ func (m *multiplexer) goRead() { // write runs in a goroutine and writes packets to conn using a buffer from the writeBufferPool, or a new buffer. func (m *multiplexer) goWrite() { - buf := make([]byte, m.mtu) + buf := make([]byte, m.maxPacketSize) for pkt := range m.pktOut { plen, err := pkt.WriteTo(buf) if err != nil { diff --git a/udt/packet/packet_handshake.go b/udt/packet/packet_handshake.go index f26166a..bfa19f0 100644 --- a/udt/packet/packet_handshake.go +++ b/udt/packet/packet_handshake.go @@ -25,10 +25,10 @@ const ( // HandshakePacket is a UDT packet used to negotiate a new connection type HandshakePacket struct { ctrlHeader - UdtVer uint32 // UDT version - SockType SocketType // Socket Type (1 = STREAM or 2 = DGRAM) - InitPktSeq PacketID // initial packet sequence number - MaxPktSize uint32 // maximum packet size (including UDP/IP headers) + UdtVer uint32 // UDT version + SockType SocketType // Socket Type (1 = STREAM or 2 = DGRAM) + InitPktSeq PacketID // initial packet sequence number + //MaxPktSize uint32 // maximum packet size (including UDP/IP headers) MaxFlowWinSize uint32 // maximum flow window size ReqType HandshakeReqType // connection type (regular(1), rendezvous(0), -1/-2 response) SockID uint32 // socket ID @@ -48,7 +48,7 @@ func (p *HandshakePacket) WriteTo(buf []byte) (uint, error) { endianness.PutUint32(buf[16:20], p.UdtVer) endianness.PutUint32(buf[20:24], uint32(p.SockType)) endianness.PutUint32(buf[24:28], p.InitPktSeq.Seq) - endianness.PutUint32(buf[28:32], p.MaxPktSize) + //endianness.PutUint32(buf[28:32], p.MaxPktSize) endianness.PutUint32(buf[32:36], p.MaxFlowWinSize) endianness.PutUint32(buf[36:40], uint32(p.ReqType)) endianness.PutUint32(buf[40:44], p.SockID) @@ -72,7 +72,7 @@ func (p *HandshakePacket) readFrom(data []byte) error { p.UdtVer = endianness.Uint32(data[16:20]) p.SockType = SocketType(endianness.Uint32(data[20:24])) p.InitPktSeq = PacketID{endianness.Uint32(data[24:28])} - p.MaxPktSize = endianness.Uint32(data[28:32]) + //p.MaxPktSize = endianness.Uint32(data[28:32]) p.MaxFlowWinSize = endianness.Uint32(data[32:36]) p.ReqType = HandshakeReqType(endianness.Uint32(data[36:40])) p.SockID = endianness.Uint32(data[40:44]) diff --git a/udt/readme.md b/udt/readme.md index e9a63c0..8e80dc1 100644 --- a/udt/readme.md +++ b/udt/readme.md @@ -2,7 +2,7 @@ UDT (UDP-based Data Transfer Protocol) is a transfer protocol on top of UDP. See https://udt.sourceforge.io/ for the original spec and the reference implementation. -This project is a fork from https://github.com/odysseus654/go-udt which itself is a fork. +This code is a fork from https://github.com/odysseus654/go-udt which itself is a fork. ## Stream vs Datagram @@ -18,3 +18,11 @@ messaging. The data streaming semantics is similar to that of TCP, while the messaging semantics can be regarded as a subset of SCTP [RFC4960]. ``` + +## Deviations + +MTU negotiation is disabled. Peernet uses a hardcoded max packet size (see protocol package). Packets may be routed through any network adapter, therefore pinning a MTU specific to a network adapter would not make much sense. + +The "rendezvous" functionality has been removed since Peernet supports native Traverse messages for UDP hole punching. + +Multiplexing multiple UDT sockets to a single UDT connection is removed. It added complexity without benefits in this case. Peernet uses a single UDP port and UDP connection between two peers. Multiplexing has no effect other than breaking the concept and the security of Peernet message sequences. diff --git a/udt/udt.go b/udt/udt.go index d414511..edb6ece 100644 --- a/udt/udt.go +++ b/udt/udt.go @@ -17,7 +17,7 @@ import ( // 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.MTU) + m := newMultiplexer(packetConn, config.MaxPacketSize) s := m.newSocket(config, false, !isStream) err := s.startConnect() diff --git a/udt/udtsocket.go b/udt/udtsocket.go index 692a3da..0737e0c 100644 --- a/udt/udtsocket.go +++ b/udt/udtsocket.go @@ -62,14 +62,14 @@ type udtSocket struct { initPktSeq packet.PacketID // initial packet sequence to start the connection with connectWait *sync.WaitGroup // released when connection is complete (or failed) - sockState sockState // socket state - used mostly during handshakes - mtu atomicUint32 // the negotiated maximum packet size - maxFlowWinSize uint // receiver: maximum unacknowledged packet count - currPartialRead []byte // stream connections: currently reading message (for partial reads). Owned by client caller (Read) - readDeadline *time.Timer // if set, then calls to Read() will return "timeout" after this time - readDeadlinePassed bool // if set, then calls to Read() will return "timeout" - writeDeadline *time.Timer // if set, then calls to Write() will return "timeout" after this time - writeDeadlinePassed bool // if set, then calls to Write() will return "timeout" + sockState sockState // socket state - used mostly during handshakes + maxPacketSize uint32 // the maximum packet size + maxFlowWinSize uint // receiver: maximum unacknowledged packet count + currPartialRead []byte // stream connections: currently reading message (for partial reads). Owned by client caller (Read) + readDeadline *time.Timer // if set, then calls to Read() will return "timeout" after this time + readDeadlinePassed bool // if set, then calls to Read() will return "timeout" + writeDeadline *time.Timer // if set, then calls to Write() will return "timeout" after this time + writeDeadlinePassed bool // if set, then calls to Write() will return "timeout" rttProt sync.RWMutex // lock must be held before referencing rtt/rttVar rtt uint // receiver: estimated roundtrip time. (in microseconds) @@ -404,11 +404,6 @@ func (s *udtSocket) SetWriteDeadline(t time.Time) error { func newSocket(m *multiplexer, config *Config, sockID uint32, isServer bool, isDatagram bool) (s *udtSocket) { now := time.Now() - mtu := m.mtu - if config.MaxPacketSize > 0 && config.MaxPacketSize < mtu { - mtu = config.MaxPacketSize - } - maxFlowWinSize := config.MaxFlowWinSize if maxFlowWinSize == 0 { maxFlowWinSize = DefaultConfig().MaxFlowWinSize @@ -425,7 +420,7 @@ func newSocket(m *multiplexer, config *Config, sockID uint32, isServer bool, isD sockState: sockStateInit, udtVer: 4, isServer: isServer, - mtu: atomicUint32{val: uint32(mtu)}, + maxPacketSize: uint32(config.MaxPacketSize), maxFlowWinSize: maxFlowWinSize, isDatagram: isDatagram, sockID: sockID, @@ -510,10 +505,10 @@ func (s *udtSocket) sendHandshake(reqType packet.HandshakeReqType) { } p := &packet.HandshakePacket{ - UdtVer: uint32(s.udtVer), - SockType: sockType, - InitPktSeq: s.initPktSeq, - MaxPktSize: s.mtu.get(), // maximum packet size (including UDP/IP headers) + UdtVer: uint32(s.udtVer), + SockType: sockType, + InitPktSeq: s.initPktSeq, + //MaxPktSize: s.maxPacketSize, // maximum packet size (including UDP/IP headers) MaxFlowWinSize: uint32(s.maxFlowWinSize), // maximum flow window size ReqType: reqType, SockID: s.sockID, @@ -543,9 +538,10 @@ func (s *udtSocket) readHandshake(m *multiplexer, p *packet.HandshakePacket) boo s.farSockID = p.SockID s.isDatagram = p.SockType == packet.TypeDGRAM - if s.mtu.get() > p.MaxPktSize { - s.mtu.set(p.MaxPktSize) - } + // MTU negotiation is disabled. Packets may be sent across any network adapter; it would be impossible to use a per-adapter MTU. + //if s.mtu.get() > p.MaxPktSize { + // s.mtu.set(p.MaxPktSize) + //} s.launchProcessors() s.recv.configureHandshake(p) s.send.configureHandshake(p, true) @@ -581,9 +577,10 @@ func (s *udtSocket) readHandshake(m *multiplexer, p *packet.HandshakePacket) boo } s.farSockID = p.SockID - if s.mtu.get() > p.MaxPktSize { - s.mtu.set(p.MaxPktSize) - } + // See documentation above MTU negotation above. + //if s.mtu.get() > p.MaxPktSize { + // s.mtu.set(p.MaxPktSize) + //} s.launchProcessors() s.recv.configureHandshake(p) s.send.configureHandshake(p, true) diff --git a/udt/udtsocket_cc.go b/udt/udtsocket_cc.go index d6eab9c..f006817 100644 --- a/udt/udtsocket_cc.go +++ b/udt/udtsocket_cc.go @@ -208,7 +208,7 @@ func (s *udtSocketCc) GetRTT() time.Duration { // GetMSS is the largest packet size we can currently send (in bytes) func (s *udtSocketCc) GetMSS() uint { - return uint(s.socket.mtu.get()) + return uint(s.socket.maxPacketSize) } // SetACKPerid sets the time between ACKs sent to the peer diff --git a/udt/udtsocket_send.go b/udt/udtsocket_send.go index 3c5bc9f..5681503 100644 --- a/udt/udtsocket_send.go +++ b/udt/udtsocket_send.go @@ -84,7 +84,7 @@ func (s *udtSocketSend) SetPacketSendPeriod(snd time.Duration) { // check to see if we have a bandwidth limit here maxBandwidth := s.socket.Config.MaxBandwidth if maxBandwidth > 0 { - minSP := time.Second / time.Duration(float64(maxBandwidth)/float64(s.socket.mtu.get())) + minSP := time.Second / time.Duration(float64(maxBandwidth)/float64(s.socket.maxPacketSize)) if snd < minSP { snd = minSP } @@ -201,7 +201,7 @@ func (s *udtSocketSend) processDataMsg(isFirst bool, inChan <-chan sendMessage) s.msgSeq++ } - mtu := int(s.socket.mtu.get()) + mtu := int(s.socket.maxPacketSize) msgLen := len(partialSend.content) if msgLen >= mtu { // we are full -- send what we can and leave the rest