diff --git a/cmd/server/chunked_load.go b/cmd/server/chunked_load.go index 6420d0eca..864c995fb 100644 --- a/cmd/server/chunked_load.go +++ b/cmd/server/chunked_load.go @@ -629,8 +629,7 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold rpStr := nullStrVal(resolvedPathStr) if rpStr != "" { - rp := unmarshalResolvedPath(rpStr) - pks := extractResolvedPubkeys(rp) + pks := resolvedPubkeysFromJSON(rpStr) s.indexResolvedPathHops(tx, pks, hopsSeen) } else if relayPM != nil && obsPJ != "" && obsPJ != "[]" { // resolved_path is NULL on live (since #1287 relay data is diff --git a/cmd/server/eviction_resolved_pubkeys_test.go b/cmd/server/eviction_resolved_pubkeys_test.go new file mode 100644 index 000000000..078ca351f --- /dev/null +++ b/cmd/server/eviction_resolved_pubkeys_test.go @@ -0,0 +1,54 @@ +package main + +import ( + "fmt" + "runtime" + "testing" +) + +func evictionResolvedStore(t testing.TB, n int) (*PacketStore, []int) { + t.Helper() + db := setupTestDB(t) + t.Cleanup(func() { db.Close() }) + pk := func(i int) string { return fmt.Sprintf("%064x", i) } + ids := make([]int, 0, n) + for i := 1; i <= n; i++ { + if _, err := db.conn.Exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen) VALUES (?, '00', ?, '2026-01-01T00:00:00Z')`, i, fmt.Sprintf("h%015d", i)); err != nil { + t.Fatal(err) + } + for o := 0; o < 3; o++ { + rp := `["` + pk(i%9) + `",null,"` + pk(o+20) + `","` + pk(i%9) + `"]` + if _, err := db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp, resolved_path) VALUES (?, 1, '["aa","bb","cc","dd"]', 0, ?)`, i, rp); err != nil { + t.Fatal(err) + } + } + ids = append(ids, i) + } + return &PacketStore{db: db}, ids +} + +// Eviction looks up the resolved pubkeys of every packet it drops, once a +// minute in steady state. It decoded each resolved_path into a pointer per +// hop and built a de-dup map per packet: about 130MB of allocations per +// five minutes on a busy week. +func TestResolvedPubkeysForEvictionBatch(t *testing.T) { + s, ids := evictionResolvedStore(t, 300) + got := s.resolvedPubkeysForEvictionBatch(ids) + want := []string{fmt.Sprintf("%064x", 1), fmt.Sprintf("%064x", 20), fmt.Sprintf("%064x", 21), fmt.Sprintf("%064x", 22)} + if fmt.Sprint(got[1]) != fmt.Sprint(want) { + t.Fatalf("tx 1: %v, want %v (first-seen order, de-duplicated)", got[1], want) + } + if len(got) != 300 { + t.Fatalf("%d packets resolved, want 300", len(got)) + } + var a, b runtime.MemStats + runtime.GC() + runtime.ReadMemStats(&a) + _ = s.resolvedPubkeysForEvictionBatch(ids) + runtime.ReadMemStats(&b) + per := float64(b.Mallocs-a.Mallocs) / float64(len(ids)) + t.Logf("%.1f allocations per evicted packet (3 observations, 4 hops each)", per) + if per > 20 { + t.Fatalf("%.1f allocations per evicted packet, limit 20", per) + } +} diff --git a/cmd/server/neighbor_persist.go b/cmd/server/neighbor_persist.go index d73b972cf..21eca9aad 100644 --- a/cmd/server/neighbor_persist.go +++ b/cmd/server/neighbor_persist.go @@ -221,3 +221,102 @@ func unmarshalResolvedPath(s string) []*string { } return result } + +// resolvedPubkeysFromJSON returns the non-empty pubkeys of a resolved_path +// JSON array, exactly extractResolvedPubkeys(unmarshalResolvedPath(s)). The +// bulk loaders call it for most observations; the usual array of plain +// strings and nulls is read in place, one string per pubkey, instead of +// decoding a string and a pointer per hop. Anything else is decoded. +// +// Pubkeys go through internStr: relays repeat across nearly every +// observation, so a pubkey seen before costs no allocation. Interned keys +// live for the process, evicted nodes' keys included; the table is bounded +// (internMax), and past it a pubkey is returned uninterned. +func resolvedPubkeysFromJSON(s string) []string { + if s == "" { + return nil + } + if out, ok := scanResolvedPath(s); ok { + return out + } + return extractResolvedPubkeys(unmarshalResolvedPath(s)) +} + +// scanResolvedPath reads s as a JSON array of escape-free strings and nulls +// with JSON whitespace between tokens. ok is false for anything else. +func scanResolvedPath(s string) (out []string, ok bool) { + i := 0 + ws := func() { + for i < len(s) && (s[i] == ' ' || s[i] == '\t' || s[i] == '\n' || s[i] == '\r') { + i++ + } + } + ws() + if i >= len(s) { + return nil, false + } + if strings.HasPrefix(s[i:], "null") { + // unmarshalResolvedPath gives nil for a bare null. + i += 4 + ws() + return nil, i == len(s) + } + if s[i] != '[' { + return nil, false + } + i++ + ws() + if i < len(s) && s[i] == ']' { + i++ + ws() + return nil, i == len(s) + } + for { + switch { + case i < len(s) && s[i] == '"': + i++ + start := i + for i < len(s) && s[i] != '"' { + if s[i] == '\\' || s[i] < 0x20 || s[i] >= 0x80 { + return nil, false + } + i++ + } + if i >= len(s) { + return nil, false + } + if i > start { + if out == nil { + out = make([]string, 0, 8) + } + out = append(out, internStr(s[start:i])) + } + i++ + case strings.HasPrefix(s[i:], "null"): + i += 4 + default: + return nil, false + } + ws() + if i >= len(s) { + return nil, false + } + switch s[i] { + case ',': + i++ + ws() + case ']': + i++ + ws() + if i != len(s) { + return nil, false + } + if out == nil { + out = []string{} + } + return out, true + default: + return nil, false + } + } +} diff --git a/cmd/server/resolved_index.go b/cmd/server/resolved_index.go index 5b0f05a72..940733dd5 100644 --- a/cmd/server/resolved_index.go +++ b/cmd/server/resolved_index.go @@ -516,38 +516,35 @@ func (s *PacketStore) resolvedPubkeysForEvictionBatch(txIDs []int) map[int][]str continue } + var txID int + var rpJSON sql.NullString for rows.Next() { - var txID int - var rpJSON sql.NullString if err := rows.Scan(&txID, &rpJSON); err != nil { continue } if !rpJSON.Valid || rpJSON.String == "" { continue } - rp := unmarshalResolvedPath(rpJSON.String) - for _, p := range rp { - if p != nil && *p != "" { - result[txID] = append(result[txID], *p) + // De-duplicate as we go: a packet has a handful of keys, so a + // scan of its list is cheaper than a map per packet. + pks := result[txID] + for _, pk := range resolvedPubkeysFromJSON(rpJSON.String) { + dup := false + for _, have := range pks { + if have == pk { + dup = true + break + } + } + if !dup { + pks = append(pks, pk) } } + result[txID] = pks } rows.Close() } - // Deduplicate per-txID - for txID, pks := range result { - seen := make(map[string]bool, len(pks)) - deduped := pks[:0] - for _, pk := range pks { - if !seen[pk] { - seen[pk] = true - deduped = append(deduped, pk) - } - } - result[txID] = deduped - } - return result } diff --git a/cmd/server/resolved_pubkeys_bench_test.go b/cmd/server/resolved_pubkeys_bench_test.go new file mode 100644 index 000000000..70adcb8bd --- /dev/null +++ b/cmd/server/resolved_pubkeys_bench_test.go @@ -0,0 +1,37 @@ +package main + +import ( + "fmt" + "strings" + "testing" +) + +// One resolved_path per observation: 5 relays and 3 unresolved hops, drawn +// from a pool of 2,000 relays, as on a busy instance. +func BenchmarkResolvedPubkeysFromJSON(b *testing.B) { + const pool = 2000 + paths := make([]string, 256) + for i := range paths { + hops := make([]string, 0, 8) + for h := 0; h < 5; h++ { + hops = append(hops, fmt.Sprintf(`"%064x"`, (i*7+h*131)%pool)) + } + hops = append(hops, "null", "null", "null") + paths[i] = "[" + strings.Join(hops, ",") + "]" + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + _ = resolvedPubkeysFromJSON(paths[i%len(paths)]) + } +} + +// An eviction batch of 1,000 packets with 3 observations each. +func BenchmarkResolvedPubkeysForEvictionBatch(b *testing.B) { + s, ids := evictionResolvedStore(b, 1000) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + _ = s.resolvedPubkeysForEvictionBatch(ids) + } +} diff --git a/cmd/server/resolved_pubkeys_json_test.go b/cmd/server/resolved_pubkeys_json_test.go new file mode 100644 index 000000000..10e1b689f --- /dev/null +++ b/cmd/server/resolved_pubkeys_json_test.go @@ -0,0 +1,52 @@ +package main + +import ( + "math/rand" + "strings" + "testing" +) + +// The bulk loaders decode each observation's resolved_path (a JSON array of +// pubkeys and nulls, ~420 chars on live data, set on most observations) +// only to collect its non-empty pubkeys. Decoding into []*string allocated +// a string and a pointer per hop plus two slices: about 210MB over one +// startup on a busy week. resolvedPubkeysFromJSON reads the usual array in +// place, and a pubkey it has seen before costs nothing: only the result is +// allocated. +func TestResolvedPubkeysFromJSONAllocations(t *testing.T) { + pk := strings.Repeat("ab", 32) + rp := `["` + pk + `",null,"` + strings.Repeat("cd", 32) + `",null,"` + strings.Repeat("ef", 32) + `"]` + _ = resolvedPubkeysFromJSON(rp) // first sight keeps one copy of each + if n := testing.AllocsPerRun(100, func() { _ = resolvedPubkeysFromJSON(rp) }); n > 1 { + t.Fatalf("resolvedPubkeysFromJSON allocates %.0f times for 3 known pubkeys, want at most 1 (the result)", n) + } +} + +// It returns exactly what decoding and extracting would, odd inputs included. +func TestResolvedPubkeysFromJSONMatchesDecoding(t *testing.T) { + cases := []string{ + "", "[]", "null", "[null]", `["aa"]`, `["aa",null,"bb"]`, ` [ "aa" , null ] `, `[""]`, `["",null,""]`, + `["aa",]`, `["aa"`, `["aa"]x`, `[1]`, `["aa",1]`, `[nul]`, `[nulll]`, `["a\"b"]`, `["é"]`, + `[["aa"]]`, `{"a":1}`, "[\"a\x01\"]", `["aa" "bb"]`, `[,]`, `[null,]`, "[\"\xff\"]", `[true]`, + } + rng := rand.New(rand.NewSource(2)) + alphabet := []string{`[`, `]`, `"`, `,`, ` `, `a`, `1`, `\`, `null`, `nul`, `{`, `:`} + for i := 0; i < 20000; i++ { + var b strings.Builder + for j := rng.Intn(12); j >= 0; j-- { + b.WriteString(alphabet[rng.Intn(len(alphabet))]) + } + cases = append(cases, b.String()) + } + for _, s := range cases { + got, want := resolvedPubkeysFromJSON(s), extractResolvedPubkeys(unmarshalResolvedPath(s)) + if len(got) != len(want) { + t.Fatalf("%q: got %q, want %q", s, got, want) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("%q: got %q, want %q", s, got, want) + } + } + } +} diff --git a/cmd/server/store.go b/cmd/server/store.go index 3c8ffbc18..ec7b73300 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -1074,8 +1074,7 @@ func (s *PacketStore) Load() error { // Decode-window: extract resolved pubkeys for index, don't store on struct. rpStr := nullStrVal(resolvedPathStr) if rpStr != "" { - rp := unmarshalResolvedPath(rpStr) - pks := extractResolvedPubkeys(rp) + pks := resolvedPubkeysFromJSON(rpStr) // Single point of truth — see indexResolvedPathHops doc + #1558. s.indexResolvedPathHops(tx, pks, hopsSeen) } else if relayPM != nil && obsPJ != "" && obsPJ != "[]" { @@ -1419,8 +1418,7 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { // they were heard via — Load() does the equivalent inline. rpStr := nullStrVal(resolvedPathStr) if rpStr != "" { - rp := unmarshalResolvedPath(rpStr) - pks := extractResolvedPubkeys(rp) + pks := resolvedPubkeysFromJSON(rpStr) accumulateDedup(localResolvedPKsByTx, localResolvedSeenByTx, txID, pks) } else if relayPM != nil && obsPJ != "" && obsPJ != "[]" { // resolved_path not persisted — reconstruct relay hops from