The Pedigree Project 0.1
affinity-regressions.cc
1/* Copyright (c) 2026, Pedigree Developers. */
2#include <config.h>
3
4#if PEDIGREE_AFFINITY_TESTS && THREADS
5#include "pedigree/kernel/Atomic.h"
6#include "pedigree/kernel/LockGuard.h"
7#include "pedigree/kernel/Log.h"
8#include "pedigree/kernel/process/Process.h"
9#include "pedigree/kernel/process/Scheduler.h"
10#include "pedigree/kernel/process/Semaphore.h"
11#include "pedigree/kernel/process/Thread.h"
12#include "pedigree/kernel/process/WaitQueue.h"
13#include "pedigree/kernel/processor/Processor.h"
14#include "pedigree/kernel/processor/ProcessorInformation.h"
15#include "pedigree/kernel/time/Time.h"
16
17namespace {
18bool check(bool condition, const char* detail) {
19 if (!condition)
20 ERROR("AFFINITY-CORE: FAIL " << detail);
21 return condition;
22}
23
24bool sameMask(const CpuAffinityMask& left, const CpuAffinityMask& right) {
25 for (size_t i = 0; i < CpuAffinityMask::WordCount; ++i) {
26 if (left.words[i] != right.words[i])
27 return false;
28 }
29 return true;
30}
31
32struct WaitState {
33 WaitQueue queue;
34 Semaphore entered{0}, firstDone{0}, secondDone{0};
35 Atomic<size_t> returns{0};
36 Atomic<size_t> beforeGateCpu{CpuAffinityMask::MaximumCpus};
37 Atomic<size_t> firstCpu{CpuAffinityMask::MaximumCpus};
38 Atomic<size_t> secondCpu{CpuAffinityMask::MaximumCpus};
39 Atomic<size_t> firstReason{static_cast<size_t>(WaitQueue::WakeReason::Waiting)};
40 Atomic<size_t> secondReason{static_cast<size_t>(WaitQueue::WakeReason::Waiting)};
41 Atomic<bool> gatePassed{false};
42 bool repeat = false;
43};
44
45int waitEntry(void* parameter) {
46 auto& state = *static_cast<WaitState*>(parameter);
47 state.entered.release();
48 WaitQueue::WakeReason reason;
49 {
50 auto guard = state.queue.acquire();
51 reason = guard.wait(WaitQueue::Channel(&state, 1), Thread::EventWait);
52 }
53 state.firstReason = static_cast<size_t>(reason);
54 state.beforeGateCpu = Processor::index();
55 if (reason == WaitQueue::WakeReason::Signalled) {
56 // The ordinary wait and its guards have retired. This owned test entry
57 // has no CPU-local continuation or cleanup scope across the return gate.
58 const auto result = Processor::information().getCurrentThread()->completeAffinityAtSafePoint();
59 state.gatePassed = result == AffinityResult::Success && !Processor::getInterrupts();
60 state.firstCpu = Processor::index();
61 Processor::setInterrupts(true);
62 }
63 state.returns += 1;
64 state.firstDone.release();
65 if (!state.repeat || reason != WaitQueue::WakeReason::Signalled || !state.gatePassed)
66 return 0;
67 {
68 auto guard = state.queue.acquire();
69 reason = guard.wait(WaitQueue::Channel(&state, 2), Thread::EventWait);
70 }
71 state.secondReason = static_cast<size_t>(reason);
72 state.secondCpu = Processor::index();
73 state.returns += 1;
74 state.secondDone.release();
75 return 0;
76}
77
78Thread* startWaiter(WaitState& state, size_t cpu) {
79 ThreadPlacement placement;
80 placement.allowed.set(cpu);
81 placement.migratable = true;
82 Thread* current = Processor::information().getCurrentThread();
83 auto* peer =
84 new Thread(current->getParent(), waitEntry, &state, nullptr, false, false, true, &placement);
85 if (!peer)
86 return nullptr;
87 peer->setPriority(current->getPriority());
88 if (!peer->start()) {
89 peer->setUnwindState(Thread::TerminateThread);
90 if (!peer->joinForCompletion())
91 FATAL("AFFINITY-CORE: failed startup could not be joined");
92 return nullptr;
93 }
94 return peer;
95}
96
97bool waitForSleeping(Thread& peer, WaitState& state) {
98 const Time::Timestamp start = Time::getTicks();
99 while (Time::getTicks() - start < 5 * Time::Multiplier::Second) {
100 bool sleeping;
101 {
102 // A successful acquisition also waits for the outgoing-stack release.
103 LockGuard<Spinlock> guard(peer.getLock());
104 sleeping = peer.getStatus() == Thread::Sleeping;
105 }
106 if (sleeping && state.queue.waiterCount() == 1)
107 return true;
108 Scheduler::instance().yield();
109 }
110 return false;
111}
112
113void retireWaiter(Thread* peer, WaitState& state, bool completed) {
114 if (!completed) {
115 peer->setUnwindState(Thread::TerminateThread);
116 state.queue.wakeAll();
117 }
118 if (!peer->joinForCompletion())
119 FATAL("AFFINITY-CORE: owned waiter could not be joined");
120}
121
122struct ReadyContext {
123 Thread* target = nullptr;
124 WaitState* state = nullptr;
125 CpuAffinityMask destination;
126 PerProcessorScheduler* source = nullptr;
127 bool moving = false;
128 size_t calls = 0;
129 bool passed = true;
130};
131
132ReadyContext* g_ReadyContext = nullptr;
133
134void beforeReadyPublication(Thread* peer) {
135 auto* context = __atomic_load_n(&g_ReadyContext, __ATOMIC_ACQUIRE);
136 if (!context || peer != context->target)
137 return;
138 ++context->calls;
139 if (context->calls != 1) {
140 context->passed = false;
141 return;
142 }
143 context->passed &=
144 check(Processor::getInterrupts() && !Processor::inDeviceHardIrq(), "ready hook context");
145 if (!context->passed)
146 return;
147
148 uint64_t generation = 0;
149 const AffinityResult requested = peer->requestAffinity(context->destination, generation);
150 context->passed &=
151 check(requested == AffinityResult::Success, "request while ready publication is held");
152 if (requested == AffinityResult::Success) {
153 context->passed &= check(peer->waitAffinity(generation) == AffinityResult::Success,
154 "affinity completion while wake is held");
155 }
156 ThreadPlacement placement;
157 peer->snapshotPlacement(placement);
158 bool correctOwner;
159 {
160 LockGuard<Spinlock> guard(peer->getLock());
161 correctOwner = peer->getScheduler() == context->source;
162 }
163 context->passed &= check(correctOwner && sameMask(placement.allowed, context->destination),
164 "mask acknowledged without moving the blocked kernel continuation");
165
166 // This is another real status publisher. It must not bypass the extracted
167 // notification, including when no second CPU exists to perform migration.
168 Scheduler::instance().threadStatusChanged(peer);
169 for (size_t i = 0; i < 8; ++i)
170 Scheduler::instance().yield();
171 context->passed &= check(context->state->returns == 0 && context->state->queue.waiterCount() == 1,
172 "waiter executed or was reused before wake publication retired");
173}
174
175bool heldWake(size_t sourceCpu, size_t destinationCpu) {
176 WaitState state;
177 state.repeat = true;
178 Thread* peer = startWaiter(state, sourceCpu);
179 if (!check(peer != nullptr, "wake waiter allocation/start"))
180 return false;
181 bool passed = check(state.entered.acquireForCompletion(1, 5) && waitForSleeping(*peer, state),
182 "first wait enrollment");
183 ReadyContext context;
184 context.target = peer;
185 context.state = &state;
186 context.destination.set(destinationCpu);
187 context.moving = sourceCpu != destinationCpu;
188 {
189 LockGuard<Spinlock> guard(peer->getLock());
190 context.source = peer->getScheduler();
191 }
192 if (passed) {
193 __atomic_store_n(&g_ReadyContext, &context, __ATOMIC_RELEASE);
194 WaitQueue::setReadyPublicationHookForTest(peer, beforeReadyPublication);
195 const bool woke =
196 state.queue.wakeOne(WaitQueue::WakeReason::Signalled, WaitQueue::Channel(&state, 1));
197 WaitQueue::setReadyPublicationHookForTest(nullptr, nullptr);
198 __atomic_store_n(&g_ReadyContext, static_cast<ReadyContext*>(nullptr), __ATOMIC_RELEASE);
199 passed &= check(woke && context.calls == 1 && context.passed, "held wake admission");
200 }
201 bool completed = false;
202 if (passed) {
203 passed &=
204 check(state.firstDone.acquireForCompletion(1, 5) && state.returns == 1 &&
205 state.beforeGateCpu == sourceCpu && state.gatePassed &&
206 state.firstCpu == destinationCpu &&
207 state.firstReason == static_cast<size_t>(WaitQueue::WakeReason::Signalled) &&
208 waitForSleeping(*peer, state),
209 "source wait return, clean gate migration and second wait enrollment");
210 if (passed) {
211 const bool woke =
212 state.queue.wakeOne(WaitQueue::WakeReason::Signalled, WaitQueue::Channel(&state, 2));
213 completed = woke && state.secondDone.acquireForCompletion(1, 5);
214 passed &=
215 check(completed && state.returns == 2 && state.secondCpu == destinationCpu &&
216 state.secondReason == static_cast<size_t>(WaitQueue::WakeReason::Signalled) &&
217 state.queue.waiterCount() == 0,
218 "waiter reuse, single execution and final wake reason");
219 }
220 }
221 retireWaiter(peer, state, completed);
222 if (passed)
223 NOTICE("AFFINITY-CORE: PASS extracted wake " << (context.moving ? "migration" : "same CPU"));
224 return passed;
225}
226
227struct CancelContext {
228 Thread* target = nullptr;
229 Atomic<size_t> calls{0};
230 Atomic<bool> safeContext{true};
231};
232
233CancelContext* g_CancelContext = nullptr;
234
235void cancelBeforeCommit(Thread* peer) {
236 auto* context = __atomic_load_n(&g_CancelContext, __ATOMIC_ACQUIRE);
237 if (!context || peer != context->target)
238 return;
239 context->calls += 1;
240 context->safeContext = Processor::getInterrupts() && !Processor::inDeviceHardIrq();
241 peer->setUnwindState(Thread::TerminateThread);
242}
243
244bool cancelPending(size_t sourceCpu, size_t destinationCpu) {
245 WaitState state;
246 Thread* peer = startWaiter(state, sourceCpu);
247 if (!check(peer != nullptr, "cancel waiter allocation/start"))
248 return false;
249 bool passed = check(state.entered.acquireForCompletion(1, 5) && waitForSleeping(*peer, state),
250 "cancel wait enrollment");
251 CancelContext context;
252 context.target = peer;
253 CpuAffinityMask destination;
254 destination.set(destinationCpu);
255 if (passed) {
256 __atomic_store_n(&g_CancelContext, &context, __ATOMIC_RELEASE);
257 Thread::setAffinityCommitHookForTest(peer, cancelBeforeCommit);
258 uint64_t generation = 0;
259 const AffinityResult requested = peer->requestAffinity(destination, generation);
260 passed &= check(requested == AffinityResult::Success, "pending request admission");
261 if (requested == AffinityResult::Success) {
262 passed &= check(peer->waitAffinity(generation) == AffinityResult::Terminal,
263 "pending generation terminal completion");
264 }
265 Thread::setAffinityCommitHookForTest(nullptr, nullptr);
266 __atomic_store_n(&g_CancelContext, static_cast<CancelContext*>(nullptr), __ATOMIC_RELEASE);
267 passed &= check(context.calls == 1 && context.safeContext, "terminal precommit hook");
268 }
269 retireWaiter(peer, state, false);
270 if (passed)
271 NOTICE("AFFINITY-CORE: PASS pending terminal cancellation");
272 return passed;
273}
274
275struct PinContext {
276 Thread* target = nullptr;
277 Atomic<size_t> calls{0};
278 Atomic<bool> rejected{false};
279};
280
281PinContext* g_PinContext = nullptr;
282
283void pinBeforeCommit(Thread* peer) {
284 auto* context = __atomic_load_n(&g_PinContext, __ATOMIC_ACQUIRE);
285 if (!context || peer != context->target)
286 return;
287 context->calls += 1;
288 const bool pinned = peer->tryPinLegacyUserCallbacks();
289 context->rejected = !pinned;
290 if (pinned)
291 peer->unpinLegacyUserCallbacks();
292}
293
294bool legacyPins(size_t sourceCpu, size_t destinationCpu) {
295 WaitState state;
296 Thread* peer = startWaiter(state, sourceCpu);
297 if (!check(peer != nullptr, "legacy pin waiter allocation/start"))
298 return false;
299 bool passed = check(state.entered.acquireForCompletion(1, 5) && waitForSleeping(*peer, state),
300 "legacy pin wait enrollment");
301 CpuAffinityMask source, destination;
302 source.set(sourceCpu);
303 destination.set(destinationCpu);
304 size_t pins = 0;
305 for (size_t i = 0; passed && i < 2; ++i) {
306 const bool pinned = peer->tryPinLegacyUserCallbacks();
307 pins += pinned;
308 passed &= check(pinned, "counted legacy pin admission");
309 }
310 const bool moving = sourceCpu != destinationCpu;
311 auto rejectsExclusion = [&] {
312 uint64_t generation = ~uint64_t(0);
313 const auto result = peer->requestAffinity(destination, generation);
314 if (result == AffinityResult::Success || result == AffinityResult::Busy)
315 peer->waitAffinity(generation);
316 ThreadPlacement placement;
317 peer->snapshotPlacement(placement);
318 return check(result == AffinityResult::Unsupported && !generation && placement.migratable &&
319 sameMask(placement.allowed, source),
320 "legacy pin rejected exclusion without changing policy");
321 };
322 if (passed && moving)
323 passed &= rejectsExclusion();
324 if (passed) {
325 uint64_t generation = 0;
326 const auto result = peer->requestAffinity(source, generation);
327 passed &= check(result == AffinityResult::Success, "legacy pin compatible mask admission");
328 if (result == AffinityResult::Success)
329 passed &= check(peer->waitAffinity(generation) == AffinityResult::Success,
330 "legacy pin compatible mask acknowledgement");
331 }
332 if (pins) {
333 peer->unpinLegacyUserCallbacks();
334 --pins;
335 }
336 if (passed && moving)
337 passed &= rejectsExclusion();
338 while (pins) {
339 peer->unpinLegacyUserCallbacks();
340 --pins;
341 }
342 PinContext context;
343 context.target = peer;
344 if (passed) {
345 if (moving) {
346 __atomic_store_n(&g_PinContext, &context, __ATOMIC_RELEASE);
347 Thread::setAffinityCommitHookForTest(peer, pinBeforeCommit);
348 }
349 uint64_t generation = 0;
350 const auto result = peer->requestAffinity(destination, generation);
351 passed &= check(result == AffinityResult::Success, "last legacy pin release permits policy");
352 if (result == AffinityResult::Success)
353 passed &= check(peer->waitAffinity(generation) == AffinityResult::Success,
354 "unpinned policy acknowledgement");
355 if (moving) {
356 Thread::setAffinityCommitHookForTest(nullptr, nullptr);
357 __atomic_store_n(&g_PinContext, static_cast<PinContext*>(nullptr), __ATOMIC_RELEASE);
358 passed &= check(context.calls == 1 && context.rejected,
359 "accepted excluding request rejects legacy pin admission");
360 const bool pinned = peer->tryPinLegacyUserCallbacks();
361 passed &= check(!pinned, "committed excluding mask rejects legacy pin admission");
362 if (pinned)
363 peer->unpinLegacyUserCallbacks();
364 }
365 ThreadPlacement placement;
366 peer->snapshotPlacement(placement);
367 bool sourceOwned;
368 {
369 LockGuard<Spinlock> guard(peer->getLock());
370 sourceOwned = peer->getScheduler() == Scheduler::schedulerForCpu(sourceCpu);
371 }
372 passed &= check(sourceOwned && state.returns == 0 && sameMask(placement.allowed, destination),
373 "sleeping target acknowledges policy on original owner");
374 }
375 bool completed = false;
376 if (passed) {
377 completed =
378 state.queue.wakeOne(WaitQueue::WakeReason::Signalled, WaitQueue::Channel(&state, 1)) &&
379 state.firstDone.acquireForCompletion(1, 5);
380 passed &= check(completed && state.beforeGateCpu == sourceCpu && state.gatePassed &&
381 state.firstCpu == destinationCpu && state.returns == 1,
382 "unpinned target moves only at clean gate");
383 }
384 retireWaiter(peer, state, completed);
385 if (passed)
386 NOTICE("AFFINITY-CORE: PASS legacy callback pins " << (moving ? "migration" : "same CPU"));
387 return passed;
388}
389
390int forbiddenEntry(void* parameter) {
391 *static_cast<bool*>(parameter) = true;
392 return 0;
393}
394
395bool pinnedAdmission() {
396 bool entered = false;
397 Thread* current = Processor::information().getCurrentThread();
398 auto* peer =
399 new Thread(current->getParent(), forbiddenEntry, &entered, nullptr, false, true, true);
400 if (!check(peer != nullptr, "pinned waiter allocation"))
401 return false;
402 ThreadPlacement before, after;
403 peer->snapshotPlacement(before);
404 uint64_t generation = 0;
405 const AffinityResult result = peer->requestAffinity(before.allowed, generation);
406 peer->snapshotPlacement(after);
407 bool passed = check(!before.migratable && result == AffinityResult::Pinned && !after.migratable &&
408 sameMask(before.allowed, after.allowed),
409 "fixed placement admission changed state");
410 peer->setUnwindState(Thread::TerminateThread);
411 if (!peer->joinForCompletion())
412 FATAL("AFFINITY-CORE: unstarted pinned waiter could not be joined");
413 passed &= check(!entered, "unstarted pinned waiter executed");
414 if (passed)
415 NOTICE("AFFINITY-CORE: PASS pinned admission");
416 return passed;
417}
418} // namespace
419
420EXPORTED_PUBLIC bool runAffinityRegressions() {
421 NOTICE("AFFINITY-CORE: BEGIN");
422 if (!check(Processor::getInterrupts() && !Processor::inDeviceHardIrq() &&
423 Processor::information().getCurrentThread(),
424 "fixture entry context"))
425 return false;
426 const size_t source = Processor::index();
427 const CpuAffinityMask online = Scheduler::onlineAffinity();
428 if (!check(online.contains(source), "source CPU online"))
429 return false;
430 size_t destination = source;
431 for (size_t cpu = 0; cpu < CpuAffinityMask::MaximumCpus; ++cpu) {
432 if (cpu != source && online.contains(cpu)) {
433 destination = cpu;
434 break;
435 }
436 }
437 bool passed = pinnedAdmission() && legacyPins(source, source) && heldWake(source, source) &&
438 cancelPending(source, source);
439 if (destination != source) {
440 passed = passed && legacyPins(source, destination) && heldWake(source, destination) &&
441 cancelPending(source, destination);
442 } else {
443 NOTICE("AFFINITY-CORE: SKIP cross-CPU migration (one online CPU)");
444 }
445 if (passed)
446 NOTICE("AFFINITY-CORE: END PASS");
447 return passed;
448}
449#endif
ProcessorBase::getInterrupts
static bool getInterrupts()
Definition hosted/Processor.cc:506
ProcessorBase::information
static ProcessorInformation & information()
Definition x64/Processor.h:30
ProcessorBase::inDeviceHardIrq
static bool inDeviceHardIrq()
Definition Processor.h:559
ProcessorBase::setInterrupts
static void setInterrupts(bool bEnable)
Definition hosted/Processor.cc:458
ProcessorBase::index
static size_t index()
Definition x64/Processor.h:36
Scheduler::instance
static Scheduler & instance()
Definition Scheduler.h:96
Scheduler::yield
void yield()
Definition Scheduler.cc:226
Thread::setUnwindState
void setUnwindState(UnwindType ut)
Definition Thread.cc:3628
Thread::TerminateThread
@ TerminateThread
Exit only this thread during Process exit.
Definition Thread.h:515
Thread::tryPinLegacyUserCallbacks
bool tryPinLegacyUserCallbacks()
Definition Thread-affinity.cc:72
Thread::waitAffinity
AffinityResult waitAffinity(uint64_t generation)
Definition Thread-affinity.cc:156
Thread::joinForCompletion
bool joinForCompletion()
Definition Thread.cc:2771
Thread::getStatus
Status getStatus() const
Definition Thread.h:431
Thread::getParent
Process * getParent() const
Definition Thread.h:338
Thread::getScheduler
class PerProcessorScheduler * getScheduler() const
Definition Thread.h:925
Thread::requestAffinity
AffinityResult requestAffinity(const CpuAffinityMask &mask, uint64_t &generation)
Definition Thread-affinity.cc:90
Thread::getLock
Spinlock & getLock()
Definition Thread.h:609
Generated on Tue Sep 22 2026 06:05:48 for The Pedigree Project by doxygen 1.9.8