发送接收窗可视化Debug

2025-10-15 17:30:38 +08:00 · 2022-12-16 12:23:30 +08:00
parent c16d437fb8
commit 407f4c0b42
9 changed files with 125 additions and 29 deletions
--- a/cmd/netstack/main.go
+++ b/cmd/netstack/main.go
@@ -171,7 +171,7 @@ func main() {

 		time.Sleep(time.Second)
 		log.Printf("\n\n客户端 写入数据")
-		buf := make([]byte, 1<<21)
+		buf := make([]byte, 1<<17)
 		conn.Write(buf)
 		time.Sleep(1 * time.Minute)
 		conn.Close()
--- a/logger/logger.go
+++ b/logger/logger.go
@@ -1,7 +1,9 @@
 package logger

 import (
+	"fmt"
 	"log"
+	"strings"
 	"sync"
 )

@@ -56,14 +58,32 @@ func (l *logger) info(f func()) {
 	f()
 }

-func TODO(msg string) {
+func TODO(msg string, v ...string) {
 	GetInstance().info(func() {
-		log.Println("TODO: " + msg)
+		log.Printf("\033[1;37;41mTODO: %s\033[0m\n", msg+" "+strings.Join(v, " "))
 	})
 }

-func FIXME(msg string) {
+func FIXME(msg string, v ...string) {
 	GetInstance().info(func() {
-		log.Fatal("FIXME: " + msg)
+		log.Fatalf("\033[1;37;41mFIXME: %s\033[0m\n", msg+" "+strings.Join(v, " "))
 	})
 }
+
+func NOTICE(msg string, v ...string) {
+	GetInstance().info(func() {
+		log.Printf("\033[1;37;41mNOTICE: %s\033[0m\n", msg+" "+strings.Join(v, " "))
+	})
+}
+
+func COLORS() {
+	for b := 40; b <= 47; b++ { // 背景色彩 = 40-47
+		for f := 30; f <= 37; f++ { // 前景色彩 = 30-37
+			for d := range []int{0, 1, 4, 5, 7, 8} { // 显示方式 = 0,1,4,5,7,8
+				fmt.Printf(" %c[%d;%d;%dm%s(f=%d,b=%d,d=%d)%c[0m ", 0x1B, d, b, f, "", f, b, d, 0x1B)
+			}
+			fmt.Println("")
+		}
+		fmt.Println("")
+	}
+}
--- a/tcpip/transport/tcp/README.md
+++ b/tcpip/transport/tcp/README.md
@@ -264,6 +264,17 @@ K = 4

 ![img](https://doc.shiyanlou.com/document-uid949121labid10418timestamp1555574248321.png)

+发送窗口的上限值 = Min [rwnd, cwnd]，cwnd 拥塞窗口
+
+上图中分成了四个部分，分别是：（其中那个黑模型就是滑动窗口）
+
+1. 已收到 ack 确认的数据
+2. 已经发送，但还没收到 ack 的数据
+3. 在窗口中还没有发出的（接收方还有空间）
+4. 窗口以外的数据（接收方没空间）
+
+滑动： 当发送端收到数据 ack 确认时，窗口向右滑
+
 ## Zero Window

 如果一个处理缓慢的 Server（接收端）是怎么把 Client（发送端）的 TCP Sliding Window 给降成 0 的。此时，你一定会问，如果 Window 变成 0 了，TCP 会怎么样？是不是发送端就不发数据了？是的，发送端就不发数据了，你可以想像成“Window Closed”，那你一定还会问，如果发送端不发数据了，接收方一会儿 Window size 可用了，怎么通知发送端呢？
--- a/tcpip/transport/tcp/accept.go
+++ b/tcpip/transport/tcp/accept.go
@@ -212,6 +212,9 @@ func (l *listenContext) createConnectedEndpoint(s *segment, iss seqnum.Value,
 	// The receiver at least temporarily has a zero receive window scale,
 	// but the caller may change it (before starting the protocol loop).
 	n.snd = newSender(n, iss, irs, s.window, rcvdSynOpts.MSS, rcvdSynOpts.WS)
+	logger.GetInstance().Info(logger.HANDSHAKE, func() {
+		//log.Println("服务端握手成功 服务端的sender", n.snd)
+	})
 	n.rcv = newReceiver(n, irs, l.rcvWnd, 0)

 	return n, nil
@@ -251,7 +254,7 @@ func (l *listenContext) createEndpointAndPerformHandshake(s *segment, opts *head
 	// TODO 更新接收窗口扩张因子
 	ep.rcv.rcvWndScale = h.effectiveRcvWndScale()
 	logger.GetInstance().Info(logger.HANDSHAKE, func() {
-		log.Println("rp.rcv.rcvWndScale", ep.rcv.rcvWndScale)
+		log.Println("ep.rcv.rcvWndScale", ep.rcv.rcvWndScale)
 	})

 	return ep, nil
--- a/tcpip/transport/tcp/connect.go
+++ b/tcpip/transport/tcp/connect.go
@@ -247,6 +247,7 @@ func (h *handshake) synSentState(s *segment) *tcpip.Error {
 	h.flags |= flagAck
 	h.mss = rcvSynOpts.MSS
 	h.sndWndScale = rcvSynOpts.WS
+	logger.NOTICE(atoi(h.sndWnd), atoi(h.sndWndScale), atoi(h.rcvWnd))

 	// If this is a SYN ACK response, we only need to acknowledge the SYN
 	// and the handshake is completed.
@@ -255,6 +256,7 @@ func (h *handshake) synSentState(s *segment) *tcpip.Error {
 		// 客户端握手完成，发送 ack 报文给服务端
 		h.state = handshakeCompleted
 		// 最后依次 ack 报文丢了也没关系，因为后面一但发送任何数据包都是带ack的
+		// 这里要求对端缩减窗口
 		h.ep.sendRaw(buffer.VectorisedView{}, flagAck, h.iss+1, h.ackNum, h.rcvWnd>>h.effectiveRcvWndScale())
 		return nil
 	}
@@ -286,7 +288,6 @@ func (h *handshake) synSentState(s *segment) *tcpip.Error {
 // 正常情况下，会调用该函数来处理第三次 ack 报文
 func (h *handshake) synRcvdState(s *segment) *tcpip.Error {
 	if s.flagIsSet(flagRst) {
-		// TODO 需要根据窗口返回 等理解了窗口后再写
 		// RFC 793, page 37, states that in the SYN-RCVD state, a reset
 		// is acceptable if the sequence number is in the window.
 		if s.sequenceNumber.InWindow(h.ackNum, h.rcvWnd) {
@@ -302,13 +303,41 @@ func (h *handshake) synRcvdState(s *segment) *tcpip.Error {
 	// 如果是syn报文，且序列号对应不上，那么返回 rst
 	if s.flagIsSet(flagSyn) && s.sequenceNumber != h.ackNum-1 {
 		// TODO 返回RST报文
+		// We received two SYN segments with different sequence
+		// numbers, so we reset this and restart the whole
+		// process, except that we don't reset the timer.
+		ack := s.sequenceNumber.Add(s.logicalLen())
+		seq := seqnum.Value(0)
+		if s.flagIsSet(flagAck) {
+			seq = s.ackNumber
+		}
+		h.ep.sendRaw(buffer.VectorisedView{}, flagRst|flagAck, seq, ack, 0)
+
+		if !h.active {
+			return tcpip.ErrInvalidEndpointState
+		}
+
+		if err := h.resetState(); err != nil {
+			return err
+		}
+		synOpts := header.TCPSynOptions{
+			WS:            h.rcvWndScale,
+			TS:            h.ep.sendTSOk,
+			TSVal:         h.ep.timestamp(),
+			TSEcr:         h.ep.recentTS,
+			SACKPermitted: h.ep.sackPermitted,
+		}
+		sendSynTCP(&s.route, h.ep.id, h.flags, h.iss, h.ackNum, h.rcvWnd, synOpts)
 		return nil
 	}

 	// 如果时ack报文 表示三次握手已经完成
 	if s.flagIsSet(flagAck) {
 		log.Println("TCP STATE ESTABLISHED")
-		// TODO 修改时间戳
+		if h.ep.sendTSOk && !s.parsedOptions.TS {
+			h.ep.stack.Stats().DroppedPackets.Increment()
+			return nil
+		}
 		h.state = handshakeCompleted
 		return nil
 	}
@@ -320,7 +349,10 @@ func (h *handshake) synRcvdState(s *segment) *tcpip.Error {
 func (h *handshake) handleSegment(s *segment) *tcpip.Error {
 	h.sndWnd = s.window
 	if !s.flagIsSet(flagSyn) && h.sndWndScale > 0 {
-		h.sndWnd <<= uint8(h.sndWndScale) // 收紧窗口
+		h.sndWnd <<= uint8(h.sndWndScale) // 收紧发送窗口
+		logger.NOTICE("扩张发送窗口到", atoi(h.sndWnd))
+	} else {
+		logger.NOTICE("原有发送窗口与服务端的接收窗口大小相同", atoi(h.sndWnd))
 	}

 	switch h.state {
@@ -543,6 +575,7 @@ func sendTCP(r *stack.Route, id stack.TransportEndpointID, data buffer.Vectorise

 	if rcvWnd > 0xffff { // 65535
 		rcvWnd = 0xffff
+		logger.NOTICE("告诉对端 我的接收窗口为", atoi(rcvWnd))
 	}

 	// Initialize the header.
@@ -576,7 +609,9 @@ func sendTCP(r *stack.Route, id stack.TransportEndpointID, data buffer.Vectorise
 		r.Stats().TCP.ResetsSent.Increment()
 	}

-	log.Printf("TCP 发送 [%s] 报文片段到 %s, seq: |%d|, ack: %d, rcvWnd: %d",
+	logger.GetInstance().Info(logger.TCP, func() {
+	})
+	log.Printf("TCP 发送 [%s] 报文片段到 %s, seq: %d, ack: %d, 可接收rcvWnd: %d",
 		flagString(flags), fmt.Sprintf("%s:%d", id.RemoteAddress, id.RemotePort),
 		seq, ack, rcvWnd)

@@ -703,6 +738,7 @@ func (e *endpoint) handleSegments() *tcpip.Error {
 			// Patch the window size in the segment according to the
 			// send window scale.
 			s.window <<= e.snd.sndWndScale
+			logger.NOTICE("这里进行了发送窗口的扩张", atoi(s.window))
 			// If the timestamp option is negotiated and the segment
 			// does not carry a timestamp option then the segment
 			// must be dropped as per
@@ -781,6 +817,9 @@ func (e *endpoint) protocolMainLoop(handshake bool) *tcpip.Error {

 		// 到这里就表示三次握手已经成功了，那么初始化发送器和接收器
 		e.snd = newSender(e, h.iss, h.ackNum-1, h.sndWnd, h.mss, h.sndWndScale)
+		logger.GetInstance().Info(logger.HANDSHAKE, func() {
+			//log.Println("客户端握手成功 客户端的sender", e.snd)
+		})

 		e.rcvListMu.Lock()
 		e.rcv = newReceiver(e, h.ackNum-1, h.rcvWnd, h.effectiveRcvWndScale())
--- a/tcpip/transport/tcp/endpoint.go
+++ b/tcpip/transport/tcp/endpoint.go
@@ -194,9 +194,6 @@ func newEndpoint(stack *stack.Stack, netProto tcpip.NetworkProtocolNumber, waite
 		e.cc = cs
 	}

-	log.Println(e.sndBufSize, e.rcvBufSize, e.cc)
-
-	// TODO 需要添加
 	e.segmentQueue.setLimit(2 * e.rcvBufSize)
 	e.workMu.Init()
 	e.workMu.Lock()
@@ -800,9 +797,11 @@ func (e *endpoint) HandlePacket(r *stack.Route, id stack.TransportEndpointID, vv
 		if _, err := e.GetRemoteAddress(); err != nil {
 			prefix = "监听者"
 		}
-		log.Printf(prefix+"收到 tcp [%s] 报文片段 from %s, seq: %d, ack: |%d|",
-			flagString(s.flags), fmt.Sprintf("%s:%d", s.id.RemoteAddress, s.id.RemotePort),
-			s.sequenceNumber, s.ackNumber)
+		logger.GetInstance().Info(logger.TCP, func() {
+			log.Printf(prefix+"收到 tcp [%s] 报文片段 from %s, seq: %d, ack: |%d|",
+				flagString(s.flags), fmt.Sprintf("%s:%d", s.id.RemoteAddress, s.id.RemotePort),
+				s.sequenceNumber, s.ackNumber)
+		})

 		// 对于 端口监听者 listener 而言这里唤醒的是 protocolListenLoop
 		// 对于普通tcp连接 conn 而言这里唤醒的是 protocolMainLoop
@@ -827,7 +826,7 @@ func (e *endpoint) updateSndBufferUsage(v int) {
 	e.sndBufMu.Unlock()
 	if notify { // 如果缓存中剩余的数据过多是不需要补充的
 		e.waiterQueue.Notify(waiter.EventOut)
-		//log.Println("提醒 用户层的 Write() 继续写入")
+		log.Println("提醒 用户层的 Write() 继续写入")
 	}
 }

--- a/tcpip/transport/tcp/rcv.go
+++ b/tcpip/transport/tcp/rcv.go
@@ -2,6 +2,7 @@ package tcp

 import (
 	"container/heap"
+	"fmt"
 	"log"
 	"netstack/logger"
 	"netstack/tcpip/seqnum"
@@ -59,7 +60,7 @@ func (r *receiver) getSendParams() (rcvNxt seqnum.Value, rcvWnd seqnum.Size) {
 		r.rcvAcc = acc
 	}

-	log.Println("-------------", n, acc, r.rcvNxt.Size(r.rcvAcc)>>r.rcvWndScale)
+	//log.Println("-------------", n, acc, r.rcvNxt.Size(r.rcvAcc)>>r.rcvWndScale)
 	return r.rcvNxt, r.rcvNxt.Size(r.rcvAcc) >> r.rcvWndScale
 }

@@ -90,6 +91,7 @@ func (r *receiver) consumeSegment(s *segment, segSeq seqnum.Value, segLen seqnum
 	r.rcvNxt = segSeq.Add(segLen)
 	logger.GetInstance().Info(logger.TCP, func() {
 	})
+	log.Println(r)

 	// 修剪SACK块以删除任何涵盖已消耗序列号的SACK信息。
 	TrimSACKBlockList(&r.ep.sack, r.rcvNxt)
@@ -132,6 +134,7 @@ func (r *receiver) handleRcvdSegment(s *segment) {

 	// tcp流量控制：判断该数据段的序列号是否在接收窗口内，如果不在，立即返回ack给对端。
 	if !r.acceptable(segSeq, segLen) {
+		log.Fatal("发了太多")
 		r.ep.snd.sendAck()
 		return
 	}
@@ -180,3 +183,21 @@ func (r *receiver) handleRcvdSegment(s *segment) {
 |                                            |
 LastByteRead                          LastByteRecv
 */
+
+var fmtRecver string = `%s
+                                             +------>% 10s <-----+
+                                             |                        |
+-----------------+-------------+-------------+------------------------+
+| ANR %10s | not revived |  rcvd unack |   able rcv             |
+-----------------+-------------+-------------+------------------------+
+^                                            ^                        ^
+|                                            |                        |
+% 10s                              % 10s               % 10s`
+
+func (r receiver) String() string {
+	return fmt.Sprintf(fmtRecver,
+		atoi(r.ep.id.LocalPort),
+		atoi(r.rcvNxt.Size(r.rcvAcc)),
+		atoi(r.ep.rcvBufUsed),
+		atoi(r.rcvNxt-seqnum.Value(r.ep.rcvBufUsed)), atoi(r.rcvNxt), atoi(r.rcvAcc))
+}
--- a/tcpip/transport/tcp/segment.go
+++ b/tcpip/transport/tcp/segment.go
@@ -62,7 +62,7 @@ type segment struct {
 	sequenceNumber seqnum.Value // tcp序号 第一个字节在整个报文的位置
 	ackNumber      seqnum.Value // 确认号 希望继续获取的下一个字节序号
 	flags          uint8
-	window         seqnum.Size
+	window         seqnum.Size // NOTE 这里是本地的接收窗口大小 不是发送窗口
 	// parsedOptions stores the parsed values from the options in the segment.
 	parsedOptions header.TCPOptions
 	options       []byte
--- a/tcpip/transport/tcp/snd.go
+++ b/tcpip/transport/tcp/snd.go
@@ -181,7 +181,7 @@ type fastRecovery struct {
 func newSender(ep *endpoint, iss, irs seqnum.Value, sndWnd seqnum.Size, mss uint16, sndWndScale int) *sender {
 	s := &sender{
 		ep:             ep,
-		sndCwnd:        InitialCwnd,
+		sndCwnd:        InitialCwnd, // TODO 暂时写死 tcp拥塞窗口 决定了发送窗口的初始大小
 		sndWnd:         sndWnd,
 		sndUna:         iss + 1,
 		sndNxt:         iss + 1, // 缓存长度为0
@@ -402,9 +402,9 @@ func (s *sender) handleRcvdSegment(seg *segment) {

 	// TODO tcp拥塞控制
 	if s.writeList.Front() != nil {
-		log.Println(s)
 		//log.Fatal(s.sndNxt, " 确认成功 继续发送 ", seg.sequenceNumber)
 	}
+	log.Println(s)

 	s.sendData()
 }
@@ -462,8 +462,9 @@ func (s *sender) sendData() {

 			s.outstanding++
 			segEnd = seg.sequenceNumber.Add(seqnum.Size(seg.data.Size()))
-			log.Println("发送窗口一开始是", s.sndWnd,
+			log.Println("发送窗口是", s.sndWnd,
 				"最多发送数据", available,
+				"缓存数据头", seg.sequenceNumber,
 				"缓存数据尾", segEnd,
 				"发送端缓存包数量", s.outstanding)
 		}
@@ -497,9 +498,9 @@ func (s *sender) sendData() {

 }

-var fmtSender string = `
-                 +----->  % 10s  <------+
-                 |                           |
+var fmtSender string = `%s
+    	         +----->  % 10s  <------+
+                 |    Scale  % 4s            |
 -----------------+-------------+-------------+------------------
 |      已确认    |UAC% 10s|NXT% 10s|   不可发送
 -----------------+-------------+-------------+------------------
@@ -508,11 +509,13 @@ var fmtSender string = `
             % 10s    % 10s`

 func (s sender) String() string {
-	return fmt.Sprintf(fmtSender, atoi(uint32(s.sndWnd)),
-		atoi(uint32(s.sndNxt-s.sndUna)), atoi(s.ep.sndBufUsed),
-		atoi(uint32(s.sndUna)), atoi(uint32(s.sndNxt)))
+	return fmt.Sprintf(fmtSender,
+		atoi(s.ep.id.LocalPort),
+		atoi(s.sndWnd), atoi(s.sndWndScale),
+		atoi(s.sndNxt-s.sndUna), atoi(s.ep.sndBufSize-s.ep.sndBufUsed),
+		atoi(s.sndUna), atoi(s.sndNxt))
 }

-func atoi[T int | int8 | int16 | int32 | int64 | uint | uint8 | uint16 | uint32](i T) string {
+func atoi[T int | int8 | int16 | int32 | int64 | uint | uint8 | uint16 | uint32 | seqnum.Size | seqnum.Value](i T) string {
 	return fmt.Sprintf("%d", i)
 }