@@ -3,6 +3,7 @@ package fracmanager
33import (
44 "bufio"
55 "io"
6+ "math/rand"
67 "os"
78 "path/filepath"
89 "strconv"
@@ -11,112 +12,112 @@ import (
1112 "time"
1213
1314 insaneJSON "github.com/ozontech/insane-json"
15+ "github.com/pkg/profile"
1416 "github.com/stretchr/testify/assert"
1517
1618 "github.com/ozontech/seq-db/consts"
17- "github.com/ozontech/seq-db/disk"
1819 "github.com/ozontech/seq-db/frac"
1920 "github.com/ozontech/seq-db/seq"
2021 "github.com/ozontech/seq-db/tests/common"
2122)
2223
23- func fillActiveFraction (active * frac.Active , wg * sync. WaitGroup ) error {
24+ func fillActiveFraction (active * frac.Active ) error {
2425 const muliplier = 10
2526
2627 docRoot := insaneJSON .Spawn ()
2728 defer insaneJSON .Release (docRoot )
2829
29- dp := frac .NewDocProvider ()
30-
3130 file , err := os .Open (filepath .Join (common .TestDataDir , "k8s.logs" ))
3231 if err != nil {
3332 return err
3433 }
3534 defer file .Close ()
35+
36+ k := 0
37+ wg := sync.WaitGroup {}
38+ dp := frac .NewDocProvider ()
3639 for i := 0 ; i < muliplier ; i ++ {
3740 dp .TryReset ()
38- _ , err := file . Seek ( 0 , io . SeekStart )
39- if err != nil {
41+
42+ if _ , err := file . Seek ( 0 , io . SeekStart ); err != nil {
4043 return err
4144 }
45+
4246 scanner := bufio .NewScanner (file )
4347 for scanner .Scan () {
48+ k ++
4449 doc := scanner .Bytes ()
45- err := docRoot .DecodeBytes (doc )
46- if err != nil {
50+ if err := docRoot .DecodeBytes (doc ); err != nil {
4751 return err
4852 }
49- tokens := seq .Tokens ("_all_:" , "service:100500" , "k8s_pod:" + strconv .Itoa (i ))
50- dp .Append (doc , docRoot , seq .SimpleID (0 ), tokens )
53+
54+ id := seq .NewID (time .Now (), uint64 (rand .Int63 ()))
55+ dp .Append (doc , docRoot , id , seq .Tokens (
56+ "_all_:" ,
57+ "service:service" + strconv .Itoa (rand .Intn (200 )),
58+ "k8s_pod1:" + strconv .Itoa (k % 100000 ),
59+ "k8s_pod2:" + strconv .Itoa (k % 1000000 ),
60+ ))
5161 }
5262 docs , metas := dp .Provide ()
5363 wg .Add (1 )
54- if err := active .Append (docs , metas , wg ); err != nil {
64+ if err := active .Append (docs , metas , & wg ); err != nil {
5565 return err
5666 }
5767 }
5868
69+ wg .Wait ()
5970 return nil
6071}
6172
62- func getCacheMaintainer () (* CacheMaintainer , func ()) {
63- done := make (chan struct {})
64- cm := NewCacheMaintainer (32 * consts .MB , 24 * consts .MB , nil )
65- wg := cm .RunCleanLoop (done , time .Second , time .Second )
66- return cm , func () {
67- close (done )
68- wg .Wait ()
73+ func defaultSealingParams () frac.SealParams {
74+ const minZstdLevel = 1
75+ return frac.SealParams {
76+ IDsZstdLevel : minZstdLevel ,
77+ LIDsZstdLevel : minZstdLevel ,
78+ TokenListZstdLevel : minZstdLevel ,
79+ DocsPositionsZstdLevel : minZstdLevel ,
80+ TokenTableZstdLevel : minZstdLevel ,
81+ DocBlocksZstdLevel : minZstdLevel ,
82+ DocBlockSize : 128 * consts .KB ,
6983 }
7084}
7185
72- func BenchmarkSealing (b * testing.B ) {
73- b .ResetTimer ()
74- b .StopTimer ()
75- b .ReportAllocs ()
86+ func Benchmark_SealingNoSort (b * testing.B ) {
87+ runSealingBench (b , & frac.Config {SkipSortDocs : true })
88+ }
89+
90+ func Benchmark_SealingWithSort (b * testing.B ) {
91+ runSealingBench (b , & frac.Config {})
92+ }
7693
77- cm , stopFn := getCacheMaintainer ()
78- defer stopFn ()
94+ func runSealingBench (b * testing.B , cfg * frac.Config ) {
95+ cm := NewCacheMaintainer (consts .MB * 64 , consts .MB * 64 , nil )
96+ fp := newFractionProvider (cfg , cm , 1 , 1 )
97+ defer fp .Stop ()
7998
8099 dataDir := filepath .Join (b .TempDir (), "BenchmarkSealing" )
81100 common .RecreateDir (dataDir )
82101
83- readLimiter := disk . NewReadLimiter ( 1 , nil )
84-
85- activeIndexer := frac . NewActiveIndexer ( 10 , 10 )
102+ active := fp . NewActive ( filepath . Join ( dataDir , "test" ) )
103+ err := fillActiveFraction ( active )
104+ assert . NoError ( b , err )
86105
87- activeIndexer .Start ()
88- defer activeIndexer .Stop ()
106+ params := defaultSealingParams ()
107+ // The first sealing will sort all the LIDs, so we take this load out of the measurement range
108+ _ , err = frac .Seal (active , params )
109+ assert .NoError (b , err )
89110
90- const minZstdLevel = - 5
91- defaultSealParams := frac.SealParams {
92- IDsZstdLevel : minZstdLevel ,
93- LIDsZstdLevel : minZstdLevel ,
94- TokenListZstdLevel : minZstdLevel ,
95- DocsPositionsZstdLevel : minZstdLevel ,
96- TokenTableZstdLevel : minZstdLevel ,
97- DocBlocksZstdLevel : minZstdLevel ,
98- DocBlockSize : consts .MB * 4 ,
99- }
100- for i := 0 ; i < b .N ; i ++ {
101- wg := sync.WaitGroup {}
102- active := frac .NewActive (
103- filepath .Join (dataDir , "test_" + strconv .Itoa (i )),
104- activeIndexer ,
105- readLimiter ,
106- cm .CreateDocBlockCache (),
107- cm .CreateSortDocsCache (),
108- & frac.Config {},
109- )
110- err := fillActiveFraction (active , & wg )
111- assert .NoError (b , err )
111+ b .ReportAllocs ()
112112
113- wg .Wait ()
114- active .GetAllDocuments () // emulate search-pre-sorted LIDs
113+ defer profile .Start ( // turn on profiling (look for the file fracmanager/cpu.pprof)
114+ profile .CPUProfile ,
115+ profile .ProfilePath ("." ),
116+ profile .NoShutdownHook ,
117+ ).Stop ()
115118
116- b . StartTimer ()
117- _ , err = frac .Seal (active , defaultSealParams )
119+ for b . Loop () {
120+ _ , err = frac .Seal (active , params )
118121 assert .NoError (b , err )
119-
120- b .StopTimer ()
121122 }
122123}
0 commit comments