lib/machine/src/explore/distributed/workload.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const canon = @import("canon.zig");
2 const explore = @import("../root.zig");
3 const fault = @import("../../fault/root.zig");
4 const property = @import("property.zig");
5 const state_owner = @import("state.zig");
6 const std = @import("std");
7 const types = @import("types.zig");
8
9 const State = state_owner.State;
10 const golden: u64 = 0x9e37_79b9_7f4a_7c15;
11
12 pub fn evaluate(
13 config: types.Config,
14 value: *State,
15 decision: explore.SearchDecision,
16 events: *types.Events,
17 leaf: bool,
18 ) types.Error!property.Verdicts {
19 try property.declare(events, value.now);
20 try step(config, value, decision, events);
21 events.finish(if (leaf) .exhausted else .incomplete) catch unreachable;
22 const verdicts = property.evaluate(events);
23 if (property.unexplained(verdicts)) return error.TraceCapacityExceeded;
24 return verdicts;
25 }
26
27 pub fn step(
28 config: types.Config,
29 value: *State,
30 decision: explore.SearchDecision,
31 events: *types.Events,
32 ) types.Error!void {
33 std.debug.assert(decision.valid());
34 value.assertValid();
35 const previous = value.now;
36 const entry_frontier = value.fabric.entry_frontier;
37 value.now = @max(value.now, choiceTick(decision));
38 std.debug.assert(value.now >= previous);
39 try record(events, value.now, decision);
40 const advanced = switch (decision.choice) {
41 .input => |choice| try applyInput(value, choice, events),
42 .schedule => |choice| try applySchedule(config, value, choice, events),
43 .fault => |choice| try applyFault(config, value, choice, events),
44 .topology => return error.DecisionStreamMismatch,
45 };
46 if (advanced) value.fabric = canon.advance(value.fabric, decision, value);
47 std.debug.assert(value.fabric.entry_frontier == entry_frontier + @intFromBool(advanced));
48 value.assertValid();
49 }
50
51 fn applyInput(
52 value: *State,
53 choice: explore.GeneratedInput,
54 events: *types.Events,
55 ) types.Error!bool {
56 switch (choice.value) {
57 .terminal => |word| {
58 if (value.open) return false;
59 std.debug.assert(value.round < types.depth);
60 value.request = word;
61 try openRound(value, events, word);
62 std.debug.assert(value.open);
63 return true;
64 },
65 .service_result, .effect_result => |outcome| {
66 try append(events, value.now, .{ .operation = .{
67 .id = types.Semantic.host_operation.id(),
68 .outcome = outcome,
69 } });
70 return false;
71 },
72 .wait, .entropy, .packet => return false,
73 }
74 }
75
76 fn applySchedule(
77 config: types.Config,
78 value: *State,
79 choice: explore.GeneratedSchedule,
80 events: *types.Events,
81 ) types.Error!bool {
82 const node = switch (choice.value) {
83 .idle => return false,
84 .turn => |selected| selected,
85 };
86 if (node >= types.node_count) return error.DecisionStreamMismatch;
87 std.debug.assert(node < value.nodes.len);
88 value.turn_node = node;
89 if (value.nodes[node].hung) {
90 try diagnose(events, value.now, .turn_skipped, node);
91 return true;
92 }
93 if (node == types.coordinator) _ = try suspect(value, events);
94 _ = try deliver(config, value, node, events);
95 return true;
96 }
97
98 fn applyFault(
99 config: types.Config,
100 value: *State,
101 choice: explore.GeneratedFault,
102 events: *types.Events,
103 ) types.Error!bool {
104 switch (choice.action) {
105 .healthy => {
106 if (value.hung == null and !value.delivery_blocked) return false;
107 clear(value);
108 return true;
109 },
110 .inject => |kind| {
111 if (value.hung != null or value.delivery_blocked) {
112 return error.FaultStateMismatch;
113 }
114 try inject(value, kind);
115 return true;
116 },
117 .persist => |kind| {
118 try persist(config, value, kind, events);
119 return true;
120 },
121 .recover => |kind| {
122 if (!active(value, kind)) return error.FaultStateMismatch;
123 clear(value);
124 return true;
125 },
126 }
127 }
128
129 fn inject(value: *State, kind: fault.Kind) types.Error!void {
130 std.debug.assert(value.hang_streak == 0);
131 switch (kind) {
132 .process_crash => {
133 const victim = value.turn_node orelse 1;
134 std.debug.assert(victim < types.node_count);
135 value.hung = victim;
136 value.nodes[victim].hung = true;
137 },
138 .packet_delay => value.delivery_blocked = true,
139 else => return error.FaultStateMismatch,
140 }
141 value.hang_streak = 1;
142 std.debug.assert(active(value, kind));
143 }
144
145 fn persist(
146 config: types.Config,
147 value: *State,
148 kind: fault.Kind,
149 events: *types.Events,
150 ) types.Error!void {
151 if (!active(value, kind)) return error.FaultStateMismatch;
152 std.debug.assert(value.hang_streak < types.depth);
153 if (config.persistence == .retained) value.hang_streak += 1;
154 if (value.hung != null and value.hang_streak >= types.suspicion_steps) {
155 try observe(events, value.now, .persistent_hang, value.hang_streak);
156 }
157 }
158
159 fn clear(value: *State) void {
160 if (value.hung) |node| value.nodes[node].hung = false;
161 value.hung = null;
162 value.delivery_blocked = false;
163 value.hang_streak = 0;
164 std.debug.assert(!value.suspecting());
165 }
166
167 fn active(value: *const State, kind: fault.Kind) bool {
168 return switch (kind) {
169 .process_crash => value.hung != null,
170 .packet_delay => value.delivery_blocked,
171 else => false,
172 };
173 }
174
175 fn suspect(value: *State, events: *types.Events) types.Error!bool {
176 if (!value.suspecting()) return false;
177 const opened = value.round;
178 value.conjunct.persistent_hang = true;
179 try observe(events, value.now, .node_suspected, value.hung.?);
180 try openRound(value, events, nextProposal(value));
181 std.debug.assert(value.round == opened + 1);
182 return true;
183 }
184
185 fn openRound(
186 value: *State,
187 events: *types.Events,
188 proposal: u64,
189 ) types.Error!void {
190 std.debug.assert(value.round < types.depth);
191 value.messages.cancel(types.coordinator, .propose);
192 value.round += 1;
193 value.proposal = proposal;
194 value.open = true;
195 value.acks = 0;
196 value.ack_mask = 0;
197 value.nodes[types.coordinator].accepted = true;
198 value.nodes[types.coordinator].accepted_round = value.round;
199 value.nodes[types.coordinator].accepted_value = proposal;
200 var node: u8 = 0;
201 while (node < types.node_count) : (node += 1) {
202 if (node == types.coordinator) continue;
203 try value.messages.push(.{
204 .kind = .propose,
205 .round = value.round,
206 .value = proposal,
207 .sender = types.coordinator,
208 .destination = node,
209 });
210 }
211 std.debug.assert(value.messages.count >= types.node_count - 1);
212 std.debug.assert(value.backing() >= 1);
213 try observe(events, value.now, .round_opened, value.round);
214 }
215
216 fn deliver(
217 config: types.Config,
218 value: *State,
219 node: u8,
220 events: *types.Events,
221 ) types.Error!bool {
222 if (value.delivery_blocked) {
223 try observe(events, value.now, .delivery_deferred, node);
224 return false;
225 }
226 const index = value.messages.find(node) orelse {
227 try diagnose(events, value.now, .inbox_empty, node);
228 return false;
229 };
230 const queued = value.messages.count;
231 const message = value.messages.take(index);
232 std.debug.assert(value.messages.count == queued - 1);
233 std.debug.assert(message.destination == node);
234 switch (message.kind) {
235 .propose => try accept(config, value, node, message, events),
236 .ack => {
237 value.conjunct.queued_acknowledgement = true;
238 try tally(config, value, message, events);
239 },
240 }
241 return true;
242 }
243
244 fn accept(
245 config: types.Config,
246 value: *State,
247 node: u8,
248 message: types.Message,
249 events: *types.Events,
250 ) types.Error!void {
251 if (node == types.coordinator) return error.DecisionStreamMismatch;
252 std.debug.assert(message.kind == .propose);
253 std.debug.assert(message.sender == types.coordinator);
254 if (message.round != value.round) {
255 try observe(events, value.now, .stale_accept, pack(node, message.round));
256 }
257 value.nodes[node].accepted = true;
258 value.nodes[node].accepted_round = message.round;
259 value.nodes[node].accepted_value = message.value;
260 const acknowledgement: types.Message = .{
261 .kind = .ack,
262 .round = message.round,
263 .value = message.value,
264 .sender = node,
265 .destination = types.coordinator,
266 };
267 try observe(events, value.now, .proposal_accepted, pack(node, message.round));
268 switch (config.delivery) {
269 .queued => try value.messages.push(acknowledgement),
270 .immediate => try tally(config, value, acknowledgement, events),
271 }
272 }
273
274 fn tally(
275 config: types.Config,
276 value: *State,
277 message: types.Message,
278 events: *types.Events,
279 ) types.Error!void {
280 std.debug.assert(message.kind == .ack);
281 std.debug.assert(message.destination == types.coordinator);
282 const stale = message.round != value.round;
283 if (stale) {
284 value.conjunct.delayed_delivery = true;
285 std.debug.assert(message.round < value.round);
286 const distance = value.round - message.round;
287 try observe(events, value.now, .delayed_delivery, distance);
288 }
289 try observe(events, value.now, .acknowledged, pack(message.sender, message.round));
290 const bit = @as(u8, 1) << @intCast(message.sender);
291 const counted = switch (config.variant) {
292 .defective => true,
293 .repaired => !stale and (value.ack_mask & bit) == 0,
294 };
295 if (counted) {
296 value.ack_mask |= bit;
297 std.debug.assert(value.acks < types.depth);
298 value.acks += 1;
299 }
300 try announce(value, events);
301 }
302
303 fn announce(value: *State, events: *types.Events) types.Error!void {
304 if (!value.open) return;
305 if (value.votes() < types.quorum) return;
306 std.debug.assert(value.round > 0);
307 value.open = false;
308 value.committed = true;
309 value.committed_round = value.round;
310 value.committed_value = value.proposal;
311 try observe(events, value.now, .commit_announced, value.proposal);
312 std.debug.assert(value.committed_round == value.round);
313 const backed = value.backing();
314 std.debug.assert(backed >= 1);
315 std.debug.assert(backed <= types.node_count);
316 if (backed < types.quorum) {
317 try observe(events, value.now, .unbacked_commit, pack(backed, value.round));
318 }
319 }
320
321 fn nextProposal(value: *const State) u64 {
322 const round: u64 = value.round;
323 const next = value.proposal ^ ((round +% 1) *% golden);
324 std.debug.assert(next != value.proposal);
325 return next;
326 }
327
328 fn record(
329 events: *types.Events,
330 tick: u64,
331 decision: explore.SearchDecision,
332 ) types.Error!void {
333 try append(events, tick, switch (decision.choice) {
334 .input => |choice| .{ .controlled_input = choice },
335 .schedule => |choice| .{ .schedule_choice = choice },
336 .topology => |choice| .{ .topology_choice = choice },
337 .fault => |choice| .{ .injected_fault = choice },
338 });
339 }
340
341 fn observe(
342 events: *types.Events,
343 tick: u64,
344 id: types.Semantic,
345 value: u64,
346 ) types.Error!void {
347 try append(events, tick, .{ .observation = .{ .id = id.id(), .value = value } });
348 }
349
350 fn diagnose(
351 events: *types.Events,
352 tick: u64,
353 code: types.Diagnostic,
354 node: u8,
355 ) types.Error!void {
356 try append(events, tick, .{ .diagnostic = .{
357 .id = pack(node, @backingInt(code)),
358 .code = @backingInt(code),
359 } });
360 }
361
362 fn append(
363 events: *types.Events,
364 tick: u64,
365 payload: explore.EventValue,
366 ) types.Error!void {
367 events.append(tick, payload) catch |failure| switch (failure) {
368 error.CapacityExceeded => return error.TraceCapacityExceeded,
369 error.SequenceClosed, error.VirtualTimeRegressed => unreachable,
370 };
371 }
372
373 fn choiceTick(decision: explore.SearchDecision) u64 {
374 return switch (decision.choice) {
375 .input => |choice| choice.virtual_time_tick,
376 .schedule => |choice| choice.virtual_time_tick,
377 .topology => |choice| choice.virtual_time_tick,
378 .fault => |choice| choice.virtual_time_tick,
379 };
380 }
381
382 fn pack(high: anytype, low: anytype) u64 {
383 return (@as(u64, high) << 32) | @as(u64, low);
384 }