diff --git a/sess.go b/sess.go index abcf29b..5fb20ac 100644 --- a/sess.go +++ b/sess.go @@ -90,8 +90,8 @@ type ( nonce Entropy // packets waiting to be sent on wire - txqueue []ipv4.Message - chTxQueue chan []ipv4.Message + txqueue []ipv4.Message + bconn batchConn // for casting of batchConn mu sync.Mutex } @@ -118,7 +118,14 @@ func newUDPSession(conv uint32, dataShards, parityShards int, l *Listener, conn sess.l = l sess.block = block sess.recvbuf = make([]byte, mtuLimit) - sess.chTxQueue = make(chan []ipv4.Message) + + // cast to writebatch conn + addr, _ := net.ResolveUDPAddr("udp", conn.LocalAddr().String()) + if addr.IP.To4() != nil { + sess.bconn = ipv4.NewPacketConn(conn) + } else { + sess.bconn = ipv6.NewPacketConn(conn) + } // FEC codec initialization sess.fecDecoder = newFECDecoder(rxFECMulti*(dataShards+parityShards), dataShards, parityShards) @@ -149,7 +156,6 @@ func newUDPSession(conv uint32, dataShards, parityShards int, l *Listener, conn if sess.l == nil { // it's a client connection go sess.readLoop() - go sess.txLoop() atomic.AddUint64(&DefaultSnmp.ActiveOpens, 1) } else { atomic.AddUint64(&DefaultSnmp.PassiveOpens, 1) @@ -266,8 +272,8 @@ func (s *UDPSession) WriteBuffers(v [][]byte) (n int, err error) { if s.kcp.WaitSnd() >= int(s.kcp.snd_wnd) || !s.writeDelay { s.kcp.flush(false) } - s.mu.Unlock() s.uncork() + s.mu.Unlock() atomic.AddUint64(&DefaultSnmp.BytesSent, uint64(n)) return n, nil } @@ -304,25 +310,9 @@ func (s *UDPSession) WriteBuffers(v [][]byte) (n int, err error) { // uncork sends data in txqueue if there is any func (s *UDPSession) uncork() { - s.mu.Lock() - s.uncorkInternal() - s.mu.Unlock() -} - -// uncork sends data in txqueue if there is any -func (s *UDPSession) uncorkInternal() { if len(s.txqueue) > 0 { - if s.l != nil { - select { - case s.l.chTxQueue <- s.txqueue: - case <-s.l.die: - } - } else { - select { - case s.chTxQueue <- s.txqueue: - case <-s.die: - } - } + s.tx(s.txqueue) + s.txqueue = nil } s.txqueue = nil } @@ -526,9 +516,6 @@ func (s *UDPSession) output(buf []byte) { msg.Buffers = [][]byte{bts} msg.Addr = s.remote s.txqueue = append(s.txqueue, msg) - if len(s.txqueue) >= batchSize { - s.uncorkInternal() - } } for k := range ecc { @@ -537,9 +524,6 @@ func (s *UDPSession) output(buf []byte) { msg.Buffers = [][]byte{bts} msg.Addr = s.remote s.txqueue = append(s.txqueue, msg) - if len(s.txqueue) >= batchSize { - s.uncorkInternal() - } } } @@ -551,8 +535,8 @@ func (s *UDPSession) update() (interval time.Duration) { if s.kcp.WaitSnd() < waitsnd { s.notifyWriteEvent() } - s.mu.Unlock() s.uncork() + s.mu.Unlock() return } @@ -642,6 +626,7 @@ func (s *UDPSession) kcpInput(data []byte) { if s.kcp.WaitSnd() < waitsnd { s.notifyWriteEvent() } + s.uncork() s.mu.Unlock() } else { atomic.AddUint64(&DefaultSnmp.InErrs, 1) @@ -661,6 +646,7 @@ func (s *UDPSession) kcpInput(data []byte) { if s.kcp.WaitSnd() < waitsnd { s.notifyWriteEvent() } + s.uncork() s.mu.Unlock() } @@ -679,8 +665,6 @@ func (s *UDPSession) kcpInput(data []byte) { atomic.AddUint64(&DefaultSnmp.FECRecovered, fecRecovered) } - // input may trigger fast-resend and acks - s.uncork() } type ( @@ -703,8 +687,6 @@ type ( socketError atomic.Value rd atomic.Value // read deadline for Accept() - - chTxQueue chan []ipv4.Message } ) @@ -894,7 +876,6 @@ func ServeConn(block BlockCrypt, dataShards, parityShards int, conn net.PacketCo l.parityShards = parityShards l.block = block l.fecDecoder = newFECDecoder(rxFECMulti*(dataShards+parityShards), dataShards, parityShards) - l.chTxQueue = make(chan []ipv4.Message) // calculate header size if l.block != nil { @@ -905,7 +886,6 @@ func ServeConn(block BlockCrypt, dataShards, parityShards int, conn net.PacketCo } go l.monitor() - go l.txLoop() return l, nil } diff --git a/tx_generic.go b/tx_generic.go index daebb28..999256b 100644 --- a/tx_generic.go +++ b/tx_generic.go @@ -6,50 +6,21 @@ import ( "sync/atomic" "github.com/pkg/errors" + "golang.org/x/net/ipv4" ) -func (s *UDPSession) txLoop() { - for { - select { - case txqueue := <-s.chTxQueue: - nbytes := 0 - for k := range txqueue { - if n, err := s.conn.WriteTo(txqueue[k].Buffers[0], txqueue[k].Addr); err == nil { - nbytes += n - xmitBuf.Put(txqueue[k].Buffers[0]) - } else { - s.socketError.Store(errors.WithStack(err)) - s.Close() - return - } - } - atomic.AddUint64(&DefaultSnmp.OutPkts, uint64(len(txqueue))) - atomic.AddUint64(&DefaultSnmp.OutBytes, uint64(nbytes)) - case <-s.die: - return - } - } -} - -func (l *Listener) txLoop() { - for { - select { - case txqueue := <-l.chTxQueue: - nbytes := 0 - for k := range txqueue { - if n, err := l.conn.WriteTo(txqueue[k].Buffers[0], txqueue[k].Addr); err == nil { - nbytes += n - xmitBuf.Put(txqueue[k].Buffers[0]) - } else { - l.socketError.Store(errors.WithStack(err)) - l.Close() - return - } - } - atomic.AddUint64(&DefaultSnmp.OutPkts, uint64(len(txqueue))) - atomic.AddUint64(&DefaultSnmp.OutBytes, uint64(nbytes)) - case <-l.die: +func (s *UDPSession) tx(txqueue []ipv4.Message) { + nbytes := 0 + for k := range txqueue { + if n, err := s.conn.WriteTo(txqueue[k].Buffers[0], txqueue[k].Addr); err == nil { + nbytes += n + xmitBuf.Put(txqueue[k].Buffers[0]) + } else { + s.socketError.Store(errors.WithStack(err)) + s.Close() return } } + atomic.AddUint64(&DefaultSnmp.OutPkts, uint64(len(txqueue))) + atomic.AddUint64(&DefaultSnmp.OutBytes, uint64(nbytes)) } diff --git a/tx_linux.go b/tx_linux.go index fd7270b..8bf6ce9 100644 --- a/tx_linux.go +++ b/tx_linux.go @@ -3,90 +3,30 @@ package kcp import ( - "net" "sync/atomic" "github.com/pkg/errors" "golang.org/x/net/ipv4" - "golang.org/x/net/ipv6" ) -func (s *UDPSession) txLoop() { - addr, _ := net.ResolveUDPAddr("udp", s.conn.LocalAddr().String()) - var conn batchConn - - if addr.IP.To4() != nil { - conn = ipv4.NewPacketConn(s.conn) - } else { - conn = ipv6.NewPacketConn(s.conn) - } - - for { - select { - case txqueue := <-s.chTxQueue: - if len(txqueue) > 0 { - nbytes := 0 - vec := txqueue - for len(vec) > 0 { - if n, err := conn.WriteBatch(vec, 0); err == nil { - vec = vec[n:] - } else { - s.socketError.Store(errors.WithStack(err)) - s.Close() - return - } - } - - for k := range txqueue { - nbytes += len(txqueue[k].Buffers[0]) - xmitBuf.Put(txqueue[k].Buffers[0]) - } - - atomic.AddUint64(&DefaultSnmp.OutPkts, uint64(len(txqueue))) - atomic.AddUint64(&DefaultSnmp.OutBytes, uint64(nbytes)) - } - case <-s.die: +func (s *UDPSession) tx(txqueue []ipv4.Message) { + nbytes := 0 + vec := txqueue + for len(vec) > 0 { + if n, err := s.bconn.WriteBatch(vec, 0); err == nil { + vec = vec[n:] + } else { + s.socketError.Store(errors.WithStack(err)) + s.Close() return } } -} -func (l *Listener) txLoop() { - addr, _ := net.ResolveUDPAddr("udp", l.conn.LocalAddr().String()) - var conn batchConn - - if addr.IP.To4() != nil { - conn = ipv4.NewPacketConn(l.conn) - } else { - conn = ipv6.NewPacketConn(l.conn) + for k := range txqueue { + nbytes += len(txqueue[k].Buffers[0]) + xmitBuf.Put(txqueue[k].Buffers[0]) } - for { - select { - case txqueue := <-l.chTxQueue: - if len(txqueue) > 0 { - nbytes := 0 - vec := txqueue - for len(vec) > 0 { - if n, err := conn.WriteBatch(vec, 0); err == nil { - vec = vec[n:] - } else { - l.socketError.Store(errors.WithStack(err)) - l.Close() - return - } - } - - for k := range txqueue { - nbytes += len(txqueue[k].Buffers[0]) - xmitBuf.Put(txqueue[k].Buffers[0]) - } - - atomic.AddUint64(&DefaultSnmp.OutPkts, uint64(len(txqueue))) - atomic.AddUint64(&DefaultSnmp.OutBytes, uint64(nbytes)) - } - case <-l.die: - return - } - } + atomic.AddUint64(&DefaultSnmp.OutPkts, uint64(len(txqueue))) + atomic.AddUint64(&DefaultSnmp.OutBytes, uint64(nbytes)) }