UDT: Record transfer metrics.

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

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

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

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

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++
+		}
+	}
+}

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 {

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)

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

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

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)

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,