usagetracker: skip groups with nothing to remove during Cleanup

CHANGELOG.md

@@ -8,6 +8,7 @@
 * [CHANGE] Removed the following deprecated config: `-querier.filter-queryables-enabled`, `-query-frontend.cache-samples-processed-stats`, `-ingest-storage.kafka.write-clients`, `-blocks-storage.tsdb.head-postings-for-matchers-cache-size`, `-blocks-storage.tsdb.block-postings-for-matchers-cache-size`. #16352
 * [CHANGE] The `bucket` label of the `thanos_objstore_bucket_*` metrics, previously always empty, is now set to the name of the bucket the metrics refer to. This lets a component that accesses more than one bucket report each of them separately. The `thanos_store_bucket_cache_*` and `cortex_bucket_index_load*` metrics gained a `bucket` label carrying the same bucket name, for the same reason. #16265
 * [CHANGE] Compactor: Stabilize `-compactor.first-level-compaction-skip-future-max-time` to `true` and `-compactor.first-level-compaction-ooo-wait-period` to 5 minutes, both of which have been shown to improve batching during the split phase and reduce the total volume of L2 blocks in deployments with lots of out-of-order writes. #16464
+* [ENHANCEMENT] Usage-tracker: Speed up the periodic removal of idle series. The scan that locates expired series now runs a whole group of slots at a time, using SSE2 on amd64 and NEON on arm64, so groups holding nothing to remove are skipped without inspecting each slot. This shortens the time a shard mutex is held during cleanup, which was showing up as latency spikes on series tracking in large partitions. Build with `-tags nosimd` to use the portable implementation. #16473
 * [ENHANCEMENT] Compactor: Add the experimental `-compactor.block-health-validation-concurrency` option to limit how many blocks are validated concurrently within a compaction job. #16269
 * [ENHANCEMENT] Query-frontend: Improve the stability of cardinality estimates and therefore sharding factors for queries when running splitting and caching inside MQE is enabled, or range vector splitting is enabled. #16274 #16301 #16305 #16311
   * When running splitting and caching inside MQE is enabled, the `cortex_query_frontend_cardinality_estimation_difference` metric will no longer be emitted.

pkg/usagetracker/clock/minutes.go

