5. Communicate between tasks#

pw_async2: Cooperative async tasks for embedded

Your vending machine is ready to handle more functionality. In this step, you’ll write code to handle a dispenser mechanism. The vending machine uses a motor to push the selected product into a chute. A sensor detects when the item has dropped, then the motor is turned off.

The dispenser mechanism is complex enough to merit a task of its own. The VendingMachineTask will notify the new DispenserTask about which items to dispense, and DispenserTask will send confirmation back.

Set up the item drop sensor#

The ItemDropSensor class that we’ve provided is similar to the CoinSlot and Keypad classes.

Integrate the item drop sensor into main.cc:

Include item_drop_sensor.h
Create a global codelab::ItemDropSensor item_drop_sensor instance
Call the item drop sensor’s Drop() method in the interrupt handler (item_drop_sensor_isr())

When an item is dispensed successfully, the item drop sensor triggers an interrupt, which is handled by the item_drop_sensor_isr() function.

Set up communication channels#

We’ll be adding a new DispenserTask soon. To get ready for that, let’s set up communications channels between VendingMachineTask and the new task.

The mechanism through which async tasks communicate in pw_async2 is pw::async2::Channel. At its core, a channel is a asynchronous, threadsafe queue that can have one or more senders and one or more receivers.

For our vending machine, we’ll use two channels:

A Channel<int> for the vending machine to send dispense requests (item numbers) to the dispenser task.
A Channel<bool> for the dispenser task to send dispense responses (success or failure) back to the vending machine task.

Update vending_machine.h to allow communications:

Include pw_async2/channel.h
Update your VendingMachineTask to accept and store a pw::async2::Sender<int> to send requests and a pw::async2::Receiver<bool> to receive responses.

Create the channels in main.cc:
- Define a local pw::async2::ChannelStorage<int, 1> dispense_requests and a pw::async2::ChannelStorage<bool, 1> dispense_responses in your main() function. These serve as the buffers for the channels.
  
  Note
  
  Channels optionally can dynamically allocate their storage, removing the need for a ChannelStorage declaration. Refer to the Channel docs for more details.
- Create the channels themselves using the pw::async2::CreateSpscChannel() function. Spsc means “single producer, single consumer”, which is what we need to communicate between the two tasks.
  
  CreateSpscChannel returns a tuple of three items: a handle to the channel, a sender, and a receiver.
  
  The channel handle is used if you need to close the channel early, and to create additional senders and receivers if the channel supports it.
  
  Since we have an SPSC channel, no new senders or receivers can be created, and we don’t need to close our channels, we don’t need the handle here. To indicate this, we immediately call Release on the two handles we created, telling the channel that it should manage its lifetime automatically via its active senders and receivers.
- Pass the sender for the requests channel and the receiver for the responses channel to the VendingMachineTask constructor
- Create a DispenserTask instance (we’ll define this in the next section), passing the drop sensor and the opposite sender/receiver pair to its constructor
- Post the new DispenserTask to the dispatcher
Hint
1int main() { 2 pw::async2::BasicDispatcher dispatcher; 3 codelab::HardwareInit(&dispatcher); 4 5 pw::async2::ChannelStorage<int, 1> dispense_requests; 6 pw::async2::ChannelStorage<bool, 1> dispense_responses; 7 8 auto [req_handle, dispense_requests_sender, dispense_requests_receiver] = 9 pw::async2::CreateSpscChannel(dispense_requests); 10 auto [res_handle, dispense_responses_sender, dispense_responses_receiver] = 11 pw::async2::CreateSpscChannel(dispense_responses); 12 req_handle.Release(); 13 res_handle.Release(); 14 15 codelab::VendingMachineTask task(coin_slot, 16 keypad, 17 std::move(dispense_requests_sender), 18 std::move(dispense_responses_receiver)); 19 dispatcher.Post(task); 20 21 codelab::DispenserTask dispenser_task(item_drop_sensor, 22 std::move(dispense_requests_receiver), 23 std::move(dispense_responses_sender)); 24 dispatcher.Post(dispenser_task); 25 26 dispatcher.RunToCompletion(); 27 28 return 0; 29}

