pw_async2#
Cooperative async tasks for embedded
Experimental C++17
Simple Ownership: Say goodbye to that jumble of callbacks and shared state! Complex tasks with many concurrent elements can be expressed by simply combining smaller tasks.
Efficient: No dynamic memory allocation required.
Pluggable: Your existing event loop, work queue, or task scheduler can run the
Dispatcherwithout any extra threads.Coroutine-capable: C++20 coroutines work just like other tasks, and can easily plug into an existing
pw_async2system.
pw::async2::Task is Pigweed’s async primitive. Task objects
are cooperatively-scheduled “threads” which yield to the
pw::async2::Dispatcher when waiting. When the Task is able to
make progress, the Dispatcher will run it again. For example:
1#include "pw_async2/dispatcher.h"
2#include "pw_async2/poll.h"
3#include "pw_async2/try.h"
4#include "pw_log/log.h"
5#include "pw_result/result.h"
6
7namespace {
8
9using ::pw::async2::Context;
10using ::pw::async2::Pending;
11using ::pw::async2::Poll;
12using ::pw::async2::Ready;
13using ::pw::async2::Task;
14
15class MyReceiver;
16class MySender;
17
18// Receive then send that data asynchronously. If the reader or writer aren't
19// ready, the task suspends when TRY_READY invokes ``return Pending()``.
20class ReceiveAndSend final : public Task {
21 public:
22 ReceiveAndSend(MyReceiver receiver, MySender sender)
23 : receiver_(receiver), sender_(sender), state_(kReceiving) {}
24
25 Poll<> DoPend(Context& cx) final {
26 switch (state_) {
27 case kReceiving: {
28 PW_TRY_READY_ASSIGN(auto new_data, receiver_.PendReceive(cx));
29 if (!new_data.ok()) {
30 PW_LOG_ERROR("Receiving failed: %s", new_data.status().str());
31 return Ready(); // Completes the task.
32 }
33 // Start transmitting and switch to transmitting state.
34 send_future_ = sender_.Send(std::move(*new_data));
35 state_ = kTransmitting;
36 }
37 [[fallthrough]];
38 case kTransmitting: {
39 PW_TRY_READY_ASSIGN(auto sent, send_future_->Pend(cx));
40 if (!sent.ok()) {
41 PW_LOG_ERROR("Sending failed: %s", sent.str());
42 }
43 return Ready(); // Completes the task.
44 }
45 }
46 }
47
48 private:
49 MyReceiver receiver_; // Can receive data async.
50 MySender sender_; // Can send data async.
51 std::optional<SendFuture> send_future_ = std::nullopt;
52
53 enum State { kTransmitting, kReceiving };
54 State state_;
55};
56
57} // namespace
1#include "pw_allocator/allocator.h"
2#include "pw_async2/coro.h"
3#include "pw_log/log.h"
4#include "pw_result/result.h"
5
6namespace {
7
8using ::pw::OkStatus;
9using ::pw::Result;
10using ::pw::Status;
11using ::pw::async2::Coro;
12using ::pw::async2::CoroContext;
13
14class MyReceiver;
15class MySender;
16
17/// Create a coroutine which asynchronously receives a value from
18/// ``receiver`` and forwards it to ``sender``.
19///
20/// Note: the ``CoroContext`` argument is used by the ``Coro<T>`` internals to
21/// allocate the coroutine state. If this allocation fails, ``Coro<Status>``
22/// will return ``Status::Internal()``.
23Coro<Status> ReceiveAndSendCoro(CoroContext&,
24 MyReceiver receiver,
25 MySender sender) {
26 Result<MyData> data = co_await receiver.Receive();
27 if (!data.ok()) {
28 PW_LOG_ERROR("Receiving failed: %s", data.status().str());
29 co_return Status::Unavailable();
30 }
31 Status sent = co_await sender.Send(std::move(*data));
32 if (!sent.ok()) {
33 PW_LOG_ERROR("Sending failed: %s", sent.str());
34 co_return Status::Unavailable();
35 }
36 co_return OkStatus();
37}
38
39} // namespace
Tasks can then be run on a pw::async2::Dispatcher using the pw::async2::Dispatcher::Post() method:
1 auto task = ReceiveAndSend(std::move(receiver), std::move(sender));
2 Dispatcher dispatcher;
3 // Registers `task` to run on the dispatcher.
4 dispatcher.Post(task);
5 // Sets the dispatcher to run until all `Post`ed tasks have completed.
6 dispatcher.RunToCompletion();