fix: transport.UDPConn 新增了 WithUDPLinuxTimestamping(false) 开关

This commit is contained in:
2026-03-27 01:59:17 +08:00
parent 5b231141a7
commit 8cec6a0766
7 changed files with 154 additions and 37 deletions

View File

@@ -22,6 +22,7 @@ type clientOptions struct {
kcpPacketDebugLogger transport.KCPPacketDebugLogger kcpPacketDebugLogger transport.KCPPacketDebugLogger
kcpSessionStatsLogger transport.KCPSessionStatsLogger kcpSessionStatsLogger transport.KCPSessionStatsLogger
kcpSessionStatsInterval time.Duration kcpSessionStatsInterval time.Duration
udpLinuxTimestamping bool
bindIP string bindIP string
bindDevice string bindDevice string
} }
@@ -72,6 +73,13 @@ func WithBindDevice(device string) Option {
} }
} }
// WithUDPLinuxTimestamping controls whether UDP clients enable Linux timestamping.
func WithUDPLinuxTimestamping(enabled bool) Option {
return func(options *clientOptions) {
options.udpLinuxTimestamping = enabled
}
}
// Client 表示一个已经连接到 server 的 peer。 // Client 表示一个已经连接到 server 的 peer。
type Client struct { type Client struct {
id string id string
@@ -84,7 +92,8 @@ type Client struct {
// Dial 连接到 server并立即发送 register 消息完成身份注册。 // Dial 连接到 server并立即发送 register 消息完成身份注册。
func Dial(serverAddr, peerID string, opts ...Option) (*Client, error) { func Dial(serverAddr, peerID string, opts ...Option) (*Client, error) {
options := clientOptions{ options := clientOptions{
logger: latencylog.NoopLogger{}, logger: latencylog.NoopLogger{},
udpLinuxTimestamping: true,
} }
for _, opt := range opts { for _, opt := range opts {
opt(&options) opt(&options)

View File

@@ -24,7 +24,8 @@ type UDPClient struct {
// DialUDP 通过 UDP 连接到 server并发送 register 消息完成身份注册。 // DialUDP 通过 UDP 连接到 server并发送 register 消息完成身份注册。
func DialUDP(serverAddr, peerID string, opts ...Option) (*UDPClient, error) { func DialUDP(serverAddr, peerID string, opts ...Option) (*UDPClient, error) {
options := clientOptions{ options := clientOptions{
logger: latencylog.NoopLogger{}, logger: latencylog.NoopLogger{},
udpLinuxTimestamping: true,
} }
for _, opt := range opts { for _, opt := range opts {
opt(&options) opt(&options)
@@ -56,6 +57,7 @@ func DialUDP(serverAddr, peerID string, opts ...Option) (*UDPClient, error) {
rawConn, rawConn,
nil, // peer 侧已连接模式,不需要指定 peerAddr nil, // peer 侧已连接模式,不需要指定 peerAddr
transport.WithUDPLogger(options.logger, latencylog.NodeRolePeer, peerID), transport.WithUDPLogger(options.logger, latencylog.NodeRolePeer, peerID),
transport.WithUDPLinuxTimestamping(options.udpLinuxTimestamping),
transport.WithUDPTXTimestampDebugLogger(options.txTimestampDebugLogger), transport.WithUDPTXTimestampDebugLogger(options.txTimestampDebugLogger),
) )
if err != nil { if err != nil {

View File

@@ -28,23 +28,32 @@ func WithUDPTXTimestampDebugLogger(logger transport.TXTimestampDebugLogger) UDPO
} }
} }
// WithUDPLinuxTimestamping controls whether the UDP hub enables Linux timestamping.
func WithUDPLinuxTimestamping(enabled bool) UDPOption {
return func(hub *UDPHub) {
hub.linuxTimestampingEnabled = enabled
}
}
// UDPHub 管理通过 UDP 注册的 peer并负责在它们之间转发消息。 // UDPHub 管理通过 UDP 注册的 peer并负责在它们之间转发消息。
type UDPHub struct { type UDPHub struct {
mu sync.RWMutex mu sync.RWMutex
peers map[string]*net.UDPAddr peers map[string]*net.UDPAddr
addrs map[string]string addrs map[string]string
conn *transport.UDPConn conn *transport.UDPConn
logger latencylog.Logger logger latencylog.Logger
txTimestampDebugLogger transport.TXTimestampDebugLogger txTimestampDebugLogger transport.TXTimestampDebugLogger
linuxTimestampingEnabled bool
} }
// NewUDPHub 创建一个新的 UDP 连接中心。 // NewUDPHub 创建一个新的 UDP 连接中心。
func NewUDPHub(conn *net.UDPConn, opts ...UDPOption) (*UDPHub, error) { func NewUDPHub(conn *net.UDPConn, opts ...UDPOption) (*UDPHub, error) {
hub := &UDPHub{ hub := &UDPHub{
peers: make(map[string]*net.UDPAddr), peers: make(map[string]*net.UDPAddr),
addrs: make(map[string]string), addrs: make(map[string]string),
logger: latencylog.NoopLogger{}, logger: latencylog.NoopLogger{},
linuxTimestampingEnabled: true,
} }
for _, opt := range opts { for _, opt := range opts {
@@ -59,6 +68,7 @@ func NewUDPHub(conn *net.UDPConn, opts ...UDPOption) (*UDPHub, error) {
conn, conn,
nil, nil,
transport.WithUDPLogger(hub.logger, latencylog.NodeRoleServer, "hub"), transport.WithUDPLogger(hub.logger, latencylog.NodeRoleServer, "hub"),
transport.WithUDPLinuxTimestamping(hub.linuxTimestampingEnabled),
transport.WithUDPTXTimestampDebugLogger(hub.txTimestampDebugLogger), transport.WithUDPTXTimestampDebugLogger(hub.txTimestampDebugLogger),
) )
if err != nil { if err != nil {

View File

@@ -10,23 +10,22 @@ import (
"omnisocketgo/cmd/internal/protocol" "omnisocketgo/cmd/internal/protocol"
) )
// UDPConn 是对 UDP 连接的轻量封装。 // UDPConn wraps a UDP socket for protocol message send/receive.
// server 侧共享一个 net.UDPConn通过 SendTo 指定目标地址;
// peer 侧使用已连接的 net.UDPConn直接 Send 即可。
type UDPConn struct { type UDPConn struct {
conn *net.UDPConn conn *net.UDPConn
peerAddr *net.UDPAddr peerAddr *net.UDPAddr
raw syscall.RawConn raw syscall.RawConn
logger latencylog.Logger linuxTimestampingEnabled bool
txTimestampDebugLogger TXTimestampDebugLogger logger latencylog.Logger
txPacketSeq uint32 txTimestampDebugLogger TXTimestampDebugLogger
pendingTX map[uint32]udpTXPendingRecord txPacketSeq uint32
nodeRole string pendingTX map[uint32]udpTXPendingRecord
nodeID string nodeRole string
writeMu sync.Mutex nodeID string
closeOnce sync.Once writeMu sync.Mutex
closeErr error closeOnce sync.Once
closeErr error
} }
type udpTXPendingRecord struct { type udpTXPendingRecord struct {
@@ -37,10 +36,10 @@ type udpTXPendingRecord struct {
observedTimestamps map[string]int64 observedTimestamps map[string]int64
} }
// UDPOption 用于为 UDPConn 注入可选行为。 // UDPOption configures an optional behavior on UDPConn.
type UDPOption func(*UDPConn) type UDPOption func(*UDPConn)
// WithUDPLogger 为 UDP 连接注入业务消息日志上下文。 // WithUDPLogger attaches latency logging context to a UDP connection.
func WithUDPLogger(logger latencylog.Logger, nodeRole, nodeID string) UDPOption { func WithUDPLogger(logger latencylog.Logger, nodeRole, nodeID string) UDPOption {
return func(conn *UDPConn) { return func(conn *UDPConn) {
conn.logger = logger conn.logger = logger
@@ -49,20 +48,28 @@ func WithUDPLogger(logger latencylog.Logger, nodeRole, nodeID string) UDPOption
} }
} }
// WithUDPTXTimestampDebugLogger 为 UDP 连接注入 TX errqueue 调试日志器。 // WithUDPTXTimestampDebugLogger attaches a TX errqueue debug logger.
func WithUDPTXTimestampDebugLogger(logger TXTimestampDebugLogger) UDPOption { func WithUDPTXTimestampDebugLogger(logger TXTimestampDebugLogger) UDPOption {
return func(conn *UDPConn) { return func(conn *UDPConn) {
conn.txTimestampDebugLogger = logger conn.txTimestampDebugLogger = logger
} }
} }
// NewUDPConn 创建 UDP transport 连接封装。 // WithUDPLinuxTimestamping controls whether Linux UDP timestamping is enabled.
func WithUDPLinuxTimestamping(enabled bool) UDPOption {
return func(conn *UDPConn) {
conn.linuxTimestampingEnabled = enabled
}
}
// NewUDPConn creates a UDP transport wrapper.
func NewUDPConn(conn *net.UDPConn, peerAddr *net.UDPAddr, opts ...UDPOption) (*UDPConn, error) { func NewUDPConn(conn *net.UDPConn, peerAddr *net.UDPAddr, opts ...UDPOption) (*UDPConn, error) {
udpConn := &UDPConn{ udpConn := &UDPConn{
conn: conn, conn: conn,
peerAddr: peerAddr, peerAddr: peerAddr,
logger: latencylog.NoopLogger{}, linuxTimestampingEnabled: true,
pendingTX: make(map[uint32]udpTXPendingRecord), logger: latencylog.NoopLogger{},
pendingTX: make(map[uint32]udpTXPendingRecord),
} }
for _, opt := range opts { for _, opt := range opts {
@@ -73,14 +80,16 @@ func NewUDPConn(conn *net.UDPConn, peerAddr *net.UDPAddr, opts ...UDPOption) (*U
udpConn.logger = latencylog.NoopLogger{} udpConn.logger = latencylog.NoopLogger{}
} }
if err := udpConn.initUDPLinuxTimestamping(); err != nil { if udpConn.linuxTimestampingEnabled {
return nil, err if err := udpConn.initUDPLinuxTimestamping(); err != nil {
return nil, err
}
} }
return udpConn, nil return udpConn, nil
} }
// Send 将一条协议消息编码为 UDP 数据报并发送。 // Send encodes and sends one protocol message over UDP.
func (c *UDPConn) Send(msg protocol.Message) error { func (c *UDPConn) Send(msg protocol.Message) error {
c.writeMu.Lock() c.writeMu.Lock()
defer c.writeMu.Unlock() defer c.writeMu.Unlock()
@@ -94,7 +103,7 @@ func (c *UDPConn) Send(msg protocol.Message) error {
return nil return nil
} }
// SendTo 将一条协议消息编码为 UDP 数据报并发送到指定地址。 // SendTo encodes and sends one protocol message to a specific UDP address.
func (c *UDPConn) SendTo(msg protocol.Message, addr *net.UDPAddr) error { func (c *UDPConn) SendTo(msg protocol.Message, addr *net.UDPAddr) error {
c.writeMu.Lock() c.writeMu.Lock()
defer c.writeMu.Unlock() defer c.writeMu.Unlock()
@@ -108,7 +117,7 @@ func (c *UDPConn) SendTo(msg protocol.Message, addr *net.UDPAddr) error {
return nil return nil
} }
// Receive 从 UDP 连接读取一条完整协议消息。 // Receive reads one full protocol message from UDP.
func (c *UDPConn) Receive() (protocol.Message, *net.UDPAddr, error) { func (c *UDPConn) Receive() (protocol.Message, *net.UDPAddr, error) {
msg, addr, err := c.receiveMessageLinux() msg, addr, err := c.receiveMessageLinux()
if err != nil { if err != nil {
@@ -118,7 +127,7 @@ func (c *UDPConn) Receive() (protocol.Message, *net.UDPAddr, error) {
return msg, addr, nil return msg, addr, nil
} }
// ReceiveLoop 持续从 UDP 连接读取消息并交给 handler 处理。 // ReceiveLoop continuously receives messages and passes them to handler.
func (c *UDPConn) ReceiveLoop(handler func(protocol.Message, *net.UDPAddr) error) error { func (c *UDPConn) ReceiveLoop(handler func(protocol.Message, *net.UDPAddr) error) error {
for { for {
msg, addr, err := c.Receive() msg, addr, err := c.Receive()
@@ -132,7 +141,7 @@ func (c *UDPConn) ReceiveLoop(handler func(protocol.Message, *net.UDPAddr) error
} }
} }
// Close 关闭底层 UDP 连接,保证重复调用安全。 // Close closes the underlying UDP socket.
func (c *UDPConn) Close() error { func (c *UDPConn) Close() error {
c.closeOnce.Do(func() { c.closeOnce.Do(func() {
c.closeErr = c.conn.Close() c.closeErr = c.conn.Close()

View File

@@ -246,3 +246,87 @@ func TestUDPLinuxTimestampingDebugLoggerCapturesDatagramAndErrqueueEvents(t *tes
} }
} }
} }
func TestUDPLinuxTimestampingCanBeDisabled(t *testing.T) {
senderLogger := &recordingLogger{}
receiverLogger := &recordingLogger{}
serverAddr, err := net.ResolveUDPAddr("udp", "127.0.0.1:0")
if err != nil {
t.Fatalf("ResolveUDPAddr() error = %v", err)
}
serverRaw, err := net.ListenUDP("udp", serverAddr)
if err != nil {
t.Fatalf("ListenUDP() error = %v", err)
}
receiver, err := NewUDPConn(
serverRaw,
nil,
WithUDPLogger(receiverLogger, latencylog.NodeRolePeer, "peer-b"),
WithUDPLinuxTimestamping(false),
)
if err != nil {
_ = serverRaw.Close()
t.Fatalf("NewUDPConn(receiver) error = %v", err)
}
t.Cleanup(func() { _ = receiver.Close() })
peerRaw, err := net.DialUDP("udp", nil, serverRaw.LocalAddr().(*net.UDPAddr))
if err != nil {
t.Fatalf("DialUDP() error = %v", err)
}
sender, err := NewUDPConn(
peerRaw,
nil,
WithUDPLogger(senderLogger, latencylog.NodeRolePeer, "peer-a"),
WithUDPLinuxTimestamping(false),
)
if err != nil {
_ = peerRaw.Close()
t.Fatalf("NewUDPConn(sender) error = %v", err)
}
t.Cleanup(func() { _ = sender.Close() })
msg := protocol.Message{
Type: protocol.MessageTypeText,
ID: 123,
From: "peer-a",
To: "peer-b",
Body: []byte("hello without udp timestamping"),
}
sendErr := make(chan error, 1)
go func() {
sendErr <- sender.Send(msg)
}()
got, _, err := receiver.Receive()
if err != nil {
t.Fatalf("Receive() error = %v", err)
}
if err := <-sendErr; err != nil {
t.Fatalf("Send() error = %v", err)
}
if !reflect.DeepEqual(got, msg) {
t.Fatalf("message mismatch: got %+v want %+v", got, msg)
}
if sender.raw != nil {
t.Fatal("sender.raw != nil, want nil when linux timestamping is disabled")
}
if receiver.raw != nil {
t.Fatal("receiver.raw != nil, want nil when linux timestamping is disabled")
}
assertMissingEvent(t, senderLogger.Events(), latencylog.EventATXSched, msg.ID)
assertMissingEvent(t, senderLogger.Events(), latencylog.EventATXSoftware, msg.ID)
assertMissingEvent(t, receiverLogger.Events(), latencylog.EventBRXSoftware, msg.ID)
}
func assertMissingEvent(t *testing.T, events []latencylog.Event, wantEvent string, wantMessageID uint64) {
t.Helper()
for _, event := range events {
if event.Event == wantEvent && event.MessageID == wantMessageID {
t.Fatalf("unexpected event %s for message %d: %+v", wantEvent, wantMessageID, event)
}
}
}

View File

@@ -35,8 +35,9 @@ func runPlatform(cfg config, stdout, stderr io.Writer, now func() time.Time) err
} }
func dialUDPClient(cfg config) (*peerpkg.UDPClient, func(), error) { func dialUDPClient(cfg config) (*peerpkg.UDPClient, func(), error) {
options := make([]peerpkg.Option, 0, 2) options := make([]peerpkg.Option, 0, 3)
closeLogger := func() {} closeLogger := func() {}
options = append(options, peerpkg.WithUDPLinuxTimestamping(false))
if cfg.latencyLog != "" { if cfg.latencyLog != "" {
logger, err := latencylog.NewJSONLLogger(cfg.latencyLog) logger, err := latencylog.NewJSONLLogger(cfg.latencyLog)

View File

@@ -13,10 +13,11 @@ import (
func main() { func main() {
listenAddr := flag.String("listen", ":9001", "UDP server listen address") listenAddr := flag.String("listen", ":9001", "UDP server listen address")
logPath := flag.String("latency-log", "", "optional JSONL file path for latency timestamp logs") logPath := flag.String("latency-log", "", "optional JSONL file path for latency timestamp logs")
txTimestampDebugLogPath := flag.String("tx-ts-debug-log", "", "optional JSONL file path for TX errqueue debug records") txTimestampDebugLogPath := flag.String("tx-ts-debug-log", "", "optional JSONL file path for TX errqueue debug records; enables Linux UDP timestamping")
flag.Parse() flag.Parse()
hubOptions := make([]server.UDPOption, 0, 2) hubOptions := make([]server.UDPOption, 0, 2)
hubOptions = append(hubOptions, server.WithUDPLinuxTimestamping(false))
if *logPath != "" { if *logPath != "" {
logger, err := latencylog.NewJSONLLogger(*logPath) logger, err := latencylog.NewJSONLLogger(*logPath)
if err != nil { if err != nil {
@@ -31,6 +32,7 @@ func main() {
log.Fatalf("create tx timestamp debug logger %s: %v", *txTimestampDebugLogPath, err) log.Fatalf("create tx timestamp debug logger %s: %v", *txTimestampDebugLogPath, err)
} }
defer logger.Close() defer logger.Close()
hubOptions = append(hubOptions, server.WithUDPLinuxTimestamping(true))
hubOptions = append(hubOptions, server.WithUDPTXTimestampDebugLogger(logger)) hubOptions = append(hubOptions, server.WithUDPTXTimestampDebugLogger(logger))
} }