summaryrefslogtreecommitdiffhomepage
path: root/syncs/shardedint.go
blob: 28c4168d54c79a8a21343e45de8ec449b8674ab9 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
// Copyright (c) Tailscale Inc & AUTHORS
// SPDX-License-Identifier: BSD-3-Clause

package syncs

import (
	"encoding/json"
	"strconv"
	"sync/atomic"

	"golang.org/x/sys/cpu"
)

// ShardedInt provides a sharded atomic int64 value that optimizes high
// frequency (Mhz range and above) writes in highly parallel workloads.
// The zero value is not safe for use; use [NewShardedInt].
// ShardedInt implements the expvar.Var interface.
type ShardedInt struct {
	sv *ShardValue[intShard]
}

// NewShardedInt returns a new [ShardedInt].
func NewShardedInt() *ShardedInt {
	return &ShardedInt{
		sv: NewShardValue[intShard](),
	}
}

// Add adds delta to the value.
func (m *ShardedInt) Add(delta int64) {
	m.sv.One(func(v *intShard) {
		v.Add(delta)
	})
}

type intShard struct {
	atomic.Int64
	_ cpu.CacheLinePad // avoid false sharing of neighboring shards
}

// Value returns the current value.
func (m *ShardedInt) Value() int64 {
	var v int64
	for s := range m.sv.All {
		v += s.Load()
	}
	return v
}

// GetDistribution returns the current value in each shard.
// This is intended for observability/debugging only.
func (m *ShardedInt) GetDistribution() []int64 {
	v := make([]int64, 0, m.sv.Len())
	for s := range m.sv.All {
		v = append(v, s.Load())
	}
	return v
}

// String implements the expvar.Var interface
func (m *ShardedInt) String() string {
	v, _ := json.Marshal(m.Value())
	return string(v)
}

// AppendText implements the encoding.TextAppender interface
func (m *ShardedInt) AppendText(b []byte) ([]byte, error) {
	return strconv.AppendInt(b, m.Value(), 10), nil
}