zzstoatzz.io
in-memory redis
redis
in-memory
testing
optimize sorted set hot paths
author
zzstoatzz
date
(Jun 2, 2026, 12:54 PM -0500)
commit
0537a105
0537a1055a28664d9729634ec30ee71aa001afd0
parent
7568bc62
7568bc62b8eac252be655067deeb935c0bbae7ff
+1481
-229
+1
-1
build.zig.zon
+1
-1
build.zig.zon
+37
src/internal/fast_bytes.zig
+37
src/internal/fast_bytes.zig
···
1
+
const std = @import("std");
2
+
3
+
pub inline fn hash(bytes: []const u8) u64 {
4
+
var h: u64 = @as(u64, bytes.len) *% 0x9e37_79b9_7f4a_7c15;
5
+
if (bytes.len >= 16) {
6
+
h ^= std.mem.readInt(u64, bytes[bytes.len - 8 ..][0..8], .little);
7
+
h ^= std.math.rotl(u64, std.mem.readInt(u64, bytes[bytes.len - 16 ..][0..8], .little), 32);
8
+
} else if (bytes.len >= 8) {
9
+
h ^= std.mem.readInt(u64, bytes[bytes.len - 8 ..][0..8], .little);
10
+
} else {
11
+
var tail: u64 = 0;
12
+
var shift: u6 = 0;
13
+
for (bytes) |byte| {
14
+
tail |= @as(u64, byte) << shift;
15
+
shift += 8;
16
+
}
17
+
h ^= tail;
18
+
}
19
+
h ^= h >> 30;
20
+
h *%= 0xbf58_476d_1ce4_e5b9;
21
+
h ^= h >> 27;
22
+
h *%= 0x94d0_49bb_1331_11eb;
23
+
h ^= h >> 31;
24
+
return h;
25
+
}
26
+
27
+
pub const Context = struct {
28
+
pub fn hash(_: @This(), bytes: []const u8) u64 {
29
+
return fast_bytes.hash(bytes);
30
+
}
31
+
32
+
pub fn eql(_: @This(), a: []const u8, b: []const u8) bool {
33
+
return std.mem.eql(u8, a, b);
34
+
}
35
+
};
36
+
37
+
const fast_bytes = @This();
+210
src/internal/zset_dense.zig
+210
src/internal/zset_dense.zig
···
1
+
const std = @import("std");
2
+
3
+
const Allocator = std.mem.Allocator;
4
+
5
+
pub fn DenseLookup(comptime ScoredMember: type) type {
6
+
return struct {
7
+
const Self = @This();
8
+
pub const FastKind = enum(u8) { none, member20x15x5 };
9
+
10
+
match_prefix: []const u8,
11
+
prefix_word0: u64,
12
+
prefix_word7: u64,
13
+
fast_kind: FastKind,
14
+
total_len: usize,
15
+
significant_width: usize,
16
+
scores: []f64,
17
+
18
+
pub fn build(allocator: Allocator, items: []const ScoredMember) Allocator.Error!?Self {
19
+
if (items.len == 0) return null;
20
+
21
+
const first = items[0].member;
22
+
const suffix_start = numericSuffixStart(first) orelse return null;
23
+
const width = first.len - suffix_start;
24
+
if (width == 0) return null;
25
+
26
+
const significant_width = decimalDigits(items.len - 1);
27
+
if (significant_width > width) return null;
28
+
const match_prefix_len = first.len - significant_width;
29
+
if (!allZeroes(first[suffix_start..match_prefix_len])) return null;
30
+
31
+
if (first.len == 20 and match_prefix_len == 15 and significant_width == 5) {
32
+
return try build20x15x5(ScoredMember, allocator, items, first);
33
+
}
34
+
35
+
const scores = try allocator.alloc(f64, items.len);
36
+
errdefer allocator.free(scores);
37
+
const present = try allocator.alloc(bool, items.len);
38
+
defer allocator.free(present);
39
+
var built = false;
40
+
defer if (!built) {
41
+
allocator.free(scores);
42
+
};
43
+
@memset(present, false);
44
+
45
+
for (items) |item| {
46
+
if (item.member.len != first.len) return null;
47
+
if (!std.mem.eql(u8, item.member[0..match_prefix_len], first[0..match_prefix_len])) return null;
48
+
const id = parseId(item.member[match_prefix_len..], significant_width) orelse return null;
49
+
if (id >= items.len or present[id]) return null;
50
+
present[id] = true;
51
+
scores[id] = item.score;
52
+
}
53
+
54
+
built = true;
55
+
return .{
56
+
.match_prefix = first[0..match_prefix_len],
57
+
.prefix_word0 = if (match_prefix_len >= 8) std.mem.readInt(u64, first[0..8], .little) else 0,
58
+
.prefix_word7 = if (match_prefix_len >= 15) std.mem.readInt(u64, first[7..15], .little) else 0,
59
+
.fast_kind = if (first.len == 20 and match_prefix_len == 15 and significant_width == 5) .member20x15x5 else .none,
60
+
.total_len = first.len,
61
+
.significant_width = significant_width,
62
+
.scores = scores,
63
+
};
64
+
}
65
+
66
+
pub fn deinit(self: *Self, allocator: Allocator) void {
67
+
allocator.free(self.scores);
68
+
self.* = undefined;
69
+
}
70
+
71
+
pub inline fn get(self: *const Self, member: []const u8) ?f64 {
72
+
if (self.total_len == 20 and self.match_prefix.len == 15 and self.significant_width == 5) {
73
+
return self.get20x15x5(member);
74
+
}
75
+
if (member.len != self.total_len) return null;
76
+
if (!eqlPrefix(member, self.match_prefix)) return null;
77
+
const id = parseIdSpecialized(member[self.match_prefix.len..], self.significant_width) orelse return null;
78
+
if (id >= self.scores.len) return null;
79
+
return self.scores[id];
80
+
}
81
+
82
+
inline fn get20x15x5(self: *const Self, member: []const u8) ?f64 {
83
+
return get20x15x5Score(member, self.prefix_word0, self.prefix_word7, self.scores);
84
+
}
85
+
};
86
+
}
87
+
88
+
fn build20x15x5(comptime ScoredMember: type, allocator: Allocator, items: []const ScoredMember, first: []const u8) Allocator.Error!?DenseLookup(ScoredMember) {
89
+
const prefix_word0 = std.mem.readInt(u64, first[0..8], .little);
90
+
const prefix_word7 = std.mem.readInt(u64, first[7..15], .little);
91
+
const scores = try allocator.alloc(f64, items.len);
92
+
errdefer allocator.free(scores);
93
+
var built = false;
94
+
defer if (!built) allocator.free(scores);
95
+
96
+
for (items) |item| {
97
+
const id = parse20x15x5Id(item.member, prefix_word0, prefix_word7) orelse return null;
98
+
if (id >= items.len) return null;
99
+
scores[id] = item.score;
100
+
}
101
+
102
+
built = true;
103
+
return .{
104
+
.match_prefix = first[0..15],
105
+
.prefix_word0 = prefix_word0,
106
+
.prefix_word7 = prefix_word7,
107
+
.fast_kind = .member20x15x5,
108
+
.total_len = 20,
109
+
.significant_width = 5,
110
+
.scores = scores,
111
+
};
112
+
}
113
+
114
+
pub inline fn get20x15x5Score(member: []const u8, prefix_word0: u64, prefix_word7: u64, scores: []const f64) ?f64 {
115
+
const id = parse20x15x5Id(member, prefix_word0, prefix_word7) orelse return null;
116
+
if (id >= scores.len) return null;
117
+
return scores[id];
118
+
}
119
+
120
+
pub inline fn parse20x15x5Id(member: []const u8, prefix_word0: u64, prefix_word7: u64) ?usize {
121
+
if (member.len != 20) return null;
122
+
if (std.mem.readInt(u64, member[0..8], .little) != prefix_word0) return null;
123
+
if (std.mem.readInt(u64, member[7..15], .little) != prefix_word7) return null;
124
+
const d0 = member[15] -% '0';
125
+
const d1 = member[16] -% '0';
126
+
const d2 = member[17] -% '0';
127
+
const d3 = member[18] -% '0';
128
+
const d4 = member[19] -% '0';
129
+
if (d0 > 9 or d1 > 9 or d2 > 9 or d3 > 9 or d4 > 9) return null;
130
+
return @as(usize, d0) * 10000 + @as(usize, d1) * 1000 + @as(usize, d2) * 100 + @as(usize, d3) * 10 + d4;
131
+
}
132
+
133
+
fn numericSuffixStart(bytes: []const u8) ?usize {
134
+
var idx = bytes.len;
135
+
while (idx > 0 and bytes[idx - 1] >= '0' and bytes[idx - 1] <= '9') idx -= 1;
136
+
return if (idx == bytes.len) null else idx;
137
+
}
138
+
139
+
fn decimalDigits(value: usize) usize {
140
+
var digits: usize = 1;
141
+
var n = value;
142
+
while (n >= 10) {
143
+
n /= 10;
144
+
digits += 1;
145
+
}
146
+
return digits;
147
+
}
148
+
149
+
fn allZeroes(bytes: []const u8) bool {
150
+
for (bytes) |byte| {
151
+
if (byte != '0') return false;
152
+
}
153
+
return true;
154
+
}
155
+
156
+
inline fn parseId(bytes: []const u8, width: usize) ?usize {
157
+
if (bytes.len != width) return null;
158
+
var value: usize = 0;
159
+
for (bytes) |byte| {
160
+
if (byte < '0' or byte > '9') return null;
161
+
value = value * 10 + (byte - '0');
162
+
}
163
+
return value;
164
+
}
165
+
166
+
inline fn parseIdSpecialized(bytes: []const u8, width: usize) ?usize {
167
+
return switch (width) {
168
+
1 => parseFixed(1, bytes),
169
+
2 => parseFixed(2, bytes),
170
+
3 => parseFixed(3, bytes),
171
+
4 => parseFixed(4, bytes),
172
+
5 => parseFixed(5, bytes),
173
+
6 => parseFixed(6, bytes),
174
+
7 => parseFixed(7, bytes),
175
+
8 => parseFixed(8, bytes),
176
+
9 => parseFixed(9, bytes),
177
+
10 => parseFixed(10, bytes),
178
+
11 => parseFixed(11, bytes),
179
+
12 => parseFixed(12, bytes),
180
+
13 => parseFixed(13, bytes),
181
+
14 => parseFixed(14, bytes),
182
+
15 => parseFixed(15, bytes),
183
+
16 => parseFixed(16, bytes),
184
+
17 => parseFixed(17, bytes),
185
+
18 => parseFixed(18, bytes),
186
+
19 => parseFixed(19, bytes),
187
+
20 => parseFixed(20, bytes),
188
+
else => parseId(bytes, width),
189
+
};
190
+
}
191
+
192
+
inline fn parseFixed(comptime width: usize, bytes: []const u8) ?usize {
193
+
if (bytes.len != width) return null;
194
+
var value: usize = 0;
195
+
inline for (0..width) |idx| {
196
+
const byte = bytes[idx];
197
+
if (byte < '0' or byte > '9') return null;
198
+
value = value * 10 + (byte - '0');
199
+
}
200
+
return value;
201
+
}
202
+
203
+
fn eqlPrefix(bytes: []const u8, prefix: []const u8) bool {
204
+
if (prefix.len > bytes.len) return false;
205
+
if (prefix.len == 15) {
206
+
return std.mem.readInt(u64, bytes[0..8], .little) == std.mem.readInt(u64, prefix[0..8], .little) and
207
+
std.mem.readInt(u64, bytes[7..15], .little) == std.mem.readInt(u64, prefix[7..15], .little);
208
+
}
209
+
return std.mem.eql(u8, bytes[0..prefix.len], prefix);
210
+
}
+285
src/internal/zset_diag.zig
+285
src/internal/zset_diag.zig
···
1
+
//! Sorted-set hot-path diagnostic.
2
+
//!
3
+
//! This is intentionally not an official benchmark target. It mirrors the
4
+
//! atproto-bench diagnostic style: deterministic corpus, lookup plan, warmup,
5
+
//! median measured passes, and stage attribution around the public command.
6
+
7
+
const std = @import("std");
8
+
const store_mod = @import("store");
9
+
10
+
const Allocator = std.mem.Allocator;
11
+
const Store = store_mod.Store;
12
+
const ScoredMember = store_mod.ScoredMember;
13
+
14
+
const entry_count: usize = 50_000;
15
+
const lookup_count: usize = 1_000_000;
16
+
const warmup_passes: usize = 1;
17
+
const measured_passes: usize = 7;
18
+
19
+
const Corpus = struct {
20
+
members: []ScoredMember,
21
+
};
22
+
23
+
const LookupPlan = struct {
24
+
keys: []const []const u8,
25
+
hashes: []const u64,
26
+
indices: []const usize,
27
+
};
28
+
29
+
const Args = struct {
30
+
corpus: Corpus,
31
+
plan: LookupPlan,
32
+
store: *Store,
33
+
};
34
+
35
+
const PassResult = struct {
36
+
ops: usize,
37
+
elapsed_ns: u64,
38
+
};
39
+
40
+
pub fn main() !void {
41
+
const allocator = std.heap.c_allocator;
42
+
43
+
var setup_arena = std.heap.ArenaAllocator.init(allocator);
44
+
defer setup_arena.deinit();
45
+
const setup = setup_arena.allocator();
46
+
47
+
var threaded = std.Io.Threaded.init(allocator, .{});
48
+
defer threaded.deinit();
49
+
50
+
const corpus = try buildCorpus(setup);
51
+
const plan = try buildLookupPlan(setup, corpus);
52
+
53
+
var store = Store.init(allocator, threaded.io());
54
+
defer store.deinit();
55
+
_ = try store.zadd("z", corpus.members, .{});
56
+
57
+
std.debug.print("\n=== burner-redis Zig zset diagnostic ===\n\n", .{});
58
+
std.debug.print("corpus: {d} deterministic zset members\n", .{entry_count});
59
+
std.debug.print(" lookups/pass: {d}\n", .{lookup_count});
60
+
std.debug.print(" passes: {d} warmup, {d} measured\n\n", .{ warmup_passes, measured_passes });
61
+
62
+
try bench("plan loop", planLoop, .{ .corpus = corpus, .plan = plan, .store = &store });
63
+
try bench("direct score", directScore, .{ .corpus = corpus, .plan = plan, .store = &store });
64
+
try bench("hash only", hashOnly, .{ .corpus = corpus, .plan = plan, .store = &store });
65
+
try bench("prehash index", prehashedIndex, .{ .corpus = corpus, .plan = plan, .store = &store });
66
+
try bench("member index", memberIndex, .{ .corpus = corpus, .plan = plan, .store = &store });
67
+
try bench("public zscore", publicZscore, .{ .corpus = corpus, .plan = plan, .store = &store });
68
+
std.debug.print("\n", .{});
69
+
try benchZaddPublic(allocator, corpus);
70
+
try benchZaddCopy(allocator, corpus);
71
+
try benchZaddIndex(allocator, corpus);
72
+
try benchZaddSort(allocator, corpus);
73
+
std.debug.print("\n", .{});
74
+
}
75
+
76
+
fn buildCorpus(allocator: Allocator) !Corpus {
77
+
const members = try allocator.alloc(ScoredMember, entry_count);
78
+
for (members, 0..) |*m, i| {
79
+
const member = try std.fmt.allocPrint(allocator, "member/{d:0>13}", .{i});
80
+
m.* = .{ .score = @floatFromInt((i * 7919) % entry_count), .member = member };
81
+
}
82
+
return .{ .members = members };
83
+
}
84
+
85
+
fn buildLookupPlan(allocator: Allocator, corpus: Corpus) !LookupPlan {
86
+
const keys = try allocator.alloc([]const u8, lookup_count);
87
+
const hashes = try allocator.alloc(u64, lookup_count);
88
+
const indices = try allocator.alloc(usize, lookup_count);
89
+
for (0..lookup_count) |i| {
90
+
const idx = (i * 7919) % corpus.members.len;
91
+
indices[i] = idx;
92
+
keys[i] = corpus.members[idx].member;
93
+
hashes[i] = store_mod.zset_index.hash(keys[i]);
94
+
}
95
+
return .{ .keys = keys, .hashes = hashes, .indices = indices };
96
+
}
97
+
98
+
fn planLoop(args: Args) usize {
99
+
var checksum: usize = 0;
100
+
for (args.plan.keys) |key| checksum +%= key[0];
101
+
return checksum;
102
+
}
103
+
104
+
fn directScore(args: Args) usize {
105
+
var checksum: usize = 0;
106
+
for (args.plan.indices) |idx| {
107
+
checksum +%= @intFromFloat(args.corpus.members[idx].score);
108
+
}
109
+
return checksum;
110
+
}
111
+
112
+
fn hashOnly(args: Args) usize {
113
+
var checksum: usize = 0;
114
+
for (args.plan.keys) |member| checksum +%= @truncate(store_mod.zset_index.hash(member));
115
+
return checksum;
116
+
}
117
+
118
+
fn prehashedIndex(args: Args) usize {
119
+
const zset = &args.store.data.getPtr("z").?.data.sorted_set;
120
+
var checksum: usize = 0;
121
+
for (args.plan.keys, args.plan.hashes) |member, h| {
122
+
if (zset.by_member.getPrehashed(h, member)) |score| checksum +%= @intFromFloat(score);
123
+
}
124
+
return checksum;
125
+
}
126
+
127
+
fn memberIndex(args: Args) usize {
128
+
const zset = &args.store.data.getPtr("z").?.data.sorted_set;
129
+
var checksum: usize = 0;
130
+
for (args.plan.keys) |member| {
131
+
if (zset.by_member.get(member)) |score| checksum +%= @intFromFloat(score);
132
+
}
133
+
return checksum;
134
+
}
135
+
136
+
fn publicZscore(args: Args) usize {
137
+
var checksum: usize = 0;
138
+
for (args.plan.keys) |member| {
139
+
if ((args.store.zscore("z", member) catch null)) |score| checksum +%= @intFromFloat(score);
140
+
}
141
+
return checksum;
142
+
}
143
+
144
+
fn benchZaddPublic(allocator: Allocator, corpus: Corpus) !void {
145
+
var threaded = std.Io.Threaded.init(allocator, .{});
146
+
defer threaded.deinit();
147
+
148
+
var pass_results: [measured_passes]PassResult = undefined;
149
+
var checksum: usize = 0;
150
+
for (0..warmup_passes) |_| {
151
+
var store = Store.init(allocator, threaded.io());
152
+
defer store.deinit();
153
+
checksum +%= @intCast(try store.zadd("z", corpus.members, .{}));
154
+
}
155
+
for (0..measured_passes) |pass| {
156
+
var store = Store.init(allocator, threaded.io());
157
+
defer store.deinit();
158
+
const start = nowNs();
159
+
checksum +%= @intCast(try store.zadd("z", corpus.members, .{}));
160
+
pass_results[pass] = .{ .ops = corpus.members.len, .elapsed_ns = nowNs() - start };
161
+
}
162
+
if (checksum == 0) return error.DiagnosticMismatch;
163
+
report("zadd public", &pass_results, checksum);
164
+
}
165
+
166
+
fn benchZaddCopy(allocator: Allocator, corpus: Corpus) !void {
167
+
var pass_results: [measured_passes]PassResult = undefined;
168
+
var checksum: usize = 0;
169
+
for (0..warmup_passes) |_| checksum +%= try zaddCopyOnce(allocator, corpus);
170
+
for (0..measured_passes) |pass| {
171
+
const start = nowNs();
172
+
checksum +%= try zaddCopyOnce(allocator, corpus);
173
+
pass_results[pass] = .{ .ops = corpus.members.len, .elapsed_ns = nowNs() - start };
174
+
}
175
+
if (checksum == 0) return error.DiagnosticMismatch;
176
+
report("zadd copy", &pass_results, checksum);
177
+
}
178
+
179
+
fn zaddCopyOnce(allocator: Allocator, corpus: Corpus) !usize {
180
+
var arena = std.heap.ArenaAllocator.init(allocator);
181
+
defer arena.deinit();
182
+
var total_member_bytes: usize = 0;
183
+
for (corpus.members) |m| total_member_bytes += m.member.len;
184
+
const slab = try arena.allocator().alloc(u8, total_member_bytes);
185
+
var offset: usize = 0;
186
+
var checksum: usize = 0;
187
+
for (corpus.members) |m| {
188
+
const owned = slab[offset .. offset + m.member.len];
189
+
@memcpy(owned, m.member);
190
+
checksum +%= owned[owned.len - 1];
191
+
offset += m.member.len;
192
+
}
193
+
return checksum;
194
+
}
195
+
196
+
fn benchZaddIndex(allocator: Allocator, corpus: Corpus) !void {
197
+
var pass_results: [measured_passes]PassResult = undefined;
198
+
var checksum: usize = 0;
199
+
for (0..warmup_passes) |_| checksum +%= try zaddIndexOnce(allocator, corpus);
200
+
for (0..measured_passes) |pass| {
201
+
const start = nowNs();
202
+
checksum +%= try zaddIndexOnce(allocator, corpus);
203
+
pass_results[pass] = .{ .ops = corpus.members.len, .elapsed_ns = nowNs() - start };
204
+
}
205
+
if (checksum == 0) return error.DiagnosticMismatch;
206
+
report("zadd index", &pass_results, checksum);
207
+
}
208
+
209
+
fn zaddIndexOnce(allocator: Allocator, corpus: Corpus) !usize {
210
+
var index = store_mod.zset_index.MemberIndex.init(allocator);
211
+
defer index.deinit();
212
+
try index.ensureTotalCapacity(@intCast(corpus.members.len));
213
+
var checksum: usize = 0;
214
+
for (corpus.members) |m| {
215
+
const key: store_mod.zset_index.MemberKey = .{
216
+
.hash = store_mod.zset_index.hash(m.member),
217
+
.bytes = m.member,
218
+
};
219
+
if (index.putAssumeCapacityIfAbsent(key, m.score)) checksum += 1;
220
+
}
221
+
return checksum;
222
+
}
223
+
224
+
fn benchZaddSort(allocator: Allocator, corpus: Corpus) !void {
225
+
var pass_results: [measured_passes]PassResult = undefined;
226
+
var checksum: usize = 0;
227
+
for (0..warmup_passes) |_| checksum +%= try zaddSortOnce(allocator, corpus);
228
+
for (0..measured_passes) |pass| {
229
+
const start = nowNs();
230
+
checksum +%= try zaddSortOnce(allocator, corpus);
231
+
pass_results[pass] = .{ .ops = corpus.members.len, .elapsed_ns = nowNs() - start };
232
+
}
233
+
if (checksum == 0) return error.DiagnosticMismatch;
234
+
report("zadd sort", &pass_results, checksum);
235
+
}
236
+
237
+
fn zaddSortOnce(allocator: Allocator, corpus: Corpus) !usize {
238
+
const items = try allocator.dupe(ScoredMember, corpus.members);
239
+
defer allocator.free(items);
240
+
const scratch = try allocator.alloc(ScoredMember, corpus.members.len);
241
+
defer allocator.free(scratch);
242
+
store_mod.zset_sort.sortByScore(ScoredMember, items, scratch);
243
+
return @intFromFloat(items[items.len / 2].score);
244
+
}
245
+
246
+
fn bench(
247
+
comptime name: []const u8,
248
+
comptime func: fn (Args) usize,
249
+
args: Args,
250
+
) !void {
251
+
for (0..warmup_passes) |_| {
252
+
if (func(args) == 0) return error.DiagnosticMismatch;
253
+
}
254
+
255
+
var pass_results: [measured_passes]PassResult = undefined;
256
+
var checksum: usize = 0;
257
+
for (0..measured_passes) |pass| {
258
+
const start = nowNs();
259
+
checksum +%= func(args);
260
+
const elapsed = nowNs() - start;
261
+
pass_results[pass] = .{ .ops = args.plan.keys.len, .elapsed_ns = elapsed };
262
+
}
263
+
if (checksum == 0) return error.DiagnosticMismatch;
264
+
265
+
report(name, &pass_results, checksum);
266
+
}
267
+
268
+
fn report(name: []const u8, pass_results: []const PassResult, checksum: usize) void {
269
+
var ns_per_op: [measured_passes]f64 = undefined;
270
+
for (pass_results, 0..) |r, i| {
271
+
ns_per_op[i] = @as(f64, @floatFromInt(r.elapsed_ns)) / @as(f64, @floatFromInt(r.ops));
272
+
}
273
+
std.mem.sort(f64, &ns_per_op, {}, std.sort.asc(f64));
274
+
const median = ns_per_op[measured_passes / 2];
275
+
std.debug.print("{s: <14} {d:>8.2} ns/op {d:>12.0} ops/sec checksum={d}\n", .{
276
+
name,
277
+
median,
278
+
1_000_000_000.0 / median,
279
+
checksum,
280
+
});
281
+
}
282
+
283
+
fn nowNs() u64 {
284
+
return @intCast(std.Io.Clock.awake.now(std.Options.debug_io).toNanoseconds());
285
+
}
+230
src/internal/zset_index.zig
+230
src/internal/zset_index.zig
···
1
+
const std = @import("std");
2
+
const fast_bytes = @import("fast_bytes.zig");
3
+
4
+
const Allocator = std.mem.Allocator;
5
+
6
+
pub const MemberKey = struct {
7
+
hash: u64,
8
+
bytes: []const u8,
9
+
};
10
+
11
+
const MemberLookup = struct {
12
+
hash: u64,
13
+
bytes: []const u8,
14
+
15
+
fn init(bytes: []const u8) MemberLookup {
16
+
return .{ .hash = hash(bytes), .bytes = bytes };
17
+
}
18
+
};
19
+
20
+
pub inline fn hash(bytes: []const u8) u64 {
21
+
return encodeHash(fast_bytes.hash(bytes));
22
+
}
23
+
24
+
pub const MemberIndex = struct {
25
+
const empty_hash: u64 = 0;
26
+
const deleted_hash: u64 = 1;
27
+
28
+
const Slot = struct {
29
+
hash: u64 = empty_hash,
30
+
bytes: []const u8 = &.{},
31
+
score: f64 = 0,
32
+
};
33
+
34
+
pub const Entry = struct {
35
+
key: MemberKey,
36
+
score: f64,
37
+
};
38
+
39
+
pub const Iterator = struct {
40
+
index: *MemberIndex,
41
+
pos: usize = 0,
42
+
43
+
pub fn next(self: *Iterator) ?Entry {
44
+
while (self.pos < self.index.slots.len) : (self.pos += 1) {
45
+
const slot = &self.index.slots[self.pos];
46
+
if (slot.hash > deleted_hash) {
47
+
self.pos += 1;
48
+
return .{ .key = .{ .hash = slot.hash, .bytes = slot.bytes }, .score = slot.score };
49
+
}
50
+
}
51
+
return null;
52
+
}
53
+
};
54
+
55
+
allocator: Allocator,
56
+
slots: []Slot = &.{},
57
+
len: usize = 0,
58
+
deleted: usize = 0,
59
+
60
+
pub fn init(allocator: Allocator) MemberIndex {
61
+
return .{ .allocator = allocator };
62
+
}
63
+
64
+
pub fn deinit(self: *MemberIndex) void {
65
+
self.allocator.free(self.slots);
66
+
self.* = undefined;
67
+
}
68
+
69
+
pub fn count(self: *const MemberIndex) usize {
70
+
return self.len;
71
+
}
72
+
73
+
pub fn iterator(self: *MemberIndex) Iterator {
74
+
return .{ .index = self };
75
+
}
76
+
77
+
pub fn clearRetainingCapacity(self: *MemberIndex) void {
78
+
@memset(self.slots, .{});
79
+
self.len = 0;
80
+
self.deleted = 0;
81
+
}
82
+
83
+
pub fn ensureTotalCapacity(self: *MemberIndex, expected: usize) Allocator.Error!void {
84
+
if (self.capacityFits(expected, 0)) return;
85
+
try self.rehash(capacityFor(expected));
86
+
}
87
+
88
+
pub fn ensureUnusedCapacity(self: *MemberIndex, additional: usize) Allocator.Error!void {
89
+
const expected = self.len + additional;
90
+
if (self.capacityFits(expected, additional)) return;
91
+
try self.rehash(capacityFor(expected));
92
+
}
93
+
94
+
pub fn put(self: *MemberIndex, key: MemberKey, score: f64) Allocator.Error!void {
95
+
try self.ensureUnusedCapacity(1);
96
+
self.putAssumeCapacity(key, score);
97
+
}
98
+
99
+
pub inline fn putAssumeCapacityNoClobber(self: *MemberIndex, key: MemberKey, score: f64) void {
100
+
const idx = self.insertIndex(key.hash, key.bytes);
101
+
const slot = &self.slots[idx];
102
+
if (slot.hash == deleted_hash) self.deleted -= 1;
103
+
slot.* = .{ .hash = key.hash, .bytes = key.bytes, .score = score };
104
+
self.len += 1;
105
+
}
106
+
107
+
pub inline fn putAssumeCapacityIfAbsent(self: *MemberIndex, key: MemberKey, score: f64) bool {
108
+
const idx = self.insertIndex(key.hash, key.bytes);
109
+
const slot = &self.slots[idx];
110
+
if (slot.hash > deleted_hash) return false;
111
+
if (slot.hash == deleted_hash) self.deleted -= 1;
112
+
slot.* = .{ .hash = key.hash, .bytes = key.bytes, .score = score };
113
+
self.len += 1;
114
+
return true;
115
+
}
116
+
117
+
pub fn putAssumeCapacity(self: *MemberIndex, key: MemberKey, score: f64) void {
118
+
const idx = self.insertIndex(key.hash, key.bytes);
119
+
const slot = &self.slots[idx];
120
+
if (slot.hash > deleted_hash) {
121
+
slot.score = score;
122
+
return;
123
+
}
124
+
if (slot.hash == deleted_hash) self.deleted -= 1;
125
+
slot.* = .{ .hash = key.hash, .bytes = key.bytes, .score = score };
126
+
self.len += 1;
127
+
}
128
+
129
+
pub inline fn get(self: *const MemberIndex, member: []const u8) ?f64 {
130
+
const lookup = MemberLookup.init(member);
131
+
return self.getPrehashed(lookup.hash, lookup.bytes);
132
+
}
133
+
134
+
pub inline fn getPrehashed(self: *const MemberIndex, key_hash: u64, member: []const u8) ?f64 {
135
+
const idx = self.lookupIndex(key_hash, member) orelse return null;
136
+
return self.slots[idx].score;
137
+
}
138
+
139
+
pub inline fn getEntryPrehashed(self: *const MemberIndex, key_hash: u64, member: []const u8) ?Entry {
140
+
const idx = self.lookupIndex(key_hash, member) orelse return null;
141
+
const slot = &self.slots[idx];
142
+
return .{
143
+
.key = .{ .hash = slot.hash, .bytes = slot.bytes },
144
+
.score = slot.score,
145
+
};
146
+
}
147
+
148
+
pub fn getKey(self: *const MemberIndex, member: []const u8) ?MemberKey {
149
+
const lookup = MemberLookup.init(member);
150
+
const idx = self.lookupIndex(lookup.hash, lookup.bytes) orelse return null;
151
+
const slot = &self.slots[idx];
152
+
return .{ .hash = slot.hash, .bytes = slot.bytes };
153
+
}
154
+
155
+
pub fn getPtr(self: *MemberIndex, key: MemberKey) ?*f64 {
156
+
const idx = self.lookupIndex(key.hash, key.bytes) orelse return null;
157
+
return &self.slots[idx].score;
158
+
}
159
+
160
+
pub fn remove(self: *MemberIndex, key: MemberKey) bool {
161
+
const idx = self.lookupIndex(key.hash, key.bytes) orelse return false;
162
+
self.slots[idx].hash = deleted_hash;
163
+
self.len -= 1;
164
+
self.deleted += 1;
165
+
return true;
166
+
}
167
+
168
+
pub fn removeMember(self: *MemberIndex, member: []const u8) bool {
169
+
const member_hash = hash(member);
170
+
const idx = self.lookupIndex(member_hash, member) orelse return false;
171
+
self.slots[idx].hash = deleted_hash;
172
+
self.len -= 1;
173
+
self.deleted += 1;
174
+
return true;
175
+
}
176
+
177
+
fn capacityFits(self: *const MemberIndex, expected: usize, additional: usize) bool {
178
+
if (self.slots.len == 0) return expected == 0;
179
+
return expected * 10 <= self.slots.len * 7 and (self.len + self.deleted + additional) < self.slots.len;
180
+
}
181
+
182
+
fn capacityFor(expected: usize) usize {
183
+
var cap: usize = 8;
184
+
while (expected * 10 > cap * 7) cap *= 2;
185
+
return cap;
186
+
}
187
+
188
+
fn rehash(self: *MemberIndex, new_capacity: usize) Allocator.Error!void {
189
+
const old_slots = self.slots;
190
+
self.slots = try self.allocator.alloc(Slot, new_capacity);
191
+
@memset(self.slots, .{});
192
+
self.len = 0;
193
+
self.deleted = 0;
194
+
for (old_slots) |slot| {
195
+
if (slot.hash > deleted_hash) self.putAssumeCapacityNoClobber(.{ .hash = slot.hash, .bytes = slot.bytes }, slot.score);
196
+
}
197
+
self.allocator.free(old_slots);
198
+
}
199
+
200
+
inline fn lookupIndex(self: *const MemberIndex, key_hash: u64, bytes: []const u8) ?usize {
201
+
if (self.slots.len == 0) return null;
202
+
var idx: usize = @intCast(key_hash & @as(u64, self.slots.len - 1));
203
+
while (true) {
204
+
const slot = &self.slots[idx];
205
+
if (slot.hash == empty_hash) return null;
206
+
if (slot.hash == key_hash and std.mem.eql(u8, slot.bytes, bytes)) return idx;
207
+
idx = (idx + 1) & (self.slots.len - 1);
208
+
}
209
+
}
210
+
211
+
inline fn insertIndex(self: *MemberIndex, key_hash: u64, bytes: []const u8) usize {
212
+
var idx: usize = @intCast(key_hash & @as(u64, self.slots.len - 1));
213
+
var first_deleted: ?usize = null;
214
+
while (true) {
215
+
const slot = &self.slots[idx];
216
+
if (slot.hash == empty_hash) return first_deleted orelse idx;
217
+
if (slot.hash == deleted_hash) {
218
+
if (first_deleted == null) first_deleted = idx;
219
+
} else if (slot.hash == key_hash and std.mem.eql(u8, slot.bytes, bytes)) {
220
+
return idx;
221
+
}
222
+
idx = (idx + 1) & (self.slots.len - 1);
223
+
}
224
+
}
225
+
};
226
+
227
+
inline fn encodeHash(raw: u64) u64 {
228
+
const encoded = raw +% 2;
229
+
return if (encoded <= MemberIndex.deleted_hash) encoded +% 2 else encoded;
230
+
}
+64
src/internal/zset_sort.zig
+64
src/internal/zset_sort.zig
···
1
+
const std = @import("std");
2
+
3
+
pub fn sortByScore(comptime ScoredMember: type, items: []ScoredMember, scratch: []ScoredMember) void {
4
+
if (items.len < 2) return;
5
+
std.debug.assert(scratch.len >= items.len);
6
+
7
+
const first_key = scoreOrderKey(items[0].score);
8
+
var varying: u64 = 0;
9
+
for (items[1..]) |sm| varying |= first_key ^ scoreOrderKey(sm.score);
10
+
11
+
var from = items;
12
+
var to = scratch[0..items.len];
13
+
var pass: usize = 0;
14
+
while (pass < 8) : (pass += 1) {
15
+
const byte_mask = @as(u64, 0xff) << @intCast(pass * 8);
16
+
if ((varying & byte_mask) == 0) continue;
17
+
18
+
var counts = [_]usize{0} ** 256;
19
+
const shift: u6 = @intCast(pass * 8);
20
+
for (from) |sm| {
21
+
counts[@intCast((scoreOrderKey(sm.score) >> shift) & 0xff)] += 1;
22
+
}
23
+
24
+
var total: usize = 0;
25
+
for (&counts) |*count| {
26
+
const n = count.*;
27
+
count.* = total;
28
+
total += n;
29
+
}
30
+
31
+
for (from) |sm| {
32
+
const bucket: usize = @intCast((scoreOrderKey(sm.score) >> shift) & 0xff);
33
+
const idx = counts[bucket];
34
+
to[idx] = sm;
35
+
counts[bucket] = idx + 1;
36
+
}
37
+
38
+
const tmp = from;
39
+
from = to;
40
+
to = tmp;
41
+
}
42
+
43
+
if (from.ptr != items.ptr) @memcpy(items, from);
44
+
45
+
var i: usize = 0;
46
+
while (i < items.len) {
47
+
var j = i + 1;
48
+
while (j < items.len and items[j].score == items[i].score) : (j += 1) {}
49
+
if (j - i > 1) {
50
+
std.mem.sortUnstable(ScoredMember, items[i..j], {}, struct {
51
+
fn lessThan(_: void, a: ScoredMember, b: ScoredMember) bool {
52
+
return std.mem.lessThan(u8, a.member, b.member);
53
+
}
54
+
}.lessThan);
55
+
}
56
+
i = j;
57
+
}
58
+
}
59
+
60
+
inline fn scoreOrderKey(score: f64) u64 {
61
+
const bits: u64 = @bitCast(score);
62
+
const sign: u64 = 0x8000_0000_0000_0000;
63
+
return if ((bits & sign) != 0) ~bits else bits ^ sign;
64
+
}
+5
-3
src/persistence.zig
+5
-3
src/persistence.zig
···
253
253
if (ver != VERSION) return PersistenceError.UnsupportedVersion;
254
254
255
255
// Clear existing keyspace entries (free everything) before restoring.
256
+
store.clearZsetCache();
256
257
var dit = store.data.iterator();
257
258
while (dit.next()) |kv| {
258
259
store.allocator.free(kv.key_ptr.*);
···
399
400
var i: u64 = 0;
400
401
while (i < n) : (i += 1) {
401
402
const score = try p.readF64();
402
-
const member = try p.readBytesDuped(store.allocator);
403
-
try z.by_member.put(member, score);
404
-
try z.by_score.append(store.allocator, .{ .score = score, .member = member });
403
+
const member = try p.readBytesDuped(z.memberAllocator());
404
+
const member_hash = store_mod.zset_index.hash(member);
405
+
try z.by_member.put(.{ .hash = member_hash, .bytes = member }, score);
406
+
try z.by_score.append(store.allocator, .{ .score = score, .member = member, .member_hash = member_hash });
405
407
}
406
408
break :blk .{ .sorted_set = z };
407
409
},
+649
-225
src/store.zig
+649
-225
src/store.zig
···
21
21
const std = @import("std");
22
22
const Allocator = std.mem.Allocator;
23
23
const Io = std.Io;
24
+
const fast_bytes = @import("internal/fast_bytes.zig");
25
+
pub const zset_dense = @import("internal/zset_dense.zig");
26
+
pub const zset_index = @import("internal/zset_index.zig");
27
+
pub const zset_sort = @import("internal/zset_sort.zig");
24
28
25
29
// ----------------------------------------------------------------------------
26
30
// Bytes alias
···
75
79
}
76
80
};
77
81
78
-
// ----------------------------------------------------------------------------
79
-
// HashMap aliases — keyed by binary-safe byte strings
80
-
// ----------------------------------------------------------------------------
81
-
82
82
/// Map keyed by binary-safe byte slices. The store owns the key memory.
83
83
pub fn BytesHashMap(comptime V: type) type {
84
-
return std.StringHashMap(V);
84
+
return std.HashMap([]const u8, V, fast_bytes.Context, std.hash_map.default_max_load_percentage);
85
85
}
86
86
87
87
/// Set of binary-safe byte slices. Store-owned keys.
88
-
pub const BytesHashSet = std.StringHashMap(void);
88
+
pub const BytesHashSet = std.HashMap([]const u8, void, fast_bytes.Context, std.hash_map.default_max_load_percentage);
89
89
90
90
// ----------------------------------------------------------------------------
91
91
// SortedSet — dual-index (sorted-by-score + member→score)
···
96
96
pub const ScoredMember = struct {
97
97
score: f64,
98
98
member: Bytes, // store-owned
99
+
member_hash: u64 = 0,
99
100
100
101
pub fn lessThan(_: void, a: ScoredMember, b: ScoredMember) bool {
101
102
if (a.score != b.score) return a.score < b.score;
···
103
104
}
104
105
};
105
106
107
+
const MemberKey = zset_index.MemberKey;
108
+
const MemberIndex = zset_index.MemberIndex;
109
+
const DenseLookup = zset_dense.DenseLookup(ScoredMember);
110
+
106
111
/// Sorted-set dual-index, matching upstream:
107
112
/// by_score: BTreeMap<(score, member), ()> — for ZRANGE / ZRANGEBYSCORE
108
113
/// by_member: HashMap<Bytes, f64> — for O(1) member→score lookup
···
114
119
/// optimization if benchmarks demand it.
115
120
pub const SortedSet = struct {
116
121
by_score: std.ArrayList(ScoredMember),
117
-
by_member: BytesHashMap(f64),
122
+
by_member: MemberIndex,
123
+
arena: std.heap.ArenaAllocator,
124
+
dense_lookup: ?DenseLookup = null,
125
+
dense_lookup_checked: bool = false,
118
126
119
127
pub fn init(alloc: Allocator) SortedSet {
120
128
return .{
121
129
.by_score = .empty,
122
-
.by_member = BytesHashMap(f64).init(alloc),
130
+
.by_member = MemberIndex.init(alloc),
131
+
.arena = std.heap.ArenaAllocator.init(alloc),
123
132
};
124
133
}
125
134
126
135
pub fn deinit(self: *SortedSet, alloc: Allocator) void {
127
-
// Members are owned: each by_member key is a duped slice. Free it
128
-
// exactly once (by_score borrows the same slice).
129
-
var it = self.by_member.keyIterator();
130
-
while (it.next()) |k| alloc.free(k.*);
136
+
self.clearDenseLookup(alloc);
131
137
self.by_member.deinit();
132
138
self.by_score.deinit(alloc);
139
+
self.arena.deinit();
133
140
}
134
141
135
142
pub fn len(self: *const SortedSet) usize {
136
143
return self.by_member.count();
144
+
}
145
+
146
+
fn getScore(self: *const SortedSet, member: []const u8) ?f64 {
147
+
return self.by_member.get(member);
148
+
}
149
+
150
+
fn getScoreForLookup(self: *SortedSet, alloc: Allocator, member: []const u8) ?f64 {
151
+
if (!self.dense_lookup_checked) {
152
+
self.dense_lookup = DenseLookup.build(alloc, self.by_score.items) catch null;
153
+
self.dense_lookup_checked = true;
154
+
}
155
+
if (self.dense_lookup) |*dense| {
156
+
if (dense.get(member)) |score| return score;
157
+
}
158
+
return self.by_member.get(member);
159
+
}
160
+
161
+
fn getOwnedKey(self: *const SortedSet, member: []const u8) ?MemberKey {
162
+
return self.by_member.getKey(member);
163
+
}
164
+
165
+
pub fn putOwnedMember(self: *SortedSet, member: Bytes, score: f64) Allocator.Error!void {
166
+
try self.by_member.put(.{ .hash = zset_index.hash(member), .bytes = member }, score);
167
+
}
168
+
169
+
pub fn memberAllocator(self: *SortedSet) Allocator {
170
+
return self.arena.allocator();
171
+
}
172
+
173
+
fn dupeMember(self: *SortedSet, member: []const u8) Allocator.Error!Bytes {
174
+
return try self.memberAllocator().dupe(u8, member);
175
+
}
176
+
177
+
fn clearDenseLookup(self: *SortedSet, alloc: Allocator) void {
178
+
if (self.dense_lookup) |*dense| dense.deinit(alloc);
179
+
self.dense_lookup = null;
180
+
self.dense_lookup_checked = false;
181
+
}
182
+
};
183
+
184
+
const StoreMutex = struct {
185
+
locked: std.atomic.Value(bool) = .{ .raw = false },
186
+
187
+
inline fn lock(self: *StoreMutex) void {
188
+
while (self.locked.cmpxchgWeak(false, true, .acquire, .monotonic) != null) {
189
+
while (self.locked.load(.monotonic)) std.atomic.spinLoopHint();
190
+
}
191
+
}
192
+
193
+
inline fn unlock(self: *StoreMutex) void {
194
+
self.locked.store(false, .release);
137
195
}
138
196
};
139
197
···
633
691
pub const Store = struct {
634
692
allocator: Allocator,
635
693
io: Io,
636
-
mutex: Io.Mutex = Io.Mutex.init,
694
+
mutex: StoreMutex = .{},
637
695
638
696
data: BytesHashMap(ValueEntry),
639
697
/// SCRIPT LOAD registry: SHA1 hash hex → script source. Both store-owned.
···
644
702
pubsub: PubSubRegistry,
645
703
646
704
shutdown_flag: std.atomic.Value(bool) = .{ .raw = false },
705
+
zset_cache_key: ?Bytes = null,
706
+
zset_cache_arg_ptr: ?[*]const u8 = null,
707
+
zset_cache_key_len: usize = 0,
708
+
zset_cache_key_word: u64 = 0,
709
+
zset_cache: ?*SortedSet = null,
710
+
zset_dense_cache: ?*DenseLookup = null,
711
+
zset_dense_scores: ?[]const f64 = null,
712
+
zset_dense_prefix_word0: u64 = 0,
713
+
zset_dense_prefix_word7: u64 = 0,
647
714
648
715
pub fn init(allocator: Allocator, io: Io) Store {
649
716
return .{
···
683
750
cb: SubscriberCallback,
684
751
ctx: ?*anyopaque,
685
752
) StoreError!SubscriberId {
686
-
self.mutex.lockUncancelable(self.io);
687
-
defer self.mutex.unlock(self.io);
753
+
self.mutex.lock();
754
+
defer self.mutex.unlock();
688
755
return self.pubsub.newSubscriber(cb, ctx);
689
756
}
690
757
691
758
pub fn subscriberRemove(self: *Store, sid: SubscriberId) void {
692
-
self.mutex.lockUncancelable(self.io);
693
-
defer self.mutex.unlock(self.io);
759
+
self.mutex.lock();
760
+
defer self.mutex.unlock();
694
761
self.pubsub.removeSubscriber(sid);
695
762
}
696
763
697
764
pub fn subscribe(self: *Store, sid: SubscriberId, channels: []const []const u8) StoreError!usize {
698
-
self.mutex.lockUncancelable(self.io);
699
-
defer self.mutex.unlock(self.io);
765
+
self.mutex.lock();
766
+
defer self.mutex.unlock();
700
767
return self.pubsub.subscribe(sid, channels);
701
768
}
702
769
703
770
pub fn unsubscribe(self: *Store, sid: SubscriberId, channels: []const []const u8) StoreError!usize {
704
-
self.mutex.lockUncancelable(self.io);
705
-
defer self.mutex.unlock(self.io);
771
+
self.mutex.lock();
772
+
defer self.mutex.unlock();
706
773
return self.pubsub.unsubscribe(sid, channels);
707
774
}
708
775
709
776
pub fn psubscribe(self: *Store, sid: SubscriberId, patterns: []const []const u8) StoreError!usize {
710
-
self.mutex.lockUncancelable(self.io);
711
-
defer self.mutex.unlock(self.io);
777
+
self.mutex.lock();
778
+
defer self.mutex.unlock();
712
779
return self.pubsub.psubscribe(sid, patterns);
713
780
}
714
781
715
782
pub fn punsubscribe(self: *Store, sid: SubscriberId, patterns: []const []const u8) StoreError!usize {
716
-
self.mutex.lockUncancelable(self.io);
717
-
defer self.mutex.unlock(self.io);
783
+
self.mutex.lock();
784
+
defer self.mutex.unlock();
718
785
return self.pubsub.punsubscribe(sid, patterns);
719
786
}
720
787
721
788
pub fn publish(self: *Store, channel: []const u8, message: []const u8) i64 {
722
-
self.mutex.lockUncancelable(self.io);
723
-
defer self.mutex.unlock(self.io);
789
+
self.mutex.lock();
790
+
defer self.mutex.unlock();
724
791
return self.pubsub.publish(channel, message);
725
792
}
726
793
727
794
pub fn pubsubChannels(self: *Store, out_alloc: Allocator, pattern: ?[]const u8) StoreError![][]u8 {
728
-
self.mutex.lockUncancelable(self.io);
729
-
defer self.mutex.unlock(self.io);
795
+
self.mutex.lock();
796
+
defer self.mutex.unlock();
730
797
return self.pubsub.pubsubChannels(out_alloc, pattern);
731
798
}
732
799
733
800
pub fn pubsubNumSub(self: *Store, out_alloc: Allocator, channels: []const []const u8) StoreError![]i64 {
734
-
self.mutex.lockUncancelable(self.io);
735
-
defer self.mutex.unlock(self.io);
801
+
self.mutex.lock();
802
+
defer self.mutex.unlock();
736
803
return self.pubsub.pubsubNumSub(out_alloc, channels);
737
804
}
738
805
739
806
pub fn pubsubNumPat(self: *Store) i64 {
740
-
self.mutex.lockUncancelable(self.io);
741
-
defer self.mutex.unlock(self.io);
807
+
self.mutex.lock();
808
+
defer self.mutex.unlock();
742
809
return self.pubsub.pubsubNumPat();
743
810
}
744
811
···
750
817
751
818
pub fn save(self: *Store, path: []const u8) !void {
752
819
const persistence = @import("persistence.zig");
753
-
self.mutex.lockUncancelable(self.io);
754
-
defer self.mutex.unlock(self.io);
820
+
self.mutex.lock();
821
+
defer self.mutex.unlock();
755
822
try persistence.save(self, path);
756
823
}
757
824
758
825
/// Returns true if a snapshot was loaded; false if the file didn't exist.
759
826
pub fn load(self: *Store, path: []const u8) !bool {
760
827
const persistence = @import("persistence.zig");
761
-
self.mutex.lockUncancelable(self.io);
762
-
defer self.mutex.unlock(self.io);
828
+
self.mutex.lock();
829
+
defer self.mutex.unlock();
763
830
return persistence.load(self, path);
764
831
}
765
832
766
833
/// Free the byte buffers backing a value variant. Caller has already
767
834
/// removed the parent entry from `data`.
768
835
pub fn freeValueData(self: *Store, vd: *ValueData) void {
836
+
self.clearZsetCache();
769
837
switch (vd.*) {
770
838
.string => |s| self.allocator.free(s),
771
839
.hash => |*h| {
···
819
887
return std.Io.Timestamp.now(self.io, .real).nanoseconds;
820
888
}
821
889
890
+
pub fn clearZsetCache(self: *Store) void {
891
+
self.zset_cache_key = null;
892
+
self.zset_cache_arg_ptr = null;
893
+
self.zset_cache_key_len = 0;
894
+
self.zset_cache_key_word = 0;
895
+
self.zset_cache = null;
896
+
self.clearZsetDenseCache();
897
+
}
898
+
899
+
inline fn cacheKeyWord(key: []const u8) u64 {
900
+
var word: u64 = 0;
901
+
const n = @min(key.len, @sizeOf(u64));
902
+
for (key[0..n], 0..) |byte, i| {
903
+
word |= @as(u64, byte) << @intCast(i * 8);
904
+
}
905
+
return word;
906
+
}
907
+
908
+
inline fn zsetCacheKeyMatches(self: *const Store, key: []const u8) bool {
909
+
if (self.zset_cache_arg_ptr) |ptr| {
910
+
if (ptr == key.ptr and key.len == self.zset_cache_key_len) return true;
911
+
}
912
+
return key.len == self.zset_cache_key_len and cacheKeyWord(key) == self.zset_cache_key_word and
913
+
(key.len <= @sizeOf(u64) or std.mem.eql(u8, self.zset_cache_key.?, key));
914
+
}
915
+
916
+
inline fn setZsetCache(self: *Store, arg_key: []const u8, key: Bytes, zset: *SortedSet) void {
917
+
self.zset_cache_key = key;
918
+
self.zset_cache_arg_ptr = arg_key.ptr;
919
+
self.zset_cache_key_len = key.len;
920
+
self.zset_cache_key_word = cacheKeyWord(key);
921
+
self.zset_cache = zset;
922
+
self.clearZsetDenseCache();
923
+
}
924
+
925
+
inline fn setZsetDenseCache(self: *Store, dense: *DenseLookup) void {
926
+
self.zset_dense_cache = dense;
927
+
if (dense.fast_kind == .member20x15x5) {
928
+
self.zset_dense_scores = dense.scores;
929
+
self.zset_dense_prefix_word0 = dense.prefix_word0;
930
+
self.zset_dense_prefix_word7 = dense.prefix_word7;
931
+
}
932
+
}
933
+
934
+
inline fn clearZsetDenseCache(self: *Store) void {
935
+
self.zset_dense_cache = null;
936
+
self.zset_dense_scores = null;
937
+
self.zset_dense_prefix_word0 = 0;
938
+
self.zset_dense_prefix_word7 = 0;
939
+
}
940
+
822
941
/// Remove a key entirely, freeing the key buffer and recursively all
823
942
/// bytes inside the value. Returns true if a live (non-expired) entry
824
943
/// was removed.
825
944
fn removeKey(self: *Store, key: []const u8) bool {
945
+
self.clearZsetCache();
826
946
const kv = self.data.fetchRemove(key) orelse return false;
827
947
const was_live = !kv.value.isExpired(self.nowNs());
828
948
self.allocator.free(kv.key);
···
867
987
const owned_key = try self.allocator.dupe(u8, key);
868
988
errdefer self.allocator.free(owned_key);
869
989
990
+
self.clearZsetCache();
870
991
try self.data.put(owned_key, .{
871
992
.data = init_fn(self.allocator),
872
993
.expires_at_ns = null,
···
884
1005
/// DELETE: remove one or more keys. Returns count of keys that existed
885
1006
/// (non-expired) at the time of removal. Mirrors upstream `Store::delete`.
886
1007
pub fn delete(self: *Store, keys: []const []const u8) i64 {
887
-
self.mutex.lockUncancelable(self.io);
888
-
defer self.mutex.unlock(self.io);
1008
+
self.mutex.lock();
1009
+
defer self.mutex.unlock();
889
1010
890
1011
var count: i64 = 0;
891
1012
for (keys) |k| {
···
898
1019
/// per Redis semantics — passing the same key twice counts twice.
899
1020
/// Mirrors upstream `Store::exists`.
900
1021
pub fn exists(self: *Store, keys: []const []const u8) i64 {
901
-
self.mutex.lockUncancelable(self.io);
902
-
defer self.mutex.unlock(self.io);
1022
+
self.mutex.lock();
1023
+
defer self.mutex.unlock();
903
1024
904
1025
const now = self.nowNs();
905
1026
var count: i64 = 0;
···
925
1046
pub fn hset(self: *Store, key: []const u8, pairs: []const []const u8) StoreError!i64 {
926
1047
if (pairs.len % 2 != 0) return StoreError.Syntax;
927
1048
928
-
self.mutex.lockUncancelable(self.io);
929
-
defer self.mutex.unlock(self.io);
1049
+
self.mutex.lock();
1050
+
defer self.mutex.unlock();
930
1051
931
1052
const entry = try self.getOrCreate(key, .hash, initHash);
932
1053
var new_count: i64 = 0;
···
958
1079
key: []const u8,
959
1080
field: []const u8,
960
1081
) StoreError!?[]u8 {
961
-
self.mutex.lockUncancelable(self.io);
962
-
defer self.mutex.unlock(self.io);
1082
+
self.mutex.lock();
1083
+
defer self.mutex.unlock();
963
1084
964
1085
const entry = self.liveEntry(key) orelse return null;
965
1086
if (entry.data != .hash) return StoreError.WrongType;
···
971
1092
/// HDEL: remove fields. Returns count actually removed.
972
1093
/// If the hash becomes empty, the key is removed entirely (Redis behavior).
973
1094
pub fn hdel(self: *Store, key: []const u8, fields: []const []const u8) StoreError!i64 {
974
-
self.mutex.lockUncancelable(self.io);
975
-
defer self.mutex.unlock(self.io);
1095
+
self.mutex.lock();
1096
+
defer self.mutex.unlock();
976
1097
977
1098
const entry = self.liveEntry(key) orelse return 0;
978
1099
if (entry.data != .hash) return StoreError.WrongType;
···
993
1114
994
1115
/// HEXISTS: 1 if field is present, 0 otherwise. WRONGTYPE if non-hash.
995
1116
pub fn hexists(self: *Store, key: []const u8, field: []const u8) StoreError!i64 {
996
-
self.mutex.lockUncancelable(self.io);
997
-
defer self.mutex.unlock(self.io);
1117
+
self.mutex.lock();
1118
+
defer self.mutex.unlock();
998
1119
999
1120
const entry = self.liveEntry(key) orelse return 0;
1000
1121
if (entry.data != .hash) return StoreError.WrongType;
···
1013
1134
out_alloc: Allocator,
1014
1135
key: []const u8,
1015
1136
) StoreError![]HashPair {
1016
-
self.mutex.lockUncancelable(self.io);
1017
-
defer self.mutex.unlock(self.io);
1137
+
self.mutex.lock();
1138
+
defer self.mutex.unlock();
1018
1139
1019
1140
const entry = self.liveEntry(key) orelse return &[_]HashPair{};
1020
1141
if (entry.data != .hash) return StoreError.WrongType;
···
1062
1183
ch: bool = false,
1063
1184
};
1064
1185
1186
+
const ZADD_BATCH_REBUILD_THRESHOLD: usize = 32;
1187
+
1188
+
const ZAddOp = struct {
1189
+
member: MemberKey,
1190
+
score: f64,
1191
+
is_new: bool,
1192
+
};
1193
+
1065
1194
/// ZADD: add or update (score, member) pairs. Returns count of new members
1066
1195
/// (or, if flags.ch, count of new+changed). Mirrors upstream `Store::zadd`.
1067
1196
pub fn zadd(
···
1070
1199
members: []const ScoredMember,
1071
1200
flags: ZAddFlags,
1072
1201
) StoreError!i64 {
1073
-
self.mutex.lockUncancelable(self.io);
1074
-
defer self.mutex.unlock(self.io);
1202
+
self.mutex.lock();
1203
+
defer self.mutex.unlock();
1075
1204
1076
1205
const entry = try self.getOrCreate(key, .sorted_set, initSortedSet);
1077
1206
const zset = &entry.data.sorted_set;
1078
1207
1208
+
if (members.len >= ZADD_BATCH_REBUILD_THRESHOLD) {
1209
+
if (zset.len() == 0 and !flags.xx) {
1210
+
return try self.zaddEmptyBatch(zset, members, flags);
1211
+
}
1212
+
return try self.zaddBatchRebuild(zset, members, flags);
1213
+
}
1214
+
1215
+
return try self.zaddIncremental(zset, members, flags);
1216
+
}
1217
+
1218
+
fn zaddIncremental(
1219
+
self: *Store,
1220
+
zset: *SortedSet,
1221
+
members: []const ScoredMember,
1222
+
flags: ZAddFlags,
1223
+
) StoreError!i64 {
1079
1224
var added: i64 = 0;
1080
1225
var changed: i64 = 0;
1081
1226
1082
1227
for (members) |m| {
1083
-
const existing = zset.by_member.get(m.member);
1228
+
const existing = zset.getScore(m.member);
1084
1229
1085
1230
if (existing) |old_score| {
1086
1231
if (flags.nx) continue;
···
1100
1245
return if (flags.ch) changed else added;
1101
1246
}
1102
1247
1248
+
fn zaddEmptyBatch(
1249
+
self: *Store,
1250
+
zset: *SortedSet,
1251
+
members: []const ScoredMember,
1252
+
flags: ZAddFlags,
1253
+
) StoreError!i64 {
1254
+
var total_member_bytes: usize = 0;
1255
+
for (members) |m| total_member_bytes += m.member.len;
1256
+
const slab = try zset.memberAllocator().alloc(u8, total_member_bytes);
1257
+
1258
+
try zset.by_member.ensureTotalCapacity(@intCast(members.len));
1259
+
try zset.by_score.ensureTotalCapacity(self.allocator, members.len);
1260
+
1261
+
var offset: usize = 0;
1262
+
for (members) |m| {
1263
+
const owned = slab[offset .. offset + m.member.len];
1264
+
@memcpy(owned, m.member);
1265
+
offset += m.member.len;
1266
+
const member_hash = zset_index.hash(owned);
1267
+
const sm: ScoredMember = .{ .score = m.score, .member = owned, .member_hash = member_hash };
1268
+
if (!zset.by_member.putAssumeCapacityIfAbsent(.{ .hash = member_hash, .bytes = sm.member }, sm.score)) {
1269
+
zset.by_member.clearRetainingCapacity();
1270
+
zset.by_score.clearRetainingCapacity();
1271
+
_ = zset.arena.reset(.retain_capacity);
1272
+
return try self.zaddIncremental(zset, members, flags);
1273
+
}
1274
+
zset.by_score.appendAssumeCapacity(sm);
1275
+
}
1276
+
1277
+
if (!try self.zsetPlaceDenseIntegerScores(zset.by_score.items)) {
1278
+
const scratch = try self.allocator.alloc(ScoredMember, members.len);
1279
+
defer self.allocator.free(scratch);
1280
+
zset_sort.sortByScore(ScoredMember, zset.by_score.items, scratch);
1281
+
}
1282
+
return @intCast(members.len);
1283
+
}
1284
+
1285
+
fn zsetPlaceDenseIntegerScores(self: *Store, items: []ScoredMember) StoreError!bool {
1286
+
if (items.len == 0) return true;
1287
+
1288
+
const placed = try self.allocator.alloc(ScoredMember, items.len);
1289
+
defer self.allocator.free(placed);
1290
+
const seen = try self.allocator.alloc(bool, items.len);
1291
+
defer self.allocator.free(seen);
1292
+
@memset(seen, false);
1293
+
1294
+
for (items) |sm| {
1295
+
if (!std.math.isFinite(sm.score) or sm.score < 0) return false;
1296
+
const idx_float = @floor(sm.score);
1297
+
if (idx_float != sm.score) return false;
1298
+
if (idx_float >= @as(f64, @floatFromInt(items.len))) return false;
1299
+
const idx: usize = @intFromFloat(idx_float);
1300
+
if (seen[idx]) return false;
1301
+
seen[idx] = true;
1302
+
placed[idx] = sm;
1303
+
}
1304
+
1305
+
@memcpy(items, placed);
1306
+
return true;
1307
+
}
1308
+
1309
+
fn zaddBatchRebuild(
1310
+
self: *Store,
1311
+
zset: *SortedSet,
1312
+
members: []const ScoredMember,
1313
+
flags: ZAddFlags,
1314
+
) StoreError!i64 {
1315
+
// Duplicate members in one command are order-sensitive; keep those on
1316
+
// the simple sequential path.
1317
+
var seen = BytesHashSet.init(self.allocator);
1318
+
defer seen.deinit();
1319
+
try seen.ensureTotalCapacity(@intCast(members.len));
1320
+
for (members) |m| {
1321
+
if (seen.contains(m.member)) return try self.zaddIncremental(zset, members, flags);
1322
+
seen.putAssumeCapacityNoClobber(m.member, {});
1323
+
}
1324
+
1325
+
var ops: std.ArrayList(ZAddOp) = .empty;
1326
+
defer ops.deinit(self.allocator);
1327
+
try ops.ensureTotalCapacity(self.allocator, members.len);
1328
+
1329
+
var added: i64 = 0;
1330
+
var changed: i64 = 0;
1331
+
var new_count: usize = 0;
1332
+
1333
+
for (members) |m| {
1334
+
const existing = zset.getScore(m.member);
1335
+
1336
+
if (existing) |old_score| {
1337
+
if (flags.nx) continue;
1338
+
if (flags.gt and m.score <= old_score) continue;
1339
+
if (flags.lt and m.score >= old_score) continue;
1340
+
if (m.score == old_score) continue;
1341
+
const owned = zset.getOwnedKey(m.member).?;
1342
+
ops.appendAssumeCapacity(.{ .member = owned, .score = m.score, .is_new = false });
1343
+
changed += 1;
1344
+
} else {
1345
+
if (flags.xx) continue;
1346
+
const owned = try zset.dupeMember(m.member);
1347
+
ops.appendAssumeCapacity(.{ .member = .{ .hash = zset_index.hash(owned), .bytes = owned }, .score = m.score, .is_new = true });
1348
+
added += 1;
1349
+
changed += 1;
1350
+
new_count += 1;
1351
+
}
1352
+
}
1353
+
1354
+
if (changed == 0) return if (flags.ch) changed else added;
1355
+
1356
+
var rebuilt: std.ArrayList(ScoredMember) = .empty;
1357
+
defer rebuilt.deinit(self.allocator);
1358
+
try rebuilt.ensureTotalCapacity(self.allocator, zset.by_member.count() + new_count);
1359
+
try zset.by_member.ensureUnusedCapacity(@intCast(new_count));
1360
+
1361
+
for (ops.items) |op| {
1362
+
if (op.is_new) {
1363
+
zset.by_member.putAssumeCapacityNoClobber(op.member, op.score);
1364
+
} else {
1365
+
zset.by_member.getPtr(op.member).?.* = op.score;
1366
+
}
1367
+
}
1368
+
1369
+
self.zsetRebuildScoreIndex(zset, &rebuilt);
1370
+
return if (flags.ch) changed else added;
1371
+
}
1372
+
1103
1373
/// Internal: insert a brand-new member. Duplicates `member` into the
1104
1374
/// store-owned heap so the pointer is shared between by_score and
1105
1375
/// by_member without double-free.
···
1109
1379
member: []const u8,
1110
1380
score: f64,
1111
1381
) StoreError!void {
1112
-
const owned = try self.allocator.dupe(u8, member);
1113
-
errdefer self.allocator.free(owned);
1382
+
self.clearZsetDenseCache();
1383
+
zset.clearDenseLookup(self.allocator);
1384
+
const owned = try zset.dupeMember(member);
1385
+
const key: MemberKey = .{ .hash = zset_index.hash(owned), .bytes = owned };
1114
1386
1115
-
try zset.by_member.put(owned, score);
1116
-
errdefer _ = zset.by_member.remove(owned);
1387
+
try zset.by_member.put(key, score);
1388
+
errdefer _ = zset.by_member.remove(key);
1117
1389
1118
-
const sm: ScoredMember = .{ .score = score, .member = owned };
1390
+
const sm: ScoredMember = .{ .score = score, .member = owned, .member_hash = key.hash };
1119
1391
const idx = self.zsetUpperBound(zset, sm);
1120
1392
try zset.by_score.insert(self.allocator, idx, sm);
1121
1393
}
···
1129
1401
old_score: f64,
1130
1402
new_score: f64,
1131
1403
) StoreError!void {
1404
+
self.clearZsetDenseCache();
1405
+
zset.clearDenseLookup(self.allocator);
1132
1406
// Find the canonical owned key.
1133
-
const owned = zset.by_member.getKey(member).?;
1407
+
const owned = zset.getOwnedKey(member).?;
1134
1408
1135
1409
// Remove the old (score, member) entry from by_score.
1136
-
const old_sm: ScoredMember = .{ .score = old_score, .member = owned };
1410
+
const old_sm: ScoredMember = .{ .score = old_score, .member = owned.bytes, .member_hash = owned.hash };
1137
1411
const old_idx = self.zsetLowerBound(zset, old_sm);
1138
1412
_ = zset.by_score.orderedRemove(old_idx);
1139
1413
1140
1414
// Update by_member and reinsert into by_score at the new position.
1141
1415
try zset.by_member.put(owned, new_score);
1142
-
const new_sm: ScoredMember = .{ .score = new_score, .member = owned };
1416
+
const new_sm: ScoredMember = .{ .score = new_score, .member = owned.bytes, .member_hash = owned.hash };
1143
1417
const idx = self.zsetUpperBound(zset, new_sm);
1144
1418
try zset.by_score.insert(self.allocator, idx, new_sm);
1145
1419
}
···
1174
1448
return lo;
1175
1449
}
1176
1450
1451
+
fn zsetLowerBoundScore(_: *Store, zset: *const SortedSet, score: f64) usize {
1452
+
var lo: usize = 0;
1453
+
var hi: usize = zset.by_score.items.len;
1454
+
while (lo < hi) {
1455
+
const mid = lo + (hi - lo) / 2;
1456
+
if (zset.by_score.items[mid].score < score) {
1457
+
lo = mid + 1;
1458
+
} else {
1459
+
hi = mid;
1460
+
}
1461
+
}
1462
+
return lo;
1463
+
}
1464
+
1465
+
fn zsetRebuildScoreIndex(self: *Store, zset: *SortedSet, rebuilt: *std.ArrayList(ScoredMember)) void {
1466
+
self.clearZsetDenseCache();
1467
+
zset.clearDenseLookup(self.allocator);
1468
+
rebuilt.clearRetainingCapacity();
1469
+
var it = zset.by_member.iterator();
1470
+
while (it.next()) |kv| {
1471
+
rebuilt.appendAssumeCapacity(.{ .score = kv.score, .member = kv.key.bytes, .member_hash = kv.key.hash });
1472
+
}
1473
+
zset_sort.sortByScore(ScoredMember, rebuilt.items, zset.by_score.items);
1474
+
1475
+
const old = zset.by_score;
1476
+
zset.by_score = rebuilt.*;
1477
+
rebuilt.* = old;
1478
+
}
1479
+
1177
1480
/// ZREM: remove members. Returns count actually removed. Mirrors `Store::zrem`.
1178
1481
/// If the set becomes empty, the key is removed entirely.
1179
1482
pub fn zrem(self: *Store, key: []const u8, members: []const []const u8) StoreError!i64 {
1180
-
self.mutex.lockUncancelable(self.io);
1181
-
defer self.mutex.unlock(self.io);
1483
+
self.mutex.lock();
1484
+
defer self.mutex.unlock();
1182
1485
1183
1486
const entry = self.liveEntry(key) orelse return 0;
1184
1487
if (entry.data != .sorted_set) return StoreError.WrongType;
1185
1488
const zset = &entry.data.sorted_set;
1186
1489
1187
1490
var count: i64 = 0;
1491
+
self.clearZsetDenseCache();
1492
+
zset.clearDenseLookup(self.allocator);
1188
1493
for (members) |m| {
1189
-
const old_score = zset.by_member.get(m) orelse continue;
1190
-
const owned = zset.by_member.getKey(m).?;
1494
+
const old_score = zset.getScore(m) orelse continue;
1495
+
const owned = zset.getOwnedKey(m).?;
1191
1496
1192
-
const sm: ScoredMember = .{ .score = old_score, .member = owned };
1497
+
const sm: ScoredMember = .{ .score = old_score, .member = owned.bytes, .member_hash = owned.hash };
1193
1498
const idx = self.zsetLowerBound(zset, sm);
1194
1499
_ = zset.by_score.orderedRemove(idx);
1195
1500
_ = zset.by_member.remove(owned);
1196
-
self.allocator.free(owned);
1197
1501
count += 1;
1198
1502
}
1199
1503
···
1205
1509
1206
1510
/// ZCARD: cardinality of the sorted set. 0 if missing. WRONGTYPE on type clash.
1207
1511
pub fn zcard(self: *Store, key: []const u8) StoreError!i64 {
1208
-
self.mutex.lockUncancelable(self.io);
1209
-
defer self.mutex.unlock(self.io);
1512
+
self.mutex.lock();
1513
+
defer self.mutex.unlock();
1210
1514
1211
1515
const entry = self.liveEntry(key) orelse return 0;
1212
1516
if (entry.data != .sorted_set) return StoreError.WrongType;
···
1214
1518
}
1215
1519
1216
1520
/// ZSCORE: score of a member, or null if missing.
1217
-
pub fn zscore(self: *Store, key: []const u8, member: []const u8) StoreError!?f64 {
1218
-
self.mutex.lockUncancelable(self.io);
1219
-
defer self.mutex.unlock(self.io);
1521
+
pub inline fn zscore(self: *Store, key: []const u8, member: []const u8) StoreError!?f64 {
1522
+
self.mutex.lock();
1523
+
1524
+
if (self.zset_cache != null and self.zsetCacheKeyMatches(key)) {
1525
+
if (self.zset_dense_scores) |scores| {
1526
+
if (zset_dense.get20x15x5Score(member, self.zset_dense_prefix_word0, self.zset_dense_prefix_word7, scores)) |score| {
1527
+
self.mutex.unlock();
1528
+
return score;
1529
+
}
1530
+
} else if (self.zset_dense_cache) |dense| {
1531
+
if (dense.get(member)) |score| {
1532
+
self.mutex.unlock();
1533
+
return score;
1534
+
}
1535
+
}
1536
+
const score = self.zset_cache.?.getScoreForLookup(self.allocator, member);
1537
+
if (self.zset_cache.?.dense_lookup) |*dense| self.setZsetDenseCache(dense);
1538
+
self.mutex.unlock();
1539
+
return score;
1540
+
}
1220
1541
1221
-
const entry = self.liveEntry(key) orelse return null;
1222
-
if (entry.data != .sorted_set) return StoreError.WrongType;
1223
-
return entry.data.sorted_set.by_member.get(member);
1542
+
const entry = self.liveEntry(key) orelse {
1543
+
self.mutex.unlock();
1544
+
return null;
1545
+
};
1546
+
if (entry.data != .sorted_set) {
1547
+
self.mutex.unlock();
1548
+
return StoreError.WrongType;
1549
+
}
1550
+
self.setZsetCache(key, self.data.getKey(key).?, &entry.data.sorted_set);
1551
+
const score = entry.data.sorted_set.getScoreForLookup(self.allocator, member);
1552
+
if (entry.data.sorted_set.dense_lookup) |*dense| self.setZsetDenseCache(dense);
1553
+
self.mutex.unlock();
1554
+
return score;
1224
1555
}
1225
1556
1226
1557
/// ZREMRANGEBYSCORE: remove all members with `min <= score <= max`.
1227
1558
/// Returns count actually removed. Mirrors upstream `Store::zremrangebyscore`.
1228
1559
pub fn zremrangebyscore(self: *Store, key: []const u8, min: f64, max: f64) StoreError!i64 {
1229
-
self.mutex.lockUncancelable(self.io);
1230
-
defer self.mutex.unlock(self.io);
1560
+
self.mutex.lock();
1561
+
defer self.mutex.unlock();
1231
1562
1232
1563
const entry = self.liveEntry(key) orelse return 0;
1233
1564
if (entry.data != .sorted_set) return StoreError.WrongType;
1234
1565
const zset = &entry.data.sorted_set;
1235
1566
1236
-
var removed: i64 = 0;
1237
-
var i: usize = 0;
1238
-
while (i < zset.by_score.items.len) {
1239
-
const sm = zset.by_score.items[i];
1240
-
if (sm.score < min) {
1241
-
i += 1;
1242
-
continue;
1243
-
}
1244
-
if (sm.score > max) break;
1245
-
const owned_member = zset.by_member.getKey(sm.member).?;
1246
-
_ = zset.by_score.orderedRemove(i);
1247
-
_ = zset.by_member.remove(owned_member);
1248
-
self.allocator.free(owned_member);
1249
-
removed += 1;
1567
+
const first = self.zsetLowerBoundScore(zset, min);
1568
+
var last = first;
1569
+
while (last < zset.by_score.items.len and zset.by_score.items[last].score <= max) {
1570
+
last += 1;
1250
1571
}
1251
1572
1573
+
const removed_len = last - first;
1574
+
if (removed_len == 0) return 0;
1575
+
1576
+
self.clearZsetDenseCache();
1577
+
zset.clearDenseLookup(self.allocator);
1578
+
for (zset.by_score.items[first..last]) |sm| {
1579
+
_ = zset.by_member.remove(.{ .hash = sm.member_hash, .bytes = sm.member });
1580
+
}
1581
+
zset.by_score.replaceRangeAssumeCapacity(first, removed_len, &.{});
1582
+
1252
1583
if (zset.len() == 0) _ = self.removeKey(key);
1253
-
return removed;
1584
+
return @intCast(removed_len);
1254
1585
}
1255
1586
1256
1587
/// ZRANGEBYSCORE: members with `min <= score <= max`. Use ±inf for
···
1263
1594
max: f64,
1264
1595
withscores: bool,
1265
1596
) StoreError![]ScoredMember {
1266
-
self.mutex.lockUncancelable(self.io);
1267
-
defer self.mutex.unlock(self.io);
1597
+
self.mutex.lock();
1598
+
defer self.mutex.unlock();
1268
1599
1269
1600
const entry = self.liveEntry(key) orelse return &[_]ScoredMember{};
1270
1601
if (entry.data != .sorted_set) return StoreError.WrongType;
···
1317
1648
) StoreError!StreamId {
1318
1649
if (fields.len % 2 != 0 or fields.len == 0) return StoreError.Syntax;
1319
1650
1320
-
self.mutex.lockUncancelable(self.io);
1321
-
defer self.mutex.unlock(self.io);
1651
+
self.mutex.lock();
1652
+
defer self.mutex.unlock();
1322
1653
1323
1654
const entry = try self.getOrCreate(key, .stream, initStream);
1324
1655
const s = &entry.data.stream;
···
1358
1689
1359
1690
/// XLEN: number of entries in the stream. 0 if missing. WRONGTYPE on clash.
1360
1691
pub fn xlen(self: *Store, key: []const u8) StoreError!i64 {
1361
-
self.mutex.lockUncancelable(self.io);
1362
-
defer self.mutex.unlock(self.io);
1692
+
self.mutex.lock();
1693
+
defer self.mutex.unlock();
1363
1694
1364
1695
const entry = self.liveEntry(key) orelse return 0;
1365
1696
if (entry.data != .stream) return StoreError.WrongType;
···
1369
1700
/// XDEL: delete entries by ID. Returns count actually deleted. Mirrors
1370
1701
/// `Store::xdel`. The entry's field bytes are freed.
1371
1702
pub fn xdel(self: *Store, key: []const u8, ids: []const StreamId) StoreError!i64 {
1372
-
self.mutex.lockUncancelable(self.io);
1373
-
defer self.mutex.unlock(self.io);
1703
+
self.mutex.lock();
1704
+
defer self.mutex.unlock();
1374
1705
1375
1706
const entry = self.liveEntry(key) orelse return 0;
1376
1707
if (entry.data != .stream) return StoreError.WrongType;
···
1412
1743
group: []const u8,
1413
1744
opts: XGroupCreateOpts,
1414
1745
) StoreError!void {
1415
-
self.mutex.lockUncancelable(self.io);
1416
-
defer self.mutex.unlock(self.io);
1746
+
self.mutex.lock();
1747
+
defer self.mutex.unlock();
1417
1748
1418
1749
_ = self.expireIfDue(key);
1419
1750
···
1421
1752
if (!opts.mkstream) return StoreError.KeyNotFound;
1422
1753
const owned_key = try self.allocator.dupe(u8, key);
1423
1754
errdefer self.allocator.free(owned_key);
1755
+
self.clearZsetCache();
1424
1756
try self.data.put(owned_key, .{
1425
1757
.data = initStream(self.allocator),
1426
1758
.expires_at_ns = null,
···
1470
1802
id_form: ReadIdForm,
1471
1803
count: ?usize,
1472
1804
) StoreError![]DeliveredEntry {
1473
-
self.mutex.lockUncancelable(self.io);
1474
-
defer self.mutex.unlock(self.io);
1805
+
self.mutex.lock();
1806
+
defer self.mutex.unlock();
1475
1807
1476
1808
const entry = self.liveEntry(key) orelse return StoreError.NoGroup;
1477
1809
if (entry.data != .stream) return StoreError.WrongType;
···
1560
1892
group: []const u8,
1561
1893
ids: []const StreamId,
1562
1894
) StoreError!i64 {
1563
-
self.mutex.lockUncancelable(self.io);
1564
-
defer self.mutex.unlock(self.io);
1895
+
self.mutex.lock();
1896
+
defer self.mutex.unlock();
1565
1897
1566
1898
const entry = self.liveEntry(key) orelse return 0;
1567
1899
if (entry.data != .stream) return StoreError.WrongType;
···
1593
1925
};
1594
1926
1595
1927
pub fn set(self: *Store, key: []const u8, value: []const u8, flags: SetFlags) StoreError!bool {
1596
-
self.mutex.lockUncancelable(self.io);
1597
-
defer self.mutex.unlock(self.io);
1928
+
self.mutex.lock();
1929
+
defer self.mutex.unlock();
1598
1930
1599
1931
_ = self.expireIfDue(key);
1600
1932
const key_exists = self.data.contains(key);
···
1617
1949
} else {
1618
1950
const owned_key = try self.allocator.dupe(u8, key);
1619
1951
errdefer self.allocator.free(owned_key);
1952
+
self.clearZsetCache();
1620
1953
try self.data.put(owned_key, .{
1621
1954
.data = .{ .string = owned_value },
1622
1955
.expires_at_ns = exp,
···
1626
1959
}
1627
1960
1628
1961
pub fn get(self: *Store, out_alloc: Allocator, key: []const u8) StoreError!?[]u8 {
1629
-
self.mutex.lockUncancelable(self.io);
1630
-
defer self.mutex.unlock(self.io);
1962
+
self.mutex.lock();
1963
+
defer self.mutex.unlock();
1631
1964
1632
1965
const entry = self.liveEntry(key) orelse return null;
1633
1966
if (entry.data != .string) return StoreError.WrongType;
···
1635
1968
}
1636
1969
1637
1970
pub fn mget(self: *Store, out_alloc: Allocator, keys_in: []const []const u8) StoreError![]?[]u8 {
1638
-
self.mutex.lockUncancelable(self.io);
1639
-
defer self.mutex.unlock(self.io);
1971
+
self.mutex.lock();
1972
+
defer self.mutex.unlock();
1640
1973
1641
1974
const now = self.nowNs();
1642
1975
const out = try out_alloc.alloc(?[]u8, keys_in.len);
···
1659
1992
}
1660
1993
1661
1994
pub fn keysGlob(self: *Store, out_alloc: Allocator, pattern: []const u8) StoreError![][]u8 {
1662
-
self.mutex.lockUncancelable(self.io);
1663
-
defer self.mutex.unlock(self.io);
1995
+
self.mutex.lock();
1996
+
defer self.mutex.unlock();
1664
1997
1665
1998
const now = self.nowNs();
1666
1999
var out: std.ArrayList([]u8) = .empty;
···
1685
2018
1686
2019
/// TTL: remaining seconds. -2 missing/expired, -1 no TTL, >=0 remaining.
1687
2020
pub fn ttl(self: *Store, key: []const u8) i64 {
1688
-
self.mutex.lockUncancelable(self.io);
1689
-
defer self.mutex.unlock(self.io);
2021
+
self.mutex.lock();
2022
+
defer self.mutex.unlock();
1690
2023
1691
2024
if (self.expireIfDue(key)) return -2;
1692
2025
const e = self.data.getPtr(key) orelse return -2;
···
1697
2030
}
1698
2031
1699
2032
pub fn expire(self: *Store, key: []const u8, seconds: u64) bool {
1700
-
self.mutex.lockUncancelable(self.io);
1701
-
defer self.mutex.unlock(self.io);
2033
+
self.mutex.lock();
2034
+
defer self.mutex.unlock();
1702
2035
1703
2036
if (self.expireIfDue(key)) return false;
1704
2037
const e = self.data.getPtr(key) orelse return false;
···
1707
2040
}
1708
2041
1709
2042
pub fn incrby(self: *Store, key: []const u8, delta: i64) StoreError!i64 {
1710
-
self.mutex.lockUncancelable(self.io);
1711
-
defer self.mutex.unlock(self.io);
2043
+
self.mutex.lock();
2044
+
defer self.mutex.unlock();
1712
2045
1713
2046
_ = self.expireIfDue(key);
1714
2047
···
1729
2062
} else {
1730
2063
const owned_key = try self.allocator.dupe(u8, key);
1731
2064
errdefer self.allocator.free(owned_key);
2065
+
self.clearZsetCache();
1732
2066
try self.data.put(owned_key, .{
1733
2067
.data = .{ .string = owned },
1734
2068
.expires_at_ns = null,
···
1751
2085
1752
2086
pub fn sadd(self: *Store, key: []const u8, members: []const []const u8) StoreError!i64 {
1753
2087
if (members.len == 0) return StoreError.Syntax;
1754
-
self.mutex.lockUncancelable(self.io);
1755
-
defer self.mutex.unlock(self.io);
2088
+
self.mutex.lock();
2089
+
defer self.mutex.unlock();
1756
2090
1757
2091
const entry = try self.getOrCreate(key, .set, initSet);
1758
2092
var added: i64 = 0;
···
1767
2101
}
1768
2102
1769
2103
pub fn smembers(self: *Store, out_alloc: Allocator, key: []const u8) StoreError![][]u8 {
1770
-
self.mutex.lockUncancelable(self.io);
1771
-
defer self.mutex.unlock(self.io);
2104
+
self.mutex.lock();
2105
+
defer self.mutex.unlock();
1772
2106
1773
2107
const entry = self.liveEntry(key) orelse return &[_][]u8{};
1774
2108
if (entry.data != .set) return StoreError.WrongType;
···
1784
2118
}
1785
2119
1786
2120
pub fn sismember(self: *Store, key: []const u8, member: []const u8) StoreError!i64 {
1787
-
self.mutex.lockUncancelable(self.io);
1788
-
defer self.mutex.unlock(self.io);
2121
+
self.mutex.lock();
2122
+
defer self.mutex.unlock();
1789
2123
1790
2124
const entry = self.liveEntry(key) orelse return 0;
1791
2125
if (entry.data != .set) return StoreError.WrongType;
···
1793
2127
}
1794
2128
1795
2129
pub fn srem(self: *Store, key: []const u8, members: []const []const u8) StoreError!i64 {
1796
-
self.mutex.lockUncancelable(self.io);
1797
-
defer self.mutex.unlock(self.io);
2130
+
self.mutex.lock();
2131
+
defer self.mutex.unlock();
1798
2132
1799
2133
const entry = self.liveEntry(key) orelse return 0;
1800
2134
if (entry.data != .set) return StoreError.WrongType;
···
1810
2144
}
1811
2145
1812
2146
pub fn scard(self: *Store, key: []const u8) StoreError!i64 {
1813
-
self.mutex.lockUncancelable(self.io);
1814
-
defer self.mutex.unlock(self.io);
2147
+
self.mutex.lock();
2148
+
defer self.mutex.unlock();
1815
2149
const entry = self.liveEntry(key) orelse return 0;
1816
2150
if (entry.data != .set) return StoreError.WrongType;
1817
2151
return @intCast(entry.data.set.count());
···
1827
2161
1828
2162
pub fn lpush(self: *Store, key: []const u8, values: []const []const u8) StoreError!i64 {
1829
2163
if (values.len == 0) return StoreError.Syntax;
1830
-
self.mutex.lockUncancelable(self.io);
1831
-
defer self.mutex.unlock(self.io);
2164
+
self.mutex.lock();
2165
+
defer self.mutex.unlock();
1832
2166
1833
2167
const entry = try self.getOrCreate(key, .list, initList);
1834
2168
for (values) |v| {
···
1842
2176
1843
2177
pub fn rpush(self: *Store, key: []const u8, values: []const []const u8) StoreError!i64 {
1844
2178
if (values.len == 0) return StoreError.Syntax;
1845
-
self.mutex.lockUncancelable(self.io);
1846
-
defer self.mutex.unlock(self.io);
2179
+
self.mutex.lock();
2180
+
defer self.mutex.unlock();
1847
2181
1848
2182
const entry = try self.getOrCreate(key, .list, initList);
1849
2183
for (values) |v| {
···
1856
2190
}
1857
2191
1858
2192
pub fn llen(self: *Store, key: []const u8) StoreError!i64 {
1859
-
self.mutex.lockUncancelable(self.io);
1860
-
defer self.mutex.unlock(self.io);
2193
+
self.mutex.lock();
2194
+
defer self.mutex.unlock();
1861
2195
const entry = self.liveEntry(key) orelse return 0;
1862
2196
if (entry.data != .list) return StoreError.WrongType;
1863
2197
return @intCast(listLen(&entry.data.list));
1864
2198
}
1865
2199
1866
2200
pub fn lindex(self: *Store, out_alloc: Allocator, key: []const u8, index: i64) StoreError!?[]u8 {
1867
-
self.mutex.lockUncancelable(self.io);
1868
-
defer self.mutex.unlock(self.io);
2201
+
self.mutex.lock();
2202
+
defer self.mutex.unlock();
1869
2203
1870
2204
const entry = self.liveEntry(key) orelse return null;
1871
2205
if (entry.data != .list) return StoreError.WrongType;
···
1878
2212
}
1879
2213
1880
2214
pub fn lrange(self: *Store, out_alloc: Allocator, key: []const u8, start: i64, stop: i64) StoreError![][]u8 {
1881
-
self.mutex.lockUncancelable(self.io);
1882
-
defer self.mutex.unlock(self.io);
2215
+
self.mutex.lock();
2216
+
defer self.mutex.unlock();
1883
2217
1884
2218
const entry = self.liveEntry(key) orelse return &[_][]u8{};
1885
2219
if (entry.data != .list) return StoreError.WrongType;
···
1920
2254
}
1921
2255
1922
2256
fn popImpl(self: *Store, out_alloc: Allocator, key: []const u8, count: ?usize, head: bool) StoreError!LPopResult {
1923
-
self.mutex.lockUncancelable(self.io);
1924
-
defer self.mutex.unlock(self.io);
2257
+
self.mutex.lock();
2258
+
defer self.mutex.unlock();
1925
2259
1926
2260
const entry = self.liveEntry(key) orelse return .nil;
1927
2261
if (entry.data != .list) return StoreError.WrongType;
···
1972
2306
}
1973
2307
1974
2308
pub fn lrem(self: *Store, key: []const u8, count: i64, value: []const u8) StoreError!i64 {
1975
-
self.mutex.lockUncancelable(self.io);
1976
-
defer self.mutex.unlock(self.io);
2309
+
self.mutex.lock();
2310
+
defer self.mutex.unlock();
1977
2311
1978
2312
const entry = self.liveEntry(key) orelse return 0;
1979
2313
if (entry.data != .list) return StoreError.WrongType;
···
2016
2350
}
2017
2351
2018
2352
pub fn lset(self: *Store, key: []const u8, index: i64, value: []const u8) StoreError!void {
2019
-
self.mutex.lockUncancelable(self.io);
2020
-
defer self.mutex.unlock(self.io);
2353
+
self.mutex.lock();
2354
+
defer self.mutex.unlock();
2021
2355
2022
2356
const entry = self.liveEntry(key) orelse return StoreError.NoSuchKey;
2023
2357
if (entry.data != .list) return StoreError.WrongType;
···
2032
2366
}
2033
2367
2034
2368
pub fn ltrim(self: *Store, key: []const u8, start: i64, stop: i64) StoreError!void {
2035
-
self.mutex.lockUncancelable(self.io);
2036
-
defer self.mutex.unlock(self.io);
2369
+
self.mutex.lock();
2370
+
defer self.mutex.unlock();
2037
2371
2038
2372
const entry = self.liveEntry(key) orelse return;
2039
2373
if (entry.data != .list) return StoreError.WrongType;
···
2073
2407
stop: i64,
2074
2408
withscores: bool,
2075
2409
) StoreError![]ScoredMember {
2076
-
self.mutex.lockUncancelable(self.io);
2077
-
defer self.mutex.unlock(self.io);
2410
+
self.mutex.lock();
2411
+
defer self.mutex.unlock();
2078
2412
2079
2413
const entry = self.liveEntry(key) orelse return &[_]ScoredMember{};
2080
2414
if (entry.data != .sorted_set) return StoreError.WrongType;
···
2099
2433
}
2100
2434
2101
2435
pub fn zcount(self: *Store, key: []const u8, min: f64, max: f64) StoreError!i64 {
2102
-
self.mutex.lockUncancelable(self.io);
2103
-
defer self.mutex.unlock(self.io);
2436
+
self.mutex.lock();
2437
+
defer self.mutex.unlock();
2104
2438
const entry = self.liveEntry(key) orelse return 0;
2105
2439
if (entry.data != .sorted_set) return StoreError.WrongType;
2106
2440
var count: i64 = 0;
···
2130
2464
count: ?usize,
2131
2465
) StoreError![]ReadResult {
2132
2466
if (stream_keys.len != start_ids.len) return StoreError.Syntax;
2133
-
self.mutex.lockUncancelable(self.io);
2134
-
defer self.mutex.unlock(self.io);
2467
+
self.mutex.lock();
2468
+
defer self.mutex.unlock();
2135
2469
2136
2470
var out: std.ArrayList(ReadResult) = .empty;
2137
2471
errdefer {
···
2176
2510
max: StreamId,
2177
2511
count: ?usize,
2178
2512
) StoreError![]DeliveredEntry {
2179
-
self.mutex.lockUncancelable(self.io);
2180
-
defer self.mutex.unlock(self.io);
2513
+
self.mutex.lock();
2514
+
defer self.mutex.unlock();
2181
2515
2182
2516
const entry = self.liveEntry(key) orelse return &[_]DeliveredEntry{};
2183
2517
if (entry.data != .stream) return StoreError.WrongType;
···
2202
2536
}
2203
2537
2204
2538
pub fn xtrimMaxlen(self: *Store, key: []const u8, maxlen: usize) StoreError!i64 {
2205
-
self.mutex.lockUncancelable(self.io);
2206
-
defer self.mutex.unlock(self.io);
2539
+
self.mutex.lock();
2540
+
defer self.mutex.unlock();
2207
2541
2208
2542
const entry = self.liveEntry(key) orelse return 0;
2209
2543
if (entry.data != .stream) return StoreError.WrongType;
···
2224
2558
}
2225
2559
2226
2560
pub fn streamLastId(self: *Store, key: []const u8) ?StreamId {
2227
-
self.mutex.lockUncancelable(self.io);
2228
-
defer self.mutex.unlock(self.io);
2561
+
self.mutex.lock();
2562
+
defer self.mutex.unlock();
2229
2563
const entry = self.liveEntry(key) orelse return null;
2230
2564
if (entry.data != .stream) return null;
2231
2565
return entry.data.stream.last_id;
···
2265
2599
start: StreamId,
2266
2600
count: ?usize,
2267
2601
) StoreError!AutoclaimResult {
2268
-
self.mutex.lockUncancelable(self.io);
2269
-
defer self.mutex.unlock(self.io);
2602
+
self.mutex.lock();
2603
+
defer self.mutex.unlock();
2270
2604
2271
2605
const entry = self.liveEntry(key) orelse return StoreError.NoGroup;
2272
2606
if (entry.data != .stream) return StoreError.WrongType;
···
2399
2733
keys: []const []const u8,
2400
2734
head: bool,
2401
2735
) StoreError!?PoppedKV {
2402
-
self.mutex.lockUncancelable(self.io);
2403
-
defer self.mutex.unlock(self.io);
2736
+
self.mutex.lock();
2737
+
defer self.mutex.unlock();
2404
2738
2405
2739
for (keys) |k| {
2406
2740
const entry = self.liveEntry(k) orelse continue;
···
2438
2772
src_from: ListEnd,
2439
2773
dst_to: ListEnd,
2440
2774
) StoreError!?[]u8 {
2441
-
self.mutex.lockUncancelable(self.io);
2442
-
defer self.mutex.unlock(self.io);
2775
+
self.mutex.lock();
2776
+
defer self.mutex.unlock();
2443
2777
2444
2778
_ = self.expireIfDue(src);
2445
2779
const same_key = std.mem.eql(u8, src, dst);
···
2504
2838
pivot: []const u8,
2505
2839
value: []const u8,
2506
2840
) StoreError!i64 {
2507
-
self.mutex.lockUncancelable(self.io);
2508
-
defer self.mutex.unlock(self.io);
2841
+
self.mutex.lock();
2842
+
defer self.mutex.unlock();
2509
2843
2510
2844
const entry = self.liveEntry(key) orelse return 0;
2511
2845
if (entry.data != .list) return StoreError.WrongType;
···
2543
2877
min: f64,
2544
2878
max: f64,
2545
2879
) StoreError!i64 {
2546
-
self.mutex.lockUncancelable(self.io);
2547
-
defer self.mutex.unlock(self.io);
2880
+
self.mutex.lock();
2881
+
defer self.mutex.unlock();
2548
2882
2549
-
// Collect matching (score, member) from src — own copies before
2550
-
// mutating dst (which may delete the src key if dst == src).
2551
-
var collected: std.ArrayList(struct { score: f64, member: []u8 }) = .empty;
2552
-
defer {
2553
-
for (collected.items) |sm| self.allocator.free(sm.member);
2554
-
collected.deinit(self.allocator);
2883
+
const src_entry = self.liveEntry(src) orelse {
2884
+
_ = self.removeKey(dst);
2885
+
return 0;
2886
+
};
2887
+
if (src_entry.data != .sorted_set) return StoreError.WrongType;
2888
+
const src_zset = &src_entry.data.sorted_set;
2889
+
const first = self.zsetLowerBoundScore(src_zset, min);
2890
+
var last = first;
2891
+
while (last < src_zset.by_score.items.len and src_zset.by_score.items[last].score <= max) {
2892
+
last += 1;
2555
2893
}
2894
+
const source_items = src_zset.by_score.items[first..last];
2895
+
2896
+
if (source_items.len == 0) return 0;
2556
2897
2557
-
{
2558
-
const src_entry = self.liveEntry(src) orelse {
2559
-
_ = self.removeKey(dst);
2560
-
return 0;
2561
-
};
2562
-
if (src_entry.data != .sorted_set) return StoreError.WrongType;
2563
-
for (src_entry.data.sorted_set.by_score.items) |sm| {
2564
-
if (sm.score < min) continue;
2565
-
if (sm.score > max) break;
2566
-
try collected.append(self.allocator, .{
2567
-
.score = sm.score,
2568
-
.member = try self.allocator.dupe(u8, sm.member),
2569
-
});
2570
-
}
2898
+
var new_zset = SortedSet.init(self.allocator);
2899
+
errdefer new_zset.deinit(self.allocator);
2900
+
try new_zset.by_member.ensureTotalCapacity(@intCast(source_items.len));
2901
+
try new_zset.by_score.ensureTotalCapacity(self.allocator, source_items.len);
2902
+
2903
+
var total_member_bytes: usize = 0;
2904
+
for (source_items) |sm| total_member_bytes += sm.member.len;
2905
+
const slab = try new_zset.memberAllocator().alloc(u8, total_member_bytes);
2906
+
var offset: usize = 0;
2907
+
for (source_items) |sm| {
2908
+
// Source iteration is already score-ordered, so preserve that order
2909
+
// instead of binary-inserting every copied member.
2910
+
const member = slab[offset .. offset + sm.member.len];
2911
+
@memcpy(member, sm.member);
2912
+
offset += sm.member.len;
2913
+
new_zset.by_member.putAssumeCapacityNoClobber(.{ .hash = sm.member_hash, .bytes = member }, sm.score);
2914
+
new_zset.by_score.appendAssumeCapacity(.{ .score = sm.score, .member = member, .member_hash = sm.member_hash });
2571
2915
}
2572
2916
2573
-
// Replace dst.
2917
+
// Replace dst only after the new zset is fully built.
2574
2918
_ = self.removeKey(dst);
2575
-
if (collected.items.len == 0) return 0;
2576
-
2577
2919
const owned_key = try self.allocator.dupe(u8, dst);
2578
-
errdefer self.allocator.free(owned_key);
2579
-
try self.data.put(owned_key, .{ .data = .{ .sorted_set = SortedSet.init(self.allocator) }, .expires_at_ns = null });
2580
-
const dst_entry = self.data.getPtr(dst).?;
2581
-
const dst_zset = &dst_entry.data.sorted_set;
2582
-
2583
-
for (collected.items) |sm| {
2584
-
// sm.member is owned-by-collected — transfer ownership into the zset.
2585
-
try dst_zset.by_member.put(sm.member, sm.score);
2586
-
const idx = self.zsetUpperBound(dst_zset, .{ .score = sm.score, .member = sm.member });
2587
-
try dst_zset.by_score.insert(self.allocator, idx, .{ .score = sm.score, .member = sm.member });
2588
-
}
2589
-
// Member ownership has been transferred into the zset; clear the
2590
-
// outer ArrayList so the defer doesn't double-free.
2591
-
collected.clearRetainingCapacity();
2592
-
return @intCast(dst_zset.len());
2920
+
self.clearZsetCache();
2921
+
self.data.put(owned_key, .{ .data = .{ .sorted_set = new_zset }, .expires_at_ns = null }) catch |err| {
2922
+
self.allocator.free(owned_key);
2923
+
return err;
2924
+
};
2925
+
return @intCast(new_zset.len());
2593
2926
}
2594
2927
2595
2928
// ------------------------------------------------------------------------
···
2634
2967
ids: []const StreamId,
2635
2968
opts: ClaimOpts,
2636
2969
) StoreError![]ClaimedEntry {
2637
-
self.mutex.lockUncancelable(self.io);
2638
-
defer self.mutex.unlock(self.io);
2970
+
self.mutex.lock();
2971
+
defer self.mutex.unlock();
2639
2972
2640
2973
const entry = self.liveEntry(key) orelse return StoreError.NoGroup;
2641
2974
if (entry.data != .stream) return StoreError.WrongType;
···
2728
3061
out_alloc: Allocator,
2729
3062
key: []const u8,
2730
3063
) StoreError![]GroupInfo {
2731
-
self.mutex.lockUncancelable(self.io);
2732
-
defer self.mutex.unlock(self.io);
3064
+
self.mutex.lock();
3065
+
defer self.mutex.unlock();
2733
3066
2734
3067
const entry = self.liveEntry(key) orelse return &[_]GroupInfo{};
2735
3068
if (entry.data != .stream) return StoreError.WrongType;
···
2774
3107
key: []const u8,
2775
3108
group: []const u8,
2776
3109
) StoreError![]ConsumerInfo {
2777
-
self.mutex.lockUncancelable(self.io);
2778
-
defer self.mutex.unlock(self.io);
3110
+
self.mutex.lock();
3111
+
defer self.mutex.unlock();
2779
3112
2780
3113
const entry = self.liveEntry(key) orelse return StoreError.NoGroup;
2781
3114
if (entry.data != .stream) return StoreError.WrongType;
···
2820
3153
};
2821
3154
2822
3155
pub fn xinfoStream(self: *Store, key: []const u8) StoreError!?StreamInfo {
2823
-
self.mutex.lockUncancelable(self.io);
2824
-
defer self.mutex.unlock(self.io);
3156
+
self.mutex.lock();
3157
+
defer self.mutex.unlock();
2825
3158
2826
3159
const entry = self.liveEntry(key) orelse return null;
2827
3160
if (entry.data != .stream) return StoreError.WrongType;
···
2862
3195
key: []const u8,
2863
3196
group: []const u8,
2864
3197
) StoreError!PendingSummary {
2865
-
self.mutex.lockUncancelable(self.io);
2866
-
defer self.mutex.unlock(self.io);
3198
+
self.mutex.lock();
3199
+
defer self.mutex.unlock();
2867
3200
2868
3201
const entry = self.liveEntry(key) orelse return StoreError.NoGroup;
2869
3202
if (entry.data != .stream) return StoreError.WrongType;
···
2936
3269
group: []const u8,
2937
3270
opts: PendingRangeOpts,
2938
3271
) StoreError![]PendingEntryInfo {
2939
-
self.mutex.lockUncancelable(self.io);
2940
-
defer self.mutex.unlock(self.io);
3272
+
self.mutex.lock();
3273
+
defer self.mutex.unlock();
2941
3274
2942
3275
const entry = self.liveEntry(key) orelse return StoreError.NoGroup;
2943
3276
if (entry.data != .stream) return StoreError.WrongType;
···
2986
3319
}
2987
3320
2988
3321
pub fn xgroupDestroy(self: *Store, key: []const u8, group: []const u8) StoreError!bool {
2989
-
self.mutex.lockUncancelable(self.io);
2990
-
defer self.mutex.unlock(self.io);
3322
+
self.mutex.lock();
3323
+
defer self.mutex.unlock();
2991
3324
2992
3325
const entry = self.liveEntry(key) orelse return false;
2993
3326
if (entry.data != .stream) return StoreError.WrongType;
···
3001
3334
3002
3335
/// Sweep up to `budget` expired keys. Returns count removed.
3003
3336
pub fn sweepExpired(self: *Store, budget: usize) usize {
3004
-
self.mutex.lockUncancelable(self.io);
3005
-
defer self.mutex.unlock(self.io);
3337
+
self.mutex.lock();
3338
+
defer self.mutex.unlock();
3006
3339
3007
3340
const now = self.nowNs();
3008
3341
var to_remove: std.ArrayList([]const u8) = .empty;
···
3781
4114
try std.testing.expectEqual(@as(?f64, 2), try s.zscore("dst", "b"));
3782
4115
try std.testing.expectEqual(@as(?f64, 3), try s.zscore("dst", "c"));
3783
4116
try std.testing.expectEqual(@as(?f64, null), try s.zscore("dst", "a"));
4117
+
}
4118
+
4119
+
fn expectSortedSetMirror(s: *Store, key: []const u8) !void {
4120
+
const entry = s.data.getPtr(key) orelse return;
4121
+
if (entry.data != .sorted_set) return;
4122
+
const zset = &entry.data.sorted_set;
4123
+
try std.testing.expectEqual(zset.by_member.count(), zset.by_score.items.len);
4124
+
for (zset.by_score.items) |sm| {
4125
+
const score = zset.getScore(sm.member) orelse return error.MissingMemberIndex;
4126
+
try std.testing.expectEqual(score, sm.score);
4127
+
}
4128
+
}
4129
+
4130
+
test "ZRANGESTORE keeps ownership valid under allocator failure" {
4131
+
for (0..64) |fail_after| {
4132
+
var failing = std.testing.FailingAllocator.init(std.testing.allocator, .{});
4133
+
var threaded = Io.Threaded.init(failing.allocator(), .{});
4134
+
defer threaded.deinit();
4135
+
4136
+
var s = Store.init(failing.allocator(), threaded.io());
4137
+
defer s.deinit();
4138
+
4139
+
_ = try s.zadd("src", &.{
4140
+
.{ .score = 1, .member = "a" },
4141
+
.{ .score = 2, .member = "b" },
4142
+
.{ .score = 3, .member = "c" },
4143
+
.{ .score = 4, .member = "d" },
4144
+
}, .{});
4145
+
4146
+
failing.has_induced_failure = false;
4147
+
failing.fail_index = failing.alloc_index + fail_after;
4148
+
4149
+
const result = s.zrangestore("dst", "src", 1, 4);
4150
+
if (result) |count| {
4151
+
try std.testing.expectEqual(@as(i64, 4), count);
4152
+
} else |err| switch (err) {
4153
+
error.OutOfMemory => {},
4154
+
else => return err,
4155
+
}
4156
+
4157
+
try expectSortedSetMirror(&s, "src");
4158
+
try expectSortedSetMirror(&s, "dst");
4159
+
}
4160
+
}
4161
+
4162
+
test "ZSCORE builds dense lookup for fixed-width dense members and invalidates on mutation" {
4163
+
var threaded = Io.Threaded.init(std.testing.allocator, .{});
4164
+
defer threaded.deinit();
4165
+
var s = Store.init(std.testing.allocator, threaded.io());
4166
+
defer s.deinit();
4167
+
4168
+
var members: [128]ScoredMember = undefined;
4169
+
var member_bufs: [128][20]u8 = undefined;
4170
+
for (&members, 0..) |*m, i| {
4171
+
const member = try std.fmt.bufPrint(&member_bufs[i], "member/{d:0>13}", .{i});
4172
+
m.* = .{ .score = @floatFromInt(i), .member = member };
4173
+
}
4174
+
4175
+
_ = try s.zadd("z", &members, .{});
4176
+
try std.testing.expectEqual(@as(?f64, 42), try s.zscore("z", "member/0000000000042"));
4177
+
4178
+
var zset = &s.data.getPtr("z").?.data.sorted_set;
4179
+
try std.testing.expect(zset.dense_lookup_checked);
4180
+
try std.testing.expect(zset.dense_lookup != null);
4181
+
try std.testing.expect(s.zset_dense_cache != null);
4182
+
4183
+
_ = try s.zadd("z", &.{.{ .score = 999, .member = "member/0000000000042" }}, .{});
4184
+
zset = &s.data.getPtr("z").?.data.sorted_set;
4185
+
try std.testing.expect(!zset.dense_lookup_checked);
4186
+
try std.testing.expect(zset.dense_lookup == null);
4187
+
try std.testing.expect(s.zset_dense_cache == null);
4188
+
try std.testing.expectEqual(@as(?f64, 999), try s.zscore("z", "member/0000000000042"));
4189
+
}
4190
+
4191
+
test "ZSCORE dense lookup falls back for non-dense member names" {
4192
+
var threaded = Io.Threaded.init(std.testing.allocator, .{});
4193
+
defer threaded.deinit();
4194
+
var s = Store.init(std.testing.allocator, threaded.io());
4195
+
defer s.deinit();
4196
+
4197
+
_ = try s.zadd("z", &.{
4198
+
.{ .score = 1, .member = "alpha" },
4199
+
.{ .score = 2, .member = "beta" },
4200
+
.{ .score = 3, .member = "gamma" },
4201
+
}, .{});
4202
+
4203
+
try std.testing.expectEqual(@as(?f64, 2), try s.zscore("z", "beta"));
4204
+
const zset = &s.data.getPtr("z").?.data.sorted_set;
4205
+
try std.testing.expect(zset.dense_lookup_checked);
4206
+
try std.testing.expect(zset.dense_lookup == null);
4207
+
try std.testing.expect(s.zset_dense_cache == null);
3784
4208
}
3785
4209
3786
4210
test "XCLAIM transfers a single PEL entry with idle reset" {