优化payload

server/handler/link_cstp.go

@@ -20,6 +20,7 @@ func LinkCstp(conn net.Conn, cSess *sessdata.ConnSession) {
 		err     error
 		n       int
 		dataLen uint16
+		data    []byte
 		dead    = time.Duration(cSess.CstpDpd+5) * time.Second
 	)
 
@@ -34,9 +35,9 @@ func LinkCstp(conn net.Conn, cSess *sessdata.ConnSession) {
 			return
 		}
 		// hdata := make([]byte, BufferSize)
-		hb := getByteFull()
+		pl := getPayload()
-		hdata := *hb
+		data = *pl.Data
-		n, err = conn.Read(hdata)
+		n, err = conn.Read(data)
 		if err != nil {
 			base.Error("read hdata: ", err)
 			return
@@ -48,7 +49,7 @@ func LinkCstp(conn net.Conn, cSess *sessdata.ConnSession) {
 			base.Error(err)
 		}
 
-		switch hdata[6] {
+		switch data[6] {
 		case 0x07: // KEEPALIVE
 			// do nothing
 			// base.Debug("recv keepalive", cSess.IpAddr)
@@ -57,19 +58,20 @@ func LinkCstp(conn net.Conn, cSess *sessdata.ConnSession) {
 			return
 		case 0x03: // DPD-REQ
 			// base.Debug("recv DPD-REQ", cSess.IpAddr)
-			if payloadOutCstp(cSess, sessdata.LTypeIPData, 0x04, nil) {
+			pl.PType = 0x04
+			if payloadOutCstp(cSess, pl) {
 				return
 			}
 		case 0x04:
 			// log.Println("recv DPD-RESP")
 		case 0x00: // DATA
-			dataLen = binary.BigEndian.Uint16(hdata[4:6]) // 4,5
+			dataLen = binary.BigEndian.Uint16(data[4:6]) // 4,5
-			if payloadIn(cSess, sessdata.LTypeIPData, 0x00, hdata[8:8+dataLen]) {
+			copy(data, data[8:8+dataLen])
+			*pl.Data = data[:dataLen]
+			if payloadIn(cSess, pl) {
 				return
 			}
 		}
-
-		putByte(hb)
 	}
 }
 
@@ -83,38 +85,40 @@ func cstpWrite(conn net.Conn, cSess *sessdata.ConnSession) {
 	var (
 		err  error
 		n    int
-		// header  []byte
+		data []byte
-		payload *sessdata.Payload
+		pl   *sessdata.Payload
 	)
 
 	for {
 		select {
-		case payload = <-cSess.PayloadOutCstp:
+		case pl = <-cSess.PayloadOutCstp:
 		case <-cSess.CloseChan:
 			return
 		}
 
-		if payload.LType != sessdata.LTypeIPData {
+		if pl.LType != sessdata.LTypeIPData {
 			continue
 		}
 
-		h := []byte{'S', 'T', 'F', 0x01, 0x00, 0x00, payload.PType, 0x00}
+		data = *pl.Data
-		hb := getByteZero()
+		if pl.PType == 0x00 {
-		header := *hb
+			l := len(data)
-		header = append(header, h...)
+			data = data[:l+8]
-		if payload.PType == 0x00 {
+			copy(data[8:], data)
-			data := *payload.Data
+			copy(data[:8], plHeader)
-			binary.BigEndian.PutUint16(header[4:6], uint16(len(data)))
+			binary.BigEndian.PutUint16(data[4:6], uint16(l))
-			header = append(header, data...)
+		} else {
+			data = append(data[:0], plHeader...)
+			data[6] = pl.PType
 		}
-		n, err = conn.Write(header)
+		*pl.Data = data
+		n, err = conn.Write(*pl.Data)
 		if err != nil {
 			base.Error("write err", err)
 			return
 		}
 
-		putByte(hb)
+		putPayload(pl)
-		putPayload(payload)
 
 		// 限流设置
 		err = cSess.RateLimit(n, false)

server/handler/link_tun.go

@@ -70,23 +70,23 @@ func tunWrite(ifce *water.Interface, cSess *sessdata.ConnSession) {
 
 	var (
 		err error
-		payload *sessdata.Payload
+		pl  *sessdata.Payload
 	)
 
 	for {
 		select {
-		case payload = <-cSess.PayloadIn:
+		case pl = <-cSess.PayloadIn:
 		case <-cSess.CloseChan:
 			return
 		}
 
-		_, err = ifce.Write(*payload.Data)
+		_, err = ifce.Write(*pl.Data)
 		if err != nil {
 			base.Error("tun Write err", err)
 			return
 		}
 
-		putPayload(payload)
+		putPayload(pl)
 	}
 }
 
@@ -102,14 +102,15 @@ func tunRead(ifce *water.Interface, cSess *sessdata.ConnSession) {
 
 	for {
 		// data := make([]byte, BufferSize)
-		hb := getByteFull()
+		pl := getPayload()
-		data := *hb
+		n, err = ifce.Read(*pl.Data)
-		n, err = ifce.Read(data)
 		if err != nil {
 			base.Error("tun Read err", n, err)
 			return
 		}
 
+		*pl.Data = (*pl.Data)[:n]
+
 		// data = data[:n]
 		// ip_src := waterutil.IPv4Source(data)
 		// ip_dst := waterutil.IPv4Destination(data)
@@ -118,10 +119,8 @@ func tunRead(ifce *water.Interface, cSess *sessdata.ConnSession) {
 		// packet := gopacket.NewPacket(data, layers.LayerTypeIPv4, gopacket.Default)
 		// fmt.Println("read:", packet)
 
-		if payloadOut(cSess, sessdata.LTypeIPData, 0x00, data[:n]) {
+		if payloadOut(cSess, pl) {
 			return
 		}
-
-		putByte(hb)
 	}
 }

server/handler/payload.go

@@ -6,18 +6,9 @@ import (
 	"github.com/songgao/water/waterutil"
 )
 
-func payloadIn(cSess *sessdata.ConnSession, lType sessdata.LType, pType byte, data []byte) bool {
+func payloadIn(cSess *sessdata.ConnSession, pl *sessdata.Payload) bool {
-	pl := getPayload()
-	pl.LType = lType
-	pl.PType = pType
-	*pl.Data = append(*pl.Data, data...)
-
-	return payloadInData(cSess, pl)
-}
-
-func payloadInData(cSess *sessdata.ConnSession, payload *sessdata.Payload) bool {
 	// 进行Acl规则判断
-	check := checkLinkAcl(cSess.Group, payload)
+	check := checkLinkAcl(cSess.Group, pl)
 	if !check {
 		// 校验不通过直接丢弃
 		return false
@@ -25,7 +16,7 @@ func payloadInData(cSess *sessdata.ConnSession, payload *sessdata.Payload) bool
 
 	closed := false
 	select {
-	case cSess.PayloadIn <- payload:
+	case cSess.PayloadIn <- pl:
 	case <-cSess.CloseChan:
 		closed = true
 	}
@@ -33,21 +24,16 @@ func payloadInData(cSess *sessdata.ConnSession, payload *sessdata.Payload) bool
 	return closed
 }
 
-func payloadOut(cSess *sessdata.ConnSession, lType sessdata.LType, pType byte, data []byte) bool {
+func payloadOut(cSess *sessdata.ConnSession, pl *sessdata.Payload) bool {
 	dSess := cSess.GetDtlsSession()
 	if dSess == nil {
-		return payloadOutCstp(cSess, lType, pType, data)
+		return payloadOutCstp(cSess, pl)
 	} else {
-		return payloadOutDtls(cSess, dSess, lType, pType, data)
+		return payloadOutDtls(cSess, dSess, pl)
 	}
 }
 
-func payloadOutCstp(cSess *sessdata.ConnSession, lType sessdata.LType, pType byte, data []byte) bool {
+func payloadOutCstp(cSess *sessdata.ConnSession, pl *sessdata.Payload) bool {
-	pl := getPayload()
-	pl.LType = lType
-	pl.PType = pType
-	*pl.Data = append(*pl.Data, data...)
-
 	closed := false
 
 	select {
@@ -59,12 +45,7 @@ func payloadOutCstp(cSess *sessdata.ConnSession, lType sessdata.LType, pType byt
 	return closed
 }
 
-func payloadOutDtls(cSess *sessdata.ConnSession, dSess *sessdata.DtlsSession, lType sessdata.LType, pType byte, data []byte) bool {
+func payloadOutDtls(cSess *sessdata.ConnSession, dSess *sessdata.DtlsSession, pl *sessdata.Payload) bool {
-	pl := getPayload()
-	pl.LType = lType
-	pl.PType = pType
-	*pl.Data = append(*pl.Data, data...)
-
 	select {
 	case cSess.PayloadOutDtls <- pl:
 	case <-dSess.CloseChan:

server/handler/pool.go

@@ -3,13 +3,20 @@ package handler
 import (
 	"sync"
 
+	"github.com/bjdgyc/anylink/base"
 	"github.com/bjdgyc/anylink/sessdata"
 )
 
-var plPool = sync.Pool{
+// 不允许直接修改
+// [6] => PType
+var plHeader = []byte{'S', 'T', 'F', 0x01, 0x00, 0x00, 0x00, 0x00}
+
+var plPool = &sync.Pool{
 	New: func() interface{} {
-		b := make([]byte, 0, BufferSize)
+		b := make([]byte, BufferSize)
 		pl := sessdata.Payload{
+			LType: sessdata.LTypeIPData,
+			PType: 0x00,
 			Data:  &b,
 		}
 		// fmt.Println("plPool-init", len(pl.Data), cap(pl.Data))
@@ -23,13 +30,19 @@ func getPayload() *sessdata.Payload {
 }
 
 func putPayload(pl *sessdata.Payload) {
-	pl.LType = 0
+	// 错误数据丢弃
-	pl.PType = 0
+	if cap(*pl.Data) != BufferSize {
-	*pl.Data = (*pl.Data)[:0]
+		base.Warn("payload cap is err", cap(*pl.Data))
+		return
+	}
+
+	pl.LType = sessdata.LTypeIPData
+	pl.PType = 0x00
+	*pl.Data = (*pl.Data)[:BufferSize]
 	plPool.Put(pl)
 }
 
-var bytePool = sync.Pool{
+var bytePool = &sync.Pool{
 	New: func() interface{} {
 		b := make([]byte, 0, BufferSize)
 		// fmt.Println("bytePool-init")

server/sessdata/protocol.go

@@ -8,8 +8,8 @@ const (
 )
 
 type Payload struct {
-	PType byte  // payload types
 	LType LType // LinkType
+	PType byte  // payload types
 	Data  *[]byte
 }