From e2cda5b5657a0d27fb048faa4b2901c2c383c3af Mon Sep 17 00:00:00 2001 From: Kleissner Date: Mon, 14 Mar 2022 06:13:52 +0100 Subject: [PATCH] UDT: Record transfer metrics. --- Transfer UDT.go | 3 +- udt/acceptsock_heap.go | 4 +- udt/config.go | 4 +- udt/listener.go | 8 +-- udt/metrics.go | 65 +++++++++++++++++++++++ udt/multiplexer.go | 8 ++- udt/udt.go | 10 ++-- udt/udtsocket.go | 114 +++++++++++++++++++++++------------------ udt/udtsocket_cc.go | 4 +- udt/udtsocket_recv.go | 5 +- udt/udtsocket_send.go | 4 +- 11 files changed, 153 insertions(+), 76 deletions(-) create mode 100644 udt/metrics.go diff --git a/Transfer UDT.go b/Transfer UDT.go index fe99508..aca5ade 100644 --- a/Transfer UDT.go +++ b/Transfer UDT.go @@ -10,7 +10,6 @@ package core import ( "errors" - "net" "time" "github.com/PeernetOfficial/core/protocol" @@ -77,7 +76,7 @@ func (peer *PeerInfo) startFileTransferUDT(hash []byte, fileSize uint64, offset, // FileTransferRequestUDT creates a UDT server listening for incoming data transfer via the lite protocol and requests a file transfer from a remote peer. // The caller must call udtConn.Close() when done. Do not use any of the closing functions of virtualConn. // Limit is optional. 0 means the entire file. -func (peer *PeerInfo) FileTransferRequestUDT(hash []byte, offset, limit uint64) (udtConn net.Conn, virtualConn *virtualPacketConn, err error) { +func (peer *PeerInfo) FileTransferRequestUDT(hash []byte, offset, limit uint64) (udtConn *udt.UDTSocket, virtualConn *virtualPacketConn, err error) { virtualConn = newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32, transferID uuid.UUID) { peer.sendTransfer(data, protocol.TransferControlActive, protocol.TransferProtocolUDT, hash, offset, limit, sequenceNumber, transferID, transferLite) }) diff --git a/udt/acceptsock_heap.go b/udt/acceptsock_heap.go index c53aae6..2afd250 100644 --- a/udt/acceptsock_heap.go +++ b/udt/acceptsock_heap.go @@ -11,7 +11,7 @@ type acceptSockInfo struct { sockID uint32 initSeqNo packet.PacketID lastTouch time.Time - sock *udtSocket + sock *UDTSocket } // acceptSockHeap defines a list of acceptSockInfo records sorted by their peer socketID and initial sequence number @@ -61,7 +61,7 @@ func (h acceptSockHeap) compare(sockID uint32, initSeqNo packet.PacketID, idx in } // Find does a binary search of the heap for the specified packetID which is returned -func (h acceptSockHeap) Find(sockID uint32, initSeqNo packet.PacketID) (*udtSocket, int) { +func (h acceptSockHeap) Find(sockID uint32, initSeqNo packet.PacketID) (*UDTSocket, int) { for n := 0; n < len(h); n++ { if h[n].sock.sockID == sockID { return h[n].sock, n diff --git a/udt/config.go b/udt/config.go index faaf637..6f2ae74 100644 --- a/udt/config.go +++ b/udt/config.go @@ -18,7 +18,7 @@ type Config struct { SynTime time.Duration // SynTime CanAccept func(hsPacket *packet.HandshakePacket) error // can this listener accept this connection? - CongestionForSocket func(sock *udtSocket) CongestionControl // create or otherwise return the CongestionControl for this socket + CongestionForSocket func(sock *UDTSocket) CongestionControl // create or otherwise return the CongestionControl for this socket } // DefaultConfig constructs a Config with default values @@ -32,7 +32,7 @@ func DefaultConfig() *Config { MaxBandwidth: 0, MaxPacketSize: 65535, SynTime: 10000 * time.Microsecond, - CongestionForSocket: func(sock *udtSocket) CongestionControl { + CongestionForSocket: func(sock *UDTSocket) CongestionControl { return &NativeCongestionControl{} }, } diff --git a/udt/listener.go b/udt/listener.go index 129aea6..1b95cfa 100644 --- a/udt/listener.go +++ b/udt/listener.go @@ -20,14 +20,14 @@ Listener implements the io.Listener interface for UDT. */ type listener struct { m *multiplexer - accept chan *udtSocket + accept chan *UDTSocket closed chan struct{} acceptHist acceptSockHeap acceptHistProt sync.Mutex config *Config } -func (l *listener) Accept() (net.Conn, error) { +func (l *listener) Accept() (*UDTSocket, error) { socket, ok := <-l.accept if ok { return socket, nil @@ -161,12 +161,12 @@ 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 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{}) *listener { m := newMultiplexer(closer, config.MaxPacketSize, incomingData, outgoingData, terminationSignal) l := &listener{ m: m, - accept: make(chan *udtSocket, 100), + accept: make(chan *UDTSocket, 100), closed: make(chan struct{}, 1), config: config, } diff --git a/udt/metrics.go b/udt/metrics.go new file mode 100644 index 0000000..4c57767 --- /dev/null +++ b/udt/metrics.go @@ -0,0 +1,65 @@ +package udt + +import ( + "github.com/PeernetOfficial/core/udt/packet" +) + +// recordTypeOfPacket records statistics on packet related metrics +func (s *UDTSocket) recordTypeOfPacket(p packet.Packet, isSend bool) { + + if isSend { + switch packet.PacketTypeName(p.PacketType()) { + case "handshake": + s.Metrics.PktSendHandShake++ + case "keep-alive": + s.Metrics.PktSendKeepAlive++ + case "ack": + s.Metrics.PktSentACK++ + case "nak": + s.Metrics.PktSentNAK++ + case "congestion": + s.Metrics.PktSentCongestion++ + case "shutdown": + s.Metrics.PktSentShutdown++ + case "ack2": + s.Metrics.PktSentACK2++ + case "msg-drop": + s.Metrics.PktSendMessageDrop++ + case "error": + s.Metrics.PktSendError++ + case "user-defined": + s.Metrics.PktSendUserDefined++ + case "data": + s.Metrics.PktSentData++ + default: + s.Metrics.PktSentOther++ + } + } else { + switch packet.PacketTypeName(p.PacketType()) { + case "handshake": + s.Metrics.PktRecvHandShake++ + case "keep-alive": + s.Metrics.PktRecvKeepAlive++ + case "ack": + s.Metrics.PktRecvACK++ + case "nak": + s.Metrics.PktRecvNAK++ + case "congestion": + s.Metrics.PktRecvCongestion++ + case "shutdown": + s.Metrics.PktRecvShutdown++ + case "ack2": + s.Metrics.PktRecvACK2++ + case "msg-drop": + s.Metrics.PktRecvMessageDrop++ + case "error": + s.Metrics.PktRecvError++ + case "user-defined": + s.Metrics.PktRecvUserDefined++ + case "data": + s.Metrics.PktRecvData++ + default: + s.Metrics.PktRecvOther++ + } + } +} diff --git a/udt/multiplexer.go b/udt/multiplexer.go index a6789d2..027859b 100644 --- a/udt/multiplexer.go +++ b/udt/multiplexer.go @@ -10,7 +10,7 @@ import ( // A multiplexer is a single UDT socket over a single PacketConn. type multiplexer struct { - socket *udtSocket // Socket + 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 @@ -35,7 +35,7 @@ func newMultiplexer(closer Closer, maxPacketSize uint, incomingData <-chan []byt return } -func (m *multiplexer) newSocket(config *Config, isServer bool, isDatagram bool) (s *udtSocket) { +func (m *multiplexer) newSocket(config *Config, isServer bool, isDatagram bool) (s *UDTSocket) { m.socketID = rand.Uint32() m.socket = newSocket(m, config, m.socketID, isServer, isDatagram) return m.socket @@ -89,6 +89,10 @@ func (m *multiplexer) sendPacket(destSockID uint32, ts uint32, p packet.Packet) } } + if m.socket != nil { + m.socket.recordTypeOfPacket(p, true) + } + buf := make([]byte, m.maxPacketSize) plen, err := p.WriteTo(buf) // encode if err != nil { diff --git a/udt/udt.go b/udt/udt.go index c472c0b..4714b8d 100644 --- a/udt/udt.go +++ b/udt/udt.go @@ -11,10 +11,6 @@ implemented: */ -import ( - "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. @@ -27,15 +23,15 @@ const ( 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. - TerminateReasonSocketClosed = 1004 // Send: Socket closed. Called udtSocket.Close(). + TerminateReasonSocketClosed = 1004 // Send: Socket closed. Called UDTSocket.Close(). 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. - TerminateReasonSignal = 1008 // Send: Terminate signal. Called udtSocket.Terminate(). + TerminateReasonSignal = 1008 // Send: Terminate signal. Called UDTSocket.Terminate(). ) // DialUDT establishes an outbound UDT connection using the existing provided packet connection. It creates a UDT client. -func DialUDT(config *Config, closer 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) (*UDTSocket, 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 01c3948..80c2418 100644 --- a/udt/udtsocket.go +++ b/udt/udtsocket.go @@ -43,12 +43,12 @@ type shutdownMessage struct { } /* -udtSocket encapsulates a UDT socket between a local and remote address pair, as -defined by the UDT specification. udtSocket implements the net.Conn interface +UDTSocket encapsulates a UDT socket between a local and remote address pair, as +defined by the UDT specification. UDTSocket implements the net.Conn interface so that it can be used anywhere that a stream-oriented network connection (like TCP) would be used. */ -type udtSocket struct { +type UDTSocket struct { // this data not changed after the socket is initialized and/or handshaked m *multiplexer // the multiplexer that handles this socket //raddr *net.UDPAddr // the remote address @@ -101,28 +101,39 @@ type udtSocket struct { cong *udtSocketCc // reference to contestion control // performance metrics - //PktSent uint64 // number of sent data packets, including retransmissions - //PktRecv uint64 // number of received packets - //PktSndLoss uint // number of lost packets (sender side) - //PktRcvLoss uint // number of lost packets (receiver side) - //PktRetrans uint // number of retransmitted packets - //PktSentACK uint // number of sent ACK packets - //PktRecvACK uint // number of received ACK packets - //PktSentNAK uint // number of sent NAK packets - //PktRecvNAK uint // number of received NAK packets - //MbpsSendRate float64 // sending rate in Mb/s - //MbpsRecvRate float64 // receiving rate in Mb/s - //SndDuration time.Duration // busy sending time (i.e., idle time exclusive) + Metrics *Metrics +} - // instant measurements - //PktSndPeriod time.Duration // packet sending period - //PktFlowWindow uint // flow window size, in number of packets - //PktCongestionWindow uint // congestion window size, in number of packets - //PktFlightSize uint // number of packets on flight - //MsRTT time.Duration // RTT - //MbpsBandwidth float64 // estimated bandwidth, in Mb/s - //ByteAvailSndBuf uint // available UDT sender buffer size - //ByteAvailRcvBuf uint // available UDT receiver buffer size +type Metrics struct { + PktSentData uint64 // number of sent data packets, including retransmissions + PktSendHandShake uint64 // number of Handshake packets sent + PktRecvHandShake uint64 // number of Handshake packets received + PktSendKeepAlive uint64 // number of Keep-alive packets sent + PktRecvKeepAlive uint64 // number of Keep-alive packets received + PktRecvData uint64 // number of received packets + PktSentCongestion uint64 // number of Congestion Packets sent + PktRecvCongestion uint64 // number of Congestion Packets received + PktSentShutdown uint64 // number of Shutdown Packets sent + PktRecvShutdown uint64 // number of Shutdown Packets received + PktSendMessageDrop uint64 // number of Message Drop Packets sent + PktRecvMessageDrop uint64 // number of Message Drop Packets received + PktSendError uint64 // number of Error Packets sent + PktRecvError uint64 // number of Error Packets received + PktSendUserDefined uint64 // number of User Defined Packets sent + PktRecvUserDefined uint64 // number of User Defined Packets received + PktSndLoss uint // number of lost packets (sender side) + PktRcvLoss uint // number of lost packets (receiver side) + PktRetrans uint // number of retransmitted packets + PktSentACK uint // number of sent ACK packets + PktSentACK2 uint // number of sent ACK2 packets + PktRecvACK2 uint // number of received ACK2 packets + PktRecvACK uint // number of received ACK packets + PktSentNAK uint // number of sent NAK packets + PktRecvNAK uint // number of received NAK packets + PktSentOther uint // number of sent Other packets + PktRecvOther uint // number of received Other packets + MbpsSendRate float64 // sending rate in Mb/s + MbpsRecvRate float64 // receiving rate in Mb/s } /******************************************************************************* @@ -130,7 +141,7 @@ type udtSocket struct { *******************************************************************************/ // Grab the next data packet -func (s *udtSocket) fetchReadPacket(blocking bool) ([]byte, error) { +func (s *UDTSocket) fetchReadPacket(blocking bool) ([]byte, error) { var result []byte if blocking { for { @@ -170,7 +181,7 @@ func (s *udtSocket) fetchReadPacket(blocking bool) ([]byte, error) { return result, nil } -func (s *udtSocket) connectionError() error { +func (s *UDTSocket) connectionError() error { switch s.sockState { case sockStateRefused: return errors.New("Connection refused by remote host") @@ -190,7 +201,7 @@ func (s *udtSocket) connectionError() error { // Read can be made to time out and return an Error with Timeout() == true // after a fixed time limit; see SetDeadline and SetReadDeadline. // (required for net.Conn implementation) -func (s *udtSocket) Read(p []byte) (n int, err error) { +func (s *UDTSocket) Read(p []byte) (n int, err error) { connErr := s.connectionError() if s.isDatagram { // for datagram sockets, block until we have a message to return and then return it @@ -240,7 +251,7 @@ func (s *udtSocket) Read(p []byte) (n int, err error) { // Write can be made to time out and return an Error with Timeout() == true // after a fixed time limit; see SetDeadline and SetWriteDeadline. // (required for net.Conn implementation) -func (s *udtSocket) Write(p []byte) (n int, err error) { +func (s *UDTSocket) Write(p []byte) (n int, err error) { // at the moment whatever we have right now we'll shove it into a channel and return // on the other side: // for datagram sockets: this is a distinct message to be broken into as few packets as possible @@ -293,7 +304,7 @@ func (s *udtSocket) Write(p []byte) (n int, err error) { // Write operations will be permitted to send (initial packets) // Read operations will return an error // (required for net.Conn implementation). // Note: Do not simultaneously call Close() and Write(). To close while the socket is still in use, use Terminate(). -func (s *udtSocket) Close() error { +func (s *UDTSocket) Close() error { s.closeMutex.Lock() defer s.closeMutex.Unlock() @@ -311,7 +322,7 @@ func (s *udtSocket) Close() error { // Terminate terminates the connection immediately. Unlike Close, it does not permit any reading/writing. // If the connection should be ordinarily closed (after reading/writing) use Close(). -func (s *udtSocket) Terminate() error { +func (s *UDTSocket) Terminate() error { s.closeMutex.Lock() defer s.closeMutex.Unlock() @@ -325,7 +336,7 @@ func (s *udtSocket) Terminate() error { return nil } -func (s *udtSocket) isOpen() bool { +func (s *UDTSocket) isOpen() bool { switch s.sockState { case sockStateClosed, sockStateRefused, sockStateCorrupted, sockStateTimeout: return false @@ -336,14 +347,14 @@ func (s *udtSocket) isOpen() bool { // LocalAddr returns the local network address. // (required for net.Conn implementation) -func (s *udtSocket) LocalAddr() net.Addr { +func (s *UDTSocket) LocalAddr() net.Addr { //return s.m.laddr return nil } // RemoteAddr returns the remote network address. // (required for net.Conn implementation) -func (s *udtSocket) RemoteAddr() net.Addr { +func (s *UDTSocket) RemoteAddr() net.Addr { //return s.raddr return nil } @@ -371,13 +382,13 @@ func (s *udtSocket) RemoteAddr() net.Addr { // failure on I/O can be detected using // errors.Is(err, syscall.ETIMEDOUT). // (required for net.Conn implementation) -func (s *udtSocket) SetDeadline(t time.Time) error { +func (s *UDTSocket) SetDeadline(t time.Time) error { s.setDeadline(t, &s.readDeadline, &s.readDeadlinePassed) s.setDeadline(t, &s.writeDeadline, &s.writeDeadlinePassed) return nil } -func (s *udtSocket) setDeadline(dl time.Time, timer **time.Timer, timerPassed *bool) { +func (s *UDTSocket) setDeadline(dl time.Time, timer **time.Timer, timerPassed *bool) { if *timer == nil { if !dl.IsZero() { *timer = time.NewTimer(dl.Sub(time.Now())) @@ -404,7 +415,7 @@ func (s *udtSocket) setDeadline(dl time.Time, timer **time.Timer, timerPassed *b // and any currently-blocked Read call. // A zero value for t means Read will not time out. // (required for net.Conn implementation) -func (s *udtSocket) SetReadDeadline(t time.Time) error { +func (s *UDTSocket) SetReadDeadline(t time.Time) error { s.setDeadline(t, &s.readDeadline, &s.readDeadlinePassed) return nil } @@ -415,7 +426,7 @@ func (s *udtSocket) SetReadDeadline(t time.Time) error { // some of the data was successfully written. // A zero value for t means Write will not time out. // (required for net.Conn implementation) -func (s *udtSocket) SetWriteDeadline(t time.Time) error { +func (s *UDTSocket) SetWriteDeadline(t time.Time) error { s.setDeadline(t, &s.writeDeadline, &s.writeDeadlinePassed) return nil } @@ -425,7 +436,7 @@ func (s *udtSocket) SetWriteDeadline(t time.Time) error { *******************************************************************************/ // newSocket creates a new UDT socket, which will be configured afterwards as either an incoming our outgoing socket -func newSocket(m *multiplexer, config *Config, sockID uint32, isServer bool, isDatagram bool) (s *udtSocket) { +func newSocket(m *multiplexer, config *Config, sockID uint32, isServer bool, isDatagram bool) (s *UDTSocket) { now := time.Now() maxFlowWinSize := config.MaxFlowWinSize @@ -436,7 +447,7 @@ func newSocket(m *multiplexer, config *Config, sockID uint32, isServer bool, isD maxFlowWinSize = 32 } - s = &udtSocket{ + s = &UDTSocket{ m: m, Config: config, //raddr: raddr, @@ -459,19 +470,20 @@ func newSocket(m *multiplexer, config *Config, sockID uint32, isServer bool, isD bandwidth: 1, sendPacket: make(chan packet.Packet, 256), shutdownEvent: make(chan shutdownMessage, 5), + Metrics: &Metrics{}, } s.cong = newUdtSocketCc(s) return } -func (s *udtSocket) launchProcessors() { +func (s *UDTSocket) launchProcessors() { s.send = newUdtSocketSend(s) s.recv = newUdtSocketRecv(s) s.cong.init(s.initPktSeq) } -func (s *udtSocket) startConnect() error { +func (s *UDTSocket) startConnect() error { connectWait := &sync.WaitGroup{} s.connectWait = connectWait @@ -489,7 +501,7 @@ func (s *udtSocket) startConnect() error { return s.connectionError() } -func (s *udtSocket) goManageConnection() { +func (s *UDTSocket) goManageConnection() { for { select { case <-s.lingerTimer: // linger timer expired, shut everything down @@ -517,7 +529,7 @@ func (s *udtSocket) goManageConnection() { } } -func (s *udtSocket) sendHandshake(reqType packet.HandshakeReqType) { +func (s *UDTSocket) sendHandshake(reqType packet.HandshakeReqType) { sockType := packet.TypeSTREAM if s.isDatagram { sockType = packet.TypeDGRAM @@ -540,7 +552,7 @@ func (s *udtSocket) sendHandshake(reqType packet.HandshakeReqType) { } // checkValidHandshake checks to see if we want to accept a new connection with this handshake. -func (s *udtSocket) checkValidHandshake(m *multiplexer, p *packet.HandshakePacket) bool { +func (s *UDTSocket) checkValidHandshake(m *multiplexer, p *packet.HandshakePacket) bool { if s.udtVer != 4 { return false } @@ -549,7 +561,7 @@ func (s *udtSocket) checkValidHandshake(m *multiplexer, p *packet.HandshakePacke // readHandshake is received when a handshake packet is received without a destination, either as part // of a listening response or as a rendezvous connection -func (s *udtSocket) readHandshake(m *multiplexer, p *packet.HandshakePacket) bool { +func (s *UDTSocket) readHandshake(m *multiplexer, p *packet.HandshakePacket) bool { switch s.sockState { case sockStateInit: // server accepting a connection from a client s.initPktSeq = p.InitPktSeq @@ -626,7 +638,7 @@ func (s *udtSocket) readHandshake(m *multiplexer, p *packet.HandshakePacket) boo return false } -func (s *udtSocket) shutdown(sockState sockState, permitLinger bool, err error, reason int) { +func (s *UDTSocket) shutdown(sockState sockState, permitLinger bool, err error, reason int) { if !s.isOpen() { return // already closed } @@ -669,14 +681,14 @@ func absdiff(a uint, b uint) uint { return a - b } -func (s *udtSocket) applyRTT(rtt uint) { +func (s *UDTSocket) applyRTT(rtt uint) { s.rttProt.Lock() s.rttVar = (s.rttVar*3 + absdiff(s.rtt, rtt)) >> 2 s.rtt = (s.rtt*7 + rtt) >> 3 s.rttProt.Unlock() } -func (s *udtSocket) getRTT() (rtt, rttVar uint) { +func (s *UDTSocket) getRTT() (rtt, rttVar uint) { s.rttProt.RLock() rtt = s.rtt rttVar = s.rttVar @@ -685,7 +697,7 @@ func (s *udtSocket) getRTT() (rtt, rttVar uint) { } // Update Estimated Bandwidth and packet delivery rate -func (s *udtSocket) applyReceiveRates(deliveryRate uint, bandwidth uint) { +func (s *UDTSocket) applyReceiveRates(deliveryRate uint, bandwidth uint) { s.receiveRateProt.Lock() if deliveryRate > 0 { s.deliveryRate = (s.deliveryRate*7 + deliveryRate) >> 3 @@ -696,7 +708,7 @@ func (s *udtSocket) applyReceiveRates(deliveryRate uint, bandwidth uint) { s.receiveRateProt.Unlock() } -func (s *udtSocket) getRcvSpeeds() (deliveryRate uint, bandwidth uint) { +func (s *UDTSocket) getRcvSpeeds() (deliveryRate uint, bandwidth uint) { s.receiveRateProt.RLock() deliveryRate = s.deliveryRate bandwidth = s.bandwidth @@ -706,7 +718,7 @@ func (s *udtSocket) getRcvSpeeds() (deliveryRate uint, bandwidth uint) { // called by the multiplexer read loop when a packet is received for this socket. // Minimal processing is permitted but try not to stall the caller -func (s *udtSocket) readPacket(m *multiplexer, p packet.Packet) { +func (s *UDTSocket) readPacket(m *multiplexer, p packet.Packet) { now := time.Now() if s.sockState == sockStateClosed { return diff --git a/udt/udtsocket_cc.go b/udt/udtsocket_cc.go index 2d1fd95..fe3cf26 100644 --- a/udt/udtsocket_cc.go +++ b/udt/udtsocket_cc.go @@ -29,7 +29,7 @@ type congMsg struct { type udtSocketCc struct { // channels sockClosed <-chan struct{} // closed when socket is closed - socket *udtSocket + socket *UDTSocket congestion CongestionControl // congestion control object for this socket msgs chan congMsg @@ -38,7 +38,7 @@ type udtSocketCc struct { sndPeriod time.Duration // delay between sending packets } -func newUdtSocketCc(s *udtSocket) *udtSocketCc { +func newUdtSocketCc(s *UDTSocket) *udtSocketCc { newCongestion := s.Config.CongestionForSocket if newCongestion == nil { newCongestion = DefaultConfig().CongestionForSocket diff --git a/udt/udtsocket_recv.go b/udt/udtsocket_recv.go index 6d32b45..c4481e2 100644 --- a/udt/udtsocket_recv.go +++ b/udt/udtsocket_recv.go @@ -10,7 +10,7 @@ type udtSocketRecv struct { // channels messageIn chan<- []byte // inbound messages. Sender is goReceiveEvent->ingestData, Receiver is client caller (Read) sendPacket chan<- packet.Packet // send a packet out on the wire - socket *udtSocket + socket *UDTSocket nextSequenceExpect packet.PacketID // the peer's next largest packet ID expected. lastSequence packet.PacketID // the peer's last received packet ID before any loss events @@ -34,7 +34,7 @@ type udtSocketRecv struct { resendNAKLimiter rateLimiter // Doubles after every resend to prevent ddos } -func newUdtSocketRecv(s *udtSocket) *udtSocketRecv { +func newUdtSocketRecv(s *UDTSocket) *udtSocketRecv { sr := &udtSocketRecv{ socket: s, messageIn: s.messageIn, @@ -68,6 +68,7 @@ func (s *udtSocketRecv) goReceiveEvent() { for { select { case evt := <-s.socket.recvEvent: + s.socket.recordTypeOfPacket(evt.pkt, false) switch sp := evt.pkt.(type) { case *packet.Ack2Packet: s.ingestAck2(sp, evt.now) diff --git a/udt/udtsocket_send.go b/udt/udtsocket_send.go index 660bffc..3e1d5ac 100644 --- a/udt/udtsocket_send.go +++ b/udt/udtsocket_send.go @@ -27,7 +27,7 @@ type udtSocketSend struct { messageOut <-chan sendMessage // outbound data messages. Sender is client caller (Write), Receiver is goSendEvent. Closed when socket is closed sendPacket chan<- packet.Packet // send a packet out on the wire shutdownEvent chan<- shutdownMessage // channel signals the connection to be shutdown - socket *udtSocket + socket *UDTSocket sendState sendState // current sender state sendPktPend *sendPacketHeap // list of packets that have been sent but not yet acknowledged @@ -44,7 +44,7 @@ type udtSocketSend struct { resendDataTime time.Duration // Doubles after every send to prevent ddos } -func newUdtSocketSend(s *udtSocket) *udtSocketSend { +func newUdtSocketSend(s *UDTSocket) *udtSocketSend { ss := &udtSocketSend{ socket: s, sendPktSeq: s.initPktSeq,