Lifecycle of an async2 Channel

Channels begin open when created through CreateSpscChannel, CreateMpscChannel, CreateSpmcChannel, or CreateMpmcChannel. Each of these functions returns a handle to the channel, optionally alongside a sender and/or a receiver depending on the channel type.

Senders and receivers are movable but not copyable. If a specific channel type supports multiple senders or receivers, they are created by calling CreateSender() or CreateReceiver() on the handle.

In addition to creating senders and receivers, the channel handle has a Close() function which is used to close the channel early, bypassing its natural lifecycle. Channel handles are copyable, so they can be handed out to different parts of the system that need to manage the channel.

As long as handles exist, the channel remains open. When the last handle is released or destroyed, the channel’s lifecycle is controlled by the active senders and receivers. As long as at least one sender and one receiver exists, the channel is open. When either side fully hangs up, the channel closes. Receivers can drain the channel of all buffered messages even after the senders have hung up.

Since channel handles keep a channel alive, it is important to get rid of them when they’re no longer needed. The Release() function is used for this purpose. It is common to immediately release the handle returned by a channel creation function if you don’t need to close the channel early.

Create the new dispenser task#

The DispenserTask will turn the dispenser motor on and off in response to dispense requests from the VendingMachineTask.

Declare a new DispenserTask class in vending_machine.h:

The DispenserTask constructor should accept references to the drop sensor, the requests channel receiver, and the responses channel sender as args.
Create a State enum member with these states:
- kIdle: Waiting for a dispense request (motor is off)
- kDispensing: Actively dispensing an item (motor is on)
- kReportDispenseSuccess: Waiting to report success (motor is off)
Create a data member to store the current state
Remember to define your DoPend override

Implement the dispenser’s state machine in vending_machine.cc:

In kIdle, call Receive to get a future for the next dispense request and pend it to completion, then turn on the dispenser motor using SetDispenserMotorState, which is provided in hardware.h.
In kDispensing, wait for the item drop sensor to trigger then turn off the dispenser motor and transition to kReportDispenseSuccess.
In kReportDispenseSuccess, use the response channel sender to send true to the VendingMachineTask and transition to kIdle.

Note

Dispensing can’t fail yet. We’ll get to that later.

Communicate between tasks#

Now, let’s get VendingMachineTask communicating with DispenserTask.

Instead of just logging when a purchase is made, VendingMachineTask will send the selected item to the DispenserTask through the dispense requests channel. Then it will wait for a response with the dispense responses channel.

Prepare VendingMachineTask for comms in vending_machine.h:

Add new states: kAwaitingDispenseIdle (dispenser is ready for a request) and kAwaitingDispense (waiting for dispenser to finish dispensing an item)
Define member variables for the request send and response receive futures.

Update the vending machine task’s state machine in vending_machine.cc:

Transition the kAwaitingSelection state to kAwaitingDispenseIdle, storing the selected item in a member variable.
Implement the kAwaitingDispenseIdle and kAwaitingDispense states:
- In kAwaitingDispenseIdle, send the selected item to the DispenserTask using the dispense requests sender.
- In kAwaitingDispense, wait for a response from the DispenserTask and log whether the dispense was successful.
  
  Once complete, you can return to kWelcome to start the process over.

Test the dispenser#

Run the app:
```
bazelisk run //pw_async2/codelab
```
Press c Enter to input a coin.
Press 1 Enter to make a selection.

Press i Enter to trigger the item drop sensor, signaling that the item has finished dispensing.

You should see the vending machine display it’s welcome message again.

INF  Welcome to the Pigweed Vending Machine!
INF  Please insert a coin.
INF  Dispenser task awake
c
INF  Received 1 coin.
INF  Please press a keypad key.
1
INF  Keypad 1 was pressed. Dispensing an item.
INF  Dispenser task awake
INF  [Motor for item 1 set to On]
i
INF  Dispenser task awake
INF  [Motor for item 1 set to Off]
INF  Dispense succeeded. Thanks for your purchase!
INF  Welcome to the Pigweed Vending Machine!
INF  Please insert a coin.

Congratulations! You now have a fully functioning vending machine! Or do you…?

