Skip to content

Make MVEC Lib tolerant to dropped messages. #14

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Ryanbahl9 wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Make MVEC Lib tolerant to dropped messages. #14

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
98b6d62
For MVEC, replace future queue with a single future that will timeout…
Feb 25, 2026
6cb2192
fix pre-commit errors
Feb 25, 2026
5a97d6e
fix pre-commit error
Feb 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 34 additions & 16 deletions mvec/mvec_lib/include/mvec_lib/mvec_relay_socketcan.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,11 @@
#ifndef MVEC_LIB__MVEC_RELAY_SOCKETCAN_HPP_
#define MVEC_LIB__MVEC_RELAY_SOCKETCAN_HPP_

#include <chrono>
#include <future>
#include <memory>
#include <mutex>
#include <queue>
#include <optional>

#include "mvec_lib/mvec_relay.hpp"
#include "socketcan_adapter/socketcan_adapter.hpp"
Expand All @@ -31,10 +32,16 @@ namespace polymath::sygnal
class MvecRelaySocketcan
{
public:
/// @brief Constructor
/// @brief Constructor with default 500ms response timeout
/// @param socketcan_adapter Shared pointer to socketcan adapter for CAN communication
explicit MvecRelaySocketcan(std::shared_ptr<socketcan::SocketcanAdapter> socketcan_adapter);

/// @brief Constructor with custom response timeout
/// @param socketcan_adapter Shared pointer to socketcan adapter for CAN communication
/// @param response_timeout Timeout for all MVEC response types
MvecRelaySocketcan(
std::shared_ptr<socketcan::SocketcanAdapter> socketcan_adapter, std::chrono::milliseconds response_timeout);

/// @brief Parse incoming CAN frame and fulfill waiting promises
/// @param frame CAN frame to parse
/// @return Message type that was parsed
Expand All @@ -49,16 +56,16 @@ class MvecRelaySocketcan
void clear_relay();

/// @brief Query current relay states asynchronously
/// @return Future that will contain relay query reply
std::future<MvecRelayQueryReply> get_relay_state();
/// @return Future containing reply, or nullopt on timeout or rejection due to pending requests.
std::future<std::optional<MvecRelayQueryReply>> get_relay_state();

/// @brief Send relay command and wait for confirmation
/// @return Future that will contain command reply
std::future<MvecRelayCommandReply> send_relay_command();
/// @return Future containing reply, or nullopt on timeout or rejection due to pending requests.
std::future<std::optional<MvecRelayCommandReply>> send_relay_command();

/// @brief Query device population (which relays/fuses are installed)
/// @return Future that will contain population reply
std::future<MvecPopulationReply> get_relay_population();
/// @return Future containing reply, or nullopt on timeout or rejection due to pending requests.
std::future<std::optional<MvecPopulationReply>> get_relay_population();

/// @brief Get last received fuse status message
/// @return Optional containing fuse status if valid data available
Expand All @@ -72,21 +79,32 @@ class MvecRelaySocketcan
/// @return Optional containing error status if valid data available
const std::optional<MvecErrorStatusMessage> get_last_error_status();

/// @brief Default timeout for MVEC responses (500ms)
static constexpr std::chrono::milliseconds MVEC_DEFAULT_RESPONSE_TIMEOUT{500};

private:
/// @brief SocketCAN adapter for CAN communication
std::shared_ptr<socketcan::SocketcanAdapter> socketcan_adapter_;
/// @brief Core MVEC relay implementation
MvecRelay relay_impl_;

/// @brief Queue of promises waiting for relay query responses
std::queue<std::promise<MvecRelayQueryReply>> query_reply_promises_;
/// @brief Queue of promises waiting for relay command responses
std::queue<std::promise<MvecRelayCommandReply>> command_reply_promises_;
/// @brief Queue of promises waiting for population query responses
std::queue<std::promise<MvecPopulationReply>> population_reply_promises_;
/// @brief Timeout for all MVEC response types
std::chrono::milliseconds response_timeout_;

/// @brief Single-slot promise for relay query response
std::optional<std::promise<std::optional<MvecRelayQueryReply>>> query_reply_promise_;
std::chrono::steady_clock::time_point query_send_time_;
std::mutex query_mutex_;

/// @brief Single-slot promise for relay command response
std::optional<std::promise<std::optional<MvecRelayCommandReply>>> command_reply_promise_;
std::chrono::steady_clock::time_point command_send_time_;
std::mutex command_mutex_;

/// @brief Mutex protecting promise queues for thread safety
std::mutex promises_mutex_;
/// @brief Single-slot promise for population query response
std::optional<std::promise<std::optional<MvecPopulationReply>>> population_reply_promise_;
Copy link
Contributor

@zeerekahmad zeerekahmad Feb 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't know if you need optional here! Just don't fulfill the promise, the future will take care of not having a value!

std::chrono::steady_clock::time_point population_send_time_;
std::mutex population_mutex_;
};

} // namespace polymath::sygnal
Expand Down
133 changes: 76 additions & 57 deletions mvec/mvec_lib/src/mvec_relay_socketcan.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,52 +14,53 @@

