~james/core

core/spindle/nixcache/usage.go -rw-r--r-- 2.3 KiB
452f17b7 — james chore(import): snapshot james.dad.town/core at f70c71c7696a 14 hours ago
                                                                                
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
package nixcache

import (
	"context"
	"log/slog"
	"maps"
	"slices"
	"sync"
	"time"
)

const (
	useFlushInterval = 10 * time.Second
	useMaxPending    = 4096
	useFlushTimeout  = 10 * time.Second
)

type UseRecordFunc func(ctx context.Context, cache string, storeHashes []string) error

type UseBatcher struct {
	cache  string
	record UseRecordFunc
	logger *slog.Logger
	max    int

	mu      sync.Mutex
	pending map[string]struct{}
	dropped int

	stop chan struct{}
	done chan struct{}
	once sync.Once
}

func NewUseBatcher(cache string, record UseRecordFunc, logger *slog.Logger) *UseBatcher {
	return newUseBatcher(cache, record, logger, useFlushInterval, useMaxPending)
}

func newUseBatcher(cache string, record UseRecordFunc, logger *slog.Logger, interval time.Duration, maxPending int) *UseBatcher {
	if logger == nil {
		logger = slog.Default()
	}
	b := &UseBatcher{
		cache:   cache,
		record:  record,
		logger:  logger.With("where", "nix_use", "cache", cache),
		max:     maxPending,
		pending: make(map[string]struct{}),
		stop:    make(chan struct{}),
		done:    make(chan struct{}),
	}
	go b.loop(interval)
	return b
}

// Add drops the hash past the pending cap, and only counts it for the log.
func (b *UseBatcher) Add(storeHash string) {
	if b == nil {
		return
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	if _, ok := b.pending[storeHash]; ok {
		return
	}
	if len(b.pending) >= b.max {
		b.dropped++
		return
	}
	b.pending[storeHash] = struct{}{}
}

func (b *UseBatcher) Close() error {
	if b == nil {
		return nil
	}
	b.once.Do(func() { close(b.stop) })
	<-b.done
	return nil
}

func (b *UseBatcher) loop(interval time.Duration) {
	defer close(b.done)
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			b.flush()
		case <-b.stop:
			b.flush()
			return
		}
	}
}

func (b *UseBatcher) flush() {
	b.mu.Lock()
	hashes := slices.Collect(maps.Keys(b.pending))
	clear(b.pending)
	dropped := b.dropped
	b.dropped = 0
	b.mu.Unlock()

	if dropped > 0 {
		b.logger.Debug("dropped nix use reports past the pending cap", "dropped", dropped)
	}
	if len(hashes) == 0 {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), useFlushTimeout)
	defer cancel()
	if err := b.record(ctx, b.cache, hashes); err != nil {
		b.logger.Debug("report nix use failed", "hashes", len(hashes), "error", err)
	}
}