@@ -22,6 +22,10 @@ func ToMinutes(t time.Time) Minutes {
 // This value only makes sense within the last hour.
 type Minutes uint8
 
+// Cycle is the number of values Minutes takes before wrapping back to zero.
+// ToMinutes always returns a value below it, and comparisons are only defined for such values.
+const Cycle Minutes = 2 * 60
+
 // GreaterThan returns true if this value is greater than other on a four-hour clock face assuming that none of the values is ever older than 1h.
 func (m Minutes) GreaterThan(other Minutes) bool {
 	if m > other {

pkg/usagetracker/tenantshard/expiry.go

@@ -0,0 +1,106 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package tenantshard
+
+import (
+	"math/bits"
+	"unsafe"
+
+	"github.com/grafana/mimir/pkg/usagetracker/clock"
+)
+
+// Finding the entries to evict during Cleanup used to be a per-slot loop with three
+// unpredictable branches per slot, which dominated the cost of Cleanup. This file turns
+// that test into a branchless, whole-group operation, so that groups holding nothing to
+// evict are skipped without ever branching per slot.
+//
+// # Turning the clock comparison into a range check
+//
+// An entry is evicted when watermark.GreaterOrEqualThan(value) holds, where value is
+// clock.Minutes in [0, 120) and watermark is also in [0, 120). Expanding
+// clock.Minutes.GreaterThan, that condition is (watermark-value) mod 120 < 60, so the
+// values that expire are the 60 consecutive minutes ending at the watermark, wrapping
+// around the 120 minute clock face.
+//
+// Slots hold the value xor-ed (see xorData), that is x = 255-value, so the expiring
+// values map to the byte range starting at lo = 255-watermark and running upwards,
+// wrapping at 256. Two cases:
+//
+//   - watermark >= 59: the minutes do not wrap, and x lands in [lo, lo+59], entirely
+//     inside [136, 255].
+//   - watermark < 59: the minutes wrap, and x lands in [lo, 255] plus [136, 194-watermark].
+//     Going up from lo and wrapping at 256, those two pieces are joined by [0, 135], which
+//     holds no valid value, so the whole thing is still one range: [lo, lo+195].
+//
+// Both cases are therefore "uint8(x-lo) <= length", differing only in length. The second
+// case sweeps up the empty (0) and tombstone (1) markers as a side effect, so the result
+// is always intersected with a separate occupied test.
+//
+// # Domain
+//
+// This is exact for values and watermarks in [0, 120), which is what clock.ToMinutes
+// produces. Snapshots are the only path that can introduce a byte outside that range, and
+// loadSnapshot rejects those before they reach the map.
+
+// expiredRange returns the range of xorData byte values that Cleanup must evict for the
+// given watermark: a slot expires when uint8(x-lo) <= length and the slot is occupied.
+func expiredRange(watermark clock.Minutes) (lo, length uint8) {
+	lo = ^uint8(watermark) // 255 - watermark
+	if watermark >= 59 {
+		// The 60 minute window ending at the watermark does not wrap around the clock face.
+		return lo, 59
+	}
+	return lo, 195
+}
+
+// matchExpired returns a bitset with the high bit of lane j set when slot j of the group
+// is occupied and its value falls in the expired range described by lo and length.
+func (d *data) matchExpired(lo, length uint8) bitset {
+	return matchExpiredWord(castUint64Data(d), lo, length)
+}
+
+// matchExpiredWord is the portable implementation of matchExpired over one 8 byte group.
+// It mirrors, lane for lane, what the assembly implementations compute with a byte-wise
+// subtract, an unsigned compare and an occupied test.
+func matchExpiredWord(x uint64, lo, length uint8) bitset {
+	y := subBytes(x, lo)
+	// Occupied means the slot holds neither empty (0) nor tombstone (1), that is x > 1,
+	// which for bytes is the same as x &^ 1 being non-zero.
+	notOccupied := findZeroBytes(x &^ loBits)
+	return leBytes(y, length) &^ notOccupied
+}
+
+// subBytes subtracts n from every byte lane of x, wrapping at 256 per lane instead of
+// borrowing into the next lane.
+func subBytes(x uint64, n uint8) uint64 {
+	nb := uint64(n) * loBits
+	// Setting the high bit of every lane of x guarantees each lane is at least as large as
+	// the corresponding lane of nb&^hiBits, so the subtraction never borrows across lanes.
+	// The final xor restores the high bits to what a per-lane subtraction would produce.
+	return ((x | hiBits) - (nb &^ hiBits)) ^ ((x ^ ^nb) & hiBits)
+}
+
+// leBytes sets the high bit of every byte lane of x that is less than or equal to n,
+// unsigned, and clears the others.
+func leBytes(x uint64, n uint8) bitset {
+	// Compare the low 7 bits of each lane by borrowing into the spare high bit.
+	low := (((uint64(n&0x7f) * loBits) | hiBits) - (x &^ hiBits)) & hiBits
+	if n < 0x80 {
+		// x <= n also requires the high bit of the lane to be clear.
+		return bitset(low & ^x)
+	}
+	// Any lane with its high bit clear is below 0x80 and therefore below n.
+	return bitset((^x & hiBits) | low)
+}
+
+// lastMatch clears and returns the index of the highest set lane in the given bitset.
+// It is the counterpart of nextMatch and assumes the bitset is non-zero.
+func lastMatch(b *bitset) uint32 {
+	s := uint32(63 - bits.LeadingZeros64(uint64(*b)))
+	*b &= ^(1 << s)
+	return s >> 3
+}
+
+func castUint64Data(d *data) uint64 {
+	return *(*uint64)(unsafe.Pointer(d))
+}

pkg/usagetracker/tenantshard/expiry_amd64.go

@@ -0,0 +1,16 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+//go:build amd64 && !nosimd
+
+package tenantshard
+
+// simdImpl names the scanExpired implementation compiled into this binary. It is only used
+// for reporting in tests and benchmarks.
+const simdImpl = "sse2"
+
+// scanExpired returns the index of the first group in p[:n] that holds at least one
+// occupied, expired slot, or n when there is none. See matchExpiredWord for the predicate
+// it evaluates, and expiry_amd64.s for the implementation.
+//
+//go:noescape
+func scanExpired(p *data, n int, lo, length uint8) int

pkg/usagetracker/tenantshard/expiry_amd64.s

@@ -0,0 +1,87 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+//go:build amd64 && !nosimd
+
+#include "textflag.h"
+
+// func scanExpired(p *data, n int, lo, length uint8) int
+//
+// Evaluates matchExpiredWord over two groups at a time. SSE2 has no unsigned byte compare,
+// so "x-lo <= length" is done with a saturating subtract: it lands on zero exactly when the
+// lane is within range. The same trick against 1 identifies the empty and tombstone
+// markers, which PANDN then removes from the result.
+//
+// Everything here is SSE2, which is part of the amd64 baseline, so there is no feature
+// detection to do.
+TEXT ·scanExpired(SB), NOSPLIT, $0-32
+	MOVQ	p+0(FP), SI
+	MOVQ	n+8(FP), CX
+	MOVBLZX	lo+16(FP), AX
+	MOVBLZX	length+17(FP), DX
+
+	XORQ	DI, DI                  // index of the group under the cursor
+	MOVQ	$0x0101010101010101, R8
+
+	// Broadcast each constant across all 16 lanes by splatting it into a general purpose
+	// register first, which avoids the SSE2 unpack dance.
+	IMULQ	R8, AX
+	MOVQ	AX, X1                  // lo in every lane
+	MOVLHPS	X1, X1
+	IMULQ	R8, DX
+	MOVQ	DX, X2                  // length in every lane
+	MOVLHPS	X2, X2
+	MOVQ	R8, X3                  // tombstone marker in every lane
+	MOVLHPS	X3, X3
+	PXOR	X0, X0
+
+loop16:
+	CMPQ	CX, $2
+	JLT	tail
+	MOVOU	(SI), X4
+	MOVO	X4, X5
+	PSUBB	X1, X5                  // X5 = x - lo, wrapping per lane
+	PSUBUSB	X2, X5                  // saturates to zero when x-lo <= length
+	PCMPEQB	X0, X5                  // X5 = lanes within range
+	MOVO	X4, X6
+	PSUBUSB	X3, X6                  // saturates to zero when x <= 1
+	PCMPEQB	X0, X6                  // X6 = lanes holding empty or tombstone
+	PANDN	X5, X6                  // X6 = in range and occupied
+	PMOVMSKB X6, BX
+	TESTL	BX, BX
+	JNZ	found
+	ADDQ	$16, SI
+	ADDQ	$2, DI
+	SUBQ	$2, CX
+	JMP	loop16
+
+found:
+	TESTL	$0xff, BX
+	JNZ	done                    // the first of the two groups hit
+	ADDQ	$1, DI
+	JMP	done
+
+	// An odd number of groups leaves a single 8 byte group to check. Loading it into the
+	// low half of the register zeroes the high half, and zero lanes read as empty, so they
+	// can never produce a spurious hit.
+tail:
+	TESTQ	CX, CX
+	JZ	notfound
+	MOVQ	(SI), X4
+	MOVO	X4, X5
+	PSUBB	X1, X5
+	PSUBUSB	X2, X5
+	PCMPEQB	X0, X5
+	MOVO	X4, X6
+	PSUBUSB	X3, X6
+	PCMPEQB	X0, X6
+	PANDN	X5, X6
+	PMOVMSKB X6, BX
+	TESTL	$0xff, BX
+	JNZ	done
+
+notfound:
+	MOVQ	n+8(FP), DI
+
+done:
+	MOVQ	DI, ret+24(FP)
+	RET

pkg/usagetracker/tenantshard/expiry_arm64.go

@@ -0,0 +1,16 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+//go:build arm64 && !nosimd
+
+package tenantshard
+
+// simdImpl names the scanExpired implementation compiled into this binary. It is only used
+// for reporting in tests and benchmarks.
+const simdImpl = "neon"
+
+// scanExpired returns the index of the first group in p[:n] that holds at least one
+// occupied, expired slot, or n when there is none. See matchExpiredWord for the predicate
+// it evaluates, and expiry_arm64.s for the implementation.
+//
+//go:noescape
+func scanExpired(p *data, n int, lo, length uint8) int

pkg/usagetracker/tenantshard/expiry_arm64.s

@@ -0,0 +1,71 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+//go:build arm64 && !nosimd
+
+#include "textflag.h"
+
+// func scanExpired(p *data, n int, lo, length uint8) int
+//
+// Evaluates matchExpiredWord over two groups at a time: subtract lo from every byte lane,
+// keep the lanes that land within length of it, and drop the lanes holding the empty or
+// tombstone markers.
+//
+// The unsigned byte comparisons are expressed with UMIN and UMAX rather than with CMHS and
+// CMHI, because the assembler only learned the latter in Go 1.27: "y <= n" is
+// "umin(y, n) == y" and "x >= 2" is "umax(x, 2) == x". Two extra instructions per iteration,
+// and it assembles on the Go version Mimir currently builds with.
+TEXT ·scanExpired(SB), NOSPLIT|NOFRAME, $0-32
+	MOVD	p+0(FP), R0
+	MOVD	n+8(FP), R1
+	MOVBU	lo+16(FP), R2
+	MOVBU	length+17(FP), R3
+
+	VDUP	R2, V0.B16              // lo in every lane
+	VDUP	R3, V1.B16              // length in every lane
+	MOVD	$2, R4
+	VDUP	R4, V2.B16              // lowest occupied byte value in every lane
+
+	MOVD	$0, R5                  // index of the group under the cursor
+
+loop16:
+	CMP	$2, R1
+	BLT	tail
+	VLD1	(R0), [V4.B16]
+	VSUB	V0.B16, V4.B16, V5.B16  // V5 = x - lo, wrapping per lane
+	VUMIN	V1.B16, V5.B16, V6.B16
+	VCMEQ	V5.B16, V6.B16, V6.B16  // V6 = lanes where x-lo <= length, unsigned
+	VUMAX	V2.B16, V4.B16, V7.B16
+	VCMEQ	V4.B16, V7.B16, V7.B16  // V7 = lanes where x >= 2, that is occupied
+	VAND	V6.B16, V7.B16, V7.B16
+	VMOV	V7.D[0], R6             // lanes of the first group
+	VMOV	V7.D[1], R7             // lanes of the second group
+	CBNZ	R6, done
+	CBZ	R7, next16
+	ADD	$1, R5
+	B	done
+
+next16:
+	ADD	$16, R0
+	ADD	$2, R5
+	SUB	$2, R1
+	B	loop16
+
+	// An odd number of groups leaves a single 8 byte group to check.
+tail:
+	CBZ	R1, notfound
+	VLD1	(R0), [V4.B8]
+	VSUB	V0.B8, V4.B8, V5.B8
+	VUMIN	V1.B8, V5.B8, V6.B8
+	VCMEQ	V5.B8, V6.B8, V6.B8
+	VUMAX	V2.B8, V4.B8, V7.B8
+	VCMEQ	V4.B8, V7.B8, V7.B8
+	VAND	V6.B8, V7.B8, V7.B8
+	VMOV	V7.D[0], R6
+	CBNZ	R6, done
+
+notfound:
+	MOVD	n+8(FP), R5
+
+done:
+	MOVD	R5, ret+24(FP)
+	RET

pkg/usagetracker/tenantshard/expiry_generic.go

@@ -0,0 +1,22 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+//go:build (!amd64 && !arm64) || nosimd
+
+package tenantshard
+
+import "unsafe"
+
+// simdImpl names the scanExpired implementation compiled into this binary. It is only used
+// for reporting in tests and benchmarks.
+const simdImpl = "generic"
+
+// scanExpired returns the index of the first group in p[:n] that holds at least one
+// occupied, expired slot, or n when there is none.
+func scanExpired(p *data, n int, lo, length uint8) int {
+	for i, d := range unsafe.Slice(p, n) {
+		if d.matchExpired(lo, length) != 0 {
+			return i
+		}
+	}
+	return n
+}