feat(trace): 异步化 Geo 查询以降低探测延迟

tsosunchia · tsosunchia · commit fd56e8946a11 · 2025-12-21T21:45:44.000+08:00
- 将 fetchIPData 改为通过 addWithGeoAsync 异步执行
- 新增 waitGeo 机制，在打印每个 TTL 前等待 Geo 数据就绪
- 为 geoWG.Wait 添加 30 秒总超时，防止无限阻塞
- 无论是否有错误都等待 geoWG 完成，避免返回后结果仍被修改
- 在 printer 中增加 pending 状态保护，避免空指针访问
- 导出 PendingGeoSource 常量供跨包使用
- 修改 Result.add 返回 (bool, int) 以追踪插入索引

此变更将 Geo/DNS 查询从主探测路径解耦，显著降低用户感知延迟，
同时通过 sync.Map 缓存和 singleflight 去重保持结果一致性。
diff --git a/.gitignore b/.gitignore
@@ -172,3 +172,4 @@ NTrace-core
 
 .gocache
 .gomodcache
+.cache
diff --git a/printer/basic.go b/printer/basic.go
@@ -77,6 +77,9 @@ func PrintTraceRouteNav(ip net.IP, domain string, dataOrigin string, maxHops int
 }
 
 func applyLangSetting(h *trace.Hop) {
+	if h.Geo == nil || h.Geo.Source == trace.PendingGeoSource {
+		return
+	}
 	if len(h.Geo.Country) <= 1 {
 		// 打印 h.Geo
 		if h.Geo.Whois != "" {
diff --git a/printer/realtime_printer_router.go b/printer/realtime_printer_router.go
@@ -85,7 +85,7 @@ func RealtimePrinterWithRouter(res *trace.Result, ttl int) {
 			fmt.Fprintf(color.Output, " %s", color.New(color.FgHiGreen, color.Bold).Sprintf("%-16s", whoisFormat[0]))
 		}
 
-		if res.Hops[ttl][i].Geo.Country == "" {
+		if res.Hops[ttl][i].Geo.Country == "" && res.Hops[ttl][i].Geo.Source != trace.PendingGeoSource {
 			res.Hops[ttl][i].Geo.Country = "LAN Address"
 		}
 
diff --git a/trace/icmp_ipv4.go b/trace/icmp_ipv4.go
@@ -74,6 +74,7 @@ func (t *ICMPTracer) PrintFunc(ctx context.Context, cancel context.CancelCauseFu
 		// 接收的时候检查一下是不是 3 跳都齐了
 		if t.ttlComp(ttl + 1) {
 			if t.RealtimePrinter != nil {
+				t.res.waitGeo(ctx, ttl)
 				t.RealtimePrinter(&t.res, ttl)
 			}
 			ttl++
@@ -192,10 +193,7 @@ func (t *ICMPTracer) addHopWithIndex(peer net.Addr, ttl, i int, rtt time.Duratio
 		RTT:     rtt,
 		MPLS:    mpls,
 	}
-
-	_ = h.fetchIPData(t.Config) // 忽略错误，继续添加结果
-
-	t.res.add(h, i, t.NumMeasurements, t.MaxAttempts)
+	t.res.addWithGeoAsync(h, i, t.NumMeasurements, t.MaxAttempts, t.Config)
 }
 
 func (t *ICMPTracer) matchWorker(ctx context.Context) {
@@ -263,6 +261,7 @@ func (t *ICMPTracer) Execute() (res *Result, err error) {
 	// 初始化 res.Hops 和 res.tailDone，并预分配到 MaxHops
 	t.res.Hops = make([][]Hop, t.MaxHops)
 	t.res.tailDone = make([]bool, t.MaxHops)
+	t.res.setGeoWait(t.NumMeasurements)
 
 	// 解析并校验用户指定的 IPv4 源地址
 	SrcAddr := net.ParseIP(t.SrcAddr).To4()
@@ -436,7 +435,7 @@ func (t *ICMPTracer) send(ctx context.Context, s *internal.ICMPSpec, ttl, i int)
 				Error:   errHopLimitTimeout,
 			}
 
-			t.res.add(h, i, t.NumMeasurements, t.MaxAttempts)
+			_, _ = t.res.add(h, i, t.NumMeasurements, t.MaxAttempts)
 			t.dropSent(seq)
 		}
 	}(seq, ttl, i)
diff --git a/trace/icmp_ipv6.go b/trace/icmp_ipv6.go
@@ -74,6 +74,7 @@ func (t *ICMPTracerv6) PrintFunc(ctx context.Context, cancel context.CancelCause
 		// 接收的时候检查一下是不是 3 跳都齐了
 		if t.ttlComp(ttl + 1) {
 			if t.RealtimePrinter != nil {
+				t.res.waitGeo(ctx, ttl)
 				t.RealtimePrinter(&t.res, ttl)
 			}
 			ttl++
@@ -192,10 +193,7 @@ func (t *ICMPTracerv6) addHopWithIndex(peer net.Addr, ttl, i int, rtt time.Durat
 		RTT:     rtt,
 		MPLS:    mpls,
 	}
-
-	_ = h.fetchIPData(t.Config) // 忽略错误，继续添加结果
-
-	t.res.add(h, i, t.NumMeasurements, t.MaxAttempts)
+	t.res.addWithGeoAsync(h, i, t.NumMeasurements, t.MaxAttempts, t.Config)
 }
 
 func (t *ICMPTracerv6) matchWorker(ctx context.Context) {
@@ -263,6 +261,7 @@ func (t *ICMPTracerv6) Execute() (res *Result, err error) {
 	// 初始化 res.Hops 和 res.tailDone，并预分配到 MaxHops
 	t.res.Hops = make([][]Hop, t.MaxHops)
 	t.res.tailDone = make([]bool, t.MaxHops)
+	t.res.setGeoWait(t.NumMeasurements)
 
 	// 解析并校验用户指定的 IPv6 源地址
 	SrcAddr := net.ParseIP(t.SrcAddr)
@@ -439,7 +438,7 @@ func (t *ICMPTracerv6) send(ctx context.Context, s *internal.ICMPSpec, ttl, i in
 				Error:   errHopLimitTimeout,
 			}
 
-			t.res.add(h, i, t.NumMeasurements, t.MaxAttempts)
+			_, _ = t.res.add(h, i, t.NumMeasurements, t.MaxAttempts)
 			t.dropSent(seq)
 		}
 	}(seq, ttl, i)
diff --git a/trace/tcp_ipv4.go b/trace/tcp_ipv4.go
@@ -76,6 +76,7 @@ func (t *TCPTracer) PrintFunc(ctx context.Context, cancel context.CancelCauseFun
 		// 接收的时候检查一下是不是 3 跳都齐了
 		if t.ttlComp(ttl + 1) {
 			if t.RealtimePrinter != nil {
+				t.res.waitGeo(ctx, ttl)
 				t.RealtimePrinter(&t.res, ttl)
 			}
 			ttl++
@@ -180,10 +181,7 @@ func (t *TCPTracer) addHopWithIndex(peer net.Addr, ttl, i int, rtt time.Duration
 		RTT:     rtt,
 		MPLS:    mpls,
 	}
-
-	_ = h.fetchIPData(t.Config) // 忽略错误，继续添加结果
-
-	t.res.add(h, i, t.NumMeasurements, t.MaxAttempts)
+	t.res.addWithGeoAsync(h, i, t.NumMeasurements, t.MaxAttempts, t.Config)
 }
 
 func (t *TCPTracer) matchWorker(ctx context.Context) {
@@ -253,6 +251,7 @@ func (t *TCPTracer) Execute() (res *Result, err error) {
 	// 初始化 res.Hops 和 res.tailDone，并预分配到 MaxHops
 	t.res.Hops = make([][]Hop, t.MaxHops)
 	t.res.tailDone = make([]bool, t.MaxHops)
+	t.res.setGeoWait(t.NumMeasurements)
 
 	// 解析并校验用户指定的 IPv4 源地址
 	SrcAddr := net.ParseIP(t.SrcAddr).To4()
@@ -475,7 +474,7 @@ func (t *TCPTracer) send(ctx context.Context, s *internal.TCPSpec, ttl, i int) e
 				Error:   errHopLimitTimeout,
 			}
 
-			t.res.add(h, i, t.NumMeasurements, t.MaxAttempts)
+			_, _ = t.res.add(h, i, t.NumMeasurements, t.MaxAttempts)
 			t.dropSent(seq)
 		}
 	}(seq, ttl, i)
diff --git a/trace/tcp_ipv6.go b/trace/tcp_ipv6.go
@@ -76,6 +76,7 @@ func (t *TCPTracerIPv6) PrintFunc(ctx context.Context, cancel context.CancelCaus
 		// 接收的时候检查一下是不是 3 跳都齐了
 		if t.ttlComp(ttl + 1) {
 			if t.RealtimePrinter != nil {
+				t.res.waitGeo(ctx, ttl)
 				t.RealtimePrinter(&t.res, ttl)
 			}
 			ttl++
@@ -180,10 +181,7 @@ func (t *TCPTracerIPv6) addHopWithIndex(peer net.Addr, ttl, i int, rtt time.Dura
 		RTT:     rtt,
 		MPLS:    mpls,
 	}
-
-	_ = h.fetchIPData(t.Config) // 忽略错误，继续添加结果
-
-	t.res.add(h, i, t.NumMeasurements, t.MaxAttempts)
+	t.res.addWithGeoAsync(h, i, t.NumMeasurements, t.MaxAttempts, t.Config)
 }
 
 func (t *TCPTracerIPv6) matchWorker(ctx context.Context) {
@@ -253,6 +251,7 @@ func (t *TCPTracerIPv6) Execute() (res *Result, err error) {
 	// 初始化 res.Hops 和 res.tailDone，并预分配到 MaxHops
 	t.res.Hops = make([][]Hop, t.MaxHops)
 	t.res.tailDone = make([]bool, t.MaxHops)
+	t.res.setGeoWait(t.NumMeasurements)
 
 	// 解析并校验用户指定的 IPv6 源地址
 	SrcAddr := net.ParseIP(t.SrcAddr)
@@ -475,7 +474,7 @@ func (t *TCPTracerIPv6) send(ctx context.Context, s *internal.TCPSpec, ttl, i in
 				Error:   errHopLimitTimeout,
 			}
 
-			t.res.add(h, i, t.NumMeasurements, t.MaxAttempts)
+			_, _ = t.res.add(h, i, t.NumMeasurements, t.MaxAttempts)
 			t.dropSent(seq)
 		}
 	}(seq, ttl, i)
diff --git a/trace/trace.go b/trace/trace.go
diff --git a/trace/udp_ipv4.go b/trace/udp_ipv4.go
diff --git a/trace/udp_ipv6.go b/trace/udp_ipv6.go

Original file line number	Diff line number	Diff line change
`@@ -172,3 +172,4 @@ NTrace-core`
`172`	`172`
`173`	`173`	`.gocache`
`174`	`174`	`.gomodcache`
	`175`	`+.cache`
Original file line number	Diff line number	Diff line change
`@@ -77,6 +77,9 @@ func PrintTraceRouteNav(ip net.IP, domain string, dataOrigin string, maxHops int`
`77`	`77`	`}`
`78`	`78`
`79`	`79`	`func applyLangSetting(h *trace.Hop) {`
	`80`	`+ if h.Geo == nil \|\| h.Geo.Source == trace.PendingGeoSource {`
	`81`	`+ return`
	`82`	`+ }`
`80`	`83`	`if len(h.Geo.Country) <= 1 {`
`81`	`84`	`// 打印 h.Geo`
`82`	`85`	`if h.Geo.Whois != "" {`
Original file line number	Diff line number	Diff line change
`@@ -85,7 +85,7 @@ func RealtimePrinterWithRouter(res *trace.Result, ttl int) {`
`85`	`85`	`fmt.Fprintf(color.Output, " %s", color.New(color.FgHiGreen, color.Bold).Sprintf("%-16s", whoisFormat[0]))`
`86`	`86`	`}`
`87`	`87`
`88`		`- if res.Hops[ttl][i].Geo.Country == "" {`
	`88`	`+ if res.Hops[ttl][i].Geo.Country == "" && res.Hops[ttl][i].Geo.Source != trace.PendingGeoSource {`
`89`	`89`	`res.Hops[ttl][i].Geo.Country = "LAN Address"`
`90`	`90`	`}`
`91`	`91`