add txLoop to listener

sess.go

@@ -128,6 +128,7 @@ 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)
 
 	// FEC codec initialization
 	sess.fecDecoder = newFECDecoder(rxFECMulti*(dataShards+parityShards), dataShards, parityShards)
@@ -158,15 +159,12 @@ 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)
 	}
 
-	// a corked txLoop
-	sess.chTxQueue = make(chan []ipv4.Message)
-	go sess.txLoop()
-
 	currestab := atomic.AddUint64(&DefaultSnmp.CurrEstab, 1)
 	maxconn := atomic.LoadUint64(&DefaultSnmp.MaxConn)
 	if currestab > maxconn {
@@ -323,9 +321,16 @@ func (s *UDPSession) uncork() {
 	s.mu.Unlock()
 
 	if len(txqueue) > 0 {
-		select {
-		case s.chTxQueue <- txqueue:
-		case <-s.die:
+		if s.l != nil {
+			select {
+			case s.l.chTxQueue <- txqueue:
+			case <-s.die:
+			}
+		} else {
+			select {
+			case s.chTxQueue <- txqueue:
+			case <-s.die:
+			}
 		}
 	}
 }
@@ -698,6 +703,7 @@ type (
 		die             chan struct{}    // notify the listener has closed
 		rd              atomic.Value     // read deadline for Accept()
 		wd              atomic.Value
+		chTxQueue       chan []ipv4.Message
 	}
 )
 
@@ -873,6 +879,7 @@ 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 {
@@ -883,6 +890,7 @@ func ServeConn(block BlockCrypt, dataShards, parityShards int, conn net.PacketCo
 	}
 
 	go l.monitor()
+	go l.txLoop()
 	return l, nil
 }
 

tx_generic.go

@@ -27,3 +27,22 @@ func (s *UDPSession) txLoop() {
 		}
 	}
 }
+
+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])
+			}
+			atomic.AddUint64(&DefaultSnmp.OutPkts, uint64(len(txqueue)))
+			atomic.AddUint64(&DefaultSnmp.OutBytes, uint64(nbytes))
+		case <-l.die:
+			return
+		}
+	}
+}

tx_linux.go

@@ -48,3 +48,41 @@ func (s *UDPSession) txLoop() {
 		}
 	}
 }
+
+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 {
+		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 {
+						break
+					}
+				}
+
+				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
+		}
+	}
+}