go.temporal.io/server/tools/fairsim/sim.go
611 LOC · 125 covered · 486 uncovered · 26 ranges · 7 concepts · 3 introducers · 6 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
//nolint:errcheck // don't need to check fmt.Fprintf
2
package fairsim
3
4
import (
5
"bufio"
6
"cmp"
7
"container/heap"
8
"encoding/json"
9
"flag"
10
"fmt"
11
"io"
12
"math/rand/v2"
13
"os"
14
"slices"
15
"strings"
16
17
"go.temporal.io/server/service/matching/counter"
18
)
19
20
// matches service/matching.strideFactor, but we don't want to export that
21
const defaultStrideFactor = 1000
22
23
type (
24
task struct {
25
pri int
26
fkey string
27
fweight float32
28
pass int64
29
index int64
30
payload string
31
}
32
33
state struct {
34
rnd *rand.Rand
35
counterFactory func() counter.Counter
36
partitions []partitionState
37
strideFactor float32
38
}
39
40
simulator struct {
41
state *state
42
stats *latencyStats
43
w io.Writer
44
verbose bool
45
nextIndex int64
46
dispatchIndex int64
47
defaultPriority int
48
}
49
50
latencyStats struct {
51
byKey map[string][]int64 // latencies by fairness key
52
byKeyNormalized map[string][]float64 // normalized latencies by fairness key
53
overall []int64 // all latencies
54
overallNormalized []float64 // all normalized latencies
55
}
56
57
partitionState struct {
58
perPri map[int]perPriState
59
heap taskHeap
60
}
61
62
taskHeap []*task
63
64
perPriState struct {
65
c counter.Counter
66
}
67
68
unfairCounter struct{}
69
)
70
71
// parseFlags creates a FlagSet, calls setup to register flags, and parses args.
72
// Returns remaining (non-flag) arguments.
73
func parseFlags(name string, args []string, setup func(*flag.FlagSet)) ([]string, error) {
74
fs := flag.NewFlagSet(name, flag.ContinueOnError)
75
var flagErrors strings.Builder
76
fs.SetOutput(&flagErrors)
77
setup(fs)
78
if err := fs.Parse(args); err != nil {
79
if flagErrors.Len() > 0 {
80
return nil, fmt.Errorf("%s: %w\n%s", name, err, flagErrors.String())
81
}
82
return nil, fmt.Errorf("%s: %w", name, err)
83
}
84
return fs.Args(), nil
85
}
86
87
func RunTool(args []string) error {
88
var (
89
seed *int64
90
fair *bool
91
partitions *int
92
strideFactor *int
93
counterFile *string
94
scriptFile *string
95
verbose *bool
96
)
97
remainingArgs, err := parseFlags("fairsim", args, func(fs *flag.FlagSet) {
98
seed = fs.Int64("seed", rand.Int64(), "Random seed")
99
fair = fs.Bool("fair", true, "Enable fairness (false for FIFO)")
100
partitions = fs.Int("partitions", 4, "Number of partitions")
101
strideFactor = fs.Int("strideFactor", defaultStrideFactor, "Stride factor")
102
counterFile = fs.String("counter-params", "", "JSON file with CounterParams")
103
scriptFile = fs.String("script", "", "Script file to execute instead of generating tasks")
104
verbose = fs.Bool("verbose", false, "verbose output")
105
})
106
if err != nil {
107
return err
108
}
109
110
// Load counter params
111
var params counter.CounterParams
112
if *counterFile == "" {
113
params = counter.DefaultCounterParams
114
} else {
115
data, err := os.ReadFile(*counterFile)
116
if err != nil {
117
return fmt.Errorf("failed to load counter params: %w", err)
118
} else if err = json.Unmarshal(data, ¶ms); err != nil {
119
return fmt.Errorf("failed to load counter params: %w", err)
120
}
121
fmt.Printf("Using counter params: %#v\n\n", params)
122
}
123
124
src := rand.NewPCG(uint64(*seed), uint64(*seed+1))
125
rnd := rand.New(src)
126
127
counterFactory := func() counter.Counter { return unfairCounter{} }
128
if *fair {
129
counterFactory = func() counter.Counter { return counter.NewHybridCounter(params, src) }
130
}
131
132
st := newState(rnd, counterFactory, *partitions, *strideFactor)
133
stats := newLatencyStats()
134
135
const defaultPriority = 3
136
sim := newSimulator(st, stats, defaultPriority, os.Stdout, *verbose)
137
138
// Check if script mode
139
if *scriptFile != "" {
140
return sim.runScript(*scriptFile)
141
}
142
143
// Default behavior: run gentasks command with remaining args from command line
144
if err := sim.executeGenTasksCommand(remainingArgs); err != nil {
145
return err
146
}
147
148
sim.finish()
149
return nil
150
}
151
153
>
return &latencyStats{
154
>
byKey: make(map[string][]int64),
155
>
byKeyNormalized: make(map[string][]float64),
156
>
}
157
>
}
158
159
>
func newState(rnd *rand.Rand, counterFactory func() counter.Counter, partitions, strideFactor int) *state {
sim.go ×20
160
>
return &state{
161
>
rnd: rnd,
162
>
counterFactory: counterFactory,
163
>
partitions: make([]partitionState, partitions),
164
>
strideFactor: float32(strideFactor),
165
>
}
166
>
}
167
168
>
func newSimulator(state *state, stats *latencyStats, defaultPriority int, w io.Writer, verbose bool) *simulator {
sim.go ×20
169
>
return &simulator{
170
>
state: state,
171
>
stats: stats,
172
>
w: w,
173
>
verbose: verbose,
174
>
defaultPriority: defaultPriority,
175
>
}
176
>
}
177
178
// addTask adds a task to the simulator, assigning defaults and an index.
180
>
t.pri = cmp.Or(t.pri, sim.defaultPriority)
181
>
t.fweight = cmp.Or(t.fweight, 1.0)
182
>
t.index = sim.nextIndex
183
>
sim.nextIndex++
184
>
sim.state.addTask(t)
185
>
}
186
187
// processTask records stats for a dispatched task and returns the latency.
189
>
latency := sim.dispatchIndex - t.index
190
>
sim.stats.byKey[t.fkey] = append(sim.stats.byKey[t.fkey], latency)
191
>
sim.stats.overall = append(sim.stats.overall, latency)
192
>
sim.dispatchIndex++
193
>
return latency
194
>
}
195
196
// printTask writes a single task's dispatch info to the writer.
198
>
if !sim.verbose {
199
>
return
200
>
}
201
fmt.Fprintf(sim.w, "task idx:%6d dsp:%6d lat:%6d pri:%2d fkey:%10q fweight:%3g part:%2d payload:%q\n",
202
t.index, sim.dispatchIndex-1, latency, t.pri, t.fkey, t.fweight, partition, t.payload)
203
}
204
205
// drainTasks pops and processes all remaining tasks, printing each one.
207
>
for t, partition := sim.state.popTask(); t != nil; t, partition = sim.state.popTask() {
208
>
latency := sim.processTask(t)
209
>
sim.printTask(t, partition, latency)
210
>
}
211
}
212
213
// finish drains remaining tasks, calculates normalized stats, and prints stats.
214
func (sim *simulator) finish() {
215
sim.drainTasks()
216
sim.stats.calculateNormalized()
217
sim.stats.fprint(sim.w, sim.verbose)
218
}
219
220
func (sim *simulator) runScript(scriptFile string) error {
221
file, err := os.Open(scriptFile)
222
if err != nil {
223
return fmt.Errorf("failed to open script file: %w", err)
224
}
225
defer file.Close()
226
227
var commands []string
228
scanner := bufio.NewScanner(file)
229
for scanner.Scan() {
230
line := strings.TrimSpace(scanner.Text())
231
if line == "" || strings.HasPrefix(line, "#") {
232
continue
233
}
234
commands = append(commands, line)
235
}
236
if err := scanner.Err(); err != nil {
237
return fmt.Errorf("error reading script file: %w", err)
238
}
239
240
return sim.runCommands(commands)
241
}
242
243
func (sim *simulator) runCommands(commands []string) error {
244
for _, cmd := range commands {
245
if err := sim.executeCommand(cmd); err != nil {
246
return fmt.Errorf("error executing command %q: %w", cmd, err)
247
}
248
}
249
sim.finish()
250
return nil
251
}
252
253
func (sim *simulator) executeCommand(line string) error {
254
parts := strings.Fields(line)
255
if len(parts) == 0 {
256
return nil
257
}
258
259
switch parts[0] {
260
case "task":
261
return sim.executeTaskCommand(parts[1:])
262
case "poll":
263
return sim.executePollCommand()
264
case "stats":
265
return sim.executeStatsCommand()
266
case "clearstats":
267
return sim.executeClearStatsCommand()
268
case "gentasks":
269
return sim.executeGenTasksCommand(parts[1:])
270
default:
271
return fmt.Errorf("unknown command: %q", parts[0])
272
}
273
}
274
275
func (sim *simulator) executeTaskCommand(args []string) error {
276
var (
277
fkey *string
278
fweight *float64
279
pri *int
280
payload *string
281
)
282
if _, err := parseFlags("task", args, func(fs *flag.FlagSet) {
283
fkey = fs.String("fkey", "", "fairness key")
284
fweight = fs.Float64("fweight", 1.0, "fairness weight")
285
pri = fs.Int("pri", sim.defaultPriority, "priority")
286
payload = fs.String("payload", "", "payload")
287
}); err != nil {
288
return err
289
}
290
291
sim.addTask(&task{
292
fkey: *fkey,
293
fweight: float32(*fweight),
294
pri: *pri,
295
payload: *payload,
296
})
297
return nil
298
}
299
300
func (sim *simulator) executePollCommand() error {
301
t, partition := sim.state.popTask()
302
if t == nil {
303
fmt.Fprintln(sim.w, "No tasks in queue")
304
return nil
305
}
306
latency := sim.processTask(t)
307
sim.printTask(t, partition, latency)
308
return nil
309
}
310
311
func (sim *simulator) executeStatsCommand() error {
312
sim.stats.calculateNormalized()
313
sim.stats.fprint(sim.w, sim.verbose)
314
return nil
315
}
316
317
func (sim *simulator) executeClearStatsCommand() error {
318
sim.stats = newLatencyStats()
319
return nil
320
}
321
322
func (sim *simulator) executeGenTasksCommand(args []string) error {
323
var (
324
tasks *int
325
keys *int
326
keyprefix *string
327
zipfS *float64
328
zipfV *float64
329
)
330
if _, err := parseFlags("gentasks", args, func(fs *flag.FlagSet) {
331
tasks = fs.Int("tasks", 100, "number of tasks to generate")
332
keys = fs.Int("keys", 10, "number of unique fairness keys")
333
keyprefix = fs.String("keyprefix", "key", "prefix for generated fairness keys")
334
zipfS = fs.Float64("zipf-s", 2.0, "zipf distribution s parameter")
335
zipfV = fs.Float64("zipf-v", 2.0, "zipf distribution v parameter")
336
}); err != nil {
337
return err
338
}
339
340
if *tasks <= 0 {
341
return fmt.Errorf("tasks must be positive, got %d", *tasks)
342
}
343
if *keys <= 0 {
344
return fmt.Errorf("keys must be positive, got %d", *keys)
345
}
346
347
zipf := rand.NewZipf(sim.state.rnd, *zipfS, *zipfV, uint64(*keys-1))
348
for range *tasks {
349
sim.addTask(&task{
350
fkey: fmt.Sprintf("%s%d", *keyprefix, zipf.Uint64()),
351
})
352
}
353
return nil
354
}
355
356
// --- state methods ---
357
358
// addTask adds a task to a random partition, computing its pass via the counter.
360
>
partition := &s.partitions[s.rnd.IntN(len(s.partitions))]
361
>
362
>
if partition.perPri == nil {
363
>
partition.perPri = make(map[int]perPriState)
364
>
}
365
367
>
if !exists {
368
>
priState = perPriState{c: s.counterFactory()}
369
>
partition.perPri[t.pri] = priState
370
>
}
371
373
>
heap.Push(&partition.heap, t)
374
}
375
376
// popTask returns the task with minimum (pri, pass, index) from a random partition.
378
>
for _, idx := range s.rnd.Perm(len(s.partitions)) {
379
>
partition := &s.partitions[idx]
380
>
if partition.heap.Len() > 0 {
381
>
t := heap.Pop(&partition.heap).(*task) //nolint:revive
382
>
return t, idx
383
>
}
384
}
386
}
387
388
// --- unfairCounter (FIFO mode) ---
389
390
func (u unfairCounter) GetPass(key string, base int64, inc int64) int64 { return base }
391
func (u unfairCounter) EstimateDistinctKeys() int { return 0 }
392
func (u unfairCounter) TopK() []counter.TopKEntry { return nil }
393
394
// --- taskHeap (heap.Interface) ---
395
397
399
>
if h[i].pri != h[j].pri {
400
return h[i].pri < h[j].pri
401
}
403
>
return h[i].pass < h[j].pass
404
>
}
405
>
return h[i].index < h[j].index
406
}
407
409
411
413
>
old := *h
414
>
n := len(old)
415
>
item := old[n-1]
416
>
*h = old[:n-1]
417
>
return item
418
>
}
419
420
// --- latencyStats ---
421
423
>
totalTasks := len(stats.overall)
424
>
totalKeys := len(stats.byKey)
425
>
if totalTasks == 0 || totalKeys == 0 {
426
return
427
}
428
// Normalized latency: raw_displacement / count_for_key * total_keys / total_tasks.
429
// Dividing by count_for_key adjusts for volume (high-volume keys naturally have larger
430
// displacements). The total_keys/total_tasks factor further scales by the key's share
431
// of total traffic, making values comparable across different workload sizes.
432
// Values center on 0: negative = expedited, positive = delayed.
434
>
stats.overallNormalized = stats.overallNormalized[:0]
435
>
for key, latencies := range stats.byKey {
436
>
taskCount := float64(len(latencies))
437
>
normalizedLatencies := make([]float64, len(latencies))
438
>
for i, rawLatency := range latencies {
439
>
normalizedLatencies[i] = float64(rawLatency) / taskCount * fairShareFactor
440
>
stats.overallNormalized = append(stats.overallNormalized, normalizedLatencies[i])
441
>
}
442
>
stats.byKeyNormalized[key] = normalizedLatencies
443
}
444
}
445
446
func (stats *latencyStats) fprint(w io.Writer, verbose bool) {
447
if verbose {
448
if len(stats.overall) > 0 {
449
slices.Sort(stats.overall)
450
mean := float64(sum(stats.overall)) / float64(len(stats.overall))
451
median := stats.overall[len(stats.overall)/2]
452
p95 := stats.overall[int(float64(len(stats.overall))*0.95)]
453
454
fmt.Fprint(w, "\n=== Raw Latency Statistics ===\n")
455
fmt.Fprintf(w, "Overall: mean=%.2f, median=%d, p95=%d, min=%d, max=%d\n",
456
mean, median, p95, stats.overall[0], stats.overall[len(stats.overall)-1])
457
}
458
459
if len(stats.overallNormalized) > 0 {
460
slices.Sort(stats.overallNormalized)
461
mean := sum(stats.overallNormalized) / float64(len(stats.overallNormalized))
462
median := stats.overallNormalized[len(stats.overallNormalized)/2]
463
p95 := stats.overallNormalized[int(float64(len(stats.overallNormalized))*0.95)]
464
465
fmt.Fprint(w, "\n=== Normalized Latency Statistics ===\n")
466
fmt.Fprintf(w, "Overall: mean=%.4f, median=%.4f, p95=%.4f, min=%.4f, max=%.4f\n",
467
mean, median, p95, stats.overallNormalized[0], stats.overallNormalized[len(stats.overallNormalized)-1])
468
}
469
470
type keyStats struct {
471
key string
472
meanRaw float64
473
medianRaw int64
474
meanNormalized float64
475
medianNormalized float64
476
count int
477
}
478
479
var keyStatsList []keyStats
480
for key, latencies := range stats.byKey {
481
if len(latencies) == 0 {
482
continue
483
}
484
485
slices.Sort(latencies)
486
meanRaw := float64(sum(latencies)) / float64(len(latencies))
487
medianRaw := latencies[len(latencies)/2]
488
489
normalizedLatencies := stats.byKeyNormalized[key]
490
slices.Sort(normalizedLatencies)
491
meanNormalized := sum(normalizedLatencies) / float64(len(normalizedLatencies))
492
medianNormalized := normalizedLatencies[len(normalizedLatencies)/2]
493
494
keyStatsList = append(keyStatsList, keyStats{
495
key: key,
496
meanRaw: meanRaw,
497
medianRaw: medianRaw,
498
meanNormalized: meanNormalized,
499
medianNormalized: medianNormalized,
500
count: len(latencies),
501
})
502
}
503
504
slices.SortFunc(keyStatsList, func(a, b keyStats) int { return cmp.Compare(a.medianNormalized, b.medianNormalized) })
505
506
fmt.Fprint(w, "\nPer-key stats (sorted by median normalized latency):\n")
507
for _, ks := range keyStatsList {
508
fmt.Fprintf(w, " %s: raw(mean=%.2f, median=%d) norm(mean=%.4f, median=%.4f) count=%d\n",
509
ks.key, ks.meanRaw, ks.medianRaw, ks.meanNormalized, ks.medianNormalized, ks.count)
510
}
511
}
512
513
ps := []float64{20, 50, 80, 90, 95}
514
515
fmt.Fprint(w, "\nRaw fairness metrics (percentile of per-key percentiles):\n")
516
fmt.Fprintf(w, " @%2.0f @%2.0f @%2.0f @%2.0f @%2.0f\n",
517
ps[0], ps[1], ps[2], ps[3], ps[4])
518
for _, p := range ps {
519
pofps := percentileOfPercentiles(stats.byKey, p, ps)
520
fmt.Fprintf(w, " p%2.0fs: %7.0f %7.0f %7.0f %7.0f %7.0f\n",
521
p, pofps[0], pofps[1], pofps[2], pofps[3], pofps[4])
522
}
523
524
fmt.Fprint(w, "\nNormalized fairness metrics (percentile of per-key percentiles):\n")
525
fmt.Fprintf(w, " @%2.0f @%2.0f @%2.0f @%2.0f @%2.0f\n",
526
ps[0], ps[1], ps[2], ps[3], ps[4])
527
for _, p := range ps {
528
pofps := percentileOfPercentiles(stats.byKeyNormalized, p, ps)
529
fmt.Fprintf(w, " p%2.0fs: %7.3f %7.3f %7.3f %7.3f %7.3f\n",
530
p, pofps[0], pofps[1], pofps[2], pofps[3], pofps[4])
531
}
532
}
533
534
// --- latencyStats query methods (for testing) ---
535
536
func (stats *latencyStats) meanNormalized(key string) float64 {
537
values := stats.byKeyNormalized[key]
538
if len(values) == 0 {
539
return 0
540
}
541
return sum(values) / float64(len(values))
542
}
543
545
>
values := stats.byKey[key]
546
>
if len(values) == 0 {
547
return 0
548
}
550
>
slices.Sort(sorted)
551
>
idx := int(p / 100.0 * float64(len(sorted)-1))
552
>
return float64(sorted[idx])
553
}
554
555
func (stats *latencyStats) overallPercentile(p float64) float64 {
556
if len(stats.overall) == 0 {
557
return 0
558
}
559
sorted := slices.Clone(stats.overall)
560
slices.Sort(sorted)
561
idx := int(p / 100.0 * float64(len(sorted)-1))
562
return float64(sorted[idx])
563
}
564
565
func (stats *latencyStats) keyCount() int {
566
return len(stats.byKey)
567
}
568
569
func (stats *latencyStats) taskCount(key string) int {
570
return len(stats.byKey[key])
571
}
572
573
// --- generic helpers ---
574
575
func sum[T int64 | float64](slice []T) T {
576
var total T
577
for _, v := range slice {
578
total += v
579
}
580
return total
581
}
582
583
>
func percentileOfPercentiles[T int64 | float64](dataByKey map[string][]T, keyPercentile float64, crossPercentile []float64) []float64 {
sim.go ×4
584
>
var keyPercentiles []float64
585
>
586
>
for _, values := range dataByKey {
587
>
if len(values) == 0 {
588
continue
589
}
591
>
for i, v := range values {
592
>
sorted[i] = float64(v)
593
>
}
594
>
slices.Sort(sorted)
595
>
idx := int(keyPercentile / 100.0 * float64(len(sorted)-1))
596
>
keyPercentiles = append(keyPercentiles, sorted[idx])
597
}
598
600
return nil
601
}
602
604
>
605
>
out := make([]float64, len(crossPercentile))
606
>
for i, p := range crossPercentile {
607
>
idx := int(p / 100.0 * float64(len(keyPercentiles)-1))
608
>
out[i] = keyPercentiles[idx]
609
>
}
610
>
return out
611
}