From 5b66b2f42d6199645c95b49348355f43b55c20b1 Mon Sep 17 00:00:00 2001 From: impact-eintr Date: Sun, 18 Dec 2022 11:17:06 +0800 Subject: [PATCH] =?UTF-8?q?emmm=20=E6=B3=A8=E5=86=8C=E5=BC=8F=E7=BD=91?= =?UTF-8?q?=E7=BB=9C=E6=A0=88=E4=B8=8D=E5=A4=AA=E5=A5=BD=E5=86=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/netstack/main.go | 82 ++++++++++++++++++++++++++++----- cmd/netstack/tcp_server.go | 59 ++++++++++++++++-------- tcpip/link/loopback/loopback.go | 10 ++++ tcpip/transport/tcp/README.md | 6 +++ 4 files changed, 126 insertions(+), 31 deletions(-) diff --git a/cmd/netstack/main.go b/cmd/netstack/main.go index 268fa88..9c26d3e 100644 --- a/cmd/netstack/main.go +++ b/cmd/netstack/main.go @@ -1,7 +1,10 @@ package main import ( + "encoding/binary" "flag" + "fmt" + "io" "log" "net" "netstack/logger" @@ -133,6 +136,10 @@ func main() { //logger.SetFlags(logger.TCP) go func() { // echo server + //time.Sleep(1 * time.Second) + //pid := Register() + //log.Fatal(pid) + listener := tcpListen(s, proto, addr, localPort) done <- struct{}{} for { @@ -143,7 +150,6 @@ func main() { log.Println("服务端 建立连接") go func() { - time.Sleep(3 * time.Second) for { time.Sleep(50 * time.Millisecond) buf := make([]byte, 1024) @@ -153,7 +159,7 @@ func main() { break } logger.NOTICE(string(buf[:n])) - conn.Write([]byte("Hello Client")) + //conn.Write([]byte("Hello Client")) } }() } @@ -176,26 +182,80 @@ func main() { log.Printf("\n\n客户端 写入数据") - for i := 0; i < 30; i++ { + for i := 0; i < 1; i++ { conn.Write([]byte("Hello Server!")) - buf := make([]byte, 1024) - n, err := conn.Read(buf) - if err != nil { - log.Println(err) - break - } - logger.NOTICE(string(buf[:n])) + //buf := make([]byte, 1024) + //n, err := conn.Read(buf) + //if err != nil { + // log.Println(err) + // break + //} + //logger.NOTICE(string(buf[:n])) time.Sleep(1 * time.Second) } - time.Sleep(500 * time.Minute) + select {} conn.Close() }() close(done) + l, err := net.Listen("tcp", "127.0.0.1:9999") + if err != nil { + fmt.Println("Error listening:", err) + os.Exit(1) + } + rcv := &RCV{ + Stack: s, + rcvBuf: make([]byte, 1<<20), + } + + TCPServer(l, rcv) + c := make(chan os.Signal) signal.Notify(c, os.Interrupt, os.Kill, syscall.SIGUSR1, syscall.SIGUSR2) <-c } + +const ( + REGISTER byte = iota + LISTEN + CONNECT + READ + WRITE + CLOSE +) + +// Register 从netstack获取pid +func Register() PID { + // 连接本地netstack服务 + conn, err := net.Dial("tcp", "127.0.0.1:9999") + if err != nil { + fmt.Println("err : ", err) + return 0 + } + defer conn.Close() + + _, err = conn.Write(make([]byte, 1<<20)) + buf, _ := io.ReadAll(conn) + + log.Fatal(buf) + return 3 +} + +// Listen 传递 pid addr port 监听+绑定地址 +func Listen(pid PID, addr tcpip.Address, localPort int) FD { + // 连接本地netstack服务 + conn, err := net.Dial("tcp", "127.0.0.1:9999") + if err != nil { + fmt.Println("err : ", err) + return 0 + } + buf := make([]byte, 1024) + binary.BigEndian.PutUint16(buf[:4], uint16(pid)) + conn.Write(buf) + buf, _ = io.ReadAll(conn) + log.Fatal(buf) + return 0 +} diff --git a/cmd/netstack/tcp_server.go b/cmd/netstack/tcp_server.go index 24f6148..2660bf8 100644 --- a/cmd/netstack/tcp_server.go +++ b/cmd/netstack/tcp_server.go @@ -6,22 +6,28 @@ import ( "io" "log" "net" + "netstack/logger" "netstack/tcpip" - "netstack/tcpip/header" "netstack/tcpip/stack" - "netstack/tcpip/transport/udp" - "netstack/waiter" "runtime" "strings" + "sync/atomic" ) +// PID netstack PID +type PID uint16 + +var currPID uint32 = 2 // 0 1 2 用过了 + +type FD uint16 + +var fds = make(map[PID][1024]FD, 8) + type TCPHandler interface { Handle(net.Conn) } func TCPServer(listener net.Listener, handler TCPHandler) error { - log.Printf("netstack 网络解析地址: %s", listener.Addr()) - for { clientConn, err := listener.Accept() if err != nil { @@ -54,31 +60,37 @@ type RCV struct { } func (r *RCV) Handle(conn net.Conn) { + logger.NOTICE("RCV handle") var err error r.rcvBuf, err = io.ReadAll(conn) - if err != nil && len(r.rcvBuf) < 9 { // proto + ip + port + if err != nil && len(r.rcvBuf) < 1 { // 操作码 panic(err) } - switch string(r.rcvBuf[:3]) { - case "udp": - var wq waiter.Queue - // 新建一个udp端 - ep, err := r.NewEndpoint(udp.ProtocolNumber, header.IPv4ProtocolNumber, &wq) - if err != nil { - log.Fatal(err) - } - r.ep = ep - r.Bind() - r.Connect() - r.Close() - case "tcp": + logger.NOTICE("注意测试") + + switch r.rcvBuf[0] { + case REGISTER: + conn.Write(r.Register()) + case LISTEN: + goto FAULT + case CONNECT: + goto FAULT + case READ: + goto FAULT + case WRITE: + goto FAULT + case CLOSE: + goto FAULT default: return } + +FAULT: + logger.NOTICE("FAULT") } -func (r *RCV) Bind() { +func (r *RCV) Listen() { if len(r.rcvBuf) < 9 { // udp ip port log.Println("Error: too few arg") return @@ -99,3 +111,10 @@ func (r *RCV) Connect() { func (r *RCV) Close() { r.ep.Close() } + +func (r *RCV) Register() []byte { + pid := uint16(atomic.AddUint32(&currPID, 1)) + b := make([]byte, 2) + binary.BigEndian.PutUint16(b[:2], pid) + return b +} diff --git a/tcpip/link/loopback/loopback.go b/tcpip/link/loopback/loopback.go index b898a54..bb9be52 100644 --- a/tcpip/link/loopback/loopback.go +++ b/tcpip/link/loopback/loopback.go @@ -1,12 +1,15 @@ package loopback import ( + "fmt" + "netstack/logger" "netstack/tcpip" "netstack/tcpip/buffer" "netstack/tcpip/stack" ) type endpoint struct { + count int dispatcher stack.NetworkDispatcher } @@ -43,6 +46,13 @@ func (e *endpoint) WritePacket(r *stack.Route, hdr buffer.Prependable, payload b views = append(views, payload.Views()...) vv := buffer.NewVectorisedView(len(views[0])+payload.Size(), views) + // TODO 这里整点活 在特定的情况下丢掉数据报 模拟网络阻塞 + + e.count++ + if e.count == 5 || e.count == 7 || e.count == 9 { + logger.NOTICE(fmt.Sprintf("统计 %d 丢掉这个报文", e.count)) + return nil + } // Because we're immediately turning around and writing the packet back to the // rx path, we intentionally don't preserve the remote and local link // addresses from the stack.Route we're passed. diff --git a/tcpip/transport/tcp/README.md b/tcpip/transport/tcp/README.md index 9f8a6ac..5da8d75 100644 --- a/tcpip/transport/tcp/README.md +++ b/tcpip/transport/tcp/README.md @@ -299,3 +299,9 @@ Silly Window Syndrome 翻译成中文就是“糊涂窗口综合症”。正如 1. 要等到 Window Size >= MSS 或是 Data Size >= MSS 2. 收到之前发送数据的 ack 回包，他才会发数据，否则就是在攒数据 + +> 上面是通过 keepalive 实现的 + +## tcp的拥塞控制 + +节介绍 tcp 的拥塞控制，拥塞控制控制是 tcp 协议中最复杂问题之一，主要是如何探测链路已经拥塞？探测到拥塞后如何处理？