Add an API for watchers without an argument
I'm going to give ShmEventLoop a specialized implementation, which
avoids copying the actual message data.
Change-Id: Ica7beba9835286d80adcab92e8d1d300fbeaedfc
diff --git a/aos/events/event_loop.h b/aos/events/event_loop.h
index d0ca196..a74e65b 100644
--- a/aos/events/event_loop.h
+++ b/aos/events/event_loop.h
@@ -302,9 +302,7 @@
// Returns the name of the underlying queue.
const Channel *channel() const { return sender_->channel(); }
- operator bool() {
- return sender_ ? true : false;
- }
+ operator bool() { return sender_ ? true : false; }
// Returns the time_points that the last message was sent at.
aos::monotonic_clock::time_point monotonic_sent_time() const {
@@ -464,13 +462,32 @@
// This will watch messages sent to the provided channel.
//
- // Watch is a functor that have a call signature like so:
- // void Event(const MessageType& type);
+ // w must have a non-polymorphic operator() (aka it can only be called with a
+ // single set of arguments; no overloading or templates). It must be callable
+ // with this signature:
+ // void(const MessageType &);
//
- // TODO(parker): Need to support ::std::bind. For now, use lambdas.
- // TODO(austin): Do we need a functor? Or is a std::function good enough?
+ // Lambdas are a common form for w. A std::function will work too.
+ //
+ // Note that bind expressions have polymorphic call operators, so they are not
+ // allowed.
+ //
+ // We template Watch as a whole instead of using std::function<void(const T
+ // &)> to allow deducing MessageType from lambdas and other things which are
+ // implicitly convertible to std::function, but not actually std::function
+ // instantiations. Template deduction guides might allow solving this
+ // differently in newer versions of C++, but those have their own corner
+ // cases.
template <typename Watch>
- void MakeWatcher(const std::string_view name, Watch &&w);
+ void MakeWatcher(const std::string_view channel_name, Watch &&w);
+
+ // Like MakeWatcher, but doesn't have access to the message data. This may be
+ // implemented to use less resources than an equivalent MakeWatcher.
+ //
+ // The function will still have access to context().
+ template <typename MessageType>
+ void MakeNoArgWatcher(const std::string_view channel_name,
+ std::function<void()> w);
// The passed in function will be called when the event loop starts.
// Use this to run code once the thread goes into "real-time-mode",
@@ -518,6 +535,16 @@
std::function<void(const Context &context, const void *message)>
watcher) = 0;
+ // Watches channel (name, type) for new messages, without needing to extract
+ // the message contents. Default implementation simply re-uses MakeRawWatcher.
+ virtual void MakeRawNoArgWatcher(
+ const Channel *channel,
+ std::function<void(const Context &context)> watcher) {
+ MakeRawWatcher(channel, [watcher](const Context &context, const void *) {
+ watcher(context);
+ });
+ }
+
// Creates a raw sender for the provided channel. This is used for reflection
// based sending.
// Note: this ignores any node constraints. Ignore at your own peril.
diff --git a/aos/events/event_loop_param_test.cc b/aos/events/event_loop_param_test.cc
index 096d84a..b4cfeb2 100644
--- a/aos/events/event_loop_param_test.cc
+++ b/aos/events/event_loop_param_test.cc
@@ -43,6 +43,66 @@
EXPECT_TRUE(happened);
}
+// Tests that no-arg watcher can receive messages from a sender.
+// Also tests that OnRun() works.
+TEST_P(AbstractEventLoopTest, BasicNoArg) {
+ auto loop1 = Make();
+ auto loop2 = MakePrimary();
+
+ aos::Sender<TestMessage> sender = loop1->MakeSender<TestMessage>("/test");
+
+ bool happened = false;
+
+ loop2->OnRun([&]() {
+ happened = true;
+
+ aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
+ TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
+ builder.add_value(200);
+ ASSERT_TRUE(msg.Send(builder.Finish()));
+ });
+
+ aos::Fetcher<TestMessage> fetcher = loop2->MakeFetcher<TestMessage>("/test");
+ loop2->MakeNoArgWatcher<TestMessage>("/test", [&]() {
+ ASSERT_TRUE(fetcher.Fetch());
+ EXPECT_EQ(fetcher->value(), 200);
+ this->Exit();
+ });
+
+ EXPECT_FALSE(happened);
+ Run();
+ EXPECT_TRUE(happened);
+}
+
+// Tests that a watcher can be created with an std::function.
+TEST_P(AbstractEventLoopTest, BasicFunction) {
+ auto loop1 = Make();
+ auto loop2 = MakePrimary();
+
+ aos::Sender<TestMessage> sender = loop1->MakeSender<TestMessage>("/test");
+
+ bool happened = false;
+
+ loop2->OnRun([&]() {
+ happened = true;
+
+ aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
+ TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
+ builder.add_value(200);
+ ASSERT_TRUE(msg.Send(builder.Finish()));
+ });
+
+ loop2->MakeWatcher("/test", std::function<void(const TestMessage &)>(
+ [&](const TestMessage &message) {
+ EXPECT_EQ(message.value(), 200);
+ this->Exit();
+ }));
+
+ EXPECT_FALSE(happened);
+ Run();
+ EXPECT_TRUE(happened);
+}
+
// Tests that watcher can receive messages from two senders.
// Also tests that OnRun() works.
TEST_P(AbstractEventLoopTest, BasicTwoSenders) {
@@ -463,6 +523,9 @@
EXPECT_DEATH(
{ loop->MakeWatcher("/test/invalid", [&](const TestMessage &) {}); },
"/test/invalid");
+ EXPECT_DEATH(
+ { loop->MakeNoArgWatcher<TestMessage>("/test/invalid", [&]() {}); },
+ "/test/invalid");
}
// Verify that registering a watcher twice for "/test" fails.
@@ -471,6 +534,16 @@
loop->MakeWatcher("/test", [&](const TestMessage &) {});
EXPECT_DEATH(loop->MakeWatcher("/test", [&](const TestMessage &) {}),
"/test");
+ EXPECT_DEATH(loop->MakeNoArgWatcher<TestMessage>("/test", [&]() {}), "/test");
+}
+
+// Verify that registering a no-arg watcher twice for "/test" fails.
+TEST_P(AbstractEventLoopDeathTest, TwoNoArgWatcher) {
+ auto loop = Make();
+ loop->MakeNoArgWatcher<TestMessage>("/test", [&]() {});
+ EXPECT_DEATH(loop->MakeWatcher("/test", [&](const TestMessage &) {}),
+ "/test");
+ EXPECT_DEATH(loop->MakeNoArgWatcher<TestMessage>("/test", [&]() {}), "/test");
}
// Verify that SetRuntimeRealtimePriority fails while running.
@@ -512,6 +585,16 @@
EXPECT_DEATH(Run(), "running");
}
+// Verify that we can't create a no-arg watcher inside OnRun.
+TEST_P(AbstractEventLoopDeathTest, NoArgWatcherInOnRun) {
+ auto loop1 = MakePrimary();
+
+ loop1->OnRun(
+ [&]() { loop1->MakeNoArgWatcher<TestMessage>("/test", [&]() {}); });
+
+ EXPECT_DEATH(Run(), "running");
+}
+
// Verify that Quit() works when there are multiple watchers.
TEST_P(AbstractEventLoopTest, MultipleWatcherQuit) {
auto loop1 = Make();
@@ -722,7 +805,7 @@
"Channel pointer not found in configuration\\(\\)->channels\\(\\)");
}
-// Verify that the send time on a message is roughly right.
+// Verify that the send time on a message is roughly right when using a watcher.
TEST_P(AbstractEventLoopTest, MessageSendTime) {
auto loop1 = MakePrimary();
auto loop2 = Make();
@@ -737,7 +820,7 @@
});
bool triggered = false;
- loop1->MakeWatcher("/test", [&triggered, &loop1](const TestMessage &msg) {
+ loop1->MakeWatcher("/test", [&](const TestMessage &msg) {
// Confirm that the data pointer makes sense from a watcher, and all the
// timestamps look right.
EXPECT_GT(&msg, loop1->context().data);
@@ -770,7 +853,92 @@
EXPECT_TRUE(triggered);
- EXPECT_TRUE(fetcher.Fetch());
+ ASSERT_TRUE(fetcher.Fetch());
+
+ monotonic_clock::duration monotonic_time_offset =
+ fetcher.context().monotonic_event_time -
+ (loop1->monotonic_now() - ::std::chrono::seconds(1));
+ realtime_clock::duration realtime_time_offset =
+ fetcher.context().realtime_event_time -
+ (loop1->realtime_now() - ::std::chrono::seconds(1));
+
+ EXPECT_EQ(fetcher.context().realtime_event_time,
+ fetcher.context().realtime_remote_time);
+ EXPECT_EQ(fetcher.context().monotonic_event_time,
+ fetcher.context().monotonic_remote_time);
+
+ EXPECT_TRUE(monotonic_time_offset > ::std::chrono::milliseconds(-500))
+ << ": Got "
+ << fetcher.context().monotonic_event_time.time_since_epoch().count()
+ << " expected " << loop1->monotonic_now().time_since_epoch().count();
+ // Confirm that the data pointer makes sense.
+ EXPECT_GT(fetcher.get(), fetcher.context().data);
+ EXPECT_LT(fetcher.get(),
+ reinterpret_cast<void *>(
+ reinterpret_cast<char *>(fetcher.context().data) +
+ fetcher.context().size));
+ EXPECT_TRUE(monotonic_time_offset < ::std::chrono::milliseconds(500))
+ << ": Got "
+ << fetcher.context().monotonic_event_time.time_since_epoch().count()
+ << " expected " << loop1->monotonic_now().time_since_epoch().count();
+
+ EXPECT_TRUE(realtime_time_offset > ::std::chrono::milliseconds(-500))
+ << ": Got "
+ << fetcher.context().realtime_event_time.time_since_epoch().count()
+ << " expected " << loop1->realtime_now().time_since_epoch().count();
+ EXPECT_TRUE(realtime_time_offset < ::std::chrono::milliseconds(500))
+ << ": Got "
+ << fetcher.context().realtime_event_time.time_since_epoch().count()
+ << " expected " << loop1->realtime_now().time_since_epoch().count();
+}
+
+// Verify that the send time on a message is roughly right when using a no-arg
+// watcher. To get a message, we need to use a fetcher to actually access the
+// message. This is also the main use case for no-arg fetchers.
+TEST_P(AbstractEventLoopTest, MessageSendTimeNoArg) {
+ auto loop1 = MakePrimary();
+ auto loop2 = Make();
+ auto sender = loop2->MakeSender<TestMessage>("/test");
+ auto fetcher = loop1->MakeFetcher<TestMessage>("/test");
+
+ auto test_timer = loop1->AddTimer([&sender]() {
+ aos::Sender<TestMessage>::Builder msg = sender.MakeBuilder();
+ TestMessage::Builder builder = msg.MakeBuilder<TestMessage>();
+ builder.add_value(200);
+ ASSERT_TRUE(msg.Send(builder.Finish()));
+ });
+
+ bool triggered = false;
+ loop1->MakeNoArgWatcher<TestMessage>("/test", [&]() {
+ // Confirm that we can indeed use a fetcher on this channel from this
+ // context, and it results in a sane data pointer and timestamps.
+ ASSERT_TRUE(fetcher.Fetch());
+
+ EXPECT_EQ(loop1->context().monotonic_remote_time,
+ loop1->context().monotonic_event_time);
+ EXPECT_EQ(loop1->context().realtime_remote_time,
+ loop1->context().realtime_event_time);
+
+ const aos::monotonic_clock::time_point monotonic_now =
+ loop1->monotonic_now();
+ const aos::realtime_clock::time_point realtime_now = loop1->realtime_now();
+
+ EXPECT_LE(loop1->context().monotonic_event_time, monotonic_now);
+ EXPECT_LE(loop1->context().realtime_event_time, realtime_now);
+ EXPECT_GE(loop1->context().monotonic_event_time + chrono::milliseconds(500),
+ monotonic_now);
+ EXPECT_GE(loop1->context().realtime_event_time + chrono::milliseconds(500),
+ realtime_now);
+
+ triggered = true;
+ });
+
+ test_timer->Setup(loop1->monotonic_now() + ::std::chrono::seconds(1));
+
+ EndEventLoop(loop1.get(), ::std::chrono::seconds(2));
+ Run();
+
+ ASSERT_TRUE(triggered);
monotonic_clock::duration monotonic_time_offset =
fetcher.context().monotonic_event_time -
@@ -1336,6 +1504,19 @@
[](const Context &, const void *) {});
}
+// Tests that no-arg watchers work with a node setup.
+TEST_P(AbstractEventLoopTest, NodeNoArgWatcher) {
+ EnableNodes("me");
+
+ auto loop1 = Make();
+ auto loop2 = Make();
+ loop1->MakeWatcher("/test", [](const TestMessage &) {});
+ loop2->MakeRawNoArgWatcher(
+ configuration::GetChannel(configuration(), "/test", "aos.TestMessage", "",
+ nullptr),
+ [](const Context &) {});
+}
+
// Tests that fetcher work with a node setup.
TEST_P(AbstractEventLoopTest, NodeFetcher) {
EnableNodes("me");
@@ -1370,6 +1551,16 @@
[](const Context &, const void *) {});
},
"node");
+ EXPECT_DEATH({ loop1->MakeNoArgWatcher<TestMessage>("/test", []() {}); },
+ "node");
+ EXPECT_DEATH(
+ {
+ loop2->MakeRawNoArgWatcher(
+ configuration::GetChannel(configuration(), "/test",
+ "aos.TestMessage", "", nullptr),
+ [](const Context &) {});
+ },
+ "node");
}
// Tests that fetchers fail when created on the wrong node.
diff --git a/aos/events/event_loop_tmpl.h b/aos/events/event_loop_tmpl.h
index 1daf88c..6c3944c 100644
--- a/aos/events/event_loop_tmpl.h
+++ b/aos/events/event_loop_tmpl.h
@@ -6,13 +6,16 @@
#include "glog/logging.h"
namespace aos {
+namespace event_loop_internal {
-// From a watch functor, this will extract the message type of the argument.
-// This is the template forward declaration, and it extracts the call operator
-// as a PTMF to be used by the following specialization.
+// From a watch functor, specializations of this will extract the message type
+// of the template argument. If T is not a valid message type, there will be no
+// matching specialization.
+//
+// This is just the forward declaration, which will be used by one of the
+// following specializations to match valid argument types.
template <class T>
-struct watch_message_type_trait
- : watch_message_type_trait<decltype(&T::operator())> {};
+struct watch_message_type_trait;
// From a watch functor, this will extract the message type of the argument.
// This is the template specialization.
@@ -21,6 +24,8 @@
using message_type = typename std::decay<A1>::type;
};
+} // namespace event_loop_internal
+
template <typename T>
typename Sender<T>::Builder Sender<T>::MakeBuilder() {
return Builder(sender_.get(), sender_->fbb_allocator());
@@ -28,19 +33,38 @@
template <typename Watch>
void EventLoop::MakeWatcher(const std::string_view channel_name, Watch &&w) {
- using T = typename watch_message_type_trait<Watch>::message_type;
+ using MessageType =
+ typename event_loop_internal::watch_message_type_trait<decltype(
+ &Watch::operator())>::message_type;
const Channel *channel = configuration::GetChannel(
- configuration_, channel_name, T::GetFullyQualifiedName(), name(), node());
+ configuration_, channel_name, MessageType::GetFullyQualifiedName(),
+ name(), node());
CHECK(channel != nullptr)
<< ": Channel { \"name\": \"" << channel_name << "\", \"type\": \""
- << T::GetFullyQualifiedName() << "\" } not found in config.";
+ << MessageType::GetFullyQualifiedName() << "\" } not found in config.";
- return MakeRawWatcher(
- channel, [this, w](const Context &context, const void *message) {
- context_ = context;
- w(*flatbuffers::GetRoot<T>(reinterpret_cast<const char *>(message)));
- });
+ MakeRawWatcher(channel,
+ [this, w](const Context &context, const void *message) {
+ context_ = context;
+ w(*flatbuffers::GetRoot<MessageType>(
+ reinterpret_cast<const char *>(message)));
+ });
+}
+
+template <typename MessageType>
+void EventLoop::MakeNoArgWatcher(const std::string_view channel_name,
+ std::function<void()> w) {
+ const Channel *channel = configuration::GetChannel(
+ configuration_, channel_name, MessageType::GetFullyQualifiedName(),
+ name(), node());
+ CHECK(channel != nullptr)
+ << ": Channel { \"name\": \"" << channel_name << "\", \"type\": \""
+ << MessageType::GetFullyQualifiedName() << "\" } not found in config.";
+ MakeRawNoArgWatcher(channel, [this, w](const Context &context) {
+ context_ = context;
+ w();
+ });
}
inline bool RawFetcher::FetchNext() {