Peernet-Archive/core
1
0
Fork 0
mirror of https://github.com/PeernetOfficial/core.git synced 2026-07-16 18:37:51 +01:00
Code Issues Packages Projects Releases Wiki Activity

UDT: Removing MTU in favor of MaxPacketSize.

Browse Source
This commit is contained in:
Kleissner
2021-10-24 04:59:14 +02:00
parent 75e3d9a42e
commit 153ca78ed7
9 changed files with 55 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
udt/config.go
View File

@@ -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{}

2
udt/listener.go
View File

@@ -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,

26
udt/multiplexer.go
View File

@@ -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 {

12
udt/packet/packet_handshake.go
View File

@@ -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])

10
udt/readme.md
View File

@@ -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.

2
udt/udt.go
View File

@@ -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()

45
udt/udtsocket.go
View File

@@ -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)

2
udt/udtsocket_cc.go
View File

@@ -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

4
udt/udtsocket_send.go
View File

@@ -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