Line | Count | Source |
1 | | #include <silk/fibers/futex.h> |
2 | | |
3 | | #include <silk/fibers/fiber.h> |
4 | | #include <silk/util/spinlock.h> |
5 | | |
6 | | #include <atomic> |
7 | | #include <cerrno> |
8 | | |
9 | | namespace silk |
10 | | { |
11 | | |
12 | | int FiberFutex::wait(uint64_t token, uint64_t * waitCycles) noexcept |
13 | 2.34k | { |
14 | | // Spin for ~500 ns (16 x ~35 ns PAUSE on Skylake) before suspending. |
15 | 2.34k | static constexpr uint32_t SPIN_COUNT = 16; |
16 | | |
17 | 2.34k | State currentState; |
18 | 2.34k | currentState.raw = state.load(std::memory_order_relaxed); |
19 | | |
20 | | // Spin briefly before suspending: if the poster is on another CPU and fires |
21 | | // within ~500 ns, we avoid the full scheduler wakeup path. |
22 | | // Skip if there are already waiters in the queue or we are already stopped. |
23 | 2.34k | if (!currentState.hasWaiters2.34k && !currentState.stopped) |
24 | 2.34k | { |
25 | 2.34k | spinWait( |
26 | 2.34k | [this, token] |
27 | 3.75k | { |
28 | 3.75k | State s; |
29 | 3.75k | s.raw = state.load(std::memory_order_relaxed); |
30 | 3.75k | return s.counter >= token || s.hasWaiters1.66k || s.stopped1.63k ; |
31 | 3.75k | }, |
32 | 2.34k | SPIN_COUNT); |
33 | 2.34k | } |
34 | | |
35 | 2.42k | while (!waitHelper(token)) |
36 | 80 | { |
37 | 80 | SuspendCtx ctx{this, token}; |
38 | 80 | FiberScheduler::suspend(reinterpret_cast<FiberScheduler::SuspendCallback *>(suspendCallback), &ctx, waitCycles); |
39 | 80 | } |
40 | | |
41 | | // Re-read state for the cancellation decision. The token-satisfied case |
42 | | // takes precedence: a waiter that got the post it asked for sees a |
43 | | // successful wakeup even if stop fired afterwards. |
44 | 2.34k | currentState.raw = state.load(std::memory_order_acquire); |
45 | 2.34k | if (currentState.counter >= token) |
46 | 2.43k | { |
47 | 2.43k | return 0; |
48 | 2.43k | } |
49 | 18.4E | return ECANCELED; |
50 | 2.34k | } |
51 | | |
52 | | void FiberFutex::post() noexcept |
53 | 4.23k | { |
54 | 4.23k | State currentState; |
55 | 4.23k | currentState.raw = state.load(std::memory_order_relaxed); |
56 | 4.23k | for (;;) |
57 | 5.12k | { |
58 | 5.12k | if (currentState.stopped) |
59 | 4 | { |
60 | | // No waiters can be queued (stop wakes them all and bars further |
61 | | // suspension); skip the increment to leave the counter stable |
62 | | // for any concurrent get() call. |
63 | 4 | return; |
64 | 4 | } |
65 | | |
66 | 5.12k | State newState; |
67 | 5.12k | newState.counter = currentState.counter + 1; |
68 | 5.12k | newState.hasWaiters = 0; |
69 | 5.12k | if (state.compare_exchange_weak(currentState.raw, newState.raw, std::memory_order_release, std::memory_order_relaxed)) |
70 | 4.69k | { |
71 | 4.69k | break; |
72 | 4.69k | } |
73 | 5.12k | } |
74 | | |
75 | 4.22k | if (currentState.hasWaiters) |
76 | 40 | { |
77 | 40 | FiberScheduler::releaseWaiters(reinterpret_cast<uint64_t>(this)); |
78 | 40 | } |
79 | 4.22k | } |
80 | | |
81 | | void FiberFutex::stop() noexcept |
82 | 19 | { |
83 | 19 | State currentState; |
84 | 19 | currentState.raw = state.load(std::memory_order_relaxed); |
85 | 19 | for (;;) |
86 | 19 | { |
87 | 19 | if (currentState.stopped) |
88 | 1 | { |
89 | 1 | return; |
90 | 1 | } |
91 | | |
92 | 18 | State newState(currentState); |
93 | 18 | newState.hasWaiters = 0; |
94 | 18 | newState.stopped = 1; |
95 | 18 | if (state.compare_exchange_weak(currentState.raw, newState.raw, std::memory_order_release, std::memory_order_relaxed)) |
96 | 18 | { |
97 | 18 | break; |
98 | 18 | } |
99 | 18 | } |
100 | | |
101 | 18 | if (currentState.hasWaiters) |
102 | 5 | { |
103 | 5 | FiberScheduler::releaseWaiters(reinterpret_cast<uint64_t>(this)); |
104 | 5 | } |
105 | 18 | } |
106 | | |
107 | | bool FiberFutex::waitHelper(uint64_t token) noexcept |
108 | 2.38k | { |
109 | 2.38k | State currentState; |
110 | 2.38k | currentState.raw = state.load(std::memory_order_acquire); |
111 | 2.38k | for (;;) |
112 | 2.47k | { |
113 | 2.47k | if (currentState.counter >= token || currentState.stopped101 ) |
114 | 2.39k | { |
115 | 2.39k | return true; |
116 | 2.39k | } |
117 | | |
118 | 77 | if (!currentState.hasWaiters) |
119 | 48 | { |
120 | 48 | State newState(currentState); |
121 | 48 | newState.hasWaiters = 1; |
122 | 48 | if (state.compare_exchange_weak(currentState.raw, newState.raw, std::memory_order_release, std::memory_order_acquire)) |
123 | 45 | { |
124 | 45 | return false; |
125 | 45 | } |
126 | 3 | continue; |
127 | 48 | } |
128 | | |
129 | 29 | return false; |
130 | 77 | } |
131 | 2.38k | } |
132 | | |
133 | | void FiberFutex::suspendCallback(Fiber * fiber, SuspendCtx * ctx) noexcept |
134 | 81 | { |
135 | 81 | FiberFutex * event = ctx->event; |
136 | | |
137 | 81 | State currentState; |
138 | 81 | currentState.raw = event->state.load(std::memory_order_acquire); |
139 | 81 | if (currentState.counter >= ctx->token || currentState.stopped77 ) |
140 | 4 | { |
141 | 4 | FiberScheduler::schedule(fiber); |
142 | 4 | return; |
143 | 4 | } |
144 | | |
145 | 77 | FiberScheduler::enqueueWaiter(reinterpret_cast<uint64_t>(event), fiber); |
146 | | |
147 | | // Re-check after enqueue. If hasWaiters is false, post() or stop() already |
148 | | // ran their releaseWaiters but missed us (we were not yet in the table). |
149 | | // We must release now to avoid a missed wakeup. If the counter already |
150 | | // satisfies the token, or the futex is stopped, the fibers we wake will |
151 | | // return immediately from waitHelper. |
152 | 77 | currentState.raw = event->state.load(std::memory_order_acquire); |
153 | 77 | if (!currentState.hasWaiters) |
154 | 1 | { |
155 | 1 | FiberScheduler::releaseWaiters(reinterpret_cast<uint64_t>(event)); |
156 | 1 | } |
157 | 77 | } |
158 | | |
159 | | } // namespace silk |