Source browser
tests/test_async.cpp
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 test_async.cpp
2 * Queue, snapshot and completion checks that use promises as deterministic
3 * coordination points. The checks remain enabled in Release builds.
4 */
5#include "shdr.h" // Include shdr.h declarations.
6#include <atomic> // Include atomic declarations.
7#include <chrono> // Include chrono declarations.
8#include <future> // Include future declarations.
9#include <iostream> // Include iostream declarations.
10#include <stdexcept> // Include stdexcept declarations.
11
12namespace { // Group these helpers within their namespace.
13void require(bool condition, const char* message) { // Check that this operation satisfies the expected behavior.
14 if (!condition) // Check this condition before entering the following branch.
15 throw std::runtime_error(message); // Report this failed operation.
16} // End this block.
17template <class Function> void must_throw(Function function, const char* message) { // Accept the value types or dimensions used by this template.
18 bool caught = false; // Set caught from this expression.
19 try { // Run the operation while retaining the error handling below.
20 function(); // Apply function to these arguments.
21 } catch (const std::exception&) { // Report the exception from the preceding operation.
22 caught = true; // Set caught from this expression.
23 } // End this block.
24 require(caught, message); // Check that this operation satisfies the expected behavior.
25} // End this block.
26struct release_guard { // Declare release guard.
27 std::promise<void>* value; // Declare value for the following calculation.
28 explicit release_guard(std::promise<void>& promise) : value(&promise) { // Implement release guard for the operations below.
29 } // End this block.
30 void release() { // Implement release for the operations below.
31 if (value) { // Check this condition before entering the following branch.
32 value->set_value(); // Apply set value to these arguments.
33 value = 0; // Set value from this expression.
34 } // End this block.
35 } // End this block.
36 ~release_guard() { // Apply release guard to these arguments.
37 try { // Run the operation while retaining the error handling below.
38 release(); // Apply release to these arguments.
39 } catch (...) { // Report the exception from the preceding operation.
40 } // End this block.
41 } // End this block.
42}; // Finish this declaration.
43
44void queue_and_errors() { // Implement queue and errors for the operations below.
45 shdr::executor pool(1, 1); // Construct pool with these initial values.
46 std::promise<void> entered, released; // Declare entered for the following calculation.
47 release_guard unlock(released); // Construct unlock with these initial values.
48 std::shared_future<void> release = released.get_future().share(); // Set release from this expression.
49 std::future<void> active = pool.enqueue([&] { // Set active from this expression.
50 entered.set_value(); // Apply set value to these arguments.
51 release.wait(); // Apply wait to these arguments.
52 }); // Finish this declaration.
53 entered.get_future().wait(); // Apply get future to these arguments.
54 std::atomic<unsigned> calls(0); // Construct calls with these initial values.
55 std::future<void> queued = pool.enqueue([&] { // Set queued from this expression.
56 ++calls; // Continue the expression using calls.
57 }); // Finish this declaration.
58 require(pool.pending() == 1 && pool.capacity() == 1 && pool.workers() == 1, // Check that this operation satisfies the expected behavior.
59 "bounded pending queue"); // Provide this text argument for the operation above.
60 must_throw( // Check that this operation satisfies the expected behavior.
61 [&] { // Complete this expression or declaration.
62 pool.enqueue([] { // Apply enqueue to these arguments.
63 }); // Finish this declaration.
64 }, // Finish this declaration.
65 "full queue must reject a new job"); // Provide this text argument for the operation above.
66 require(queued.wait_for(std::chrono::nanoseconds(0)) == std::future_status::timeout, // Check that this operation satisfies the expected behavior.
67 "queued future is not falsely complete"); // Provide this text argument for the operation above.
68 unlock.release(); // Apply release to these arguments.
69 active.get(); // Apply get to these arguments.
70 queued.get(); // Apply get to these arguments.
71 require(calls == 1, "accepted job executes exactly once"); // Check that this operation satisfies the expected behavior.
72 std::future<void> failed = pool.enqueue([] { // Set failed from this expression.
73 throw std::runtime_error("task failure"); // Report task failure.
74 }); // Finish this declaration.
75 must_throw( // Check that this operation satisfies the expected behavior.
76 [&] { // Complete this expression or declaration.
77 failed.get(); // Apply get to these arguments.
78 }, // Finish this declaration.
79 "worker errors propagate through futures"); // Provide this text argument for the operation above.
80 pool.enqueue([&] { // Apply enqueue to these arguments.
81 ++calls; // Continue the expression using calls.
82 }) // End this block.
83 .get(); // Apply get to these arguments.
84 require(calls == 2, "worker survives a throwing job"); // Check that this operation satisfies the expected behavior.
85 std::future<void> forbidden_shutdown = pool.enqueue([&] { // Set forbidden shutdown from this expression.
86 pool.shutdown(); // Apply shutdown to these arguments.
87 }); // Finish this declaration.
88 must_throw( // Check that this operation satisfies the expected behavior.
89 [&] { // Complete this expression or declaration.
90 forbidden_shutdown.get(); // Apply get to these arguments.
91 }, // Finish this declaration.
92 "worker shutdown must report an error rather than deadlock"); // Provide this text argument for the operation above.
93 pool.shutdown(); // Apply shutdown to these arguments.
94 must_throw( // Check that this operation satisfies the expected behavior.
95 [&] { // Complete this expression or declaration.
96 pool.enqueue([] { // Apply enqueue to these arguments.
97 }); // Finish this declaration.
98 }, // Finish this declaration.
99 "shutdown rejects subsequent work"); // Provide this text argument for the operation above.
100 must_throw( // Check that this operation satisfies the expected behavior.
101 [] { // Complete this expression or declaration.
102 shdr::executor invalid(0, 1); // Construct invalid with these initial values.
103 }, // Finish this declaration.
104 "zero worker count must fail"); // Provide this text argument for the operation above.
105 must_throw( // Check that this operation satisfies the expected behavior.
106 [] { // Complete this expression or declaration.
107 shdr::executor invalid(1, 0); // Construct invalid with these initial values.
108 }, // Finish this declaration.
109 "zero capacity must fail"); // Provide this text argument for the operation above.
110} // End this block.
111
112void future_lifetime() { // Implement future lifetime for the operations below.
113 shdr::executor pool(1, 2); // Construct pool with these initial values.
114 std::promise<void> entered, released; // Declare entered for the following calculation.
115 release_guard unlock(released); // Construct unlock with these initial values.
116 std::shared_future<void> release = released.get_future().share(); // Set release from this expression.
117 std::atomic<bool> completed(false); // Construct completed with these initial values.
118 { // Begin this local scope.
119 std::future<void> dropped = pool.enqueue([&] { // Set dropped from this expression.
120 entered.set_value(); // Apply set value to these arguments.
121 release.wait(); // Apply wait to these arguments.
122 completed = true; // Set completed from this expression.
123 }); // Finish this declaration.
124 entered.get_future().wait(); // Apply get future to these arguments.
125 } // End this block.
126 // Reaching this point while release is pending verifies prompt future destruction.
127 // Explicit shutdown drains the worker after its release signal.
128 require(!completed, "destroying a future must not finish a waiting job"); // Check that this operation satisfies the expected behavior.
129 unlock.release(); // Apply release to these arguments.
130 pool.shutdown(); // Apply shutdown to these arguments.
131 require(completed, "shutdown drains work after its future was discarded"); // Check that this operation satisfies the expected behavior.
132
133 shdr::executor cleanup(1, 2); // Construct cleanup with these initial values.
134 std::shared_ptr<int> captured(new int(1)); // Construct captured with these initial values.
135 std::weak_ptr<int> lifetime = captured; // Set lifetime from this expression.
136 std::future<void> finished = cleanup.enqueue([captured] { // Set finished from this expression.
137 }); // Finish this declaration.
138 captured.reset(); // Apply reset to these arguments.
139 finished.get(); // Apply get to these arguments.
140 require(lifetime.expired(), "job captures are released before completion becomes visible"); // Check that this operation satisfies the expected behavior.
141} // End this block.
142
143void pipeline_snapshots() { // Implement pipeline snapshots for the operations below.
144 shdr::executor pool(1, 4); // Construct pool with these initial values.
145 std::promise<void> entered, released; // Declare entered for the following calculation.
146 release_guard unlock(released); // Construct unlock with these initial values.
147 std::shared_future<void> release = released.get_future().share(); // Set release from this expression.
148 std::future<void> gate = pool.enqueue([&] { // Set gate from this expression.
149 entered.set_value(); // Apply set value to these arguments.
150 release.wait(); // Apply wait to these arguments.
151 }); // Finish this declaration.
152 entered.get_future().wait(); // Apply get future to these arguments.
153 shdr::buffer<float> first(4), second(4), first_output(4), second_output(4); // Construct first with these initial values.
154 for (unsigned i = 0; i < 4; ++i) { // Iterate over i for this calculation.
155 first.data()[i] = float(i); // Apply data to these arguments.
156 second.data()[i] = float(i + 10); // Apply data to these arguments.
157 } // End this block.
158 auto shader = shdr::make_kernel("affine_kernel"); // Create the linked affine kernel instance.
159 shader->bind("input_values", first).bind("output_values", first_output); // Connect this shader port to its application resource.
160 shader->set<unsigned>("count", 4).set<float>("scale", 2).set<float>("offset", 1); // Supply this shader uniform with its typed application value.
161 shdr::pipeline pipeline(shdr::dispatch(*shader, shdr::extent(1))); // Add a shader dispatch with the required workgroup counts.
162 std::future<void> first_job = pool.submit(pipeline); // Submit work and retain its completion handle.
163 shader->bind("input_values", second).bind("output_values", second_output); // Connect this shader port to its application resource.
164 shader->set<float>("scale", 3).set<float>("offset", -1); // Supply this shader uniform with its typed application value.
165 std::future<void> second_job = pool.submit(pipeline); // Submit work and retain its completion handle.
166 shader.reset(); // The accepted jobs own their independent kernel snapshots.
167 unlock.release(); // Apply release to these arguments.
168 gate.get(); // Apply get to these arguments.
169 first_job.get(); // Apply get to these arguments.
170 second_job.get(); // Apply get to these arguments.
171 const shdr::buffer<float>& read_first = first_output; // Use a constant buffer view so GPU writes synchronize before host reads.
172 const shdr::buffer<float>& read_second = second_output; // Use a constant buffer view so GPU writes synchronize before host reads.
173 for (unsigned i = 0; i < 4; ++i) { // Iterate over i for this calculation.
174 require(read_first.data()[i] == float(i) * 2 + 1, // Check that this operation satisfies the expected behavior.
175 "uniforms and bindings of first snapshot"); // Provide this text argument for the operation above.
176 require(read_second.data()[i] == float(i + 10) * 3 - 1, // Check that this operation satisfies the expected behavior.
177 "uniforms and bindings of rebound snapshot"); // Provide this text argument for the operation above.
178 } // End this block.
179 auto gpu_shader = shdr::make_kernel("affine_kernel"); // Create the linked affine kernel instance.
180 gpu_shader->bind("input_values", first).bind("output_values", first_output); // Connect this shader port to its application resource.
181 shdr::pipeline gpu_only(shdr::dispatch(*gpu_shader, shdr::extent(1), shdr::backend::gpu)); // Add a shader dispatch with the required workgroup counts.
182 must_throw( // Check that this operation satisfies the expected behavior.
183 [&] { // Complete this expression or declaration.
184 pool.submit(gpu_only); // Submit work and retain its completion handle.
185 }, // Finish this declaration.
186 "CPU workers must reject GPU-only work"); // Provide this text argument for the operation above.
187} // End this block.
188
189void parallel_jobs() { // Implement parallel jobs for the operations below.
190 shdr::executor pool(2, 8); // Construct pool with these initial values.
191 shdr::buffer<float> input(257), output_a(257), output_b(257); // Construct input with these initial values.
192 for (unsigned i = 0; i < 257; ++i) // Iterate over i for this calculation.
193 input.data()[i] = float(i); // Apply data to these arguments.
194 auto shader = shdr::make_kernel("affine_kernel"); // Create the linked affine kernel instance.
195 shader->bind("input_values", input).bind("output_values", output_a); // Connect this shader port to its application resource.
196 shader->set<unsigned>("count", 257).set<float>("scale", 2).set<float>("offset", 0); // Supply this shader uniform with its typed application value.
197 shdr::pipeline pipeline(shdr::dispatch(*shader, shdr::extent(5))); // Add a shader dispatch with the required workgroup counts.
198 std::future<void> a = pool.submit(pipeline); // Submit work and retain its completion handle.
199 shader->bind("output_values", output_b).set<float>("scale", 3); // Connect this shader port to its application resource.
200 std::future<void> b = pool.submit(pipeline); // Submit work and retain its completion handle.
201 a.get(); // Apply get to these arguments.
202 b.get(); // Apply get to these arguments.
203 const shdr::buffer<float>& read_a = output_a; // Use a constant buffer view so GPU writes synchronize before host reads.
204 const shdr::buffer<float>& read_b = output_b; // Use a constant buffer view so GPU writes synchronize before host reads.
205 for (unsigned i = 0; i < 257; ++i) { // Iterate over i for this calculation.
206 require(read_a.data()[i] == float(i) * 2, "parallel first kernel invocation state"); // Check that this operation satisfies the expected behavior.
207 require(read_b.data()[i] == float(i) * 3, "parallel second kernel invocation state"); // Check that this operation satisfies the expected behavior.
208 } // End this block.
209 shdr::submission empty; // Declare empty for the following calculation.
210 require(empty.ready() && empty.wait(std::chrono::nanoseconds(0)) == shdr::wait_status::ready, // Check that this operation satisfies the expected behavior.
211 "empty GPU completion handle"); // Provide this text argument for the operation above.
212} // End this block.
213} // namespace
214
215int main() { // Run this example and validate its results.
216 try { // Run the operation while retaining the error handling below.
217 queue_and_errors(); // Apply queue and errors to these arguments.
218 future_lifetime(); // Apply future lifetime to these arguments.
219 pipeline_snapshots(); // Apply pipeline snapshots to these arguments.
220 parallel_jobs(); // Apply parallel jobs to these arguments.
221 std::cout << "Async CPU queue and snapshot checks passed\n"; // Write this part of the example result or diagnostic.
222 return 0; // Report successful completion.
223 } catch (const std::exception& e) { // Report the exception from the preceding operation.
224 std::cerr << e.what() << '\n'; // Write this part of the example result or diagnostic.
225 return 1; // Report a failed check or operation.
226 } // End this block.
227} // End this block.