Handle unexpected situations with timeouts#

What if you press the wrong button and accidentally buy an out-of-stock item? As of now, the dispenser will just keep running forever. The vending machine will eat your money while you go hungry.

Let’s fix this. We can add a timeout to the kDispensing state. If the ItemDropSensor hasn’t triggered after a certain amount of time, then something has gone wrong. The DispenserTask should stop the motor and tell the VendingMachineTask what happened. Pigweed provides a Timeout helper which works with various common future types.

Prepare DispenserTask to support timeouts in vending_machine.h:

Include the header that provides timeout-related features:

pw_async2/future_timeout.h

Create a new kReportDispenseFailure state to represent dispense failures, a new kDispenseTimeout data member in DispenserTask that holds the timeout duration (std::chrono::seconds(5) is a good value), and update the drop_future_ member to be a ValueFutureWithTimeout:

For testing purposes, make sure that the timeout period is long enough for a human to respond.

Implement the timeout support in vending_machine.cc:

When you start dispensing an item (in your transition from kIdle to kDispensing), initialize drop_future_ using Timeout.

Timeout takes the future to wrap and the timeout duration as arguments.
In the kDispensing state, pend the drop_future_.
Check the result of the future:
- If the result is OK, then the item dropped successfully. Proceed to the kReportDispenseSuccess state.
- If the result status is DeadlineExceeded, then the timeout occurred. Proceed to the kReportDispenseFailure state.
- In either case, be sure to turn off the motor and clear the dispense request.
Handle the dispense failure state.

Test the dispenser with timeouts#

Run the app:
```
bazelisk run //pw_async2/codelab
```
Press c Enter to input a coin.

# Press 1 Enter to make a selection.

Wait for the timeout.

After 5 seconds you should see a message about the dispense failing.

INF  Welcome to the Pigweed Vending Machine!
INF  Please insert a coin.
INF  Dispenser task awake
c
INF  Received 1 coin.
INF  Please press a keypad key.
1
INF  Keypad 1 was pressed. Dispensing an item.
INF  [Motor for item 1 set to On]
INF  [Motor for item 1 set to Off]
INF  Dispense failed. Choose another selection.

Try again, but this time press i Enter in 5 seconds or less so that the dispense succeeds.

Next steps#

Congratulations! You completed the codelab. Click DISPENSE PRIZE to retrieve your prize.

You now have a solid foundation in pw_async2 concepts, and quite a bit of hands-on experience with the framework. Try building something yourself with pw_async2! As always, if you get stuck, or if anything is unclear, or you just want to run some ideas by us, we would be happy to chat.

Checkpoint#

At this point, your code should look similar to the files below.

main.cc

#include "coin_slot.h"
#include "hardware.h"
#include "item_drop_sensor.h"
#include "pw_async2/basic_dispatcher.h"
#include "vending_machine.h"

namespace {

codelab::CoinSlot coin_slot;
codelab::Keypad keypad;
codelab::ItemDropSensor item_drop_sensor;

}  // namespace

// Interrupt handler function invoked when the user inserts a coin into the
// vending machine.
void coin_inserted_isr() { coin_slot.Deposit(); }

// Interrupt handler function invoked when the user presses a key on the
// machine's keypad. Receives the value of the pressed key (0-9).
void key_press_isr(int key) { keypad.Press(key); }

// Interrupt handler function invoked to simulate the item drop detector
// detecting confirmation that an item was successfully dispensed from the
// machine.
void item_drop_sensor_isr() { item_drop_sensor.Drop(); }

int main() {
  pw::async2::BasicDispatcher dispatcher;
  codelab::HardwareInit(&dispatcher);

  pw::async2::ChannelStorage<int, 1> dispense_requests;
  pw::async2::ChannelStorage<bool, 1> dispense_responses;

  auto [req_handle, dispense_requests_sender, dispense_requests_receiver] =
      pw::async2::CreateSpscChannel(dispense_requests);
  auto [res_handle, dispense_responses_sender, dispense_responses_receiver] =
      pw::async2::CreateSpscChannel(dispense_responses);
  req_handle.Release();
  res_handle.Release();

  codelab::VendingMachineTask task(coin_slot,
                                   keypad,
                                   std::move(dispense_requests_sender),
                                   std::move(dispense_responses_receiver));
  dispatcher.Post(task);

  codelab::DispenserTask dispenser_task(item_drop_sensor,
                                        std::move(dispense_requests_receiver),
                                        std::move(dispense_responses_sender));
  dispatcher.Post(dispenser_task);

  dispatcher.RunToCompletion();

  return 0;
}