#include "mvec_lib/mvec_relay_socketcan.hpp"

#include <memory>
#include <mutex>
#include <utility>

namespace polymath::sygnal
{

MvecRelaySocketcan::MvecRelaySocketcan(std::shared_ptr<socketcan::SocketcanAdapter> socketcan_adapter)
: MvecRelaySocketcan(socketcan_adapter, MVEC_DEFAULT_RESPONSE_TIMEOUT)
{}

MvecRelaySocketcan::MvecRelaySocketcan(
std::shared_ptr<socketcan::SocketcanAdapter> socketcan_adapter, std::chrono::milliseconds response_timeout)
: socketcan_adapter_(socketcan_adapter)
, relay_impl_()
, response_timeout_(response_timeout)
{}

MvecMessageType MvecRelaySocketcan::parse(const socketcan::CanFrame & frame)
{
MvecMessageType message_type = relay_impl_.parseMessage(frame);

// Check if we received expected response types and fulfill promises
std::lock_guard<std::mutex> lock(promises_mutex_);

switch (message_type) {
case MvecMessageType::RELAY_QUERY_RESPONSE: {
const auto & reply = relay_impl_.get_last_relay_query_reply();
if (reply.is_valid() && !query_reply_promises_.empty()) {
// Get the oldest waiting promise
auto promise = std::move(query_reply_promises_.front());
query_reply_promises_.pop();

// Fulfill the promise
promise.set_value(reply);
std::lock_guard<std::mutex> lock(query_mutex_);
if (reply.is_valid() && query_reply_promise_.has_value()) {
query_reply_promise_->set_value(std::make_optional(reply));
query_reply_promise_.reset();
}
break;
}
case MvecMessageType::RELAY_COMMAND_RESPONSE: {
const auto & reply = relay_impl_.get_last_relay_command_reply();
if (reply.is_valid() && !command_reply_promises_.empty()) {
auto promise = std::move(command_reply_promises_.front());
command_reply_promises_.pop();
promise.set_value(reply);
std::lock_guard<std::mutex> lock(command_mutex_);
if (reply.is_valid() && command_reply_promise_.has_value()) {
command_reply_promise_->set_value(std::make_optional(reply));
command_reply_promise_.reset();
}
break;
}
case MvecMessageType::POPULATION_RESPONSE: {
const auto & reply = relay_impl_.get_last_population_reply();
if (reply.is_valid() && !population_reply_promises_.empty()) {
auto promise = std::move(population_reply_promises_.front());
population_reply_promises_.pop();
promise.set_value(reply);
std::lock_guard<std::mutex> lock(population_mutex_);
if (reply.is_valid() && population_reply_promise_.has_value()) {
population_reply_promise_->set_value(std::make_optional(reply));
population_reply_promise_.reset();
}
break;
}
Expand All @@ -82,67 +83,85 @@ void MvecRelaySocketcan::clear_relay()
relay_impl_.clearRelayCommands();
}

std::future<MvecRelayQueryReply> MvecRelaySocketcan::get_relay_state()
std::future<std::optional<MvecRelayQueryReply>> MvecRelaySocketcan::get_relay_state()
{
// Get the query message from the relay implementation
/// TODO: (zeerek) Set invalid for received message
auto query_frame = relay_impl_.getRelayQueryMessage();
std::lock_guard<std::mutex> lock(query_mutex_);

// If a request is already in-flight, check if it has timed out
Copy link
Contributor

@zeerekahmad zeerekahmad Feb 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this logic necessary? Do we actually care to reject or would we rather just resend? The other promise would just time out because we resent and the new one would be used.

But since we only have 1 request in flight at a time, that should not matter?

if (query_reply_promise_.has_value()) {
auto elapsed = std::chrono::steady_clock::now() - query_send_time_;
if (elapsed < response_timeout_) {
// Still in-flight, reject this request
std::promise<std::optional<MvecRelayQueryReply>> rejected;
auto future = rejected.get_future();
rejected.set_value(std::nullopt);
return future;
}
// Timed out — fulfill the old promise with nullopt so any waiter gets a clean result
query_reply_promise_->set_value(std::nullopt);
query_reply_promise_.reset();
}

// Create a new promise and get its future
std::promise<MvecRelayQueryReply> promise;
std::promise<std::optional<MvecRelayQueryReply>> promise;
Copy link
Contributor

@zeerekahmad zeerekahmad Feb 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why promise optional? That seems redundant?

auto future = promise.get_future();
query_reply_promise_.emplace(std::move(promise));
query_send_time_ = std::chrono::steady_clock::now();

// Add promise to the queue with thread safety
{
std::lock_guard<std::mutex> lock(promises_mutex_);
query_reply_promises_.push(std::move(promise));
}

// Transmit the query message via socketcan adapter
auto query_frame = relay_impl_.getRelayQueryMessage();
socketcan_adapter_->send(query_frame);

return future;
}

std::future<MvecRelayCommandReply> MvecRelaySocketcan::send_relay_command()
std::future<std::optional<MvecRelayCommandReply>> MvecRelaySocketcan::send_relay_command()
{
// Get the command message from the relay implementation
/// TODO: (zeerek) Set invalid for received message
auto command_frame = relay_impl_.getRelayCommandMessage();
std::lock_guard<std::mutex> lock(command_mutex_);

if (command_reply_promise_.has_value()) {
auto elapsed = std::chrono::steady_clock::now() - command_send_time_;
if (elapsed < response_timeout_) {
std::promise<std::optional<MvecRelayCommandReply>> rejected;
auto future = rejected.get_future();
rejected.set_value(std::nullopt);
return future;
}
command_reply_promise_->set_value(std::nullopt);
command_reply_promise_.reset();
}

// Create a new promise and get its future
std::promise<MvecRelayCommandReply> promise;
std::promise<std::optional<MvecRelayCommandReply>> promise;
auto future = promise.get_future();
command_reply_promise_.emplace(std::move(promise));
command_send_time_ = std::chrono::steady_clock::now();

// Add promise to the queue with thread safety
{
std::lock_guard<std::mutex> lock(promises_mutex_);
command_reply_promises_.push(std::move(promise));
}

// Transmit the command message via socketcan adapter
auto command_frame = relay_impl_.getRelayCommandMessage();
socketcan_adapter_->send(command_frame);

return future;
}

std::future<MvecPopulationReply> MvecRelaySocketcan::get_relay_population()
std::future<std::optional<MvecPopulationReply>> MvecRelaySocketcan::get_relay_population()
{
// Get the population query message from the relay implementation
/// TODO: (zeerek) Set invalid for received message
auto population_frame = relay_impl_.getPopulationQueryMessage();
std::lock_guard<std::mutex> lock(population_mutex_);

if (population_reply_promise_.has_value()) {
auto elapsed = std::chrono::steady_clock::now() - population_send_time_;
if (elapsed < response_timeout_) {
std::promise<std::optional<MvecPopulationReply>> rejected;
auto future = rejected.get_future();
rejected.set_value(std::nullopt);
return future;
}
population_reply_promise_->set_value(std::nullopt);
population_reply_promise_.reset();
}

// Create a new promise and get its future
std::promise<MvecPopulationReply> promise;
std::promise<std::optional<MvecPopulationReply>> promise;
auto future = promise.get_future();
population_reply_promise_.emplace(std::move(promise));
population_send_time_ = std::chrono::steady_clock::now();

// Add promise to the queue with thread safety
{
std::lock_guard<std::mutex> lock(promises_mutex_);
population_reply_promises_.push(std::move(promise));
}

// Transmit the population query message via socketcan adapter
auto population_frame = relay_impl_.getPopulationQueryMessage();
socketcan_adapter_->send(population_frame);

return future;
Expand Down
22 changes: 15 additions & 7 deletions mvec/mvec_lib/test/mvec_socketcan_hardware.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,10 @@ TEST_CASE("MvecRelaySocketcan hardware integration test", "[hardware]")
auto status = population_future.wait_for(std::chrono::seconds(5));

if (status == std::future_status::ready) {
auto population_reply = population_future.get();
auto population_result = population_future.get();
REQUIRE(population_result.has_value());

const auto & population_reply = population_result.value();
std::cout << "Population query successful! Valid: " << population_reply.is_valid() << std::endl;

// Check that we got a valid response
Expand Down Expand Up @@ -95,7 +98,10 @@ TEST_CASE("MvecRelaySocketcan hardware integration test", "[hardware]")
auto status = relay_state_future.wait_for(std::chrono::seconds(5));

if (status == std::future_status::ready) {
auto relay_query_reply = relay_state_future.get();
auto relay_query_result = relay_state_future.get();
REQUIRE(relay_query_result.has_value());

const auto & relay_query_reply = relay_query_result.value();
std::cout << "Relay state query successful! Valid: " << relay_query_reply.is_valid() << std::endl;

// Check that we got a valid response
Expand Down Expand Up @@ -213,18 +219,20 @@ TEST_CASE("MvecRelaySocketcan hardware integration test", "[hardware]")

REQUIRE(status == std::future_status::ready);

auto response = relay_command_response_future.get();
auto response_result = relay_command_response_future.get();
REQUIRE(response_result.has_value());

// 1 is no error
REQUIRE(response.get_success() == 1);
REQUIRE(response_result->get_success() == 1);
std::cout << "Relay response message confirms success" << std::endl;

auto mvec_query_future = mvec_socketcan->get_relay_state();
status = mvec_query_future.wait_for(std::chrono::seconds(5));
auto relay_state = mvec_query_future.get();
auto relay_state_result = mvec_query_future.get();
REQUIRE(relay_state_result.has_value());

REQUIRE(relay_state.get_relay_state(8) == 1);
REQUIRE(relay_state.get_relay_state(9) == 1);
REQUIRE(relay_state_result->get_relay_state(8) == 1);
REQUIRE(relay_state_result->get_relay_state(9) == 1);
std::cout << "Relay states queried and confirm command" << std::endl;
}

Expand Down
Loading