limingjie/OmniSocketGo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' of https://106.52.207.92:9103/limingjie/OmniSocketGo

Browse Source
This commit is contained in:
Mock
2026-03-24 15:39:02 +08:00
parent c126b05961 1cb964c30b
commit 9503862eda
8 changed files with 155 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -1,3 +1,4 @@
bin/*
inbox/*
*.jsonl
*.html

33
cmd/internal/latencylog/summary.go
View File

@@ -31,14 +31,17 @@ type Summary struct {
BodySize int `json:"body_size"` //消息体大小（字节数）
Timestamps map[string]int64 `json:"timestamps"` //事件时间戳，key 是事件名称，value 是 UnixNano 时间戳
AProcessingLatencyNS *int64 `json:"a_processing_latency_ns,omitempty"` // A 处理时延：A_TX_SCHED - A_APP_PREP_BEGIN
AQueueLatencyNS *int64 `json:"a_queue_latency_ns,omitempty"` // A 排队时延：A_TX_SOFTWARE - A_TX_SCHED
ABTransportPropagationNS *int64 `json:"a_b_transport_propagation_ns,omitempty"` // A-B 传输+传播时延近似：B_APP_RECV - A_TX_SOFTWARE
BKernelReceivePathLatencyNS *int64 `json:"b_kernel_receive_path_latency_ns,omitempty"` // B 内核接收路径近似：B_APP_RECV - B_RX_SOFTWARE
BProcessingLatencyNS *int64 `json:"b_processing_latency_ns,omitempty"` // B 处理时延：B_PERSIST_END - B_APP_RECV
EndToEndLatencyNS *int64 `json:"end_to_end_latency_ns,omitempty"` // 端到端时延：B_PERSIST_END - A_APP_PREP_BEGIN
ApproxRTTNS *int64 `json:"approx_rtt_ns,omitempty"` // 近似 RTT：首条反向应答的 B_APP_RECV - 当前请求的 A_TX_SOFTWARE
MissingTimestamps []string `json:"missing_timestamps,omitempty"` // 缺失的时间戳列表，包含 requiredTimestampNames 中但在原始事件中没有的事件名称
AProcessingLatencyNS *int64 `json:"a_processing_latency_ns,omitempty"` // A 处理时延：A_TX_SCHED - A_APP_PREP_BEGIN
AQueueLatencyNS *int64 `json:"a_queue_latency_ns,omitempty"` // A 排队时延：A_TX_SOFTWARE - A_TX_SCHED
ABTransportPropagationNS *int64 `json:"a_b_transport_propagation_ns,omitempty"` // A-B 传输+传播时延近似：B_APP_RECV - A_TX_SOFTWARE
BKernelReceivePathLatencyNS *int64 `json:"b_kernel_receive_path_latency_ns,omitempty"` // B 内核接收路径近似：B_APP_RECV - B_RX_SOFTWARE
BProcessingLatencyNS *int64 `json:"b_processing_latency_ns,omitempty"` // B 处理时延：B_PERSIST_END - B_APP_RECV
EndToEndLatencyNS *int64 `json:"end_to_end_latency_ns,omitempty"` // 端到端时延：B_PERSIST_END - A_APP_PREP_BEGIN
AProcessingBitrateBPS *float64 `json:"a_processing_bitrate_bps,omitempty"` // A 处理阶段近似比特率：(BodySize * 8) / A 处理时延（秒）
ABTransportPropagationBitrateBPS *float64 `json:"a_b_transport_propagation_bitrate_bps,omitempty"` // A-B 传输+传播阶段近似比特率：(BodySize * 8) / A-B 传输+传播时延（秒）
EndToEndBitrateBPS *float64 `json:"end_to_end_bitrate_bps,omitempty"` // 端到端近似比特率：(BodySize * 8) / 端到端时延（秒）
ApproxRTTNS *int64 `json:"approx_rtt_ns,omitempty"` // 近似 RTT：首条反向应答的 B_APP_RECV - 当前请求的 A_TX_SOFTWARE
MissingTimestamps []string `json:"missing_timestamps,omitempty"` // 缺失的时间戳列表，包含 requiredTimestampNames 中但在原始事件中没有的事件名称
}
// LoadEventsFromFiles 从JSONL 原始日志文件中加载事件。
@@ -213,6 +216,10 @@ func completeSummary(summary *Summary) {
if value := subtractIfPresent(summary.Timestamps, EventBPersistEnd, EventAAppPrepBegin); value != nil {
summary.EndToEndLatencyNS = value
}
summary.AProcessingBitrateBPS = calculateBitrateBPS(summary.BodySize, summary.AProcessingLatencyNS)
summary.ABTransportPropagationBitrateBPS = calculateBitrateBPS(summary.BodySize, summary.ABTransportPropagationNS)
summary.EndToEndBitrateBPS = calculateBitrateBPS(summary.BodySize, summary.EndToEndLatencyNS)
}
type routeKey struct {
@@ -340,6 +347,16 @@ func subtractSummaryTimestamps(endSummary *Summary, endName string, beginSummary
return &value
}
// 除法函数，如果 bodySize <= 0 或 latencyNS 不存在或 <= 0，则返回 nil；否则返回 bodySize / latencyNS 的结果。
func calculateBitrateBPS(bodySize int, latencyNS *int64) *float64 {
if bodySize <= 0 || latencyNS == nil || *latencyNS <= 0 {
return nil
}
value := float64(bodySize) * 8 * 1_000_000_000 / float64(*latencyNS)
return &value
}
// 判断事件是否是业务相关的时延事件（其中一项）
func IsBusinessEvent(event Event) bool {
switch event.Event {

52
cmd/internal/latencylog/summary_chart.go
View File

@@ -131,6 +131,23 @@ const summaryChartHTMLTemplate = `<!doctype html>
font-size: 13px;
margin-bottom: 12px;
}
.ratio-list {
display: flex;
flex-wrap: wrap;
gap: 8px;
margin: -2px 0 12px;
}
.ratio-pill {
display: inline-flex;
align-items: center;
padding: 6px 10px;
border-radius: 999px;
border: 1px solid var(--border);
background: #f7f9fc;
color: var(--text);
font-size: 12px;
line-height: 1.2;
}
.bar {
height: 24px;
display: flex;
@@ -216,6 +233,13 @@ const summaryChartHTMLTemplate = `<!doctype html>
</div>
<div class="row-meta">{{.Subtitle}}</div>
<div class="row-meta">{{.ApproxRTT}}</div>
{{if .RatioMetrics}}
<div class="ratio-list">
{{range .RatioMetrics}}
<span class="ratio-pill">{{.Label}} {{.Value}}</span>
{{end}}
</div>
{{end}}
<div class="bar">
{{range .Segments}}
<div class="segment" style="width: {{printf "%.4f" .WidthPercent}}%; background: {{.Color}}" title="{{.Label}}: {{.Value}}"></div>
@@ -265,6 +289,7 @@ type summaryChartRow struct {
EndToEnd string
ApproxRTT string
MissingTimestamps string
RatioMetrics []summaryChartValue
Segments []summaryChartSegment
}
@@ -275,6 +300,11 @@ type summaryChartSegment struct {
WidthPercent float64
}
type summaryChartValue struct {
Label string
Value string
}
type summaryChartSegmentMetric struct {
label string
value *int64
@@ -366,6 +396,24 @@ func buildSummaryChartRow(summary Summary) summaryChartRow {
row.ApproxRTT = fmt.Sprintf("Approx RTT: %s", formatLatencyNS(*summary.ApproxRTTNS))
}
ratioMetrics := []struct {
label string
value *float64
}{
{label: "A processing bitrate", value: summary.AProcessingBitrateBPS},
{label: "A-B transport + propagation bitrate", value: summary.ABTransportPropagationBitrateBPS},
{label: "End-to-end bitrate", value: summary.EndToEndBitrateBPS},
}
for _, metric := range ratioMetrics {
if metric.value == nil || *metric.value <= 0 {
continue
}
row.RatioMetrics = append(row.RatioMetrics, summaryChartValue{
Label: metric.label,
Value: formatBitrateBPS(*metric.value),
})
}
if summary.EndToEndLatencyNS == nil || *summary.EndToEndLatencyNS <= 0 {
return row
}
@@ -444,3 +492,7 @@ func buildSummaryChartSubtitle(summary Summary) string {
func formatLatencyNS(ns int64) string {
return fmt.Sprintf("%.3f ms", float64(ns)/1_000_000)
}
func formatBitrateBPS(bitsPerSecond float64) string {
return fmt.Sprintf("%.3f Mb/s", bitsPerSecond/1_000_000)
}

31
cmd/internal/latencylog/summary_chart_test.go
View File

@@ -15,20 +15,26 @@ func TestWriteSummariesHTMLChart(t *testing.T) {
transport := int64(40_000_000)
bProcessing := int64(30_000_000)
endToEnd := int64(100_000_000)
aProcessingBitrate := float64(5) * 8 * 1_000_000_000 / float64(aProcessing)
transportBitrate := float64(5) * 8 * 1_000_000_000 / float64(transport)
endToEndBitrate := float64(5) * 8 * 1_000_000_000 / float64(endToEnd)
summaries := []Summary{
{
MessageType: protocol.MessageTypeText,
MessageID: 7,
From: "peer-a",
To: "peer-b",
BodySize: 5,
AProcessingLatencyNS: &aProcessing,
AQueueLatencyNS: &aQueue,
ABTransportPropagationNS: &transport,
BProcessingLatencyNS: &bProcessing,
EndToEndLatencyNS: &endToEnd,
ApproxRTTNS: &endToEnd,
MessageType: protocol.MessageTypeText,
MessageID: 7,
From: "peer-a",
To: "peer-b",
BodySize: 5,
AProcessingLatencyNS: &aProcessing,
AQueueLatencyNS: &aQueue,
ABTransportPropagationNS: &transport,
BProcessingLatencyNS: &bProcessing,
EndToEndLatencyNS: &endToEnd,
AProcessingBitrateBPS: &aProcessingBitrate,
ABTransportPropagationBitrateBPS: &transportBitrate,
EndToEndBitrateBPS: &endToEndBitrate,
ApproxRTTNS: &endToEnd,
},
{
MessageType: protocol.MessageTypeFile,
@@ -58,6 +64,9 @@ func TestWriteSummariesHTMLChart(t *testing.T) {
"peer-a -&gt; peer-b | 5 bytes",
"End-to-end: 100.000 ms",
"Approx RTT: 100.000 ms",
"A processing bitrate 0.002 Mb/s",
"A-B transport &#43; propagation bitrate 0.001 Mb/s",
"End-to-end bitrate 0.000 Mb/s",
"A processing 20.000 ms",
"A-B transport &#43; propagation 40.000 ms",
"file #8 (payload.bin)",

45
cmd/internal/latencylog/summary_test.go
View File

@@ -13,13 +13,13 @@ import (
func TestSummarizeEventsComputesLatencyMetrics(t *testing.T) {
events := []Event{
{TsUnixNano: 100, Event: EventAAppPrepBegin, MessageType: protocol.MessageTypeText, MessageID: 1, From: "peer-a", To: "peer-b"},
{TsUnixNano: 120, Event: EventATXSched, MessageType: protocol.MessageTypeText, MessageID: 1, From: "peer-a", To: "peer-b"},
{TsUnixNano: 140, Event: EventATXSoftware, MessageType: protocol.MessageTypeText, MessageID: 1, From: "peer-a", To: "peer-b"},
{TsUnixNano: 180, Event: EventBRXSoftware, MessageType: protocol.MessageTypeText, MessageID: 1, From: "peer-a", To: "peer-b"},
{TsUnixNano: 220, Event: EventBAppRecv, MessageType: protocol.MessageTypeText, MessageID: 1, From: "peer-a", To: "peer-b"},
{TsUnixNano: 230, Event: EventBPersistBegin, MessageType: protocol.MessageTypeText, MessageID: 1, From: "peer-a", To: "peer-b"},
{TsUnixNano: 260, Event: EventBPersistEnd, MessageType: protocol.MessageTypeText, MessageID: 1, From: "peer-a", To: "peer-b"},
{TsUnixNano: 100, Event: EventAAppPrepBegin, MessageType: protocol.MessageTypeText, MessageID: 1, From: "peer-a", To: "peer-b", BodySize: 320},
{TsUnixNano: 120, Event: EventATXSched, MessageType: protocol.MessageTypeText, MessageID: 1, From: "peer-a", To: "peer-b", BodySize: 320},
{TsUnixNano: 140, Event: EventATXSoftware, MessageType: protocol.MessageTypeText, MessageID: 1, From: "peer-a", To: "peer-b", BodySize: 320},
{TsUnixNano: 180, Event: EventBRXSoftware, MessageType: protocol.MessageTypeText, MessageID: 1, From: "peer-a", To: "peer-b", BodySize: 320},
{TsUnixNano: 220, Event: EventBAppRecv, MessageType: protocol.MessageTypeText, MessageID: 1, From: "peer-a", To: "peer-b", BodySize: 320},
{TsUnixNano: 230, Event: EventBPersistBegin, MessageType: protocol.MessageTypeText, MessageID: 1, From: "peer-a", To: "peer-b", BodySize: 320},
{TsUnixNano: 260, Event: EventBPersistEnd, MessageType: protocol.MessageTypeText, MessageID: 1, From: "peer-a", To: "peer-b", BodySize: 320},
}
summaries := SummarizeEvents(events)
@@ -46,6 +46,15 @@ func TestSummarizeEventsComputesLatencyMetrics(t *testing.T) {
if got := ptrValue(summary.EndToEndLatencyNS); got != 160 {
t.Fatalf("EndToEndLatencyNS = %d, want 160", got)
}
if got := ptrValueFloat(summary.AProcessingBitrateBPS); got != 128_000_000_000 {
t.Fatalf("AProcessingBitrateBPS = %v, want 128000000000", got)
}
if got := ptrValueFloat(summary.ABTransportPropagationBitrateBPS); got != 32_000_000_000 {
t.Fatalf("ABTransportPropagationBitrateBPS = %v, want 32000000000", got)
}
if got := ptrValueFloat(summary.EndToEndBitrateBPS); got != 16_000_000_000 {
t.Fatalf("EndToEndBitrateBPS = %v, want 16000000000", got)
}
if got := summary.Timestamps[EventBRXSoftware]; got != 180 {
t.Fatalf("timestamps[%q] = %d, want 180", EventBRXSoftware, got)
}
@@ -128,12 +137,12 @@ func TestLoadAndWriteSummaryFiles(t *testing.T) {
})
for _, event := range []Event{
{TsUnixNano: 100, Event: EventAAppPrepBegin, MessageType: protocol.MessageTypeText, MessageID: 3, From: "peer-a", To: "peer-b"},
{TsUnixNano: 120, Event: EventATXSched, MessageType: protocol.MessageTypeText, MessageID: 3, From: "peer-a", To: "peer-b"},
{TsUnixNano: 140, Event: EventATXSoftware, MessageType: protocol.MessageTypeText, MessageID: 3, From: "peer-a", To: "peer-b"},
{TsUnixNano: 180, Event: EventBRXSoftware, MessageType: protocol.MessageTypeText, MessageID: 3, From: "peer-a", To: "peer-b"},
{TsUnixNano: 220, Event: EventBAppRecv, MessageType: protocol.MessageTypeText, MessageID: 3, From: "peer-a", To: "peer-b"},
{TsUnixNano: 260, Event: EventBPersistEnd, MessageType: protocol.MessageTypeText, MessageID: 3, From: "peer-a", To: "peer-b"},
{TsUnixNano: 100, Event: EventAAppPrepBegin, MessageType: protocol.MessageTypeText, MessageID: 3, From: "peer-a", To: "peer-b", BodySize: 320},
{TsUnixNano: 120, Event: EventATXSched, MessageType: protocol.MessageTypeText, MessageID: 3, From: "peer-a", To: "peer-b", BodySize: 320},
{TsUnixNano: 140, Event: EventATXSoftware, MessageType: protocol.MessageTypeText, MessageID: 3, From: "peer-a", To: "peer-b", BodySize: 320},
{TsUnixNano: 180, Event: EventBRXSoftware, MessageType: protocol.MessageTypeText, MessageID: 3, From: "peer-a", To: "peer-b", BodySize: 320},
{TsUnixNano: 220, Event: EventBAppRecv, MessageType: protocol.MessageTypeText, MessageID: 3, From: "peer-a", To: "peer-b", BodySize: 320},
{TsUnixNano: 260, Event: EventBPersistEnd, MessageType: protocol.MessageTypeText, MessageID: 3, From: "peer-a", To: "peer-b", BodySize: 320},
} {
if err := rawLogger.LogEvent(event); err != nil {
t.Fatalf("LogEvent() error = %v", err)
@@ -174,6 +183,9 @@ func TestLoadAndWriteSummaryFiles(t *testing.T) {
if got := ptrValue(summary.EndToEndLatencyNS); got != 160 {
t.Fatalf("EndToEndLatencyNS = %d, want 160", got)
}
if got := ptrValueFloat(summary.EndToEndBitrateBPS); got != 16_000_000_000 {
t.Fatalf("EndToEndBitrateBPS = %v, want 16000000000", got)
}
}
func ptrValue(value *int64) int64 {
@@ -182,3 +194,10 @@ func ptrValue(value *int64) int64 {
}
return *value
}
func ptrValueFloat(value *float64) float64 {
if value == nil {
return 0
}
return *value
}

27
cmd/internal/transport/tcp_linux.go
View File

@@ -15,10 +15,11 @@ import (
)
const (
linuxTimestampControlBufferSize = 256 // 控制消息缓冲区。
linuxTXTimestampWaitTimeout = 250 * time.Millisecond // 等待 TX 时间戳的上限。
linuxTXTimestampPollInterval = time.Millisecond // 轮询 errqueue 的间隔。
linuxDataPollInterval = time.Millisecond // 轮询普通收发的间隔。
linuxTimestampControlBufferSize = 2048 // 控制消息缓冲区。
linuxSocketWriteBufferSize = 10 * 1024 * 1024 // 请求把 socket 发送缓冲区调到 10 MiB。
linuxTXTimestampWaitTimeout = 5000 * time.Millisecond // 等待 TX 时间戳的上限。
linuxTXTimestampPollInterval = time.Millisecond // 轮询 errqueue 的间隔。
linuxDataPollInterval = time.Millisecond // 轮询普通收发的间隔。
linuxSOTimestampingNew = 0x41
linuxSCMTimestampingNew = linuxSOTimestampingNew
@@ -88,6 +89,10 @@ func (c *TCPConn) initLinuxTimestamping() error {
return fmt.Errorf("transport: missing syscall conn")
}
if err := configureLinuxSocketWriteBuffer(rawConn); err != nil {
return fmt.Errorf("transport: configure socket write buffer: %w", err)
}
//socket是否可以成功打开 timestamping 取决于内核版本和配置，尝试多个 flag 组合直到成功或遇到非 EINVAL 错误。
if err := enableLinuxTimestamping(rawConn); err != nil {
return fmt.Errorf("transport: enable linux timestamping: %w", err)
@@ -97,6 +102,20 @@ func (c *TCPConn) initLinuxTimestamping() error {
return nil
}
// 设置 TCP缓冲区buffer size
func configureLinuxSocketWriteBuffer(rawConn syscall.RawConn) error {
var lastErr error
err := rawConn.Control(func(fd uintptr) {
lastErr = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_SNDBUF, linuxSocketWriteBufferSize)
})
if err != nil {
return err
}
return lastErr
}
// 给 socket开权限打开TX software timestamping。
func enableLinuxTimestamping(rawConn syscall.RawConn) error {
flagCandidates := []int{ //不同linux版本可能支持不同的 flag 组合，尝试多个组合直到成功。

2
cmd/internal/transport/tcp_linux_test.go
View File

@@ -182,7 +182,7 @@ func TestSelectTXTimestampEventsFallsBackToHighestObservedID(t *testing.T) {
func TestLinuxTimestampingDebugLoggerCapturesChunkAndErrqueueEvents(t *testing.T) {
clientConn, serverConn := newTCPPair(t)
setTCPWriteBuffer(t, clientConn, 4096)
setTCPWriteBuffer(t, clientConn, 10*1024*1024)
debugLogger := &recordingTXTimestampDebugLogger{}
senderLogger := &recordingLogger{}

BIN
latencysummary
View File

Binary file not shown.