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

UDT: Reduce complexity of multiplexer, removes a race condition, increases stability. pktOut was not needed.

Browse Source
This commit is contained in:
Kleissner
2021-10-25 09:46:36 +02:00
parent e476c7e40b
commit 13ef42ac39
1 changed files with 22 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
udt/multiplexer.go
View File

@@ -13,23 +13,21 @@ import (
// A multiplexer is a single UDT socket over a single PacketConn.
type multiplexer struct {
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
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
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
}
// 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{
maxPacketSize: maxPacketSize,
pktOut: make(chan packet.Packet, 100),
closer: closer,
incomingData: incomingData,
outgoingData: outgoingData,
@@ -37,7 +35,6 @@ func newMultiplexer(closer io.Closer, maxPacketSize uint, incomingData <-chan []
}
go m.goRead()
go m.goWrite()
return
}
@@ -54,7 +51,6 @@ func (m *multiplexer) unlistenUDT(l *listener) {
m.listenSock = nil
m.closer.Close()
close(m.pktOut)
}
func (m *multiplexer) newSocket(config *Config, isServer bool, isDatagram bool) (s *udtSocket) {
@@ -74,7 +70,6 @@ func (m *multiplexer) closeSocket(sockID uint32) {
m.socket = nil
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.
@@ -115,25 +110,6 @@ 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() {
for pkt := range m.pktOut {
buf := make([]byte, m.maxPacketSize)
plen, err := pkt.WriteTo(buf) // encode
if err != nil {
// TODO: handle write error
fmt.Printf("Error encoding UDT packet: %s\n", err.Error())
return
}
select {
case m.outgoingData <- buf[0:plen]:
case <-m.terminationSignal:
return
}
}
}
func (m *multiplexer) sendPacket(destSockID uint32, ts uint32, p packet.Packet) {
p.SetHeader(destSockID, ts)
if destSockID == 0 {
@@ -142,5 +118,17 @@ func (m *multiplexer) sendPacket(destSockID uint32, ts uint32, p packet.Packet)
return
}
}
m.pktOut <- p
buf := make([]byte, m.maxPacketSize)
plen, err := p.WriteTo(buf) // encode
if err != nil {
fmt.Printf("Error encoding UDT packet: %s\n", err.Error())
return
}
select {
case m.outgoingData <- buf[0:plen]:
case <-m.terminationSignal:
return
}
}