vending_machine.cc

#include "vending_machine.h"

#include <mutex>

#include "hardware.h"
#include "pw_async2/future_timeout.h"
#include "pw_async2/try.h"
#include "pw_log/log.h"
#include "pw_result/result.h"

namespace codelab {

pw::async2::Poll<int> KeyPressFuture::Pend(pw::async2::Context& cx) {
  return core_.DoPend(*this, cx);
}

KeyPressFuture::~KeyPressFuture() {
  std::lock_guard lock(internal::KeypadLock());
  core_.Unlist();
}

pw::async2::Poll<int> KeyPressFuture::DoPend(pw::async2::Context&) {
  std::lock_guard lock(internal::KeypadLock());
  if (key_pressed_.has_value()) {
    return pw::async2::Ready(*key_pressed_);
  }
  return pw::async2::Pending();
}

KeyPressFuture Keypad::WaitForKeyPress() {
  std::lock_guard lock(internal::KeypadLock());
  KeyPressFuture future(pw::async2::FutureState::kPending);
  futures_.Push(future.core_);
  return future;
}

void Keypad::Press(int key) {
  std::lock_guard lock(internal::KeypadLock());
  if (auto future = futures_.PopIfAvailable()) {
    future->key_pressed_ = key;
    future->core_.Wake();
  }
}

pw::async2::Poll<> VendingMachineTask::DoPend(pw::async2::Context& cx) {
  while (true) {
    switch (state_) {
      case kWelcome: {
        PW_LOG_INFO("Welcome to the Pigweed Vending Machine!");
        PW_LOG_INFO("Please insert a coin.");
        state_ = kAwaitingPayment;
        break;
      }

      case kAwaitingPayment: {
        if (!select_future_.is_pendable()) {
          select_future_ = pw::async2::Select(coin_slot_.GetCoins(),
                                              keypad_.WaitForKeyPress());
        }
        PW_TRY_READY_ASSIGN(auto result, select_future_.Pend(cx));
        if (result.has_value<0>()) {
          unsigned coins = result.value<0>();
          PW_LOG_INFO("Received %u coin%s.", coins, coins > 1 ? "s" : "");
          PW_LOG_INFO("Please press a keypad key.");
          coins_inserted_ += coins;
          state_ = kAwaitingSelection;
        } else {
          PW_LOG_ERROR("Please insert a coin before making a selection.");
        }
        break;
      }

      case kAwaitingSelection: {
        if (!select_future_.is_pendable()) {
          select_future_ = pw::async2::Select(coin_slot_.GetCoins(),
                                              keypad_.WaitForKeyPress());
        }
        PW_TRY_READY_ASSIGN(auto result, select_future_.Pend(cx));
        if (result.has_value<1>()) {
          int key = result.value<1>();
          PW_LOG_INFO("Keypad %d was pressed. Dispensing an item.", key);
          state_ = kAwaitingDispenseIdle;
          break;
        }

        coins_inserted_ += result.value<0>();
        PW_LOG_INFO("Received a coin. Your balance is currently %u coins.",
                    coins_inserted_);
        PW_LOG_INFO("Please press a keypad key.");
        break;
      }

      case kAwaitingDispenseIdle: {
        if (!dispense_request_future_.is_pendable()) {
          dispense_request_future_ = dispense_requests_.Send(item_to_dispense_);
        }
        PW_TRY_READY_ASSIGN(bool sent, dispense_request_future_.Pend(cx));
        if (!sent) {
          PW_LOG_WARN("Dispense requests channel unexpectedly closed.");
          return pw::async2::Ready();
        }
        state_ = kAwaitingDispense;
        break;
      }

      case kAwaitingDispense: {
        if (!dispense_response_future_.is_pendable()) {
          dispense_response_future_ = dispense_responses_.Receive();
        }
        std::optional<bool> received;
        PW_TRY_READY_ASSIGN(received, dispense_response_future_.Pend(cx));
        if (!received.has_value()) {
          PW_LOG_WARN("Dispense responses channel unexpectedly closed.");
          return pw::async2::Ready();
        }

        if (received.value()) {
          PW_LOG_INFO("Item dispensed successfully.");
          state_ = kWelcome;
        } else {
          PW_LOG_ERROR("Dispense failed. Choose another selection.");
          state_ = kAwaitingSelection;
        }
      }
    }
  }
}

pw::async2::Poll<> DispenserTask::DoPend(pw::async2::Context& cx) {
  while (true) {
    switch (state_) {
      case kIdle: {
        if (!dispense_request_future_.is_pendable()) {
          dispense_request_future_ = dispense_requests_.Receive();
        }
        PW_TRY_READY_ASSIGN(item_to_dispense_,
                            dispense_request_future_.Pend(cx));
        if (!item_to_dispense_.has_value()) {
          PW_LOG_WARN("Dispense requests channel unexpectedly closed.");
          return pw::async2::Ready();
        }
        SetDispenserMotorState(*item_to_dispense_, MotorState::kOn);
        state_ = kDispensing;
        break;
      }

      case kDispensing: {
        if (!drop_future_.is_pendable()) {
          drop_future_ =
              pw::async2::Timeout(drop_sensor_.Wait(), kDispenseTimeout);
        }
        PW_TRY_READY_ASSIGN(auto result, drop_future_.Pend(cx));
        SetDispenserMotorState(*item_to_dispense_, MotorState::kOff);
        item_to_dispense_ = std::nullopt;
        if (result.ok()) {
          state_ = kReportDispenseSuccess;
        } else {
          state_ = kReportDispenseFailure;
        }
        break;
      }

      case kReportDispenseSuccess: {
        if (!dispense_response_future_.is_pendable()) {
          dispense_response_future_ = dispense_responses_.Send(true);
        }
        PW_TRY_READY_ASSIGN(bool result, dispense_response_future_.Pend(cx));
        if (!result) {
          PW_LOG_WARN("Dispense responses channel unexpectedly closed.");
          return pw::async2::Ready();
        }
        state_ = kIdle;
        break;
      }

      case kReportDispenseFailure: {
        if (!dispense_response_future_.is_pendable()) {
          dispense_response_future_ = dispense_responses_.Send(false);
        }
        PW_TRY_READY_ASSIGN(bool result, dispense_response_future_.Pend(cx));
        if (!result) {
          PW_LOG_WARN("Dispense responses channel unexpectedly closed.");
          return pw::async2::Ready();
        }
        state_ = kIdle;
        break;
      }
    }
  }

  return pw::async2::Ready();
}

}  // namespace codelab

