Question

传输控制协议（TCP）是一种面向连接、可靠、基于字节流（stream）的传输层通信协议。

• 包编号，接收端确认（ACK），需要时重传。
• 数据校验，确保传输过程中不会出错。

相对完整的示例，包含 context、reuseport 等。

// server.go

package main

import (
	"context"
	"net"
	"sync"
	"time"
)

const (
	network  = "tcp"
	address  = ":8080"
	packsize = 8

	dialTimeout   = time.Second
	acceptTimeout = time.Millisecond * 10
	connTimeout   = time.Second * 20
)

func server(ctx context.Context) <-chan struct{} {
	ready, shutdown := make(chan struct{}), make(chan struct{})

	go func() {
		defer func() { recover() }()
		defer close(shutdown)

		// listen
		// addr := checkFatal(net.ResolveTCPAddr(network, address))
		// srv := checkFatal(net.ListenTCP(network, addr))
		srv := checkFatal(listenx(network, address))  // reuseport
		defer srv.Close()

		close(ready)

		// wait for all clients to end.
		var wg sync.WaitGroup
		defer wg.Wait()

		for {
			select{
			case <- ctx.Done(): return
			default:
			}

			// accept
			setDeadline(srv, acceptTimeout)
			conn, err := srv.AcceptTCP()

			if err != nil {
				if isTimeout(err) { continue }
				logFatal(err)
			}

			// client
			wg.Add(1)
			go func() {
				defer func() { recover() }()
				defer wg.Done()
				defer conn.Close()

				defer logPrint("closed:", conn.RemoteAddr())
				logPrint("connect:", conn.RemoteAddr())

				handle(ctx, conn)
			}()
		}
	}()

	select {
	case <- ready:
	case <- time.After(time.Second * 2):
	}

	return shutdown
}

func handle(ctx context.Context, conn *net.TCPConn) {
	pack := make([]byte, packsize)
	for {
		select{
		case <- ctx.Done(): return
		default:
		}

		if data := connRead(conn, pack); len(data) > 0 {
			connWrite(conn, data)
		}
	}
}

// client.go

package main

import (
	"fmt"
	"math/rand"
	"net"
	"time"
)

func client() {
	defer func(){ recover() }()

	conn := checkFatal(net.DialTimeout(network, address, dialTimeout))
	defer conn.Close()

	testdata(conn)
}

func testdata(conn net.Conn) {
	pack := make([]byte, packsize)
	for {
		// send
		rand.Read(pack)
		_ = connWrite(conn, pack)

		// recv
		b := connRead(conn, pack)
		if len(b) > 0 {
			fmt.Printf("  > % x  %v\n", b, conn.LocalAddr())
		}

		time.Sleep(time.Millisecond * 1000)
	}	
}

// util.go

package main

import (
	"errors"
	"io"
	"log"
	"net"
	// "runtime"
	"time"
	"syscall"
)

func init() {
	log.SetFlags(log.Ltime)
}

// --- log ------------------------------------

func logPrint(v ...any) {
	log.Println(v...)
}

func logFatal(v ...any) {
	logPrint(v...)
	// runtime.Goexit()
	panic("fatal")
}

func checkFatal[T any](v T, err error) T {
	if err != nil {
		logFatal(err)
	}

	return v
}

// --- conn -----------------------------------

func setDeadline(c any, d time.Duration) {
	t := time.Now().Add(d)

	switch v := c.(type) {
	case *net.TCPListener: v.SetDeadline(t)
	case net.Conn: v.SetDeadline(t)
	}
}

func isTimeout(err error) bool {
	e, ok := err.(net.Error)
	return ok && e.Timeout()
}

func connError(conn net.Conn, err error) {
	if err == nil {
		return 
	}

	// goexit: main gorotine !!!
	// panic!

	// ECONNRESET: connection reset by peer
	//      EPIPE: broken pipe

	switch {
	case isTimeout(err), 
		err == io.EOF, 
		errors.Is(err, syscall.ECONNRESET), 
		errors.Is(err, syscall.EPIPE): {    
			// runtime.Goexit()
			panic(err)
		}
	}

	logPrint(err)
}

func connRead(conn net.Conn, buf []byte) []byte {
	setDeadline(conn, connTimeout)
	n, err := conn.Read(buf)

	if err != nil {
		connError(conn, err)
	}

	return buf[:n]
}

func connWrite(conn net.Conn, b []byte) int {
	setDeadline(conn, connTimeout)
	n, err := conn.Write(b)

	if err != nil {
		connError(conn, err)
	}

	return n
}

// reuseport.go

package main

import (
	"context"
	"net"
	"syscall"
)

func listenx(network, address string) (*net.TCPListener, error) {
	const (
		SOL_SOCKET   = syscall.SOL_SOCKET
		SO_REUSEADDR = syscall.SO_REUSEADDR
		SO_REUSEPORT = 0xf
	)

	ctl := func(_, _ string, c syscall.RawConn) error {
		var err error

		c.Control(func(fd uintptr) {
			for _, v := range []int{ SO_REUSEADDR, SO_REUSEPORT } {
				err = syscall.SetsockoptInt(int(fd), SOL_SOCKET, v, 1)
				if err != nil { return }
			}
		})

		return err
	}

	// ------------------------------------

	lc := net.ListenConfig { Control: ctl }
	ln, err := lc.Listen(context.Background(), network, address)

    if err != nil {
    	return nil, err
    }

    return ln.(*net.TCPListener), nil
}

// main.go

package main

import (
	"context"
	"time"
)

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	done := server(ctx)

	go func() {
		time.Sleep(time.Second * 60)
		cancel()
	}()

	for i := 0; i < 10; i++ {
		go client()
	}

	<-done
}

参数

最受关注的莫过于 REUSEADDR 和 REUSEPORT 了。

SO_REUSEADDR just says that you can reuse local addresses.

This socket option tells the kernel that even if this port 
is busy (in the TIME_WAIT state), go ahead and reuse it anyway.  

If it is busy, but with another state, you will still get 
an address already in use error.  

It is useful if your server has been shut down, and then 
restarted right away while sockets are still active on i
ts port.  You should be aware that if any unexpected data 
comes in, it may confuse your server, but while this is 
possible, it is not likely.

One of the features merged in the 3.9 development cycle 
was TCP and UDP support for the SO_REUSEPORT socket option; 

The new socket option allows multiple sockets on the same 
host to bind to the same port, and is intended to improve 
the performance of multithreaded network server applications 
running on top of multicore systems.

缓冲

大部分操作系统（Linux, Windows 等）会动态调整缓冲区大小，通常无需手工设置。

$ cat /proc/sys/net/ipv4/tcp_rmem   # read   [min, default, max]
4096    131072  6291456

$ cat /proc/sys/net/ipv4/tcp_wmem   # write
4096    16384   4194304

# https://www.man7.org/linux/man-pages/man7/tcp.7.html

tcp_rmem (since Linux 2.4)

  This is a vector of 3 integers: [min, default, max].
  
  These parameters are used by TCP to regulate receive
  buffer sizes.  TCP dynamically adjusts the size of the
  receive buffer from the defaults listed below, in the
  range of these values, depending on memory available in
  the system.