Create an enum for sender errors
Will replace usages of bools, and will now currently only be used
for indicating that messages were sent too fast
After we merge this commit we will replace this enum with a general
Status class for all of aos, similar to absl::Status.
Change-Id: I4b5b2e7685744b3c6826a241cd3c84190eaa96ee
Signed-off-by: milind-u <milind.upadhyay@gmail.com>
diff --git a/aos/actions/action_test.cc b/aos/actions/action_test.cc
index 1c8e506..e7a3c53 100644
--- a/aos/actions/action_test.cc
+++ b/aos/actions/action_test.cc
@@ -1,8 +1,8 @@
#include <unistd.h>
+#include <chrono>
#include <memory>
#include <thread>
-#include <chrono>
#include "gtest/gtest.h"
@@ -48,8 +48,8 @@
typedef TypedActionFactory<actions::TestActionGoal> Factory;
explicit TestActorNOP(::aos::EventLoop *event_loop)
- : actions::ActorBase<actions::TestActionGoal>(
- event_loop, "/test_action") {}
+ : actions::ActorBase<actions::TestActionGoal>(event_loop,
+ "/test_action") {}
static Factory MakeFactory(::aos::EventLoop *event_loop) {
return Factory(event_loop, "/test_action");
@@ -85,8 +85,8 @@
typedef TypedActionFactory<actions::TestAction2Goal> Factory;
explicit TestActor2Nop(::aos::EventLoop *event_loop)
- : actions::ActorBase<actions::TestAction2Goal>(
- event_loop, "/test_action2") {}
+ : actions::ActorBase<actions::TestAction2Goal>(event_loop,
+ "/test_action2") {}
static Factory MakeFactory(::aos::EventLoop *event_loop) {
return Factory(event_loop, "/test_action2");
@@ -146,14 +146,13 @@
ActionQueue action_queue;
{
- ::aos::Sender<TestActionGoal>::Builder builder =
- goal_sender.MakeBuilder();
+ ::aos::Sender<TestActionGoal>::Builder builder = goal_sender.MakeBuilder();
TestActionGoal::Builder goal_builder =
builder.MakeBuilder<TestActionGoal>();
goal_builder.add_run(971);
- ASSERT_TRUE(builder.Send(goal_builder.Finish()));
+ builder.CheckOk(builder.Send(goal_builder.Finish()));
}
TestActorNOP nop_act(actor1_event_loop_.get());
diff --git a/aos/actions/actions.h b/aos/actions/actions.h
index 65dc170..9a62483 100644
--- a/aos/actions/actions.h
+++ b/aos/actions/actions.h
@@ -106,8 +106,8 @@
public:
// A convenient way to refer to the type of our goals.
typedef T GoalType;
- typedef typename std::remove_reference<
- decltype(*static_cast<GoalType *>(nullptr)->params())>::type ParamType;
+ typedef typename std::remove_reference<decltype(
+ *static_cast<GoalType *>(nullptr)->params())>::type ParamType;
TypedAction(typename ::aos::Fetcher<Status> *status_fetcher,
typename ::aos::Sender<GoalType> *goal_sender,
@@ -194,8 +194,8 @@
class TypedActionFactory {
public:
typedef T GoalType;
- typedef typename std::remove_reference<
- decltype(*static_cast<GoalType *>(nullptr)->params())>::type ParamType;
+ typedef typename std::remove_reference<decltype(
+ *static_cast<GoalType *>(nullptr)->params())>::type ParamType;
explicit TypedActionFactory(::aos::EventLoop *event_loop,
const ::std::string &name)
@@ -258,7 +258,7 @@
typename GoalType::Builder goal_builder =
builder.template MakeBuilder<GoalType>();
goal_builder.add_run(0);
- builder.Send(goal_builder.Finish());
+ builder.CheckOk(builder.Send(goal_builder.Finish()));
}
sent_cancel_ = true;
}
@@ -372,7 +372,7 @@
goal_builder.add_run(run_value_);
sent_started_ = true;
- if (!builder.Send(goal_builder.Finish())) {
+ if (builder.Send(goal_builder.Finish()) != RawSender::Error::kOk) {
AOS_LOG(ERROR,
"sending goal for action %" PRIx32 " on queue %.*s failed\n",
run_value_,
diff --git a/aos/actions/actor.h b/aos/actions/actor.h
index a000257..c4f2840 100644
--- a/aos/actions/actor.h
+++ b/aos/actions/actor.h
@@ -20,8 +20,8 @@
class ActorBase {
public:
typedef T GoalType;
- typedef typename std::remove_reference<
- decltype(*static_cast<GoalType *>(nullptr)->params())>::type ParamType;
+ typedef typename std::remove_reference<decltype(
+ *static_cast<GoalType *>(nullptr)->params())>::type ParamType;
ActorBase(::aos::EventLoop *event_loop, const ::std::string &name)
: event_loop_(event_loop),
@@ -40,9 +40,9 @@
status_builder.add_running(0);
status_builder.add_last_running(0);
status_builder.add_success(!abort_);
- if (!builder.Send(status_builder.Finish())) {
- AOS_LOG(ERROR, "Failed to send the status.\n");
- }
+ CHECK_EQ(builder.Send(status_builder.Finish()),
+ aos::RawSender::Error::kOk)
+ << "Failed to send initial status";
});
}
@@ -127,7 +127,8 @@
status_builder.add_running(0);
status_builder.add_last_running(0);
status_builder.add_success(!abort_);
- if (!builder.Send(status_builder.Finish())) {
+ if (builder.Send(status_builder.Finish()) !=
+ aos::RawSender::Error::kOk) {
AOS_LOG(ERROR, "Failed to send the status.\n");
}
break;
@@ -144,7 +145,8 @@
status_builder.add_running(running_id);
status_builder.add_last_running(0);
status_builder.add_success(!abort_);
- if (!builder.Send(status_builder.Finish())) {
+ if (builder.Send(status_builder.Finish()) !=
+ aos::RawSender::Error::kOk) {
AOS_LOG(ERROR, "Failed to send the status.\n");
}
}
@@ -161,7 +163,8 @@
status_builder.add_last_running(running_id);
status_builder.add_success(!abort_);
- if (!builder.Send(status_builder.Finish())) {
+ if (builder.Send(status_builder.Finish()) !=
+ aos::RawSender::Error::kOk) {
AOS_LOG(ERROR, "Failed to send the status.\n");
} else {
AOS_LOG(INFO, "Sending Done status %" PRIx32 "\n", running_id);
diff --git a/aos/aos_send.cc b/aos/aos_send.cc
index 91ffb1c..ffc7957 100644
--- a/aos/aos_send.cc
+++ b/aos/aos_send.cc
@@ -44,7 +44,7 @@
fbb.ForceDefaults(true);
fbb.Finish(aos::JsonToFlatbuffer(std::string_view(argv[1]), channel->schema(),
&fbb));
- sender->Send(fbb.GetSize());
+ sender->CheckOk(sender->Send(fbb.GetSize()));
return 0;
}
diff --git a/aos/events/aos_logging.cc b/aos/events/aos_logging.cc
index bc4513b..5ec7d3d 100644
--- a/aos/events/aos_logging.cc
+++ b/aos/events/aos_logging.cc
@@ -22,8 +22,11 @@
static_cast<::aos::logging::Level>(message_data.level));
builder.add_source_pid(message_data.source);
builder.add_name(name_str);
+ if (send_failure_counter_.failures() > 0) {
+ builder.add_send_failures(send_failure_counter_.failures());
+ }
- message.Send(builder.Finish());
+ send_failure_counter_.Count(message.Send(builder.Finish()));
},
name);
}
diff --git a/aos/events/aos_logging.h b/aos/events/aos_logging.h
index efb2b25..eca07bf 100644
--- a/aos/events/aos_logging.h
+++ b/aos/events/aos_logging.h
@@ -22,6 +22,7 @@
private:
Sender<logging::LogMessageFbs> log_sender_;
std::shared_ptr<logging::LogImplementation> implementation_;
+ SendFailureCounter send_failure_counter_;
};
} // namespace aos
diff --git a/aos/events/event_loop.cc b/aos/events/event_loop.cc
index 51341bb..67c4472 100644
--- a/aos/events/event_loop.cc
+++ b/aos/events/event_loop.cc
@@ -19,8 +19,28 @@
<< configuration::CleanedChannelToString(channel) << ".";
}
}
+
+std::string_view ErrorToString(const RawSender::Error err) {
+ switch (err) {
+ case RawSender::Error::kOk:
+ return "RawSender::Error::kOk";
+ case RawSender::Error::kMessagesSentTooFast:
+ return "RawSender::Error::kMessagesSentTooFast";
+ }
+ LOG(FATAL) << "Unknown error given with code " << static_cast<int>(err);
+}
} // namespace
+std::ostream &operator<<(std::ostream &os, const RawSender::Error err) {
+ os << ErrorToString(err);
+ return os;
+}
+
+void RawSender::CheckOk(const RawSender::Error err) {
+ CHECK_EQ(err, Error::kOk) << "Messages were sent too fast on channel: "
+ << configuration::CleanedChannelToString(channel_);
+}
+
RawSender::RawSender(EventLoop *event_loop, const Channel *channel)
: event_loop_(event_loop),
channel_(channel),
@@ -31,11 +51,10 @@
RawSender::~RawSender() { event_loop_->DeleteSender(this); }
-bool RawSender::DoSend(const SharedSpan data,
- monotonic_clock::time_point monotonic_remote_time,
- realtime_clock::time_point realtime_remote_time,
- uint32_t remote_queue_index,
- const UUID &source_boot_uuid) {
+RawSender::Error RawSender::DoSend(
+ const SharedSpan data, monotonic_clock::time_point monotonic_remote_time,
+ realtime_clock::time_point realtime_remote_time,
+ uint32_t remote_queue_index, const UUID &source_boot_uuid) {
return DoSend(data->data(), data->size(), monotonic_remote_time,
realtime_remote_time, remote_queue_index, source_boot_uuid);
}
@@ -261,7 +280,13 @@
for (RawFetcher *fetcher : fetchers_) {
fetcher->timing_.ResetTimingReport();
}
- timing_report_sender_->Send(timing_report_.span().size());
+ // TODO(milind): If we fail to send, we don't want to reset the timing report.
+ // We would need to move the reset after the send, and then find the correct
+ // timing report and set the reports with it instead of letting the sender do
+ // this. If we failed to send, we wouldn't reset or set the reports, so they
+ // can accumalate until the next send.
+ timing_report_failure_counter_.Count(
+ timing_report_sender_->Send(timing_report_.span().size()));
}
void EventLoop::UpdateTimingReport() {
@@ -421,6 +446,7 @@
if (fetcher_offsets.size() > 0) {
report_builder.add_fetchers(fetchers_offset);
}
+ report_builder.add_send_failures(timing_report_failure_counter_.failures());
fbb.Finish(report_builder.Finish());
timing_report_ = FlatbufferDetachedBuffer<timing::Report>(fbb.Release());
diff --git a/aos/events/event_loop.fbs b/aos/events/event_loop.fbs
index 1434c4e..051f575 100644
--- a/aos/events/event_loop.fbs
+++ b/aos/events/event_loop.fbs
@@ -83,6 +83,9 @@
fetchers:[Fetcher] (id: 4);
timers:[Timer] (id: 5);
phased_loops:[Timer] (id: 6);
+
+ // Total count of Report send failures
+ send_failures:uint64 (id: 7);
}
root_type Report;
diff --git a/aos/events/event_loop.h b/aos/events/event_loop.h
index 61849abe..ac2da4c 100644
--- a/aos/events/event_loop.h
+++ b/aos/events/event_loop.h
@@ -4,6 +4,7 @@
#include <sched.h>
#include <atomic>
+#include <ostream>
#include <string>
#include <string_view>
@@ -138,42 +139,56 @@
public:
using SharedSpan = std::shared_ptr<const absl::Span<const uint8_t>>;
+ enum class [[nodiscard]] Error{
+ // Represents success and no error
+ kOk,
+
+ // Error for messages on channels being sent faster than their
+ // frequency and channel storage duration allow
+ kMessagesSentTooFast};
+
RawSender(EventLoop *event_loop, const Channel *channel);
RawSender(const RawSender &) = delete;
RawSender &operator=(const RawSender &) = delete;
virtual ~RawSender();
- // Sends a message without copying it. The users starts by copying up to
- // size() bytes into the data backed by data(). They then call Send to send.
- // Returns true on a successful send.
- // If provided, monotonic_remote_time, realtime_remote_time, and
- // remote_queue_index are attached to the message and are available in the
- // context on the read side. If they are not populated, the read side will
- // get the sent times instead.
virtual void *data() = 0;
virtual size_t size() = 0;
- bool Send(size_t size);
- bool Send(size_t size, monotonic_clock::time_point monotonic_remote_time,
- realtime_clock::time_point realtime_remote_time,
- uint32_t remote_queue_index, const UUID &source_boot_uuid);
+
+ // Sends a message without copying it. The users starts by copying up to
+ // size() bytes into the data backed by data(). They then call Send to send.
+ // Returns Error::kOk on a successful send, or
+ // Error::kMessagesSentTooFast if messages were sent too fast. If provided,
+ // monotonic_remote_time, realtime_remote_time, and remote_queue_index are
+ // attached to the message and are available in the context on the read side.
+ // If they are not populated, the read side will get the sent times instead.
+ Error Send(size_t size);
+ Error Send(size_t size, monotonic_clock::time_point monotonic_remote_time,
+ realtime_clock::time_point realtime_remote_time,
+ uint32_t remote_queue_index, const UUID &source_boot_uuid);
// Sends a single block of data by copying it.
// The remote arguments have the same meaning as in Send above.
- bool Send(const void *data, size_t size);
- bool Send(const void *data, size_t size,
- monotonic_clock::time_point monotonic_remote_time,
- realtime_clock::time_point realtime_remote_time,
- uint32_t remote_queue_index, const UUID &source_boot_uuid);
+ // Returns Error::kMessagesSentTooFast if messages were sent too fast
+ Error Send(const void *data, size_t size);
+ Error Send(const void *data, size_t size,
+ monotonic_clock::time_point monotonic_remote_time,
+ realtime_clock::time_point realtime_remote_time,
+ uint32_t remote_queue_index, const UUID &source_boot_uuid);
+
+ // CHECKs that no sending Error occurred and logs the channel_ data if
+ // one did
+ void CheckOk(const Error err);
// Sends a single block of data by refcounting it to avoid copies. The data
// must not change after being passed into Send. The remote arguments have the
// same meaning as in Send above.
- bool Send(const SharedSpan data);
- bool Send(const SharedSpan data,
- monotonic_clock::time_point monotonic_remote_time,
- realtime_clock::time_point realtime_remote_time,
- uint32_t remote_queue_index, const UUID &remote_boot_uuid);
+ Error Send(const SharedSpan data);
+ Error Send(const SharedSpan data,
+ monotonic_clock::time_point monotonic_remote_time,
+ realtime_clock::time_point realtime_remote_time,
+ uint32_t remote_queue_index, const UUID &remote_boot_uuid);
const Channel *channel() const { return channel_; }
@@ -210,21 +225,21 @@
private:
friend class EventLoop;
- virtual bool DoSend(const void *data, size_t size,
- monotonic_clock::time_point monotonic_remote_time,
- realtime_clock::time_point realtime_remote_time,
- uint32_t remote_queue_index,
- const UUID &source_boot_uuid) = 0;
- virtual bool DoSend(size_t size,
- monotonic_clock::time_point monotonic_remote_time,
- realtime_clock::time_point realtime_remote_time,
- uint32_t remote_queue_index,
- const UUID &source_boot_uuid) = 0;
- virtual bool DoSend(const SharedSpan data,
- monotonic_clock::time_point monotonic_remote_time,
- realtime_clock::time_point realtime_remote_time,
- uint32_t remote_queue_index,
- const UUID &source_boot_uuid);
+ virtual Error DoSend(const void *data, size_t size,
+ monotonic_clock::time_point monotonic_remote_time,
+ realtime_clock::time_point realtime_remote_time,
+ uint32_t remote_queue_index,
+ const UUID &source_boot_uuid) = 0;
+ virtual Error DoSend(size_t size,
+ monotonic_clock::time_point monotonic_remote_time,
+ realtime_clock::time_point realtime_remote_time,
+ uint32_t remote_queue_index,
+ const UUID &source_boot_uuid) = 0;
+ virtual Error DoSend(const SharedSpan data,
+ monotonic_clock::time_point monotonic_remote_time,
+ realtime_clock::time_point realtime_remote_time,
+ uint32_t remote_queue_index,
+ const UUID &source_boot_uuid);
EventLoop *const event_loop_;
const Channel *const channel_;
@@ -236,6 +251,9 @@
ChannelPreallocatedAllocator fbb_allocator_{nullptr, 0, nullptr};
};
+// Needed for compatibility with glog
+std::ostream &operator<<(std::ostream &os, const RawSender::Error err);
+
// Fetches the newest message from a channel.
// This provides a polling based interface for channels.
template <typename T>
@@ -336,14 +354,17 @@
return typename T2::Builder(fbb_);
}
- bool Send(flatbuffers::Offset<T> offset) {
+ RawSender::Error Send(flatbuffers::Offset<T> offset) {
fbb_.Finish(offset);
- const bool result = sender_->Send(fbb_.GetSize());
+ const auto err = sender_->Send(fbb_.GetSize());
// Ensure fbb_ knows it shouldn't access the memory any more.
fbb_ = flatbuffers::FlatBufferBuilder();
- return result;
+ return err;
}
+ // Equivalent to RawSender::CheckOk
+ void CheckOk(const RawSender::Error err) { sender_->CheckOk(err); };
+
// CHECKs that this message was sent.
void CheckSent() {
CHECK(!allocator_->is_allocated()) << ": Message was not sent yet";
@@ -374,11 +395,14 @@
Builder MakeBuilder();
// Sends a prebuilt flatbuffer.
- bool Send(const NonSizePrefixedFlatbuffer<T> &flatbuffer);
+ RawSender::Error Send(const NonSizePrefixedFlatbuffer<T> &flatbuffer);
// Sends a prebuilt flatbuffer which was detached from a Builder created via
// MakeBuilder() on this object.
- bool SendDetached(FlatbufferDetachedBuffer<T> detached);
+ RawSender::Error SendDetached(FlatbufferDetachedBuffer<T> detached);
+
+ // Equivalent to RawSender::CheckOk
+ void CheckOk(const RawSender::Error err) { sender_->CheckOk(err); };
// Returns the name of the underlying queue.
const Channel *channel() const { return sender_->channel(); }
@@ -405,6 +429,22 @@
std::unique_ptr<RawSender> sender_;
};
+// Class for keeping a count of send failures on a certain channel
+class SendFailureCounter {
+ public:
+ inline void Count(const RawSender::Error err) {
+ failures_ += static_cast<size_t>(err != RawSender::Error::kOk);
+ just_failed_ = (err != RawSender::Error::kOk);
+ }
+
+ inline size_t failures() const { return failures_; }
+ inline bool just_failed() const { return just_failed_; }
+
+ private:
+ size_t failures_ = 0;
+ bool just_failed_ = false;
+};
+
// Interface for timers.
class TimerHandler {
public:
@@ -505,13 +545,13 @@
virtual monotonic_clock::time_point monotonic_now() = 0;
virtual realtime_clock::time_point realtime_now() = 0;
- // Returns true if the channel exists in the configuration.
template <typename T>
const Channel *GetChannel(const std::string_view channel_name) {
return configuration::GetChannel(configuration(), channel_name,
T::GetFullyQualifiedName(), name(), node(),
true);
}
+ // Returns true if the channel exists in the configuration.
template <typename T>
bool HasChannel(const std::string_view channel_name) {
return GetChannel<T>(channel_name) != nullptr;
@@ -807,6 +847,8 @@
// If true, don't send out timing reports.
bool skip_timing_report_ = false;
+ SendFailureCounter timing_report_failure_counter_;
+
absl::btree_set<const Channel *> taken_watchers_, taken_senders_;
};
diff --git a/aos/events/event_loop_param_test.cc b/aos/events/event_loop_param_test.cc
index 2749f60..30e293d 100644
--- a/aos/events/event_loop_param_test.cc
+++ b/aos/events/event_loop_param_test.cc
@@ -86,7 +86,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
});
loop2->MakeWatcher("/test", [&](const TestMessage &message) {
@@ -121,7 +121,7 @@
builder.add_value(200);
detached = msg.Detach(builder.Finish());
}
- ASSERT_TRUE(sender.SendDetached(std::move(detached)));
+ sender.CheckOk(sender.SendDetached(std::move(detached)));
auto fetcher = loop2->MakeFetcher<TestMessage>("/test");
ASSERT_TRUE(fetcher.Fetch());
@@ -142,7 +142,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
});
loop2->MakeNoArgWatcher<TestMessage>("/test", [&]() {
@@ -173,7 +173,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
});
aos::Fetcher<TestMessage> fetcher = loop2->MakeFetcher<TestMessage>("/test");
@@ -203,7 +203,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
});
loop2->MakeWatcher("/test", std::function<void(const TestMessage &)>(
@@ -235,13 +235,13 @@
aos::Sender<TestMessage>::Builder msg = sender1.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
{
aos::Sender<TestMessage>::Builder msg = sender2.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
});
@@ -285,7 +285,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
EXPECT_TRUE(fetcher.Fetch());
ASSERT_FALSE(fetcher.get() == nullptr);
@@ -339,7 +339,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(199);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
loop2->OnRun([&]() {
@@ -347,13 +347,13 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
{
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(201);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
});
@@ -376,13 +376,13 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
{
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(201);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
loop2->MakeWatcher("/test", [&](const TestMessage &message) {
@@ -413,13 +413,13 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
{
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(201);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
// Add a timer to actually quit.
@@ -451,13 +451,13 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
{
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(201);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
auto fetcher = loop2->MakeFetcher<TestMessage>("/test");
@@ -492,13 +492,13 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
{
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(201);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
auto fetcher = loop2->MakeFetcher<TestMessage>("/test");
@@ -536,13 +536,13 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
{
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(201);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
auto fetcher = loop2->MakeFetcher<TestMessage>("/test");
@@ -557,19 +557,19 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(202);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
{
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(203);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
{
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(204);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
if (fetcher.FetchNext()) {
@@ -603,14 +603,14 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(100);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
{
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
ASSERT_TRUE(fetcher.FetchNext());
@@ -637,7 +637,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(1);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
ASSERT_TRUE(fetcher.Fetch());
EXPECT_EQ(1, fetcher.get()->value());
@@ -645,7 +645,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(i + 2);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
EXPECT_EQ(1, fetcher.get()->value());
}
@@ -663,7 +663,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(i);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
};
std::vector<Fetcher<TestMessage>> fetchers;
for (int i = 0; i < 10; ++i) {
@@ -943,7 +943,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
});
Run();
@@ -988,7 +988,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
});
Run();
@@ -1393,7 +1393,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
});
bool triggered = false;
@@ -1493,7 +1493,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
});
bool triggered = false;
@@ -1821,7 +1821,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200 + i);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
});
@@ -1906,7 +1906,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200 + i);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
});
@@ -1975,7 +1975,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200 + i);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
});
@@ -2063,7 +2063,8 @@
loop3->configuration(), "/test", "aos.TestMessage", "", nullptr));
loop2->OnRun([&]() {
- EXPECT_TRUE(sender->Send(kMessage.span().data(), kMessage.span().size()));
+ EXPECT_EQ(sender->Send(kMessage.span().data(), kMessage.span().size()),
+ RawSender::Error::kOk);
});
bool happened = false;
@@ -2113,8 +2114,9 @@
loop3->configuration(), "/test", "aos.TestMessage", "", nullptr));
loop2->OnRun([&]() {
- EXPECT_TRUE(sender->Send(std::make_shared<absl::Span<const uint8_t>>(
- kMessage.span().data(), kMessage.span().size())));
+ EXPECT_EQ(sender->Send(std::make_shared<absl::Span<const uint8_t>>(
+ kMessage.span().data(), kMessage.span().size())),
+ RawSender::Error::kOk);
});
bool happened = false;
@@ -2171,9 +2173,10 @@
loop3->configuration(), "/test", "aos.TestMessage", "", nullptr));
loop2->OnRun([&]() {
- EXPECT_TRUE(sender->Send(kMessage.span().data(), kMessage.span().size(),
- monotonic_remote_time, realtime_remote_time,
- remote_queue_index, source_boot_uuid));
+ EXPECT_EQ(sender->Send(kMessage.span().data(), kMessage.span().size(),
+ monotonic_remote_time, realtime_remote_time,
+ remote_queue_index, source_boot_uuid),
+ RawSender::Error::kOk);
});
bool happened = false;
@@ -2217,7 +2220,8 @@
const aos::monotonic_clock::time_point monotonic_now = loop1->monotonic_now();
const aos::realtime_clock::time_point realtime_now = loop1->realtime_now();
- EXPECT_TRUE(sender->Send(kMessage.span().data(), kMessage.span().size()));
+ EXPECT_EQ(sender->Send(kMessage.span().data(), kMessage.span().size()),
+ RawSender::Error::kOk);
EXPECT_GE(sender->monotonic_sent_time(), monotonic_now);
EXPECT_LE(sender->monotonic_sent_time(),
@@ -2227,7 +2231,8 @@
realtime_now + chrono::milliseconds(100));
EXPECT_EQ(sender->sent_queue_index(), 0u);
- EXPECT_TRUE(sender->Send(kMessage.span().data(), kMessage.span().size()));
+ EXPECT_EQ(sender->Send(kMessage.span().data(), kMessage.span().size()),
+ RawSender::Error::kOk);
EXPECT_GE(sender->monotonic_sent_time(), monotonic_now);
EXPECT_LE(sender->monotonic_sent_time(),
@@ -2382,7 +2387,7 @@
loop1->OnRun([&]() {
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
});
loop1->MakeWatcher("/test", [&](const TestMessage &) {
@@ -2405,7 +2410,7 @@
loop1->OnRun([&]() {
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
});
loop1->MakeWatcher("/test", [&](const TestMessage &) {
@@ -2558,7 +2563,7 @@
auto builder = sender1.MakeBuilder();
FlatbufferDetachedBuffer<TestMessage> detached =
builder.Detach(builder.MakeBuilder<TestMessage>().Finish());
- EXPECT_DEATH(sender2.SendDetached(std::move(detached)),
+ EXPECT_DEATH(sender2.CheckOk(sender2.SendDetached(std::move(detached))),
"May only send the buffer detached from this Sender");
}
diff --git a/aos/events/event_loop_tmpl.h b/aos/events/event_loop_tmpl.h
index 1a02495..0b08dc0 100644
--- a/aos/events/event_loop_tmpl.h
+++ b/aos/events/event_loop_tmpl.h
@@ -132,17 +132,18 @@
return false;
}
-inline bool RawSender::Send(size_t size) {
+inline RawSender::Error RawSender::Send(size_t size) {
return Send(size, monotonic_clock::min_time, realtime_clock::min_time,
0xffffffffu, event_loop_->boot_uuid());
}
-inline bool RawSender::Send(
+inline RawSender::Error RawSender::Send(
size_t size, aos::monotonic_clock::time_point monotonic_remote_time,
aos::realtime_clock::time_point realtime_remote_time,
uint32_t remote_queue_index, const UUID &uuid) {
- if (DoSend(size, monotonic_remote_time, realtime_remote_time,
- remote_queue_index, uuid)) {
+ const auto err = DoSend(size, monotonic_remote_time, realtime_remote_time,
+ remote_queue_index, uuid);
+ if (err == RawSender::Error::kOk) {
if (timing_.sender) {
timing_.size.Add(size);
timing_.sender->mutate_count(timing_.sender->count() + 1);
@@ -152,23 +153,23 @@
static_cast<int>(ftrace_prefix_.size()), ftrace_prefix_.data(),
static_cast<int64_t>(monotonic_sent_time().time_since_epoch().count()),
sent_queue_index());
- return true;
}
- return false;
+ return err;
}
-inline bool RawSender::Send(const void *data, size_t size) {
+inline RawSender::Error RawSender::Send(const void *data, size_t size) {
return Send(data, size, monotonic_clock::min_time, realtime_clock::min_time,
0xffffffffu, event_loop_->boot_uuid());
}
-inline bool RawSender::Send(
+inline RawSender::Error RawSender::Send(
const void *data, size_t size,
aos::monotonic_clock::time_point monotonic_remote_time,
aos::realtime_clock::time_point realtime_remote_time,
uint32_t remote_queue_index, const UUID &uuid) {
- if (DoSend(data, size, monotonic_remote_time, realtime_remote_time,
- remote_queue_index, uuid)) {
+ const auto err = DoSend(data, size, monotonic_remote_time,
+ realtime_remote_time, remote_queue_index, uuid);
+ if (err == RawSender::Error::kOk) {
if (timing_.sender) {
timing_.size.Add(size);
timing_.sender->mutate_count(timing_.sender->count() + 1);
@@ -178,24 +179,24 @@
static_cast<int>(ftrace_prefix_.size()), ftrace_prefix_.data(),
static_cast<int64_t>(monotonic_sent_time().time_since_epoch().count()),
sent_queue_index());
- return true;
}
- return false;
+ return err;
}
-inline bool RawSender::Send(const SharedSpan data) {
+inline RawSender::Error RawSender::Send(const SharedSpan data) {
return Send(std::move(data), monotonic_clock::min_time,
realtime_clock::min_time, 0xffffffffu, event_loop_->boot_uuid());
}
-inline bool RawSender::Send(
+inline RawSender::Error RawSender::Send(
const SharedSpan data,
aos::monotonic_clock::time_point monotonic_remote_time,
aos::realtime_clock::time_point realtime_remote_time,
uint32_t remote_queue_index, const UUID &uuid) {
const size_t size = data->size();
- if (DoSend(std::move(data), monotonic_remote_time, realtime_remote_time,
- remote_queue_index, uuid)) {
+ const auto err = DoSend(std::move(data), monotonic_remote_time,
+ realtime_remote_time, remote_queue_index, uuid);
+ if (err == Error::kOk) {
if (timing_.sender) {
timing_.size.Add(size);
timing_.sender->mutate_count(timing_.sender->count() + 1);
@@ -205,9 +206,8 @@
static_cast<int>(ftrace_prefix_.size()), ftrace_prefix_.data(),
static_cast<int64_t>(monotonic_sent_time().time_since_epoch().count()),
sent_queue_index());
- return true;
}
- return false;
+ return err;
}
inline monotonic_clock::time_point TimerHandler::Call(
@@ -371,12 +371,13 @@
};
template <typename T>
-bool Sender<T>::Send(const NonSizePrefixedFlatbuffer<T> &flatbuffer) {
+RawSender::Error Sender<T>::Send(
+ const NonSizePrefixedFlatbuffer<T> &flatbuffer) {
return sender_->Send(flatbuffer.span().data(), flatbuffer.span().size());
}
template <typename T>
-bool Sender<T>::SendDetached(FlatbufferDetachedBuffer<T> detached) {
+RawSender::Error Sender<T>::SendDetached(FlatbufferDetachedBuffer<T> detached) {
CHECK_EQ(static_cast<void *>(detached.span().data() + detached.span().size() -
sender_->size()),
sender_->data())
diff --git a/aos/events/logging/log_reader.cc b/aos/events/logging/log_reader.cc
index a757303..63eba05 100644
--- a/aos/events/logging/log_reader.cc
+++ b/aos/events/logging/log_reader.cc
@@ -1424,7 +1424,7 @@
// Send! Use the replayed queue index here instead of the logged queue index
// for the remote queue index. This makes re-logging work.
- const bool sent = sender->Send(
+ const auto err = sender->Send(
RawSender::SharedSpan(timestamped_message.data,
×tamped_message.data->span),
timestamped_message.monotonic_remote_time.time,
@@ -1434,7 +1434,7 @@
channel_source_state_[timestamped_message.channel_index])
->event_loop_->boot_uuid()
: event_loop_->boot_uuid()));
- if (!sent) return false;
+ if (err != RawSender::Error::kOk) return false;
if (queue_index_map_[timestamped_message.channel_index]) {
CHECK_EQ(timestamped_message.monotonic_event_time.boot, boot_count());
@@ -1455,7 +1455,8 @@
ContiguousSentTimestamp *back =
&queue_index_map_[timestamped_message.channel_index]->back();
if ((back->starting_queue_index - back->actual_queue_index) ==
- (timestamped_message.queue_index.index - sender->sent_queue_index())) {
+ (timestamped_message.queue_index.index -
+ sender->sent_queue_index())) {
back->ending_queue_index = timestamped_message.queue_index.index;
back->ending_monotonic_event_time =
timestamped_message.monotonic_event_time.time;
@@ -1597,7 +1598,8 @@
// Send out all timestamps at the currently scheduled time.
while (remote_timestamps_.front().monotonic_timestamp_time ==
scheduled_time_) {
- sender_.Send(std::move(remote_timestamps_.front().remote_message));
+ CHECK_EQ(sender_.Send(std::move(remote_timestamps_.front().remote_message)),
+ RawSender::Error::kOk);
remote_timestamps_.pop_front();
if (remote_timestamps_.empty()) {
break;
diff --git a/aos/events/logging/logger_test.cc b/aos/events/logging/logger_test.cc
index 7bf9478..a1e8cc9 100644
--- a/aos/events/logging/logger_test.cc
+++ b/aos/events/logging/logger_test.cc
@@ -3,8 +3,8 @@
#include "absl/strings/str_format.h"
#include "aos/events/event_loop.h"
#include "aos/events/logging/log_reader.h"
-#include "aos/events/logging/snappy_encoder.h"
#include "aos/events/logging/log_writer.h"
+#include "aos/events/logging/snappy_encoder.h"
#include "aos/events/message_counter.h"
#include "aos/events/ping_lib.h"
#include "aos/events/pong_lib.h"
@@ -168,15 +168,15 @@
Logger logger(logger_event_loop.get());
logger.set_polling_period(std::chrono::milliseconds(100));
- logger_event_loop->OnRun(
- [base_name1, base_name2, &logger_event_loop, &logger]() {
+ logger_event_loop->OnRun([base_name1, base_name2, &logger_event_loop,
+ &logger]() {
+ logger.StartLogging(std::make_unique<LocalLogNamer>(
+ base_name1, logger_event_loop.get(), logger_event_loop->node()));
+ EXPECT_DEATH(
logger.StartLogging(std::make_unique<LocalLogNamer>(
- base_name1, logger_event_loop.get(), logger_event_loop->node()));
- EXPECT_DEATH(logger.StartLogging(std::make_unique<LocalLogNamer>(
- base_name2, logger_event_loop.get(),
- logger_event_loop->node())),
- "Already logging");
- });
+ base_name2, logger_event_loop.get(), logger_event_loop->node())),
+ "Already logging");
+ });
event_loop_factory_.RunFor(chrono::milliseconds(20000));
}
}
@@ -398,7 +398,8 @@
ping_sender.MakeBuilder();
examples::Ping::Builder ping_builder =
builder.MakeBuilder<examples::Ping>();
- CHECK(builder.Send(ping_builder.Finish()));
+ CHECK_EQ(builder.Send(ping_builder.Finish()),
+ RawSender::Error::kOk);
});
// 100 ms / 0.05 ms -> 2000 messages. Should be enough to crash it.
@@ -2613,9 +2614,9 @@
}
constexpr std::string_view kCombinedConfigSha1(
- "cad3b6838a518ab29470771a959b89945ee034bc7a738080fd1713a1dce51b1f");
+ "644a3ab079c3ddfb1ef866483cfca9ec015c4142202169081138f72664ffd589");
constexpr std::string_view kSplitConfigSha1(
- "aafdd7e43d1942cce5b3e2dd8c6b9706abf7068a43501625a33b7cdfddf6c932");
+ "20bb9f6ee89519c4bd49986f0afe497d61c71b29d846f2c51f51727c9ef37415");
INSTANTIATE_TEST_SUITE_P(
All, MultinodeLoggerTest,
diff --git a/aos/events/ping_lib.cc b/aos/events/ping_lib.cc
index 7a0bfec..900ed0d 100644
--- a/aos/events/ping_lib.cc
+++ b/aos/events/ping_lib.cc
@@ -36,7 +36,7 @@
ping_builder.add_value(count_);
ping_builder.add_send_time(
event_loop_->monotonic_now().time_since_epoch().count());
- CHECK(builder.Send(ping_builder.Finish()));
+ builder.CheckOk(builder.Send(ping_builder.Finish()));
VLOG(2) << "Sending ping";
}
diff --git a/aos/events/pong_lib.cc b/aos/events/pong_lib.cc
index 9d7731d..9f1c855 100644
--- a/aos/events/pong_lib.cc
+++ b/aos/events/pong_lib.cc
@@ -16,7 +16,7 @@
builder.MakeBuilder<examples::Pong>();
pong_builder.add_value(ping.value());
pong_builder.add_initial_send_time(ping.send_time());
- CHECK(builder.Send(pong_builder.Finish()));
+ builder.CheckOk(builder.Send(pong_builder.Finish()));
});
event_loop_->SetRuntimeRealtimePriority(5);
diff --git a/aos/events/shm_event_loop.cc b/aos/events/shm_event_loop.cc
index a215040..ed3d099 100644
--- a/aos/events/shm_event_loop.cc
+++ b/aos/events/shm_event_loop.cc
@@ -547,11 +547,12 @@
shm_event_loop()->CheckCurrentThread();
return lockless_queue_sender_.size();
}
- bool DoSend(size_t length,
- aos::monotonic_clock::time_point monotonic_remote_time,
- aos::realtime_clock::time_point realtime_remote_time,
- uint32_t remote_queue_index,
- const UUID &source_boot_uuid) override {
+
+ Error DoSend(size_t length,
+ aos::monotonic_clock::time_point monotonic_remote_time,
+ aos::realtime_clock::time_point realtime_remote_time,
+ uint32_t remote_queue_index,
+ const UUID &source_boot_uuid) override {
shm_event_loop()->CheckCurrentThread();
CHECK_LE(length, static_cast<size_t>(channel()->max_size()))
<< ": Sent too big a message on "
@@ -565,14 +566,15 @@
wake_upper_.Wakeup(event_loop()->is_running() ? event_loop()->priority()
: 0);
- return true;
+ // TODO(Milind): check for messages sent too fast
+ return Error::kOk;
}
- bool DoSend(const void *msg, size_t length,
- aos::monotonic_clock::time_point monotonic_remote_time,
- aos::realtime_clock::time_point realtime_remote_time,
- uint32_t remote_queue_index,
- const UUID &source_boot_uuid) override {
+ Error DoSend(const void *msg, size_t length,
+ aos::monotonic_clock::time_point monotonic_remote_time,
+ aos::realtime_clock::time_point realtime_remote_time,
+ uint32_t remote_queue_index,
+ const UUID &source_boot_uuid) override {
shm_event_loop()->CheckCurrentThread();
CHECK_LE(length, static_cast<size_t>(channel()->max_size()))
<< ": Sent too big a message on "
@@ -586,7 +588,7 @@
wake_upper_.Wakeup(event_loop()->is_running() ? event_loop()->priority()
: 0);
// TODO(austin): Return an error if we send too fast.
- return true;
+ return RawSender::Error::kOk;
}
absl::Span<char> GetSharedMemory() const {
diff --git a/aos/events/shm_event_loop_test.cc b/aos/events/shm_event_loop_test.cc
index 7e1444b..01ca92b 100644
--- a/aos/events/shm_event_loop_test.cc
+++ b/aos/events/shm_event_loop_test.cc
@@ -144,7 +144,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- msg.Send(builder.Finish());
+ msg.CheckOk(msg.Send(builder.Finish()));
}
EXPECT_FALSE(IsRealtime());
@@ -185,7 +185,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200);
- msg.Send(builder.Finish());
+ msg.CheckOk(msg.Send(builder.Finish()));
});
factory()->Run();
@@ -280,7 +280,7 @@
auto builder = sender.MakeBuilder();
TestMessage::Builder test_builder(*builder.fbb());
test_builder.add_value(1);
- CHECK(builder.Send(test_builder.Finish()));
+ builder.CheckOk(builder.Send(test_builder.Finish()));
});
factory()->Run();
EXPECT_TRUE(ran);
@@ -317,7 +317,7 @@
auto builder = sender.MakeBuilder();
TestMessage::Builder test_builder(*builder.fbb());
test_builder.add_value(1);
- CHECK(builder.Send(test_builder.Finish()));
+ builder.CheckOk(builder.Send(test_builder.Finish()));
}
ASSERT_TRUE(fetcher.Fetch());
@@ -336,13 +336,13 @@
EXPECT_DEATH(
{
++static_cast<char *>(sender->data())[-1 - i];
- sender->Send(0);
+ sender->CheckOk(sender->Send(0));
},
"Somebody wrote outside the buffer of their message");
EXPECT_DEATH(
{
++static_cast<char *>(sender->data())[sender->size() + i];
- sender->Send(0);
+ sender->CheckOk(sender->Send(0));
},
"Somebody wrote outside the buffer of their message");
}
@@ -361,7 +361,7 @@
auto builder = sender.MakeBuilder();
TestMessage::Builder test_builder(*builder.fbb());
test_builder.add_value(0);
- CHECK(builder.Send(test_builder.Finish()));
+ builder.CheckOk(builder.Send(test_builder.Finish()));
}
EXPECT_DEATH(fetcher.FetchNext(),
"The next message is no longer "
@@ -387,7 +387,7 @@
auto builder = sender.MakeBuilder();
TestMessage::Builder test_builder(*builder.fbb());
test_builder.add_value(0);
- CHECK(builder.Send(test_builder.Finish()));
+ builder.CheckOk(builder.Send(test_builder.Finish()));
}
EXPECT_DEATH(fetcher.FetchNext(),
"The next message is no longer "
@@ -410,7 +410,7 @@
auto builder = sender.MakeBuilder();
TestMessage::Builder test_builder(*builder.fbb());
test_builder.add_value(0);
- CHECK(builder.Send(test_builder.Finish()));
+ builder.CheckOk(builder.Send(test_builder.Finish()));
}
EXPECT_DEATH(fetcher.FetchNext(),
"The next message is no longer "
@@ -429,7 +429,7 @@
auto builder = sender.MakeBuilder();
TestMessage::Builder test_builder(*builder.fbb());
test_builder.add_value(0);
- CHECK(builder.Send(test_builder.Finish()));
+ builder.CheckOk(builder.Send(test_builder.Finish()));
}
EXPECT_DEATH(fetcher.FetchNext(),
"The next message is no longer "
diff --git a/aos/events/simulated_event_loop.cc b/aos/events/simulated_event_loop.cc
index 9e12481..083e246 100644
--- a/aos/events/simulated_event_loop.cc
+++ b/aos/events/simulated_event_loop.cc
@@ -2,6 +2,8 @@
#include <algorithm>
#include <deque>
+#include <optional>
+#include <queue>
#include <string_view>
#include <vector>
@@ -177,6 +179,10 @@
.count();
}
+ std::chrono::nanoseconds channel_storage_duration() const {
+ return channel_storage_duration_;
+ }
+
// The number of extra buffers (beyond the queue) we pretend to have.
int number_scratch_buffers() const {
// We need to start creating messages before we know how many
@@ -220,8 +226,8 @@
}
// Sends the message to all the connected receivers and fetchers. Returns the
- // sent queue index.
- uint32_t Send(std::shared_ptr<SimulatedMessage> message);
+ // sent queue index, or std::nullopt if messages were sent too fast.
+ std::optional<uint32_t> Send(std::shared_ptr<SimulatedMessage> message);
// Unregisters a fetcher.
void UnregisterFetcher(SimulatedFetcher *fetcher);
@@ -334,22 +340,22 @@
size_t size() override { return simulated_channel_->max_size(); }
- bool DoSend(size_t length, monotonic_clock::time_point monotonic_remote_time,
- realtime_clock::time_point realtime_remote_time,
- uint32_t remote_queue_index,
- const UUID &source_boot_uuid) override;
+ Error DoSend(size_t length, monotonic_clock::time_point monotonic_remote_time,
+ realtime_clock::time_point realtime_remote_time,
+ uint32_t remote_queue_index,
+ const UUID &source_boot_uuid) override;
- bool DoSend(const void *msg, size_t size,
- monotonic_clock::time_point monotonic_remote_time,
- realtime_clock::time_point realtime_remote_time,
- uint32_t remote_queue_index,
- const UUID &source_boot_uuid) override;
+ Error DoSend(const void *msg, size_t size,
+ monotonic_clock::time_point monotonic_remote_time,
+ realtime_clock::time_point realtime_remote_time,
+ uint32_t remote_queue_index,
+ const UUID &source_boot_uuid) override;
- bool DoSend(const SharedSpan data,
- aos::monotonic_clock::time_point monotonic_remote_time,
- aos::realtime_clock::time_point realtime_remote_time,
- uint32_t remote_queue_index,
- const UUID &source_boot_uuid) override;
+ Error DoSend(const SharedSpan data,
+ aos::monotonic_clock::time_point monotonic_remote_time,
+ aos::realtime_clock::time_point realtime_remote_time,
+ uint32_t remote_queue_index,
+ const UUID &source_boot_uuid) override;
int buffer_index() override {
// First, ensure message_ is allocated.
@@ -912,10 +918,11 @@
return ::std::move(fetcher);
}
-uint32_t SimulatedChannel::Send(std::shared_ptr<SimulatedMessage> message) {
- const uint32_t queue_index = next_queue_index_.index();
- message->context.queue_index = queue_index;
+std::optional<uint32_t> SimulatedChannel::Send(
+ std::shared_ptr<SimulatedMessage> message) {
+ std::optional<uint32_t> queue_index = {next_queue_index_.index()};
+ message->context.queue_index = *queue_index;
// Points to the actual data depending on the size set in context. Data may
// allocate more than the actual size of the message, so offset from the back
// of that to get the actual start of the data.
@@ -943,7 +950,6 @@
for (auto &fetcher : fetchers_) {
fetcher->Enqueue(message);
}
-
return queue_index;
}
@@ -963,11 +969,10 @@
simulated_channel_->CountSenderDestroyed();
}
-bool SimulatedSender::DoSend(size_t length,
- monotonic_clock::time_point monotonic_remote_time,
- realtime_clock::time_point realtime_remote_time,
- uint32_t remote_queue_index,
- const UUID &source_boot_uuid) {
+RawSender::Error SimulatedSender::DoSend(
+ size_t length, monotonic_clock::time_point monotonic_remote_time,
+ realtime_clock::time_point realtime_remote_time,
+ uint32_t remote_queue_index, const UUID &source_boot_uuid) {
VLOG(1) << simulated_event_loop_->distributed_now() << " "
<< NodeName(simulated_event_loop_->node())
<< simulated_event_loop_->monotonic_now() << " "
@@ -988,8 +993,31 @@
CHECK_LE(length, message_->context.size);
message_->context.size = length;
- // TODO(austin): Track sending too fast.
- sent_queue_index_ = simulated_channel_->Send(message_);
+ const std::optional<uint32_t> optional_queue_index =
+ simulated_channel_->Send(message_);
+
+ // Check that we are not sending messages too fast
+ if (!optional_queue_index) {
+ VLOG(1) << simulated_event_loop_->distributed_now() << " "
+ << NodeName(simulated_event_loop_->node())
+ << simulated_event_loop_->monotonic_now() << " "
+ << simulated_event_loop_->name()
+ << "\nMessages were sent too fast:\n"
+ << "For channel: "
+ << configuration::CleanedChannelToString(
+ simulated_channel_->channel())
+ << '\n'
+ << "Tried to send more than " << simulated_channel_->queue_size()
+ << " (queue size) messages in the last "
+ << std::chrono::duration<double>(
+ simulated_channel_->channel_storage_duration())
+ .count()
+ << " seconds (channel storage duration)"
+ << "\n\n";
+ return Error::kMessagesSentTooFast;
+ }
+
+ sent_queue_index_ = *optional_queue_index;
monotonic_sent_time_ = simulated_event_loop_->monotonic_now();
realtime_sent_time_ = simulated_event_loop_->realtime_now();
@@ -997,14 +1025,14 @@
// next time. Otherwise we will continue to reuse the same memory for all
// messages and corrupt it.
message_.reset();
- return true;
+ return Error::kOk;
}
-bool SimulatedSender::DoSend(const void *msg, size_t size,
- monotonic_clock::time_point monotonic_remote_time,
- realtime_clock::time_point realtime_remote_time,
- uint32_t remote_queue_index,
- const UUID &source_boot_uuid) {
+RawSender::Error SimulatedSender::DoSend(
+ const void *msg, size_t size,
+ monotonic_clock::time_point monotonic_remote_time,
+ realtime_clock::time_point realtime_remote_time,
+ uint32_t remote_queue_index, const UUID &source_boot_uuid) {
CHECK_LE(size, this->size())
<< ": Attempting to send too big a message on "
<< configuration::CleanedChannelToString(simulated_channel_->channel());
@@ -1021,11 +1049,11 @@
remote_queue_index, source_boot_uuid);
}
-bool SimulatedSender::DoSend(const RawSender::SharedSpan data,
- monotonic_clock::time_point monotonic_remote_time,
- realtime_clock::time_point realtime_remote_time,
- uint32_t remote_queue_index,
- const UUID &source_boot_uuid) {
+RawSender::Error SimulatedSender::DoSend(
+ const RawSender::SharedSpan data,
+ monotonic_clock::time_point monotonic_remote_time,
+ realtime_clock::time_point realtime_remote_time,
+ uint32_t remote_queue_index, const UUID &source_boot_uuid) {
CHECK_LE(data->size(), this->size())
<< ": Attempting to send too big a message on "
<< configuration::CleanedChannelToString(simulated_channel_->channel());
diff --git a/aos/events/simulated_event_loop_test.cc b/aos/events/simulated_event_loop_test.cc
index 75e3f85..696c16e 100644
--- a/aos/events/simulated_event_loop_test.cc
+++ b/aos/events/simulated_event_loop_test.cc
@@ -1,5 +1,6 @@
#include "aos/events/simulated_event_loop.h"
+#include <chrono>
#include <string_view>
#include "aos/events/event_loop_param_test.h"
@@ -174,7 +175,8 @@
TestMessage::Builder test_message_builder =
builder.MakeBuilder<TestMessage>();
test_message_builder.add_value(value);
- builder.Send(test_message_builder.Finish());
+ ASSERT_EQ(builder.Send(test_message_builder.Finish()),
+ RawSender::Error::kOk);
}
// Test that sending a message after running gets properly notified.
@@ -337,7 +339,7 @@
aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
builder.add_value(200 + i);
- ASSERT_TRUE(msg.Send(builder.Finish()));
+ msg.CheckOk(msg.Send(builder.Finish()));
}
});
@@ -1394,7 +1396,7 @@
aos::Sender<examples::Ping>::Builder builder = sender->MakeBuilder();
examples::Ping::Builder ping_builder = builder.MakeBuilder<examples::Ping>();
ping_builder.add_value(value);
- builder.Send(ping_builder.Finish());
+ builder.CheckOk(builder.Send(ping_builder.Finish()));
}
// Tests that reliable (and unreliable) ping messages get forwarded as expected.
diff --git a/aos/events/simulated_network_bridge.cc b/aos/events/simulated_network_bridge.cc
index b87182a..96a614d 100644
--- a/aos/events/simulated_network_bridge.cc
+++ b/aos/events/simulated_network_bridge.cc
@@ -61,10 +61,10 @@
timestamp_timer_ =
fetch_event_loop_->AddTimer([this]() { SendTimestamp(); });
if (send_event_loop_) {
- std::string timer_name = absl::StrCat(
- send_event_loop_->node()->name()->string_view(), " ",
- fetcher_->channel()->name()->string_view(), " ",
- fetcher_->channel()->type()->string_view());
+ std::string timer_name =
+ absl::StrCat(send_event_loop_->node()->name()->string_view(), " ",
+ fetcher_->channel()->name()->string_view(), " ",
+ fetcher_->channel()->type()->string_view());
if (timer_) {
timer_->set_name(timer_name);
}
@@ -250,11 +250,11 @@
return;
}
// Fill out the send times.
- sender_->Send(fetcher_->context().data, fetcher_->context().size,
- fetcher_->context().monotonic_event_time,
- fetcher_->context().realtime_event_time,
- fetcher_->context().queue_index,
- fetcher_->context().source_boot_uuid);
+ sender_->CheckOk(sender_->Send(
+ fetcher_->context().data, fetcher_->context().size,
+ fetcher_->context().monotonic_event_time,
+ fetcher_->context().realtime_event_time,
+ fetcher_->context().queue_index, fetcher_->context().source_boot_uuid));
// And simulate message_bridge's offset recovery.
client_status_->SampleFilter(client_index_,
@@ -338,8 +338,8 @@
while (remote_timestamps_.front().monotonic_timestamp_time ==
scheduled_time_) {
if (server_connection_->state() == State::CONNECTED) {
- timestamp_logger_->Send(
- std::move(remote_timestamps_.front().remote_message));
+ timestamp_logger_->CheckOk(timestamp_logger_->Send(
+ std::move(remote_timestamps_.front().remote_message)));
}
remote_timestamps_.pop_front();
if (remote_timestamps_.empty()) {
diff --git a/aos/logging/log_message.fbs b/aos/logging/log_message.fbs
index b68287f..db28a6f 100644
--- a/aos/logging/log_message.fbs
+++ b/aos/logging/log_message.fbs
@@ -22,6 +22,9 @@
// Application name
name:string (id: 3);
+
+ // Total number of LogMessage send failures.
+ send_failures:uint64 (id: 4);
}
root_type LogMessageFbs;
diff --git a/aos/network/message_bridge_client_lib.cc b/aos/network/message_bridge_client_lib.cc
index 2b46dc2..f535c4f 100644
--- a/aos/network/message_bridge_client_lib.cc
+++ b/aos/network/message_bridge_client_lib.cc
@@ -257,13 +257,14 @@
// Publish the message.
RawSender *sender = channel_state->sender.get();
- sender->Send(remote_data->data()->data(), remote_data->data()->size(),
- monotonic_clock::time_point(
- chrono::nanoseconds(remote_data->monotonic_sent_time())),
- realtime_clock::time_point(
- chrono::nanoseconds(remote_data->realtime_sent_time())),
- remote_data->queue_index(),
- UUID::FromVector(remote_data->boot_uuid()));
+ sender->CheckOk(sender->Send(
+ remote_data->data()->data(), remote_data->data()->size(),
+ monotonic_clock::time_point(
+ chrono::nanoseconds(remote_data->monotonic_sent_time())),
+ realtime_clock::time_point(
+ chrono::nanoseconds(remote_data->realtime_sent_time())),
+ remote_data->queue_index(),
+ UUID::FromVector(remote_data->boot_uuid())));
client_status_->SampleFilter(
client_index_,
diff --git a/aos/network/message_bridge_client_status.cc b/aos/network/message_bridge_client_status.cc
index faade70..357026a 100644
--- a/aos/network/message_bridge_client_status.cc
+++ b/aos/network/message_bridge_client_status.cc
@@ -122,7 +122,7 @@
builder.MakeBuilder<ClientStatistics>();
client_statistics_builder.add_connections(client_connections_offset);
- builder.Send(client_statistics_builder.Finish());
+ builder.CheckOk(builder.Send(client_statistics_builder.Finish()));
}
int MessageBridgeClientStatus::FindClientIndex(std::string_view node_name) {
diff --git a/aos/network/message_bridge_server.fbs b/aos/network/message_bridge_server.fbs
index cdf970c..128f1f3 100644
--- a/aos/network/message_bridge_server.fbs
+++ b/aos/network/message_bridge_server.fbs
@@ -40,6 +40,9 @@
// Statistics for all connections to all the clients.
table ServerStatistics {
connections:[ServerConnection] (id: 0);
+
+ // Count of timestamp send failures
+ timestamp_send_failures:uint64 (id: 1);
}
root_type ServerStatistics;
diff --git a/aos/network/message_bridge_server_lib.cc b/aos/network/message_bridge_server_lib.cc
index 006f2b1..41fb915 100644
--- a/aos/network/message_bridge_server_lib.cc
+++ b/aos/network/message_bridge_server_lib.cc
@@ -159,7 +159,7 @@
server_status->AddPartialDeliveries(peer.node_index,
partial_deliveries);
- builder.Send(remote_message_builder.Finish());
+ builder.CheckOk(builder.Send(remote_message_builder.Finish()));
}
break;
}
diff --git a/aos/network/message_bridge_server_status.cc b/aos/network/message_bridge_server_status.cc
index f7553e4..09355b2 100644
--- a/aos/network/message_bridge_server_status.cc
+++ b/aos/network/message_bridge_server_status.cc
@@ -87,7 +87,8 @@
filters_.resize(event_loop->configuration()->nodes()->size());
partial_deliveries_.resize(event_loop->configuration()->nodes()->size());
- boot_uuids_.resize(event_loop->configuration()->nodes()->size(), UUID::Zero());
+ boot_uuids_.resize(event_loop->configuration()->nodes()->size(),
+ UUID::Zero());
has_boot_uuids_.resize(event_loop->configuration()->nodes()->size(), false);
timestamp_fetchers_.resize(event_loop->configuration()->nodes()->size());
server_connection_.resize(event_loop->configuration()->nodes()->size());
@@ -217,7 +218,10 @@
ServerStatistics::Builder server_statistics_builder =
builder.MakeBuilder<ServerStatistics>();
server_statistics_builder.add_connections(server_connections_offset);
- builder.Send(server_statistics_builder.Finish());
+ server_statistics_builder.add_timestamp_send_failures(
+ timestamp_failure_counter_.failures());
+
+ builder.CheckOk(builder.Send(server_statistics_builder.Finish()));
}
void MessageBridgeServerStatus::Tick() {
@@ -352,20 +356,24 @@
// Send it out over shm, and using that timestamp, then send it out over sctp.
// This avoid some context switches.
if (!send_) return;
- timestamp_sender_.Send(timestamp_copy);
- Context context;
- context.monotonic_event_time = timestamp_sender_.monotonic_sent_time();
- context.realtime_event_time = timestamp_sender_.realtime_sent_time();
- context.queue_index = timestamp_sender_.sent_queue_index();
- context.size = timestamp_copy.span().size();
- context.source_boot_uuid = event_loop_->boot_uuid();
- context.data = timestamp_copy.span().data();
+ const auto err = timestamp_sender_.Send(timestamp_copy);
+ timestamp_failure_counter_.Count(err);
+ // Reply only if we successfully sent the timestamp
+ if (err == RawSender::Error::kOk) {
+ Context context;
+ context.monotonic_event_time = timestamp_sender_.monotonic_sent_time();
+ context.realtime_event_time = timestamp_sender_.realtime_sent_time();
+ context.queue_index = timestamp_sender_.sent_queue_index();
+ context.size = timestamp_copy.span().size();
+ context.source_boot_uuid = event_loop_->boot_uuid();
+ context.data = timestamp_copy.span().data();
- // Since we are building up the timestamp to send here, we need to trigger the
- // SendData call ourselves.
- if (send_data_) {
- send_data_(context);
+ // Since we are building up the timestamp to send here, we need to trigger
+ // the SendData call ourselves.
+ if (send_data_) {
+ send_data_(context);
+ }
}
}
diff --git a/aos/network/message_bridge_server_status.h b/aos/network/message_bridge_server_status.h
index c17e00a..3a1c9ed 100644
--- a/aos/network/message_bridge_server_status.h
+++ b/aos/network/message_bridge_server_status.h
@@ -113,6 +113,8 @@
// Sender for the timestamps that we are forwarding over the network.
aos::Sender<Timestamp> timestamp_sender_;
+ SendFailureCounter timestamp_failure_counter_;
+
aos::monotonic_clock::time_point last_statistics_send_time_ =
aos::monotonic_clock::min_time;
@@ -123,7 +125,6 @@
std::vector<uint32_t> partial_deliveries_;
};
-
} // namespace message_bridge
} // namespace aos
diff --git a/aos/network/message_bridge_test.cc b/aos/network/message_bridge_test.cc
index 2a43170..8246c22 100644
--- a/aos/network/message_bridge_test.cc
+++ b/aos/network/message_bridge_test.cc
@@ -450,7 +450,8 @@
examples::Ping::Builder ping_builder =
builder.MakeBuilder<examples::Ping>();
ping_builder.add_value(ping_count + 971);
- builder.Send(ping_builder.Finish());
+ EXPECT_EQ(builder.Send(ping_builder.Finish()),
+ RawSender::Error::kOk);
++ping_count;
}
});
@@ -985,7 +986,7 @@
aos::Sender<examples::Ping>::Builder builder = sender->MakeBuilder();
examples::Ping::Builder ping_builder = builder.MakeBuilder<examples::Ping>();
ping_builder.add_value(value);
- builder.Send(ping_builder.Finish());
+ builder.CheckOk(builder.Send(ping_builder.Finish()));
}
// Tests that when a message is sent before the bridge starts up, but is
@@ -1148,7 +1149,7 @@
examples::Ping::Builder ping_builder =
builder.MakeBuilder<examples::Ping>();
ping_builder.add_value(1);
- builder.Send(ping_builder.Finish());
+ builder.CheckOk(builder.Send(ping_builder.Finish()));
}
MakePi1Client();
diff --git a/aos/starter/starter_rpc_lib.cc b/aos/starter/starter_rpc_lib.cc
index b0b9db3..3007326 100644
--- a/aos/starter/starter_rpc_lib.cc
+++ b/aos/starter/starter_rpc_lib.cc
@@ -3,6 +3,7 @@
#include "aos/events/shm_event_loop.h"
#include "aos/flatbuffer_merge.h"
#include "aos/starter/starterd_lib.h"
+#include "glog/logging.h"
namespace aos {
namespace starter {
@@ -163,7 +164,7 @@
if (is_multi_node) {
command_builder.add_nodes(nodes_offset);
}
- CHECK(builder.Send(command_builder.Finish()));
+ builder.CheckOk(builder.Send(command_builder.Finish()));
}
timeout_timer_->Setup(event_loop_->monotonic_now() + timeout);
diff --git a/aos/starter/starterd_lib.cc b/aos/starter/starterd_lib.cc
index bc83768..7bf2e0d 100644
--- a/aos/starter/starterd_lib.cc
+++ b/aos/starter/starterd_lib.cc
@@ -602,7 +602,7 @@
aos::starter::Status::Builder status_builder(*builder.fbb());
status_builder.add_statuses(statuses_fbs);
- CHECK(builder.Send(status_builder.Finish()));
+ builder.CheckOk(builder.Send(status_builder.Finish()));
}
} // namespace starter