MCPcopy Create free account
hub / github.com/NVIDIA/stdexec / __consume

Method __consume

include/stdexec/__detail/__spawn_future.hpp:245–311  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

243 // NOTE: _Rcvr is unconstrained because the thing we pass doesn't satisfy receiver
244 template <class _Rcvr>
245 void __consume(_Rcvr& __rcvr) noexcept
246 {
247 // Write this before synchronizing with the producer, below.
248 __callback_ = [](__spawn_future_state* __self, void* __ptr) noexcept
249 {
250 auto& __rcvr = *static_cast<_Rcvr*>(__ptr);
251 if (__self != nullptr)
252 {
253 __self->__do_consume(__rcvr);
254 }
255 else
256 {
257 STDEXEC::set_stopped(std::move(__rcvr));
258 }
259 };
260
261 void* __sentinel = nullptr;
262 if (__registered_receiver_.compare_exchange_strong(
263 __sentinel,
264 std::addressof(__rcvr),
265 // We need store-release on success to ensure that the future completion of the
266 // producer can see the callback we wrote into __callback_, and we need load-acquire
267 // on failure in case we're about to observe that the producer has already finished
268 // so we can see the result it produced. The success order must be stronger than the
269 // failure order so success has to be acquire-release.
270 __std::memory_order_acq_rel,
271 __std::memory_order_acquire))
272 {
273 // Since our CAS succeeded, we can conclude that we observed a null __registered_receiver_
274 // and successfully updated it to point to the receiver. That means the consumer has
275 // successfully registered a receiver and it's up to the producer to complete it when the
276 // result is ready; alternatively, a stop request may arrive leading __try_cancel to try
277 // to complete us eagerly with set_stopped. In either case, __try_cancel and __complete
278 // will negotiate how to complete the future and we have nothing left to do.
279 return;
280 }
281
282 if (__sentinel == (this + 1))
283 { // NOTE: we didn't update __registered_receiver_
284 // __try_cancel ran before both __complete and __consume; now we need to negotiate with
285 // __complete to decide whether it finished in time to consume its output.
286 //
287 // We need acquire-release semantics here. If we succeed in abandoning the operation then
288 // the producer will be responsible for invoking __destroy, which means it needs to see
289 // the write to __callback_, which requires a store-release. IF we fail to abandon the
290 // operation then that means the producer finished in time for us to consume its result,
291 // which means we need a load-acquire to consume it properly.
292 __sentinel = __registered_receiver_.exchange(this, __std::memory_order_acq_rel);
293 }
294
295 if (__sentinel == this)
296 {
297 // Either the producer completed before we CAS'd, or it snuck in and completed between
298 // our CAS and our exchange; in either case, we ought to consume its result.
299 __do_consume(__rcvr);
300 __destroy();
301 }
302 else

Callers 1

__runMethod · 0.80

Calls 4

__do_consumeMethod · 0.80
exchangeMethod · 0.80
set_stoppedFunction · 0.70

Tested by

no test coverage detected