vending_machine.h

#pragma once

#include <optional>
#include <utility>

#include "coin_slot.h"
#include "item_drop_sensor.h"
#include "pw_async2/channel.h"
#include "pw_async2/context.h"
#include "pw_async2/future.h"
#include "pw_async2/future_timeout.h"
#include "pw_async2/poll.h"
#include "pw_async2/select.h"
#include "pw_async2/task.h"
#include "pw_sync/interrupt_spin_lock.h"
#include "pw_sync/lock_annotations.h"

namespace codelab {

namespace internal {

inline pw::sync::InterruptSpinLock& KeypadLock() {
  PW_CONSTINIT static pw::sync::InterruptSpinLock lock;
  return lock;
}

}  // namespace internal

class Keypad;

class KeyPressFuture {
 public:
  // The type returned by the future when it completes.
  using value_type = int;

  KeyPressFuture() = default;

  KeyPressFuture(KeyPressFuture&& other) = default;
  KeyPressFuture& operator=(KeyPressFuture&& other) = default;

  ~KeyPressFuture();

  // Pends until a key is pressed, returning the key number.
  pw::async2::Poll<value_type> Pend(pw::async2::Context& cx);

  [[nodiscard]] bool is_pendable() const { return core_.is_pendable(); }
  [[nodiscard]] bool is_complete() const { return core_.is_complete(); }

