优化封装

gold/link/link.go

@@ -22,6 +22,8 @@ type Link struct {
 	// 以秒为单位，小于等于 0 不发送
 	keepalive int64
 	// 收到的包的队列
+	// 没有下层 nic 时
+	// 包会分发到此
 	pipe chan *head.Packet
 	// peer 的虚拟 ip
 	peerip net.IP

gold/link/listen.go

@@ -68,7 +68,7 @@ func (m *Me) listen() (conn *net.UDPConn, err error) {
 											p.pipe <- packet
 											logrus.Debugln("[link] deliver to pipe of", p.peerip)
 										} else {
-											m.pipe <- packet.Data
+											m.nic.Write(packet.Data)
 											logrus.Debugln("[link] deliver", len(packet.Data), "bytes data to pipe of me")
 										}
 									default:
@@ -103,12 +103,6 @@ func (m *Me) listen() (conn *net.UDPConn, err error) {
 	return
 }
 
-// Read 接收所有发送给本机的报文
-// 需要开启 nopipe
-func (m *Me) Read() []byte {
-	return <-m.pipe
-}
-
 // 从 conn 读取 sz 字节数据
 func readAll(conn *net.UDPConn, sz int) ([]byte, error) {
 	i := 0

gold/link/me.go

@@ -1,10 +1,16 @@
 package link
 
 import (
+	"encoding/binary"
+	"io"
 	"net"
+	"strconv"
 	"sync"
 
 	"github.com/fumiama/WireGold/gold/head"
+	"github.com/fumiama/WireGold/lower"
+	"github.com/fumiama/water/waterutil"
+	"github.com/sirupsen/logrus"
 )
 
 // Me 是本机的抽象
@@ -28,8 +34,8 @@ type Me struct {
 	connmapmu sync.RWMutex
 	// 本机监听的 endpoint
 	myconn *net.UDPConn
-	// 不分目的 link 的接收队列
-	pipe chan []byte
+	// 本机网卡
+	nic lower.NICIO
 	// 本机路由表
 	router *Router
 	// 本机未接收完全分片池
@@ -39,11 +45,10 @@ type Me struct {
 	clock map[*head.Packet]uint8
 	// 本机上层配置
 	srcport, dstport, mtu uint16
-	readptr               []byte
 }
 
 // NewMe 设置本机参数
-func NewMe(privateKey *[32]byte, myipwithmask string, myEndpoint string, nopipeinlink bool, srcport, dstport, mtu uint16) (m Me) {
+func NewMe(privateKey *[32]byte, myipwithmask string, myEndpoint string, nic lower.NICIO, srcport, dstport, mtu uint16) (m Me) {
 	m.privKey = *privateKey
 	var err error
 	m.myend, err = net.ResolveUDPAddr("udp", myEndpoint)
@@ -61,15 +66,13 @@ func NewMe(privateKey *[32]byte, myipwithmask string, myEndpoint string, nopipei
 		panic(err)
 	}
 	m.connections = make(map[string]*Link)
-	if nopipeinlink {
-		m.pipe = make(chan []byte, 32)
-	}
+	m.nic = nic
 	m.router = &Router{
 		list:  make([]*net.IPNet, 1, 16),
 		table: make(map[string]*Link, 16),
 	}
 	m.router.SetDefault(nil)
-	m.loop = m.AddPeer(m.me.String(), nil, "127.0.0.1:56789", []string{myipwithmask}, 0, false, nopipeinlink)
+	m.loop = m.AddPeer(m.me.String(), nil, "127.0.0.1:56789", []string{myipwithmask}, 0, false, nic != nil)
 	m.srcport = srcport
 	m.dstport = dstport
 	m.mtu = mtu & 0xfff8
@@ -88,3 +91,91 @@ func (m *Me) DstPort() uint16 {
 func (m *Me) MTU() uint16 {
 	return m.mtu
 }
+
+func (m *Me) ListenFromNIC() {
+	// 双缓冲区
+	buf := make([]byte, m.MTU()+68)  // 增加报头长度与 TEA 冗余
+	buf2 := make([]byte, m.MTU()+68) // 增加报头长度与 TEA 冗余
+
+	off := 0
+	isrev := false
+	for { // 从 NIC 发送
+		var packet []byte
+		if off > 0 && !isrev {
+			packet = buf2
+		} else {
+			packet = buf
+		}
+		n, err := m.nic.Read(packet[off:])
+		if isrev {
+			off = 0
+		}
+		if err != nil {
+			logrus.Errorln("[lower] send read from nic err:", err)
+			break
+		}
+		if n == 0 {
+			continue
+		}
+		packet = packet[:n]
+		n, rem := m.send(m.nic, packet)
+		for len(rem) > 20 && n > 0 {
+			n, rem = m.send(m.nic, rem)
+		}
+		if len(rem) > 0 {
+			logrus.Debugln("[lower] remain", len(rem), "bytes to send")
+			if off > 0 {
+				off = copy(buf, rem)
+				isrev = true
+			} else {
+				off = copy(buf2, rem)
+			}
+		} else {
+			off = 0
+		}
+	}
+}
+
+func (m *Me) send(nc io.Reader, packet []byte) (n int, rem []byte) {
+	if !waterutil.IsIPv4(packet) {
+		if waterutil.IsIPv6(packet) {
+			n = int(binary.BigEndian.Uint16(packet[4:6])) + 40
+			if n > len(packet) {
+				rem = packet
+				logrus.Warnln("[lower] skip to send", len(packet), "bytes ipv6 packet head")
+			} else {
+				rem = packet[n:]
+				logrus.Warnln("[lower] skip to send", n, "bytes ipv6 packet")
+			}
+			return
+		}
+		logrus.Warnln("[lower] skip to send", len(packet), "bytes non-ipv4/v6 packet")
+		return len(packet), nil
+	}
+	totl := waterutil.IPv4TotalLength(packet)
+	if int(totl) > len(packet) {
+		buf := make([]byte, int(totl))
+		copy(buf, packet)
+		cnt, err := m.nic.Read(buf[len(packet):])
+		if err != nil {
+			rem = packet
+			return
+		}
+		packet = buf[:cnt+len(packet)]
+	}
+	rem = packet[totl:]
+	packet = packet[:totl]
+	n = int(totl)
+	dst := waterutil.IPv4Destination(packet)
+	logrus.Debugln("[lower] sending", len(packet), "bytes packet from :"+strconv.Itoa(int(m.SrcPort())), "to", dst.String()+":"+strconv.Itoa(int(m.DstPort())))
+	lnk, err := m.Connect(dst.String())
+	if err != nil {
+		logrus.Warnln("[lower] connect to peer", dst.String(), "err:", err)
+		return
+	}
+	_, err = lnk.Write(head.NewPacket(head.ProtoData, m.SrcPort(), dst, m.DstPort(), packet), false)
+	if err != nil {
+		logrus.Warnln("[lower] write to peer", dst.String(), "err:", err)
+	}
+	return
+}