future.hpp

Go to the documentation of this file.

// nros-cpp: Future<T> -- single-shot deferred result
// Freestanding C++ -- no exceptions, no STL required
 
#ifndef NROS_RCLCPP_FUTURE_HPP
#define NROS_RCLCPP_FUTURE_HPP
 
#include <cstdint>
#include <cstddef>
 
#include "rclcpp/result.hpp"
#include "rclcpp/size_bound.hpp" // rclcpp::rx_buffer_capacity<T> — the receive-buffer size
 
// FFI declarations
extern "C" {
typedef int nros_cpp_ret_t;
nros_cpp_ret_t nros_cpp_spin_once(void* handle, int32_t timeout_ms);
uint64_t nros_cpp_time_ns(void);
}
 
// phase-428 (RFC-0089): the entity types that reach into this one's private
// constructor are DEFINED in the upstream namespace now, so the friend
// declarations below name them there. A qualified friend cannot introduce a
// name, hence these declarations.
namespace rclcpp {
template <typename M> class Subscription;
} // namespace rclcpp
 
namespace rclcpp_action {
template <typename A> class Client;
} // namespace rclcpp_action
 
namespace rclcpp {
 
// phase-456 W9 — `send_request` and the `Future` it hands back moved off
// `rclcpp::Client<S>` onto `rclcpp::PollClient<S>`, the half that owns the
// `RmwServiceClient` this future takes from. The friend list FOLLOWS the method:
// the poll client is declared here, and `rclcpp::Client<S>` left both this
// declaration and the friend list below, because a friendship nothing uses is a
// claim about the code that is not true.
template <typename S> class PollClient;
 
template <typename T, size_t Cap = ::rclcpp::rx_buffer_capacity<T>::value> class Future {
  public:
    bool is_ready() {
        if (slot_ < 0 || !take_fn_) return false;
        if (ready_) return true;
        uint8_t buf[Cap];
        size_t len = 0;
        nros_cpp_ret_t ret = take_fn_(client_storage_, buf, sizeof(buf), &len);
        if (ret == 0 && len > 0) {
            ready_ = true;
            cached_len_ = len < sizeof(cached_buf_) ? len : sizeof(cached_buf_);
            for (size_t i = 0; i < cached_len_; ++i)
                cached_buf_[i] = buf[i];
            return true;
        }
        return false;
    }
 
    Result try_take(T& out) {
        if (slot_ < 0) return Result(ErrorCode::Error);
        if (!ready_ && !is_ready()) return Result(ErrorCode::Error);
        slot_ = -1; // consume
        if (T::ffi_deserialize(cached_buf_, cached_len_, &out) != 0) {
            return Result(ErrorCode::Error);
        }
        return Result::success();
    }
 
    Result wait(void* executor_handle, uint32_t timeout_ms, T& out, uint32_t poll_ms = 10) {
        if (slot_ < 0) return Result(ErrorCode::Error);
        if (poll_ms == 0) poll_ms = 1;
        // Phase 89.2: budget by wall-clock. Accumulating `step` per iteration
        // breaks when the underlying `zpico_spin_once` returns early on a
        // signaled condvar (keep-alives, discovery gossip) — the 500-step
        // default-timeout loop can then collapse into milliseconds and
        // return ErrorCode::Timeout before the reply has a chance to land.
        const uint64_t start_ns = nros_cpp_time_ns();
        const uint64_t budget_ns = static_cast<uint64_t>(timeout_ms) * 1000000ULL;
        while (true) {
            nros_cpp_ret_t ret = nros_cpp_spin_once(executor_handle, static_cast<int32_t>(poll_ms));
            // Transient conditions: keep polling. Anything else propagates.
            // - Ok (0): nothing to dispatch this round.
            // - Timeout (-2): spin_once returned after its timeout — normal.
            // - TryAgain (-6): transport hint to retry.
            if (ret != 0 && ret != static_cast<nros_cpp_ret_t>(ErrorCode::Timeout) &&
                ret != static_cast<nros_cpp_ret_t>(ErrorCode::TryAgain)) {
                return Result(ret);
            }
            if (is_ready()) return try_take(out);
            const uint64_t now_ns = nros_cpp_time_ns();
            if (now_ns - start_ns >= budget_ns) break;
        }
        return Result(ErrorCode::Timeout);
    }
 
    void cancel() { slot_ = -1; }
 
    bool is_consumed() const { return slot_ < 0; }
 
    // Move semantics (non-copyable, single-shot)
    Future(Future&& other) noexcept
        : client_storage_(other.client_storage_), take_fn_(other.take_fn_), slot_(other.slot_),
          ready_(other.ready_), cached_len_(other.cached_len_) {
        for (size_t i = 0; i < cached_len_; ++i)
            cached_buf_[i] = other.cached_buf_[i];
        other.slot_ = -1;
        other.ready_ = false;
    }
 
    Future& operator=(Future&& other) noexcept {
        if (this != &other) {
            client_storage_ = other.client_storage_;
            take_fn_ = other.take_fn_;
            slot_ = other.slot_;
            ready_ = other.ready_;
            cached_len_ = other.cached_len_;
            for (size_t i = 0; i < cached_len_; ++i)
                cached_buf_[i] = other.cached_buf_[i];
            other.slot_ = -1;
            other.ready_ = false;
        }
        return *this;
    }
 
    ~Future() { cancel(); }
 
    Future()
        : client_storage_(nullptr), take_fn_(nullptr), slot_(-1), ready_(false), cached_len_(0) {}
 
  private:
    Future(const Future&) = delete;
    Future& operator=(const Future&) = delete;
 
    template <typename S> friend class ::rclcpp::PollClient;
    template <typename A> friend class ::rclcpp_action::Client;
 
    using TryRecvFn = nros_cpp_ret_t (*)(void*, uint8_t*, size_t, size_t*);
 
    Future(void* storage, TryRecvFn fn, int slot)
        : client_storage_(storage), take_fn_(fn), slot_(slot), ready_(false), cached_len_(0) {}
 
    void* client_storage_;
    TryRecvFn take_fn_;
    int slot_;
    bool ready_;
    size_t cached_len_;
    uint8_t cached_buf_[Cap];
};
 
} // namespace rclcpp
#endif // NROS_RCLCPP_FUTURE_HPP