optimize(orbyte): use manual destroy

gold/link/link.go

@@ -22,7 +22,7 @@ var (
 
 type LinkData struct {
 	H head.Packet
-	D pbuf.Bytes
+	D []byte
 }
 
 // Link 是本机到 peer 的连接抽象
@@ -74,7 +74,7 @@ func (l *Link) ToLower(header *head.Packet, data pbuf.Bytes) {
 	if l.pipe != nil {
 		l.pipe <- LinkData{
 			H: *header,
-			D: data,
+			D: data.Copy().Trans(),
 		}
 		if config.ShowDebugLog {
 			logrus.Debugln("[listen] deliver to pipe of", l.peerip)

gold/link/listen.go

@@ -64,6 +64,8 @@ func (m *Me) listen() (conn p2p.Conn, err error) {
 }
 
 func (m *Me) waitordispatch(addr p2p.EndPoint, buf pbuf.Bytes, n int) {
+	defer buf.ManualDestroy()
+
 	recvtotlcnt := atomic.AddUint64(&m.recvtotlcnt, uint64(buf.Len()))
 	recvloopcnt := atomic.AddUintptr(&m.recvloopcnt, 1)
 	recvlooptime := atomic.LoadInt64(&m.recvlooptime)
@@ -100,6 +102,7 @@ func (m *Me) waitordispatch(addr p2p.EndPoint, buf pbuf.Bytes, n int) {
 			}
 			m.dispatch(p, b, addr)
 		})
+		h.ManualDestroy()
 	})
 }
 
@@ -151,7 +154,9 @@ func (m *Me) dispatch(header *head.Packet, body []byte, addr p2p.EndPoint) {
 	data = data.SliceFrom(8)
 	if p.usezstd {
 		data.V(func(b []byte) {
+			old := data
 			data, err = algo.DecodeZstd(b) // skip hash
+			old.ManualDestroy()
 		})
 		if err != nil {
 			if config.ShowDebugLog {
@@ -169,4 +174,5 @@ func (m *Me) dispatch(header *head.Packet, body []byte, addr p2p.EndPoint) {
 		return
 	}
 	fn(header, p, data)
+	data.ManualDestroy()
 }

gold/link/me.go

@@ -312,8 +312,11 @@ func (m *Me) sendAllSameDst(packet []byte) (n int) {
 	pcp.V(func(b []byte) {
 		copy(b, packet)
 	})
-	go pcp.V(func(b []byte) {
-		lnk.WritePacket(head.ProtoData, b, lnk.me.ttl)
-	})
+	go func() {
+		pcp.V(func(b []byte) {
+			lnk.WritePacket(head.ProtoData, b, lnk.me.ttl)
+		})
+		pcp.ManualDestroy()
+	}()
 	return
 }

gold/link/recv.go

@@ -46,6 +46,7 @@ func (m *Me) wait(data []byte, addr p2p.EndPoint) (h head.PacketBytes) {
 			return
 		}
 		data = w.ToBytes().Copy().Trans()
+		w.Destroy()
 		if len(data) < bound {
 			bound = len(data)
 			endl = "."
@@ -142,6 +143,9 @@ func (m *Me) wait(data []byte, addr p2p.EndPoint) (h head.PacketBytes) {
 	})
 
 	if ok {
+		if !h.HasInit() {
+			header.ManualDestroy()
+		}
 		return
 	}
 

gold/link/send.go

@@ -57,11 +57,13 @@ func (l *Link) WritePacket(proto uint8, data []byte, ttl uint8) {
 		pktb = pb.Seal(l.keys[teatype], teatype, sndcnt&0x07ff)
 	}
 	bs := pktb.Split(int(mtu), false)
+	pktb.Destroy()
 	if config.ShowDebugLog {
 		logrus.Debugln("[send] split packet into", len(bs), "parts")
 	}
 	for _, b := range bs { //TODO: impl. nofrag
 		go l.write2peer(head.BuildPacketFromBytes(b), randseq(sndcnt))
+		b.ManualDestroy()
 	}
 }
 
@@ -69,6 +71,7 @@ func (l *Link) WritePacket(proto uint8, data []byte, ttl uint8) {
 //
 // 因为不保证可达所以不返回错误。
 func (l *Link) write2peer(b pbuf.Bytes, seq uint32) {
+	defer b.ManualDestroy()
 	if l.doublepacket {
 		err := l.write2peer1(b, seq)
 		if err != nil {
@@ -96,6 +99,7 @@ func (l *Link) write2peer1(b pbuf.Bytes, seq uint32) (err error) {
 	if conn == nil {
 		return io.ErrClosedPipe
 	}
+	isnewb := false
 	b.V(func(data []byte) {
 		if config.ShowDebugLog {
 			bound := 64
@@ -107,6 +111,7 @@ func (l *Link) write2peer1(b pbuf.Bytes, seq uint32) (err error) {
 			logrus.Debugln("[send] crc seq", fmt.Sprintf("%08x", seq), "raw data bytes", hex.EncodeToString(data[:bound]), endl)
 		}
 		b = l.me.xorenc(data, seq)
+		isnewb = true
 		if config.ShowDebugLog {
 			bound := 64
 			endl := "..."
@@ -121,7 +126,12 @@ func (l *Link) write2peer1(b pbuf.Bytes, seq uint32) (err error) {
 	})
 	if l.me.base14 {
 		b.V(func(data []byte) {
+			old := b
 			b = pbuf.ParseBytes(base14.Encode(data)...)
+			if isnewb {
+				old.ManualDestroy()
+			}
+			isnewb = true
 			if config.ShowDebugLog {
 				bound := 64
 				endl := "..."
@@ -141,5 +151,8 @@ func (l *Link) write2peer1(b pbuf.Bytes, seq uint32) (err error) {
 		}
 		_, err = conn.WriteToPeer(b, peerep)
 	})
+	if isnewb {
+		b.ManualDestroy()
+	}
 	return
 }