Source browser
detail/async.hpp
Read the complete file below or download its original bytes. Line numbers provide stable links to a declaration or processing step.
File fingerprint
1/** @file async.hpp
2 * @brief Nonblocking GPU completion handles and bounded CPU work queues.
3 * Prepare GPU pipelines before a hot loop, keep a ring of independent resource
4 * sets, and poll completion before reading or recycling a result. GPU calls
5 * use the device's context thread. CPU workers execute independent snapshots
6 * and require host-ready resources. Complete GPU readback on the context thread
7 * before submitting a CPU job.
8 */
9#ifndef SHDR_ASYNC_HPP // Skip this section if it was already included.
10#define SHDR_ASYNC_HPP /* Mark this section as included. */
11#ifndef SHDR_RUNTIME_HPP // Skip this section if it was already included.
12#include "math.hpp" // Use the portable GLSL scalar, vector, and matrix types.
13#include "runtime.hpp" // Use the kernel, resource, and ordered pipeline definitions.
14#endif // Finish this conditional section.
15#include <chrono> // Express explicit waits and zero-time completion polls.
16#include <condition_variable> // Wake CPU workers when jobs arrive or shutdown begins.
17#include <deque> // Keep pending CPU jobs in submission order.
18#include <exception> // Pass worker failures back through completion futures.
19#include <future> // Give callers a completion result for each CPU job.
20#include <mutex> // Protect shared work queues and deferred deletion.
21#include <thread> // Track the context thread and start CPU workers.
22#include <utility> // Move job and resource ownership between runtime objects.
23
24namespace shdr { // Group these definitions in the shdr namespace.
25
26/** @brief Result of a completion query: completed work or an expired timeout. */
27enum class wait_status { ready, timeout }; // Name the two outcomes of a GPU completion query.
28
29namespace detail { // Group these definitions in the detail namespace.
30struct submission_resources { // Keep the submission resources values and operations together.
31 std::vector<std::shared_ptr<resource_state>> resources; // Keep the resource storage retained until queued work completes.
32 std::vector<std::shared_ptr<program_state>> programs; // Keep the compiled programs retained by a submission.
33}; // End this scope in async.
34
35struct submission_state { // Keep a GPU fence and its resources alive until retirement.
36 std::shared_ptr<device_state> owner; // Keep the device that owns the GL resource.
37 GLsync fence; // Keep the GL completion fence.
38 std::shared_ptr<submission_resources> retained; // Keep the submission resource bundle.
39 bool completed; // Cache whether this submission fence has already signaled.
40 explicit submission_state(const std::shared_ptr<device_state>& d) // Retain the device, fence, and queued resource bundle.
41 : owner(d), fence(0), retained(std::make_shared<submission_resources>()), completed(false) { // Retain the device and create an empty pending-completion resource bundle.
42 } // End this scope in submission state.
43 ~submission_state() noexcept { // Retire completed fences and defer pending fence cleanup.
44 if (!fence) // Finish immediately for an empty fence.
45 return; // Finish this operation.
46 const GLsync sync = fence; // Copy the fence handle into the deferred cleanup callback.
47 const std::shared_ptr<submission_resources> keep = retained; // Retain buffers and programs until deferred cleanup releases this bundle.
48 try { // Keep cleanup available if an allocation, shader, or driver operation fails.
49 if (completed) { // Retire a fence whose completion has already been observed.
50 owner->retire([sync, keep](gl_api& gl) { // Delete on the context thread or defer deletion to that thread.
51 gl.DeleteSync(sync); // Delete the GL completion fence.
52 (void)keep; // Keep the resource references captured for this cleanup.
53 }); // End this scope in submission state.
54 } else { // Handle the alternative resource or completion case.
55 // The context thread checks this retirement entry with a zero timeout.
56 owner->defer_poll( // Retain a completion check until its fence signals.
57 [sync, keep](gl_api& gl) -> bool { // Retain these captures until the queued callback finishes.
58 const GLenum status = gl.ClientWaitSync(sync, 0, 0); // Poll the discarded fence with zero flags and a zero timeout.
59 if (status == TIMEOUT_EXPIRED) // Check the driver result before marking GPU completion.
60 return false; // Keep the discarded submission pending while its fence is unsignaled.
61 gl.DeleteSync(sync); // Delete the GL completion fence.
62 (void)keep; // Keep the resource references captured for this cleanup.
63 return true; // Remove the retired fence and release its retained resource bundle.
64 }, // End this scope in submission state.
65 [sync](gl_api& gl) { // Keep only the fence handle for nonblocking shutdown cancellation.
66 gl.DeleteSync(sync); // Delete the GL completion fence.
67 }); // End this scope in submission state.
68 } // End this scope in submission state.
69 } catch (...) { // Preserve cleanup or exception propagation for this failed operation.
70 // Destruction defers pending fence cleanup. GL command references keep
71 // already queued operations valid if allocating a retirement entry fails.
72 } // End this scope in submission state.
73 } // End this scope in submission state.
74 wait_status query(GLuint64 timeout) { // Check completion with the requested driver timeout.
75 owner->require_thread(); // Require the device context thread and activate its headless context.
76 if (completed) // Reuse the cached completion result after checking the context thread.
77 return wait_status::ready; // Tell the caller that queued GPU commands have completed.
78 const GLenum result = owner->gl.ClientWaitSync(fence, 0, timeout); // Check a GPU fence with the caller-selected timeout.
79 if (result == ALREADY_SIGNALED || result == CONDITION_SATISFIED) { // Check the driver result before marking GPU completion.
80 completed = true; // Set whether the GPU fence has already signaled.
81 return wait_status::ready; // Tell the caller that queued GPU commands have completed.
82 } // End this scope in query.
83 if (result == TIMEOUT_EXPIRED) // Check the driver result before marking GPU completion.
84 return wait_status::timeout; // Tell the caller that the explicit wait expired before completion.
85 owner->gl.check("query GPU submission"); // Check for a driver error from the preceding GL operation.
86 throw error("GPU completion query failed"); // Report gpu completion query failed.
87 } // End this scope in query.
88}; // End this scope in async.
89} // namespace detail
90
91/** @brief A movable handle for one queued GPU pipeline.
92 * ready() performs a zero-timeout fence query. wait(timeout) bounds the requested
93 * driver wait. wait() explicitly waits for completion. Destroying a handle
94 * schedules pending fence and resource cleanup on the device context thread.
95 * Device garbage collection releases those retained resources after completion.
96 * Keep the GL context alive until retained work and resources have been retired.
97 */
98class submission { // Keep the submission values and operations together.
99 std::unique_ptr<detail::submission_state> state_; // Own this completion handle and its retained GPU resources.
100 explicit submission(std::unique_ptr<detail::submission_state> state) // Accept sole ownership of this fence and its retained resource bundle.
101 : state_(std::move(state)) { // Move the fence state into this completion handle.
102 } // End this scope in async.
103 friend class pipeline; // Allow the related pipeline or resource helper to access this retained state.
104
105 public: // Expose the operations callers can use.
106 /** @brief Construct an empty, already-complete handle. */
107 submission() { // Create an empty handle that is already complete.
108 } // End this scope in async.
109 submission(submission&&) noexcept = default; // Move the completion state into the new handle.
110 submission& operator=(submission&&) noexcept = default; // Retire the old handle and move in the new completion state.
111 submission(const submission&) = delete; // Prevent copying this object and duplicating its retained identity.
112 submission& operator=(const submission&) = delete; // Prevent copying this object and duplicating its retained identity.
113 /** @brief Return whether this handle represents submitted GPU commands. */
114 bool valid() const { // Check whether this handle represents actual submitted GPU work.
115 return bool(state_); // Return whether this handle retains a submission state.
116 } // End this scope in async.
117 /** @brief Query completion with a zero timeout on the device context thread. */
118 bool ready() { // Check whether queued work has completed using a zero-timeout poll.
119 return !state_ || state_->query(0) == wait_status::ready; // Check whether queued work has completed using a zero-timeout poll.
120 } // End this scope in ready.
121 /** @brief Request a driver wait with a nonnegative timeout in nanoseconds. */
122 wait_status wait(std::chrono::nanoseconds timeout) { // Apply the explicit completion wait selected by the caller.
123 if (timeout.count() < 0) // Require a nonnegative wait duration.
124 throw error("GPU submission timeout must be nonnegative"); // Report gpu submission timeout must be nonnegative.
125 if (!state_) // Treat an empty handle as already completed.
126 return wait_status::ready; // Tell the caller that queued GPU commands have completed.
127 return state_->query(static_cast<detail::GLuint64>(timeout.count())); // Apply the explicit completion wait selected by the caller.
128 } // End this scope in wait.
129 /** @brief Explicitly wait until submitted commands complete. */
130 wait_status wait() { // Apply the explicit completion wait selected by the caller.
131 if (!state_) // Treat an empty handle as already completed.
132 return wait_status::ready; // Tell the caller that queued GPU commands have completed.
133 while (state_->query((std::numeric_limits<detail::GLuint64>::max)()) == // Use the caller's explicit blocking wait until the fence completes.
134 wait_status::timeout) { // Repeat only if the implementation still reports an incomplete wait.
135 } // End this scope in wait.
136 return wait_status::ready; // Tell the caller that queued GPU commands have completed.
137 } // End this scope in wait.
138}; // End this scope in async.
139
140/** @brief Enqueue all GPU stages and return a fence for polling or explicit waits.
141 * Each stage must support GPU execution. Call prepare()
142 * before entering a latency-sensitive loop to compile shaders and allocate/upload
143 * resources in advance. Subsequent host edits can still require an upload.
144 * Uniform and binding changes after return affect subsequent submissions.
145 */
146inline submission pipeline::submit(device& d) const { // Queue compute work and return its completion handle.
147 d.state()->require_thread(); // Require the device context thread and activate its headless context.
148 const std::vector<stage>& commands = stages(); // Read the ordered stages that this submission will queue.
149 for (std::size_t i = 0; i < commands.size(); ++i) // Visit each ordered GPU stage in this submission.
150 if (commands[i].target == backend::cpu) // Reject explicit CPU stages before queueing any GPU commands.
151 throw error("GPU submission cannot contain CPU stages"); // Require GPU stages for GPU submission.
152 if (commands.empty()) // Skip fence creation for an empty pipeline.
153 return submission(); // Represent an empty pipeline with an already completed handle.
154 std::unique_ptr<detail::submission_state> completion(new detail::submission_state(d.state())); // Retain the device, fence, and queued resource bundle.
155 completion->retained->programs.reserve(commands.size()); // Reserve room to retain one compiled program per stage.
156 prepare(d); // Compile the program and prepare its bound GPU resources.
157 for (std::size_t i = 0; i < commands.size(); ++i) { // Visit each ordered GPU stage in this submission.
158 const std::vector<std::shared_ptr<detail::resource_state>> resources = // Set the resource storage retained until queued work completes.
159 commands[i].shader->resources(); // Collect the shared storage used by this shader.
160 completion->retained->resources.insert( // Retain this entry in the ordered resource or work collection.
161 completion->retained->resources.end(), resources.begin(), resources.end()); // Keep the end of the current range.
162 completion->retained->programs.push_back(commands[i].shader->program_state()); // Retain this entry in the ordered resource or work collection.
163 } // End this scope in submit.
164 try { // Keep cleanup available if an allocation, shader, or driver operation fails.
165 run(d); // Queue all validated GPU stages in pipeline order.
166 } catch (...) { // Preserve cleanup or exception propagation for this failed operation.
167 // A later dispatch can fail after an earlier dispatch was queued.
168 completion->fence = d.state()->gl.FenceSync(detail::SYNC_GPU_COMMANDS_COMPLETE, 0); // Insert a completion fence after queued compute work.
169 d.state()->gl.Flush(); // Flush queued GPU commands for driver execution.
170 throw; // Rethrow the original failure after arranging the necessary cleanup.
171 } // End this scope in submit.
172 completion->fence = d.state()->gl.FenceSync(detail::SYNC_GPU_COMMANDS_COMPLETE, 0); // Insert a completion fence after queued compute work.
173 if (!completion->fence) { // Check that the driver allocated a valid completion fence.
174 d.state()->gl.check("create GPU submission fence"); // Expose retained runtime state to the resource and backend helpers.
175 throw error("cannot create GPU submission fence"); // Report completion-fence allocation failure.
176 } // End this scope in submit.
177 d.state()->gl.Flush(); // Flush queued GPU commands for driver execution.
178 d.state()->gl.check("submit GPU pipeline"); // Expose retained runtime state to the resource and backend helpers.
179 return submission(std::move(completion)); // Queue compute work and return its completion handle.
180} // End this scope in submit.
181
182/** @brief A bounded CPU executor for independent pipeline jobs.
183 * Each submit() snapshots kernels, uniforms and resource bindings. Snapshots
184 * have independent invocation state, while buffers and images retain shared
185 * storage. Give concurrent jobs distinct writable resources, such as ring slots.
186 * Queue saturation throws immediately. Futures report results or exceptions.
187 * Destroying a future releases its result handle, while the executor retains
188 * the job until completion. shutdown() drains accepted jobs and joins workers.
189 * Executor destruction also drains and joins them.
190 */
191class executor { // Keep the executor values and operations together.
192 struct job { // Keep the job values and operations together.
193 std::function<void()> function; // Keep the application function executed by a CPU worker.
194 std::promise<void> completion; // Keep the completion result that the caller will receive.
195 explicit job(const std::function<void()>& fn) : function(fn) { // Keep the accepted CPU function and its promise result.
196 } // End this scope in async.
197 }; // End this scope in async.
198 struct queue_state { // Share the bounded queue between its producer and workers.
199 std::mutex mutex; // Keep the lock protecting shared state.
200 std::condition_variable changed; // Wake sleeping workers when work or shutdown state changes.
201 std::deque<std::shared_ptr<job>> jobs; // Keep accepted CPU work awaiting an available worker.
202 std::vector<std::thread::id> worker_ids; // Keep the worker thread identifiers used to reject self-joining.
203 std::size_t pending, capacity; // Keep the current pending count and the maximum queue size.
204 bool closing; // Keep whether the CPU executor is draining for shutdown.
205 explicit queue_state(std::size_t n) : pending(0), capacity(n), closing(false) { // Start an empty queue with the chosen pending-job capacity.
206 } // End this scope in async.
207 }; // End this scope in async.
208 std::shared_ptr<queue_state> state_; // Share the bounded queue between the executor and its workers.
209 std::vector<std::thread> workers_; // Keep the threads that execute accepted CPU jobs.
210 std::mutex shutdown_mutex_; // Keep the lock that serializes executor shutdown.
211
212 static void consume(const std::shared_ptr<queue_state>& state) { // Take accepted jobs from the queue and execute them on this worker.
213 for (;;) { // Keep this worker available until its queue is drained for shutdown.
214 std::shared_ptr<job> next; // Keep the CPU job selected by this worker.
215 { // Begin the protected scope for this operation.
216 std::unique_lock<std::mutex> lock(state->mutex); // Hold the mutex while this shared state is accessed.
217 state->changed.wait(lock, [&] { // Release the queue lock while sleeping until work or shutdown arrives.
218 return state->closing || !state->jobs.empty(); // Wake the worker for a queued job or an explicit shutdown.
219 }); // End this scope in consume.
220 if (state->jobs.empty()) // Exit only when shutdown has awakened an empty queue.
221 return; // Finish this operation.
222 next = state->jobs.front(); // Set the CPU job selected by this worker.
223 state->jobs.pop_front(); // Remove this accepted job from the pending queue.
224 --state->pending; // Release its pending slot before executing the job.
225 } // End this scope in consume.
226 try { // Keep cleanup available if an allocation, shader, or driver operation fails.
227 next->function(); // Execute the selected application function or CPU pipeline snapshot.
228 next->function = std::function<void()>(); // Release kernel and resource captures before publishing completion.
229 next->completion.set_value(); // Publish successful completion after the job releases its captures.
230 } catch (...) { // Preserve cleanup or exception propagation for this failed operation.
231 const std::exception_ptr failure = std::current_exception(); // Set the exception captured from a CPU job.
232 next->function = std::function<void()>(); // Release kernel and resource captures before publishing completion.
233 next->completion.set_exception(failure); // Deliver the worker failure through the caller completion future.
234 } // End this scope in consume.
235 } // End this scope in consume.
236 } // End this scope in consume.
237 void reserve() { // Reserve capacity and report a full queue immediately.
238 std::lock_guard<std::mutex> lock(state_->mutex); // Hold the mutex while this shared state is accessed.
239 if (state_->closing) // Check the queue state before accepting or executing work.
240 throw error("CPU executor is shut down"); // Report cpu executor is shut down.
241 if (state_->pending == state_->capacity) // Check the queue state before accepting or executing work.
242 throw error("CPU executor queue is full"); // Report cpu executor queue is full.
243 ++state_->pending; // Count the accepted job against the pending queue capacity.
244 } // End this scope in reserve.
245 void abandon() noexcept { // Release the reserved slot after a submission failure.
246 std::lock_guard<std::mutex> lock(state_->mutex); // Hold the mutex while this shared state is accessed.
247 --state_->pending; // Return the queue slot reserved by an unsuccessful submission.
248 } // End this scope in abandon.
249 std::future<void> append_reserved(const std::function<void()>& function) { // Attach the accepted CPU job to its reserved queue slot.
250 std::shared_ptr<job> work(new job(function)); // Keep the application function executed by a CPU worker.
251 std::future<void> result = work->completion.get_future(); // Set the value returned by this operation.
252 { // Begin the protected scope for this operation.
253 std::lock_guard<std::mutex> lock(state_->mutex); // Hold the mutex while this shared state is accessed.
254 if (state_->closing) // Check the queue state before accepting or executing work.
255 throw error("CPU executor is shut down"); // Report cpu executor is shut down.
256 state_->jobs.push_back(work); // Retain this entry in the ordered resource or work collection.
257 } // End this scope in append reserved.
258 state_->changed.notify_one(); // Wake workers so they can observe new queue or shutdown state.
259 return result; // Return the value returned by this operation.
260 } // End this scope in append reserved.
261
262 public: // Expose the operations callers can use.
263 /** @brief Start workers with room for at most capacity pending jobs.
264 * Capacity counts jobs waiting for a worker. Both arguments must be positive.
265 */
266 explicit executor(std::size_t workers = 1, std::size_t capacity = 64) // Start a fixed number of workers with a bounded pending-job queue.
267 : state_(std::make_shared<queue_state>(capacity)) { // Keep the maximum number of pending CPU jobs.
268 if (!workers || !capacity) // Require positive worker and queue counts.
269 throw error("CPU executor requires positive worker and queue counts"); // Report cpu executor requires positive worker and queue counts.
270 workers_.reserve(workers); // Allocate space for every worker thread handle.
271 state_->worker_ids.reserve(workers); // Allocate space for every worker thread identifier.
272 try { // Keep cleanup available if an allocation, shader, or driver operation fails.
273 for (std::size_t i = 0; i < workers; ++i) { // Visit this axis of the dispatch or current workgroup.
274 const std::shared_ptr<queue_state> state = state_; // Set the shared device or worker state.
275 workers_.push_back(std::thread([state] { // Retain this entry in the ordered resource or work collection.
276 consume(state); // Take accepted jobs from the queue and execute them on this worker.
277 })); // End this scope in executor.
278 state_->worker_ids.push_back(workers_.back().get_id()); // Retain this entry in the ordered resource or work collection.
279 } // End this scope in executor.
280 } catch (...) { // Preserve cleanup or exception propagation for this failed operation.
281 shutdown(); // Drain accepted CPU work and join the workers from an application thread.
282 throw; // Rethrow the original failure after arranging the necessary cleanup.
283 } // End this scope in executor.
284 } // End this scope in executor.
285 ~executor() { // Drain accepted work and join the CPU workers at explicit pool teardown.
286 shutdown(); // Drain accepted CPU work and join the workers from an application thread.
287 } // End this scope in executor.
288 executor(const executor&) = delete; // Prevent copying this object and duplicating its retained identity.
289 executor& operator=(const executor&) = delete; // Prevent copying this object and duplicating its retained identity.
290 /** @brief Snapshot a host-ready CPU pipeline, enqueue it and return a future.
291 * Each stage must support CPU execution and have current host data.
292 */
293 std::future<void> submit(const pipeline& source) { // Queue compute work and return its completion handle.
294 reserve(); // Reserve capacity and report a full queue immediately.
295 try { // Keep cleanup available if an allocation, shader, or driver operation fails.
296 source.require_host_ready(); // Validate CPU stages and host-ready resources before queueing.
297 const pipeline copy = source.snapshot(); // Capture isolated kernel interfaces and the current bindings and uniforms.
298 return append_reserved([copy] { // Queue compute work and return its completion handle.
299 copy.run(); // Execute the retained CPU snapshot on its worker thread.
300 }); // End this scope in submit.
301 } catch (...) { // Preserve cleanup or exception propagation for this failed operation.
302 abandon(); // Release the reserved slot after a submission failure.
303 throw; // Rethrow the original failure after arranging the necessary cleanup.
304 } // End this scope in submit.
305 } // End this scope in submit.
306 /** @brief Enqueue an application CPU function on the same bounded queue.
307 * Run host operations in the callback and keep device operations on their context thread.
308 */
309 template <class Function> std::future<void> enqueue(Function function) { // Select the scalar, dimension, or access type at compile time.
310 reserve(); // Reserve capacity and report a full queue immediately.
311 try { // Keep cleanup available if an allocation, shader, or driver operation fails.
312 return append_reserved(std::function<void()>(function)); // Queue the application callback on the bounded CPU executor.
313 } catch (...) { // Preserve cleanup or exception propagation for this failed operation.
314 abandon(); // Release the reserved slot after a submission failure.
315 throw; // Rethrow the original failure after arranging the necessary cleanup.
316 } // End this scope in enqueue.
317 } // End this scope in enqueue.
318 /** @brief Return the count of queued or currently being constructed jobs. */
319 std::size_t pending() const { // Read the count of jobs waiting for a worker.
320 std::lock_guard<std::mutex> lock(state_->mutex); // Hold the mutex while this shared state is accessed.
321 return state_->pending; // Read the count of jobs waiting for a worker.
322 } // End this scope in pending.
323 /** @brief Return the maximum number of pending jobs. */
324 std::size_t capacity() const { // Return the configured pending-job limit.
325 return state_->capacity; // Return the configured pending-job limit.
326 } // End this scope in capacity.
327 /** @brief Return the number of worker threads. */
328 std::size_t workers() const { // Return the number of worker threads.
329 return workers_.size(); // Return the number of worker threads.
330 } // End this scope in workers.
331 /** @brief Drain accepted jobs and join all workers. later submissions fail.
332 * Call from an application thread so it can join every worker.
333 */
334 void shutdown() { // Drain accepted CPU work and join the workers from an application thread.
335 if (std::find(state_->worker_ids.begin(), // Choose this branch when std::find(state ->worker ids.begin(),.
336 state_->worker_ids.end(), // Keep the end of the current range.
337 std::this_thread::get_id()) != state_->worker_ids.end()) // Reject joining the current executor worker from its own job.
338 throw error("CPU executor shutdown requires an application thread"); // Report cpu executor shutdown requires an application thread.
339 std::lock_guard<std::mutex> lifecycle(shutdown_mutex_); // Hold the mutex while this shared state is accessed.
340 { // Begin the protected scope for this operation.
341 std::lock_guard<std::mutex> lock(state_->mutex); // Hold the mutex while this shared state is accessed.
342 state_->closing = true; // Set whether the CPU executor is draining for shutdown.
343 } // End this scope in shutdown.
344 state_->changed.notify_all(); // Wake workers so they can observe new queue or shutdown state.
345 for (std::size_t i = 0; i < workers_.size(); ++i) // Visit each CPU worker during lifecycle handling.
346 if (workers_[i].joinable()) // Choose this branch when workers [i].joinable().
347 workers_[i].join(); // Join this CPU worker after accepted work has drained.
348 } // End this scope in shutdown.
349}; // End this scope in async.
350} // namespace shdr
351#endif // SHDR_ASYNC_HPP