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_CPP_FUTURE_HPP
#define NROS_CPP_FUTURE_HPP
 
#include <cstdint>
#include <cstddef>
 
#include "nros/result.hpp"
#include "nros/size_bound.hpp" // nros::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);
}
 
namespace nros {
 
template <typename T, size_t Cap = ::nros::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 Client;
    template <typename A> friend class ActionClient;
 
    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 nros
#endif // NROS_CPP_FUTURE_HPP