 private:
  friend class Keypad;
  friend class pw::async2::FutureCore;

  static constexpr const char kWaitReason[] = "Waiting for keypad press";

  explicit KeyPressFuture(pw::async2::FutureState::Pending)
      : core_(pw::async2::FutureState::kPending) {}

  pw::async2::Poll<value_type> DoPend(pw::async2::Context& cx);

  pw::async2::FutureCore core_;

  // When present, holds the key that was pressed.
  // If absent, the future is still pending.
  std::optional<int> key_pressed_ PW_GUARDED_BY(internal::KeypadLock());
};

// Ensure that KeyPressFuture satisfies the Future concept.
static_assert(pw::async2::Future<KeyPressFuture>);

class Keypad {
 public:
  // Returns a future that resolves when a key is pressed with the value
  // of the key.
  //
  // May only be called by one task.
  KeyPressFuture WaitForKeyPress();

  // Record a key press. Typically called from the keypad ISR.
  void Press(int key);

 private:
  friend class KeyPressFuture;

  // The list of futures waiting for a key press.
  pw::async2::FutureList<&KeyPressFuture::core_> futures_
      PW_GUARDED_BY(internal::KeypadLock());
};

// The main task that drives the vending machine.
class VendingMachineTask : public pw::async2::Task {
 public:
  VendingMachineTask(CoinSlot& coin_slot,
                     Keypad& keypad,
                     pw::async2::Sender<int> dispense_requests,
                     pw::async2::Receiver<bool> dispense_responses)
      : pw::async2::Task(PW_ASYNC_TASK_NAME("VendingMachineTask")),
        coin_slot_(coin_slot),
        keypad_(keypad),
        dispense_requests_(std::move(dispense_requests)),
        dispense_responses_(std::move(dispense_responses)),
        coins_inserted_(0),
        state_(kWelcome) {}

 private:
  // This is the core of the asynchronous task. The dispatcher calls this method
  // to give the task a chance to do work.
  pw::async2::Poll<> DoPend(pw::async2::Context& cx) override;

  CoinSlot& coin_slot_;
  Keypad& keypad_;
  pw::async2::SelectFuture<CoinFuture, KeyPressFuture> select_future_;
  pw::async2::Sender<int> dispense_requests_;
  pw::async2::Receiver<bool> dispense_responses_;
  pw::async2::SendFuture<int> dispense_request_future_;
  pw::async2::ReceiveFuture<bool> dispense_response_future_;
  unsigned coins_inserted_;
  int item_to_dispense_;

  enum State {
    kWelcome,
    kAwaitingPayment,
    kAwaitingSelection,
    kAwaitingDispenseIdle,
    kAwaitingDispense,
  };
  State state_;
};

class DispenserTask : public pw::async2::Task {
 public:
  DispenserTask(ItemDropSensor& drop_sensor,
                pw::async2::Receiver<int> dispense_requests,
                pw::async2::Sender<bool> dispense_responses)
      : pw::async2::Task(PW_ASYNC_TASK_NAME("DispenserTask")),
        drop_sensor_(drop_sensor),
        dispense_requests_(std::move(dispense_requests)),
        dispense_responses_(std::move(dispense_responses)),
        state_(kIdle) {}

 private:
  enum State {
    kIdle,
    kDispensing,
    kReportDispenseSuccess,
    kReportDispenseFailure,
  };

  pw::async2::Poll<> DoPend(pw::async2::Context& cx) override;

  ItemDropSensor& drop_sensor_;
  pw::async2::Receiver<int> dispense_requests_;
  pw::async2::Sender<bool> dispense_responses_;
  pw::async2::ReceiveFuture<int> dispense_request_future_;
  pw::async2::SendFuture<bool> dispense_response_future_;
  pw::async2::ValueFutureWithTimeout<void> drop_future_;
  std::optional<int> item_to_dispense_;
  State state_;

  static constexpr auto kDispenseTimeout = std::chrono::seconds(5);
};

}  // namespace codelab