diff --git a/udt/udtsocket.go b/udt/udtsocket.go index 0737e0c..2b580e1 100644 --- a/udt/udtsocket.go +++ b/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 } diff --git a/udt/udtsocket_recv.go b/udt/udtsocket_recv.go index 9273e5c..d7f9f06 100644 --- a/udt/udtsocket_recv.go +++ b/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: diff --git a/udt/udtsocket_send.go b/udt/udtsocket_send.go index ee55dea..fceb87f 100644 --- a/udt/udtsocket_send.go +++ b/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{}