UDT: Remove unnecessary and confusing sockShutdown. Elevate linger mode to full listen/send mode, which is necessary to combat the shutdown message race condition. Close #47 close #45

udt/udtsocket.go

@@ -2,7 +2,6 @@ package udt
 
 import (
 	"errors"
-	"fmt"
 	"math/rand"
 	"net"
 	"sync"
@@ -86,7 +85,6 @@ type udtSocket struct {
 	sendEvent     chan recvPktEvent    // sender: ingest the specified packet. Sender is readPacket, receiver is goSendEvent
 	sendPacket    chan packet.Packet   // packets to send out on the wire (once goManageConnection is running)
 	shutdownEvent chan shutdownMessage // channel signals the connection to be shutdown
-	sockShutdown  chan struct{}        // closed when socket is shutdown
 	sockClosed    chan struct{}        // closed when socket is closed
 
 	// timers
@@ -430,7 +428,6 @@ func newSocket(m *multiplexer, config *Config, sockID uint32, isServer bool, isD
 		recvEvent:      make(chan recvPktEvent, 256),
 		sendEvent:      make(chan recvPktEvent, 256),
 		sockClosed:     make(chan struct{}, 1),
-		sockShutdown:   make(chan struct{}, 1),
 		deliveryRate:   16,
 		bandwidth:      1,
 		sendPacket:     make(chan packet.Packet, 256),
@@ -467,21 +464,17 @@ func (s *udtSocket) startConnect() error {
 
 func (s *udtSocket) goManageConnection() {
 	sockClosed := s.sockClosed
-	sockShutdown := s.sockShutdown
 	for {
 		select {
 		case <-s.lingerTimer: // linger timer expired, shut everything down
-			s.m.closeSocket(s.sockID)
-			close(s.sockClosed)
+			s.shutdown(sockStateClosed, false, nil)
 			return
-		case _, _ = <-sockShutdown:
-			// catching this to force re-evaluation of this select (catching the linger timer)
 		case _, _ = <-sockClosed:
 			return
 		case p := <-s.sendPacket:
 			ts := uint32(time.Now().Sub(s.created) / time.Microsecond)
 			s.cong.onPktSent(p)
-			fmt.Printf("(id=%d) sending %s  (id=%d)\n", s.sockID, packet.PacketTypeName(p.PacketType()), s.farSockID)
+			//fmt.Printf("(id=%d) sending %s  (id=%d)\n", s.sockID, packet.PacketTypeName(p.PacketType()), s.farSockID)
 			s.m.sendPacket(s.farSockID, ts, p)
 		case sd := <-s.shutdownEvent: // connection shut down
 			s.shutdown(sd.sockState, sd.permitLinger, sd.err)
@@ -516,7 +509,7 @@ func (s *udtSocket) sendHandshake(reqType packet.HandshakeReqType) {
 
 	ts := uint32(time.Now().Sub(s.created) / time.Microsecond)
 	s.cong.onPktSent(p)
-	fmt.Printf("(id=%d) sending handshake(%d) (id=%d)\n", s.sockID, int(reqType), s.farSockID)
+	//fmt.Printf("(id=%d) sending handshake(%d) (id=%d)\n", s.sockID, int(reqType), s.farSockID)
 	s.m.sendPacket(s.farSockID, ts, p)
 }
 
@@ -611,17 +604,11 @@ func (s *udtSocket) shutdown(sockState sockState, permitLinger bool, err error)
 	if !s.isOpen() {
 		return // already closed
 	}
-	if err != nil {
-		fmt.Printf("socket shutdown (type=%d), due to error: %s\n", int(sockState), err.Error())
-	} else {
-		fmt.Printf("socket shutdown (type=%d)\n", int(sockState))
-	}
-	if s.connectWait != nil {
-		s.connectWait.Done()
-		s.connectWait = nil
-	}
-	s.sockState = sockState
-	s.cong.close()
+	//if err != nil {
+	//	fmt.Printf("socket shutdown (type=%d), due to error: %s\n", int(sockState), err.Error())
+	//} else {
+	//	fmt.Printf("socket shutdown (type=%d) (permitLinger = %t, duration = %s)\n", int(sockState), permitLinger, s.Config.LingerTime.String())
+	//}
 
 	if permitLinger {
 		linger := s.Config.LingerTime
@@ -629,16 +616,22 @@ func (s *udtSocket) shutdown(sockState sockState, permitLinger bool, err error)
 			linger = DefaultConfig().LingerTime
 		}
 		s.lingerTimer = time.After(linger)
+		return
 	}
 
+	if s.connectWait != nil {
+		s.connectWait.Done()
+		s.connectWait = nil
+	}
+	s.sockState = sockState
+	s.cong.close()
+
 	s.connTimeout = nil
 	s.connRetry = nil
-	if permitLinger {
-		close(s.sockShutdown)
-	} else {
-		s.m.closeSocket(s.sockID)
-		close(s.sockClosed)
-	}
+	s.m.closeSocket(s.sockID)
+	close(s.sockClosed)
+	close(s.recvEvent)
+
 	s.messageIn <- nil
 }
 

udt/udtsocket_recv.go

@@ -14,12 +14,11 @@ const (
 
 type udtSocketRecv struct {
 	// channels
-	sockClosed   <-chan struct{}      // closed when socket is closed
-	sockShutdown <-chan struct{}      // closed when socket is shutdown
-	recvEvent    <-chan recvPktEvent  // receiver: ingest the specified packet. Sender is readPacket, receiver is goReceiveEvent
-	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
+	sockClosed <-chan struct{}      // closed when socket is closed
+	recvEvent  <-chan recvPktEvent  // receiver: ingest the specified packet. Sender is readPacket, receiver is goReceiveEvent
+	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
 
 	farNextPktSeq      packet.PacketID // the peer's next largest packet ID expected.
 	farRecdPktSeq      packet.PacketID // the peer's last "received" packet ID (before any loss events)
@@ -46,12 +45,11 @@ type udtSocketRecv struct {
 
 func newUdtSocketRecv(s *udtSocket) *udtSocketRecv {
 	sr := &udtSocketRecv{
-		socket:       s,
-		sockClosed:   s.sockClosed,
-		sockShutdown: s.sockShutdown,
-		recvEvent:    s.recvEvent,
-		messageIn:    s.messageIn,
-		sendPacket:   s.sendPacket,
+		socket:     s,
+		sockClosed: s.sockClosed,
+		recvEvent:  s.recvEvent,
+		messageIn:  s.messageIn,
+		sendPacket: s.sendPacket,
 	}
 	go sr.goReceiveEvent()
 	return sr
@@ -67,7 +65,6 @@ func (s *udtSocketRecv) configureHandshake(p *packet.HandshakePacket) {
 func (s *udtSocketRecv) goReceiveEvent() {
 	recvEvent := s.recvEvent
 	sockClosed := s.sockClosed
-	sockShutdown := s.sockShutdown
 	for {
 		select {
 		case evt, ok := <-recvEvent:
@@ -84,8 +81,6 @@ func (s *udtSocketRecv) goReceiveEvent() {
 			case *packet.ErrPacket:
 				s.ingestError(sp)
 			}
-		case _, _ = <-sockShutdown: // socket is shut down, no need to receive any further data
-			return
 		case _, _ = <-sockClosed: // socket is closed, leave now
 			return
 		case <-s.ackSentEvent:

udt/udtsocket_send.go

@@ -15,7 +15,6 @@ const (
 	sendStateSending                      // recently sent something, waiting for SND before sending more
 	sendStateWaiting                      // destination is full, waiting for them to process something and come back
 	sendStateProcessDrop                  // immediately re-process any drop list requests
-	sendStateShutdown                     // connection is shutdown
 )
 
 const (
@@ -25,7 +24,6 @@ const (
 type udtSocketSend struct {
 	// channels
 	sockClosed    <-chan struct{}        // closed when socket is closed
-	sockShutdown  <-chan struct{}        // closed when socket is shutdown
 	sendEvent     <-chan recvPktEvent    // sender: ingest the specified packet. Sender is readPacket, receiver is goSendEvent
 	messageOut    <-chan sendMessage     // outbound 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
@@ -59,7 +57,6 @@ func newUdtSocketSend(s *udtSocket) *udtSocketSend {
 		expCount:       1,
 		sendPktSeq:     s.initPktSeq,
 		sockClosed:     s.sockClosed,
-		sockShutdown:   s.sockShutdown,
 		sendEvent:      s.sendEvent,
 		messageOut:     s.messageOut,
 		congestWindow:  atomicUint32{val: 16},
@@ -99,7 +96,6 @@ func (s *udtSocketSend) goSendEvent() {
 	sockClosed := s.sockClosed
 	for {
 		thisMsgChan := messageOut
-		sockShutdown := s.sockShutdown
 
 		switch s.sendState {
 		case sendStateIdle: // not waiting for anything, can send immediately
@@ -113,17 +109,11 @@ func (s *udtSocketSend) goSendEvent() {
 				s.processSendExpire()
 			}
 			continue
-		case sendStateShutdown:
-			sockShutdown = nil
-			thisMsgChan = nil
 		default:
 			thisMsgChan = nil
 		}
 
 		select {
-		case _, _ = <-sockShutdown:
-			s.sendState = sendStateShutdown
-			s.expTimerEvent = nil // don't process EXP events if we're shutting down
 		case msg, ok := <-thisMsgChan: // nil if we can't process outgoing messages right now
 			if !ok {
 				s.sendPacket <- &packet.ShutdownPacket{}