Add basic support for nodes
This adds the infrastructure and configuration language to describe a
multinode world. This only checks that if there are multiple nodes
setup, everything is both configured for multiple nodes, and that we are
listening and sending data on the correct node per the configuration.
Change-Id: I658ba05620337a210d677c43e5eb840e05f96051
diff --git a/aos/BUILD b/aos/BUILD
index ee0aa05..cbe31c6 100644
--- a/aos/BUILD
+++ b/aos/BUILD
@@ -310,6 +310,7 @@
":flatbuffers",
":json_to_flatbuffer",
"//aos:unique_malloc_ptr",
+ "//aos/network:team_number",
"//aos/util:file",
"@com_github_google_glog//:glog",
"@com_google_absl//absl/base",
@@ -463,6 +464,13 @@
"testdata/config2.json",
"testdata/config3.json",
"testdata/expected.json",
+ "testdata/invalid_destination_node.json",
+ "testdata/invalid_nodes.json",
+ "testdata/invalid_source_node.json",
+ "testdata/self_forward.json",
+ "testdata/expected_multinode.json",
+ "testdata/config1_multinode.json",
+ "testdata/config2_multinode.json",
"//aos/events:pingpong_config.json",
"//aos/events:pong.bfbs",
],
diff --git a/aos/config_flattener.cc b/aos/config_flattener.cc
index 71cda32..6fb4af3 100644
--- a/aos/config_flattener.cc
+++ b/aos/config_flattener.cc
@@ -27,7 +27,8 @@
&configuration::MergeConfiguration(config, schemas).message(), true);
// TODO(austin): Figure out how to squash the schemas onto 1 line so it is
- // easier to read?
+ // easier to read? Or figure out how to split them into a second file which
+ // gets included.
VLOG(1) << "Flattened config is " << merged_config;
util::WriteStringToFileOrDie(argv[1], merged_config);
return 0;
diff --git a/aos/configuration.cc b/aos/configuration.cc
index c7fa2a3..f08cf99 100644
--- a/aos/configuration.cc
+++ b/aos/configuration.cc
@@ -9,14 +9,21 @@
#include <unistd.h>
#include <string_view>
+#include "absl/base/call_once.h"
#include "absl/container/btree_set.h"
#include "aos/configuration_generated.h"
#include "aos/flatbuffer_merge.h"
#include "aos/json_to_flatbuffer.h"
+#include "aos/network/team_number.h"
#include "aos/unique_malloc_ptr.h"
#include "aos/util/file.h"
+#include "gflags/gflags.h"
#include "glog/logging.h"
-#include "absl/base/call_once.h"
+
+DEFINE_string(
+ override_hostname, "",
+ "If set, this forces the hostname of this node to be the provided "
+ "hostname.");
namespace aos {
@@ -56,50 +63,21 @@
rhs.message().name()->string_view();
}
+bool operator==(const FlatbufferDetachedBuffer<Node> &lhs,
+ const FlatbufferDetachedBuffer<Node> &rhs) {
+ return lhs.message().name()->string_view() ==
+ rhs.message().name()->string_view();
+}
+
+bool operator<(const FlatbufferDetachedBuffer<Node> &lhs,
+ const FlatbufferDetachedBuffer<Node> &rhs) {
+ return lhs.message().name()->string_view() <
+ rhs.message().name()->string_view();
+}
+
namespace configuration {
namespace {
-// TODO(brians): This shouldn't be necesary for running tests. Provide a way to
-// set the IP address when running tests from the test.
-const char *const kLinuxNetInterface = "eth0";
-
-void DoGetOwnIPAddress(in_addr *retu) {
- static const char *kOverrideVariable = "FRC971_IP_OVERRIDE";
- const char *override_ip = getenv(kOverrideVariable);
- if (override_ip != NULL) {
- LOG(INFO) << "Override IP is " << override_ip;
- if (inet_aton(override_ip, retu) != 0) {
- return;
- } else {
- LOG(WARNING) << "error parsing " << kOverrideVariable << " value '"
- << override_ip << "'";
- }
- } else {
- LOG(INFO) << "Couldn't get environmental variable.";
- }
-
- ifaddrs *addrs;
- if (getifaddrs(&addrs) != 0) {
- PLOG(FATAL) << "getifaddrs(" << &addrs << ") failed";
- }
- // Smart pointers don't work very well for iterating through a linked list,
- // but it does do a very nice job of making sure that addrs gets freed.
- unique_c_ptr<ifaddrs, freeifaddrs> addrs_deleter(addrs);
-
- for (; addrs != nullptr; addrs = addrs->ifa_next) {
- // ifa_addr tends to be nullptr on CAN interfaces.
- if (addrs->ifa_addr != nullptr && addrs->ifa_addr->sa_family == AF_INET) {
- if (strcmp(kLinuxNetInterface, addrs->ifa_name) == 0) {
- *retu = reinterpret_cast<sockaddr_in *>(__builtin_assume_aligned(
- addrs->ifa_addr, alignof(sockaddr_in)))->sin_addr;
- return;
- }
- }
- }
- LOG(FATAL) << "couldn't find an AF_INET interface named \""
- << kLinuxNetInterface << "\"";
-}
-
void DoGetRootDirectory(char** retu) {
ssize_t size = 0;
*retu = NULL;
@@ -310,6 +288,21 @@
}
}
+ // Now repeat this for the node list.
+ absl::btree_set<FlatbufferDetachedBuffer<Node>> nodes;
+ if (config.message().has_nodes()) {
+ for (const Node *n : *config.message().nodes()) {
+ if (!n->has_name()) {
+ continue;
+ }
+
+ auto result = nodes.insert(CopyFlatBuffer(n));
+ if (!result.second) {
+ *result.first = MergeFlatBuffers(*result.first, CopyFlatBuffer(n));
+ }
+ }
+ }
+
flatbuffers::FlatBufferBuilder fbb;
fbb.ForceDefaults(1);
@@ -351,39 +344,79 @@
}
}
+ // Nodes
+ flatbuffers::Offset<flatbuffers::Vector<flatbuffers::Offset<Node>>>
+ nodes_offset;
+ {
+ ::std::vector<flatbuffers::Offset<Node>> node_offsets;
+ for (const FlatbufferDetachedBuffer<Node> &n : nodes) {
+ node_offsets.emplace_back(CopyFlatBuffer<Node>(&n.message(), &fbb));
+ }
+ nodes_offset = fbb.CreateVector(node_offsets);
+ }
+
// And then build a Configuration with them all.
ConfigurationBuilder configuration_builder(fbb);
configuration_builder.add_channels(channels_offset);
if (config.message().has_maps()) {
configuration_builder.add_maps(maps_offset);
}
- configuration_builder.add_applications(applications_offset);
+ if (config.message().has_applications()) {
+ configuration_builder.add_applications(applications_offset);
+ }
+ if (config.message().has_nodes()) {
+ configuration_builder.add_nodes(nodes_offset);
+ }
fbb.Finish(configuration_builder.Finish());
- return fbb.Release();
+
+ // Now, validate that if there is a node list, every channel has a source
+ // node.
+ FlatbufferDetachedBuffer<Configuration> result(fbb.Release());
+
+ // Check that if there is a node list, all the source nodes are filled out and
+ // valid, and all the destination nodes are valid (and not the source). This
+ // is a basic consistency check.
+ if (result.message().has_nodes()) {
+ for (const Channel *c : *config.message().channels()) {
+ CHECK(c->has_source_node()) << ": Channel " << FlatbufferToJson(c)
+ << " is missing \"source_node\"";
+ CHECK(GetNode(&result.message(), c->source_node()->string_view()) !=
+ nullptr)
+ << ": Channel " << FlatbufferToJson(c)
+ << " has an unknown \"source_node\"";
+
+ if (c->has_destination_nodes()) {
+ for (const flatbuffers::String *n : *c->destination_nodes()) {
+ CHECK(GetNode(&result.message(), n->string_view()) != nullptr)
+ << ": Channel " << FlatbufferToJson(c)
+ << " has an unknown \"destination_nodes\" " << n->string_view();
+
+ CHECK_NE(n->string_view(), c->source_node()->string_view())
+ << ": Channel " << FlatbufferToJson(c)
+ << " is forwarding data to itself";
+ }
+ }
+ }
+ }
+
+ return result;
}
const char *GetRootDirectory() {
- static char* root_dir;// return value
+ static char *root_dir; // return value
static absl::once_flag once_;
absl::call_once(once_, DoGetRootDirectory, &root_dir);
return root_dir;
}
const char *GetLoggingDirectory() {
- static char* retu;// return value
+ static char *retu; // return value
static absl::once_flag once_;
absl::call_once(once_, DoGetLoggingDirectory, &retu);
return retu;
}
-const in_addr &GetOwnIPAddress() {
- static in_addr retu;// return value
- static absl::once_flag once_;
- absl::call_once(once_, DoGetOwnIPAddress, &retu);
- return retu;
-}
-
FlatbufferDetachedBuffer<Configuration> ReadConfig(
const std::string_view path) {
// We only want to read a file once. So track the visited files in a set.
@@ -416,7 +449,7 @@
HandleMaps(config->maps(), &name);
}
- VLOG(1) << "Acutally looking up { \"name\": \"" << name << "\", \"type\": \""
+ VLOG(1) << "Actually looking up { \"name\": \"" << name << "\", \"type\": \""
<< type << "\" }";
// Then look for the channel.
@@ -512,7 +545,19 @@
}
}
- // Now insert eerything else in unmodified.
+ flatbuffers::Offset<flatbuffers::Vector<flatbuffers::Offset<Node>>>
+ nodes_offset;
+ {
+ ::std::vector<flatbuffers::Offset<Node>> node_offsets;
+ if (config.message().has_nodes()) {
+ for (const Node *n : *config.message().nodes()) {
+ node_offsets.emplace_back(CopyFlatBuffer<Node>(n, &fbb));
+ }
+ nodes_offset = fbb.CreateVector(node_offsets);
+ }
+ }
+
+ // Now insert everything else in unmodified.
ConfigurationBuilder configuration_builder(fbb);
if (config.message().has_channels()) {
configuration_builder.add_channels(channels_offset);
@@ -523,10 +568,66 @@
if (config.message().has_applications()) {
configuration_builder.add_applications(applications_offset);
}
+ if (config.message().has_nodes()) {
+ configuration_builder.add_nodes(nodes_offset);
+ }
fbb.Finish(configuration_builder.Finish());
return fbb.Release();
}
+const Node *GetNodeFromHostname(const Configuration *config,
+ std::string_view hostname) {
+ for (const Node *node : *config->nodes()) {
+ if (node->hostname()->string_view() == hostname) {
+ return node;
+ }
+ }
+ return nullptr;
+}
+const Node *GetMyNode(const Configuration *config) {
+ const std::string hostname = (FLAGS_override_hostname.size() > 0)
+ ? FLAGS_override_hostname
+ : network::GetHostname();
+ const Node *node = GetNodeFromHostname(config, hostname);
+ if (node != nullptr) return node;
+
+ LOG(FATAL) << "Unknown node for host: " << hostname
+ << ". Consider using --override_hostname if hostname detection "
+ "is wrong.";
+ return nullptr;
+}
+
+const Node *GetNode(const Configuration *config, std::string_view name) {
+ for (const Node *node : *config->nodes()) {
+ if (node->name()->string_view() == name) {
+ return node;
+ }
+ }
+ return nullptr;
+}
+
+bool ChannelIsSendableOnNode(const Channel *channel, const Node *node) {
+ return (channel->source_node()->string_view() == node->name()->string_view());
+}
+
+bool ChannelIsReadableOnNode(const Channel *channel, const Node *node) {
+ if (channel->source_node()->string_view() == node->name()->string_view()) {
+ return true;
+ }
+
+ if (!channel->has_destination_nodes()) {
+ return false;
+ }
+
+ for (const flatbuffers::String *s : *channel->destination_nodes()) {
+ if (s->string_view() == node->name()->string_view()) {
+ return true;
+ }
+ }
+
+ return false;
+}
+
} // namespace configuration
} // namespace aos
diff --git a/aos/configuration.fbs b/aos/configuration.fbs
index a9e71ea..ef25959 100644
--- a/aos/configuration.fbs
+++ b/aos/configuration.fbs
@@ -22,6 +22,14 @@
// The schema for the data sent on this channel.
schema:reflection.Schema;
+
+ // The source node name for the data sent on this channel.
+ // If nodes is populated below, this needs to also be populated.
+ source_node:string;
+
+ // The destination node names for data sent on this channel.
+ // This only needs to be populated if this message is getting forwarded.
+ destination_nodes:[string];
}
// Table to support renaming channel names.
@@ -53,6 +61,22 @@
//
// will map "/foo" to "/baz", even if there is a global list of maps.
maps:[Map];
+
+ // The node that this application will be started on.
+ // TODO(austin): Teach starter how to use this for starting applications.
+ node:string;
+}
+
+// Per node data and connection information.
+table Node {
+ // Short name for the node. This provides a short hand to make it easy to
+ // setup forwarding rules as part of the channel setup.
+ name:string;
+
+ // Hostname used to identify and connect to the node.
+ hostname:string;
+ // Port to serve forwarded data from.
+ port:ushort = 9971;
}
// Overall configuration datastructure for the pubsub.
@@ -61,6 +85,11 @@
channels:[Channel] (id: 0);
// List of global maps. These are applied in reverse order.
maps:[Map] (id: 1);
+
+ // If present, this is the list of nodes in the system. If this is not
+ // present, AOS will be running in a single node configuration.
+ nodes:[Node] (id: 4);
+
// List of applications.
applications:[Application] (id: 2);
// List of imports. Imports are loaded first, and then this configuration
diff --git a/aos/configuration.h b/aos/configuration.h
index b837391..5d4a79b 100644
--- a/aos/configuration.h
+++ b/aos/configuration.h
@@ -49,10 +49,22 @@
return GetChannel(&config.message(), name, type, application_name);
}
-// TODO(austin): GetSchema<T>(const Flatbuffer<Configuration> &config);
+// Returns the Node out of the config with the matching name, or nullptr if it
+// can't be found.
+const Node *GetNode(const Configuration *config, std::string_view name);
+// Returns the Node out of the configuration which matches our hostname.
+// CHECKs if it can't be found.
+const Node *GetMyNode(const Configuration *config);
+const Node *GetNodeFromHostname(const Configuration *config,
+ std::string_view name);
-// Returns "our" IP address.
-const in_addr &GetOwnIPAddress();
+// Returns true if the provided channel is sendable on the provided node.
+bool ChannelIsSendableOnNode(const Channel *channel, const Node *node);
+// Returns true if the provided channel is able to be watched or fetched on the
+// provided node.
+bool ChannelIsReadableOnNode(const Channel *channel, const Node *node);
+
+// TODO(austin): GetSchema<T>(const Flatbuffer<Configuration> &config);
// Returns the "root directory" for this run. Under linux, this is the
// directory where the executable is located (from /proc/self/exe)
diff --git a/aos/configuration_test.cc b/aos/configuration_test.cc
index e20b308..35b9daf 100644
--- a/aos/configuration_test.cc
+++ b/aos/configuration_test.cc
@@ -35,6 +35,19 @@
FlatbufferToJson(config, true));
}
+// Tests that we can read and merge a multinode configuration.
+TEST_F(ConfigurationTest, ConfigMergeMultinode) {
+ FlatbufferDetachedBuffer<Configuration> config =
+ ReadConfig("aos/testdata/config1_multinode.json");
+ LOG(INFO) << "Read: " << FlatbufferToJson(config, true);
+
+ EXPECT_EQ(
+ std::string(absl::StripSuffix(
+ util::ReadFileToStringOrDie("aos/testdata/expected_multinode.json"),
+ "\n")),
+ FlatbufferToJson(config, true));
+}
+
// Tests that we sort the entries in a config so we can look entries up.
TEST_F(ConfigurationTest, UnsortedConfig) {
FlatbufferDetachedBuffer<Configuration> config =
@@ -88,6 +101,116 @@
kExpectedLocation);
}
+// Tests that we reject a configuration which has a nodes list, but has channels
+// withoout source_node filled out.
+TEST_F(ConfigurationDeathTest, InvalidSourceNode) {
+ EXPECT_DEATH(
+ {
+ FlatbufferDetachedBuffer<Configuration> config =
+ ReadConfig("aos/testdata/invalid_nodes.json");
+ },
+ "source_node");
+
+ EXPECT_DEATH(
+ {
+ FlatbufferDetachedBuffer<Configuration> config =
+ ReadConfig("aos/testdata/invalid_source_node.json");
+ },
+ "source_node");
+
+ EXPECT_DEATH(
+ {
+ FlatbufferDetachedBuffer<Configuration> config =
+ ReadConfig("aos/testdata/invalid_destination_node.json");
+ },
+ "destination_nodes");
+
+ EXPECT_DEATH(
+ {
+ FlatbufferDetachedBuffer<Configuration> config =
+ ReadConfig("aos/testdata/self_forward.json");
+ },
+ "forwarding data to itself");
+}
+
+// Tests that our node writeable helpers work as intended.
+TEST_F(ConfigurationTest, ChannelIsSendableOnNode) {
+ FlatbufferDetachedBuffer<Channel> good_channel(JsonToFlatbuffer(
+R"channel({
+ "name": "/test",
+ "type": "aos.examples.Ping",
+ "source_node": "foo"
+})channel",
+ Channel::MiniReflectTypeTable()));
+
+ FlatbufferDetachedBuffer<Channel> bad_channel(JsonToFlatbuffer(
+R"channel({
+ "name": "/test",
+ "type": "aos.examples.Ping",
+ "source_node": "bar"
+})channel",
+ Channel::MiniReflectTypeTable()));
+
+ FlatbufferDetachedBuffer<Node> node(JsonToFlatbuffer(
+R"node({
+ "name": "foo"
+})node",
+ Node::MiniReflectTypeTable()));
+
+ EXPECT_TRUE(
+ ChannelIsSendableOnNode(&good_channel.message(), &node.message()));
+ EXPECT_FALSE(
+ ChannelIsSendableOnNode(&bad_channel.message(), &node.message()));
+}
+
+// Tests that our node readable and writeable helpers work as intended.
+TEST_F(ConfigurationTest, ChannelIsReadableOnNode) {
+ FlatbufferDetachedBuffer<Channel> good_channel(JsonToFlatbuffer(
+R"channel({
+ "name": "/test",
+ "type": "aos.examples.Ping",
+ "source_node": "bar",
+ "destination_nodes": [
+ "baz",
+ "foo",
+ ]
+})channel",
+ Channel::MiniReflectTypeTable()));
+
+ FlatbufferDetachedBuffer<Channel> bad_channel1(JsonToFlatbuffer(
+R"channel({
+ "name": "/test",
+ "type": "aos.examples.Ping",
+ "source_node": "bar"
+})channel",
+ Channel::MiniReflectTypeTable()));
+
+ FlatbufferDetachedBuffer<Channel> bad_channel2(JsonToFlatbuffer(
+R"channel({
+ "name": "/test",
+ "type": "aos.examples.Ping",
+ "source_node": "bar",
+ "destination_nodes": [
+ "baz"
+ ]
+})channel",
+ Channel::MiniReflectTypeTable()));
+
+ FlatbufferDetachedBuffer<Node> node(JsonToFlatbuffer(
+R"node({
+ "name": "foo"
+})node",
+ Node::MiniReflectTypeTable()));
+
+ EXPECT_TRUE(
+ ChannelIsReadableOnNode(&good_channel.message(), &node.message()));
+ EXPECT_FALSE(
+ ChannelIsReadableOnNode(&bad_channel1.message(), &node.message()));
+ EXPECT_FALSE(
+ ChannelIsReadableOnNode(&bad_channel2.message(), &node.message()));
+}
+
+
} // namespace testing
} // namespace configuration
} // namespace aos
diff --git a/aos/events/event_loop.h b/aos/events/event_loop.h
index 68b1b5e..4cbb07b 100644
--- a/aos/events/event_loop.h
+++ b/aos/events/event_loop.h
@@ -318,6 +318,16 @@
<< ": Channel { \"name\": \"" << channel_name << "\", \"type\": \""
<< T::GetFullyQualifiedName() << "\" } not found in config.";
+ if (node() != nullptr) {
+ if (!configuration::ChannelIsReadableOnNode(channel, node())) {
+ LOG(FATAL)
+ << "Channel { \"name\": \"" << channel_name << "\", \"type\": \""
+ << T::GetFullyQualifiedName()
+ << "\" } is not able to be fetched on this node. Check your "
+ "configuration.";
+ }
+ }
+
return Fetcher<T>(MakeRawFetcher(channel));
}
@@ -331,6 +341,15 @@
<< ": Channel { \"name\": \"" << channel_name << "\", \"type\": \""
<< T::GetFullyQualifiedName() << "\" } not found in config.";
+ if (node() != nullptr) {
+ if (!configuration::ChannelIsSendableOnNode(channel, node())) {
+ LOG(FATAL) << "Channel { \"name\": \"" << channel_name
+ << "\", \"type\": \"" << T::GetFullyQualifiedName()
+ << "\" } is not able to be sent on this node. Check your "
+ "configuration.";
+ }
+ }
+
return Sender<T>(MakeRawSender(channel));
}
@@ -348,11 +367,13 @@
// Use this to run code once the thread goes into "real-time-mode",
virtual void OnRun(::std::function<void()> on_run) = 0;
- // Sets the name of the event loop. This is the application name.
- virtual void set_name(const std::string_view name) = 0;
- // Gets the name of the event loop.
+ // Gets the name of the event loop. This is the application name.
virtual const std::string_view name() const = 0;
+ // Returns the node that this event loop is running on. Returns nullptr if we
+ // are running in single-node mode.
+ virtual const Node *node() const = 0;
+
// Creates a timer that executes callback when the timer expires
// Returns a TimerHandle for configuration of the timer
virtual TimerHandler *AddTimer(::std::function<void()> callback) = 0;
@@ -365,7 +386,7 @@
const monotonic_clock::duration interval,
const monotonic_clock::duration offset = ::std::chrono::seconds(0)) = 0;
- // TODO(austin): OnExit
+ // TODO(austin): OnExit for cleanup.
// Threadsafe.
bool is_running() const { return is_running_.load(); }
@@ -375,31 +396,37 @@
virtual void SetRuntimeRealtimePriority(int priority) = 0;
virtual int priority() const = 0;
- // Fetches new messages from the provided channel (path, type). Note: this
- // channel must be a member of the exact configuration object this was built
- // with.
+ // Fetches new messages from the provided channel (path, type).
+ //
+ // Note: this channel must be a member of the exact configuration object this
+ // was built with.
virtual std::unique_ptr<RawFetcher> MakeRawFetcher(
const Channel *channel) = 0;
- // Will watch channel (name, type) for new messages
+ // Watches channel (name, type) for new messages.
virtual void MakeRawWatcher(
const Channel *channel,
std::function<void(const Context &context, const void *message)>
watcher) = 0;
+ // 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.
+ virtual std::unique_ptr<RawSender> MakeRawSender(const Channel *channel) = 0;
+
// Returns the context for the current callback.
const Context &context() const { return context_; }
// Returns the configuration that this event loop was built with.
const Configuration *configuration() const { return configuration_; }
- // Will send new messages from channel (path, type).
- virtual std::unique_ptr<RawSender> MakeRawSender(const Channel *channel) = 0;
-
// Prevents the event loop from sending a timing report.
void SkipTimingReport() { skip_timing_report_ = true; }
protected:
+ // Sets the name of the event loop. This is the application name.
+ virtual void set_name(const std::string_view name) = 0;
+
void set_is_running(bool value) { is_running_.store(value); }
// Validates that channel exists inside configuration_ and finds its index.
diff --git a/aos/events/event_loop_param_test.cc b/aos/events/event_loop_param_test.cc
index cf3b6df..c1256f3 100644
--- a/aos/events/event_loop_param_test.cc
+++ b/aos/events/event_loop_param_test.cc
@@ -1145,5 +1145,100 @@
EXPECT_TRUE(happened);
}
+// Tests that not setting up nodes results in no node.
+TEST_P(AbstractEventLoopTest, NoNode) {
+ auto loop1 = Make();
+ auto loop2 = MakePrimary();
+
+ EXPECT_EQ(loop1->node(), nullptr);
+ EXPECT_EQ(loop2->node(), nullptr);
+}
+
+// Tests that setting up nodes results in node being set.
+TEST_P(AbstractEventLoopTest, Node) {
+ EnableNodes("me");
+
+ auto loop1 = Make();
+ auto loop2 = MakePrimary();
+
+ EXPECT_NE(loop1->node(), nullptr);
+ EXPECT_NE(loop2->node(), nullptr);
+}
+
+// Tests that watchers work with a node setup.
+TEST_P(AbstractEventLoopTest, NodeWatcher) {
+ EnableNodes("me");
+
+ auto loop1 = Make();
+ auto loop2 = Make();
+ loop1->MakeWatcher("/test", [](const TestMessage &) {});
+ loop2->MakeRawWatcher(configuration()->channels()->Get(1),
+ [](const Context &, const void *) {});
+}
+
+// Tests that fetcher work with a node setup.
+TEST_P(AbstractEventLoopTest, NodeFetcher) {
+ EnableNodes("me");
+ auto loop1 = Make();
+
+ auto fetcher = loop1->MakeFetcher<TestMessage>("/test");
+ auto raw_fetcher = loop1->MakeRawFetcher(configuration()->channels()->Get(1));
+}
+
+// Tests that sender work with a node setup.
+TEST_P(AbstractEventLoopTest, NodeSender) {
+ EnableNodes("me");
+ auto loop1 = Make();
+
+ aos::Sender<TestMessage> sender = loop1->MakeSender<TestMessage>("/test");
+}
+
+// Tests that watchers fail when created on the wrong node.
+TEST_P(AbstractEventLoopDeathTest, NodeWatcher) {
+ EnableNodes("them");
+
+ auto loop1 = Make();
+ auto loop2 = Make();
+ EXPECT_DEATH({ loop1->MakeWatcher("/test", [](const TestMessage &) {}); },
+ "node");
+ EXPECT_DEATH(
+ {
+ loop2->MakeRawWatcher(configuration()->channels()->Get(1),
+ [](const Context &, const void *) {});
+ },
+ "node");
+}
+
+// Tests that fetchers fail when created on the wrong node.
+TEST_P(AbstractEventLoopDeathTest, NodeFetcher) {
+ EnableNodes("them");
+ auto loop1 = Make();
+
+ EXPECT_DEATH({ auto fetcher = loop1->MakeFetcher<TestMessage>("/test"); },
+ "node");
+ EXPECT_DEATH(
+ {
+ auto raw_fetcher =
+ loop1->MakeRawFetcher(configuration()->channels()->Get(1));
+ },
+ "node");
+}
+
+// Tests that senders fail when created on the wrong node.
+TEST_P(AbstractEventLoopDeathTest, NodeSender) {
+ EnableNodes("them");
+ auto loop1 = Make();
+
+ EXPECT_DEATH(
+ {
+ aos::Sender<TestMessage> sender =
+ loop1->MakeSender<TestMessage>("/test");
+ },
+ "node");
+
+ // Note: Creating raw senders is always supported. Right now, this lets us
+ // use them to create message_gateway.
+}
+
} // namespace testing
} // namespace aos
diff --git a/aos/events/event_loop_param_test.h b/aos/events/event_loop_param_test.h
index f4cf2ec..ea5fd3d 100644
--- a/aos/events/event_loop_param_test.h
+++ b/aos/events/event_loop_param_test.h
@@ -54,10 +54,53 @@
// Advances time by sleeping. Can't be called from inside a loop.
virtual void SleepFor(::std::chrono::nanoseconds duration) = 0;
+ void EnableNodes(std::string_view my_node) {
+ std::string json = std::string(R"config({
+ "channels": [
+ {
+ "name": "/aos",
+ "type": "aos.timing.Report",
+ "source_node": "me"
+ },
+ {
+ "name": "/test",
+ "type": "aos.TestMessage",
+ "source_node": "me"
+ },
+ {
+ "name": "/test1",
+ "type": "aos.TestMessage",
+ "source_node": "me"
+ },
+ {
+ "name": "/test2",
+ "type": "aos.TestMessage",
+ "source_node": "me"
+ }
+ ],
+ "nodes": [
+ {
+ "name": ")config") +
+ std::string(my_node) + R"config(",
+ "hostname": "myhostname"
+ }
+ ]
+})config";
+
+ flatbuffer_ = FlatbufferDetachedBuffer<Configuration>(
+ JsonToFlatbuffer(json, Configuration::MiniReflectTypeTable()));
+
+ my_node_ = my_node;
+ }
+
+ std::string_view my_node() const { return my_node_; }
+
const Configuration *configuration() { return &flatbuffer_.message(); }
private:
FlatbufferDetachedBuffer<Configuration> flatbuffer_;
+
+ std::string my_node_;
};
class AbstractEventLoopTestBase
@@ -79,6 +122,8 @@
return factory_->MakePrimary(name);
}
+ void EnableNodes(std::string_view my_node) { factory_->EnableNodes(my_node); }
+
void Run() { return factory_->Run(); }
void Exit() { return factory_->Exit(); }
@@ -87,6 +132,10 @@
return factory_->SleepFor(duration);
}
+ const Configuration *configuration() { return factory_->configuration(); }
+
+ std::string_view my_node() const { return factory_->my_node(); }
+
// Ends the given event loop at the given time from now.
void EndEventLoop(EventLoop *loop, ::std::chrono::milliseconds duration) {
auto end_timer = loop->AddTimer([this]() { this->Exit(); });
diff --git a/aos/events/event_loop_tmpl.h b/aos/events/event_loop_tmpl.h
index f652bb6..9e5b05d 100644
--- a/aos/events/event_loop_tmpl.h
+++ b/aos/events/event_loop_tmpl.h
@@ -36,6 +36,15 @@
<< ": Channel { \"name\": \"" << channel_name << "\", \"type\": \""
<< T::GetFullyQualifiedName() << "\" } not found in config.";
+ if (node() != nullptr) {
+ if (!configuration::ChannelIsReadableOnNode(channel, node())) {
+ LOG(FATAL) << "Channel { \"name\": \"" << channel_name
+ << "\", \"type\": \"" << T::GetFullyQualifiedName()
+ << "\" } is not able to be watched on this node. Check your "
+ "configuration.";
+ }
+ }
+
return MakeRawWatcher(
channel, [this, w](const Context &context, const void *message) {
context_ = context;
diff --git a/aos/events/shm_event_loop.cc b/aos/events/shm_event_loop.cc
index e549d57..e696f36 100644
--- a/aos/events/shm_event_loop.cc
+++ b/aos/events/shm_event_loop.cc
@@ -126,6 +126,8 @@
void *data_;
};
+namespace {
+
// Returns the portion of the path after the last /.
std::string_view Filename(std::string_view path) {
auto last_slash_pos = path.find_last_of("/");
@@ -135,15 +137,23 @@
: path.substr(last_slash_pos + 1, path.size());
}
-ShmEventLoop::ShmEventLoop(const Configuration *configuration)
- : EventLoop(configuration), name_(Filename(program_invocation_name)) {}
+const Node *MaybeMyNode(const Configuration *configuration) {
+ if (!configuration->has_nodes()) {
+ return nullptr;
+ }
-namespace {
+ return configuration::GetMyNode(configuration);
+}
namespace chrono = ::std::chrono;
} // namespace
+ShmEventLoop::ShmEventLoop(const Configuration *configuration)
+ : EventLoop(configuration),
+ name_(Filename(program_invocation_name)),
+ node_(MaybeMyNode(configuration)) {}
+
namespace internal {
class SimpleShmFetcher {
@@ -508,6 +518,16 @@
::std::unique_ptr<RawFetcher> ShmEventLoop::MakeRawFetcher(
const Channel *channel) {
+
+ if (node() != nullptr) {
+ if (!configuration::ChannelIsReadableOnNode(channel, node())) {
+ LOG(FATAL) << "Channel { \"name\": \"" << channel->name()->string_view()
+ << "\", \"type\": \"" << channel->type()->string_view()
+ << "\" } is not able to be fetched on this node. Check your "
+ "configuration.";
+ }
+ }
+
return ::std::unique_ptr<RawFetcher>(new internal::ShmFetcher(this, channel));
}
@@ -523,6 +543,15 @@
std::function<void(const Context &context, const void *message)> watcher) {
Take(channel);
+ if (node() != nullptr) {
+ if (!configuration::ChannelIsReadableOnNode(channel, node())) {
+ LOG(FATAL) << "Channel { \"name\": \"" << channel->name()->string_view()
+ << "\", \"type\": \"" << channel->type()->string_view()
+ << "\" } is not able to be watched on this node. Check your "
+ "configuration.";
+ }
+ }
+
NewWatcher(::std::unique_ptr<WatcherState>(
new internal::WatcherState(this, channel, std::move(watcher))));
}
diff --git a/aos/events/shm_event_loop.h b/aos/events/shm_event_loop.h
index 5063186..d10989e 100644
--- a/aos/events/shm_event_loop.h
+++ b/aos/events/shm_event_loop.h
@@ -67,6 +67,7 @@
UpdateTimingReport();
}
const std::string_view name() const override { return name_; }
+ const Node *node() const override { return node_; }
int priority() const override { return priority_; }
@@ -89,6 +90,7 @@
std::vector<std::function<void()>> on_run_;
int priority_ = 0;
std::string name_;
+ const Node *const node_;
std::vector<std::string> taken_;
internal::EPoll epoll_;
diff --git a/aos/events/shm_event_loop_test.cc b/aos/events/shm_event_loop_test.cc
index 1bbca2c..f2c730b 100644
--- a/aos/events/shm_event_loop_test.cc
+++ b/aos/events/shm_event_loop_test.cc
@@ -9,6 +9,7 @@
#include "aos/events/test_message_generated.h"
DECLARE_string(shm_base);
+DECLARE_string(override_hostname);
namespace aos {
namespace testing {
@@ -33,13 +34,21 @@
unlink((FLAGS_shm_base + "/aos/aos.timing.Report.v0").c_str());
}
+ ~ShmEventLoopTestFactory() { FLAGS_override_hostname = ""; }
+
::std::unique_ptr<EventLoop> Make(std::string_view name) override {
- ::std::unique_ptr<EventLoop> loop(new ShmEventLoop(configuration()));
+ if (configuration()->has_nodes()) {
+ FLAGS_override_hostname = "myhostname";
+ }
+ ::std::unique_ptr<ShmEventLoop> loop(new ShmEventLoop(configuration()));
loop->set_name(name);
- return loop;
+ return std::move(loop);
}
::std::unique_ptr<EventLoop> MakePrimary(std::string_view name) override {
+ if (configuration()->has_nodes()) {
+ FLAGS_override_hostname = "myhostname";
+ }
::std::unique_ptr<ShmEventLoop> loop =
::std::unique_ptr<ShmEventLoop>(new ShmEventLoop(configuration()));
primary_event_loop_ = loop.get();
diff --git a/aos/events/simulated_event_loop.cc b/aos/events/simulated_event_loop.cc
index 29c9edd..cd04c9f 100644
--- a/aos/events/simulated_event_loop.cc
+++ b/aos/events/simulated_event_loop.cc
@@ -304,11 +304,12 @@
const Configuration *configuration,
std::vector<std::pair<EventLoop *, std::function<void(bool)>>>
*raw_event_loops,
- pid_t tid)
+ const Node *node, pid_t tid)
: EventLoop(CHECK_NOTNULL(configuration)),
scheduler_(scheduler),
channels_(channels),
raw_event_loops_(raw_event_loops),
+ node_(node),
tid_(tid) {
raw_event_loops_->push_back(std::make_pair(this, [this](bool value) {
if (!has_setup_) {
@@ -378,6 +379,8 @@
scheduler_->ScheduleOnRun(on_run);
}
+ const Node *node() const override { return node_; }
+
void set_name(const std::string_view name) override {
name_ = std::string(name);
}
@@ -428,6 +431,7 @@
std::chrono::nanoseconds send_delay_;
+ const Node *const node_;
const pid_t tid_;
};
@@ -446,6 +450,16 @@
std::function<void(const Context &channel, const void *message)> watcher) {
ChannelIndex(channel);
Take(channel);
+
+ if (node() != nullptr) {
+ if (!configuration::ChannelIsReadableOnNode(channel, node())) {
+ LOG(FATAL) << "Channel { \"name\": \"" << channel->name()->string_view()
+ << "\", \"type\": \"" << channel->type()->string_view()
+ << "\" } is not able to be watched on this node. Check your "
+ "configuration.";
+ }
+ }
+
std::unique_ptr<SimulatedWatcher> shm_watcher(
new SimulatedWatcher(this, scheduler_, channel, std::move(watcher)));
@@ -463,6 +477,16 @@
std::unique_ptr<RawFetcher> SimulatedEventLoop::MakeRawFetcher(
const Channel *channel) {
ChannelIndex(channel);
+
+ if (node() != nullptr) {
+ if (!configuration::ChannelIsReadableOnNode(channel, node())) {
+ LOG(FATAL) << "Channel { \"name\": \"" << channel->name()->string_view()
+ << "\", \"type\": \"" << channel->type()->string_view()
+ << "\" } is not able to be fetched on this node. Check your "
+ "configuration.";
+ }
+ }
+
return GetSimulatedChannel(channel)->MakeRawFetcher(this);
}
@@ -676,7 +700,22 @@
SimulatedEventLoopFactory::SimulatedEventLoopFactory(
const Configuration *configuration)
- : configuration_(CHECK_NOTNULL(configuration)) {}
+ : configuration_(CHECK_NOTNULL(configuration)), node_(nullptr) {
+ CHECK(!configuration_->has_nodes())
+ << ": Got a configuration with multiple nodes and no node was selected.";
+}
+
+SimulatedEventLoopFactory::SimulatedEventLoopFactory(
+ const Configuration *configuration, std::string_view node_name)
+ : configuration_(CHECK_NOTNULL(configuration)),
+ node_(configuration::GetNode(configuration, node_name)) {
+ CHECK(configuration_->has_nodes())
+ << ": Got a configuration with no nodes and node \"" << node_name
+ << "\" was selected.";
+ CHECK(node_ != nullptr) << ": Can't find node \"" << node_name
+ << "\" in the configuration.";
+}
+
SimulatedEventLoopFactory::~SimulatedEventLoopFactory() {}
::std::unique_ptr<EventLoop> SimulatedEventLoopFactory::MakeEventLoop(
@@ -684,7 +723,7 @@
pid_t tid = tid_;
++tid_;
::std::unique_ptr<SimulatedEventLoop> result(new SimulatedEventLoop(
- &scheduler_, &channels_, configuration_, &raw_event_loops_, tid));
+ &scheduler_, &channels_, configuration_, &raw_event_loops_, node_, tid));
result->set_name(name);
result->set_send_delay(send_delay_);
return std::move(result);
diff --git a/aos/events/simulated_event_loop.h b/aos/events/simulated_event_loop.h
index 7b2b9c7..740a4c2 100644
--- a/aos/events/simulated_event_loop.h
+++ b/aos/events/simulated_event_loop.h
@@ -52,6 +52,8 @@
// This configuration must remain in scope for the lifetime of the factory and
// all sub-objects.
SimulatedEventLoopFactory(const Configuration *configuration);
+ SimulatedEventLoopFactory(const Configuration *configuration,
+ std::string_view node_name);
~SimulatedEventLoopFactory();
::std::unique_ptr<EventLoop> MakeEventLoop(std::string_view name);
@@ -68,6 +70,10 @@
// Sets the simulated send delay for the factory.
void set_send_delay(std::chrono::nanoseconds send_delay);
+ // Returns the node that this factory is running as, or nullptr if this is a
+ // single node setup.
+ const Node *node() const { return node_; }
+
monotonic_clock::time_point monotonic_now() const {
return scheduler_.monotonic_now();
}
@@ -76,7 +82,7 @@
}
private:
- const Configuration *configuration_;
+ const Configuration *const configuration_;
EventScheduler scheduler_;
// Map from name, type to queue.
absl::btree_map<SimpleChannel, std::unique_ptr<SimulatedChannel>> channels_;
@@ -86,6 +92,8 @@
std::chrono::nanoseconds send_delay_ = std::chrono::microseconds(50);
+ const Node *const node_;
+
pid_t tid_ = 0;
};
diff --git a/aos/events/simulated_event_loop_test.cc b/aos/events/simulated_event_loop_test.cc
index 44be581..4328d6f 100644
--- a/aos/events/simulated_event_loop_test.cc
+++ b/aos/events/simulated_event_loop_test.cc
@@ -13,28 +13,40 @@
class SimulatedEventLoopTestFactory : public EventLoopTestFactory {
public:
- SimulatedEventLoopTestFactory() : event_loop_factory_(configuration()) {}
-
::std::unique_ptr<EventLoop> Make(std::string_view name) override {
- return event_loop_factory_.MakeEventLoop(name);
+ MaybeMake();
+ return event_loop_factory_->MakeEventLoop(name);
}
::std::unique_ptr<EventLoop> MakePrimary(std::string_view name) override {
- return event_loop_factory_.MakeEventLoop(name);
+ MaybeMake();
+ return event_loop_factory_->MakeEventLoop(name);
}
- void Run() override { event_loop_factory_.Run(); }
- void Exit() override { event_loop_factory_.Exit(); }
+ void Run() override { event_loop_factory_->Run(); }
+ void Exit() override { event_loop_factory_->Exit(); }
// TODO(austin): Implement this. It's used currently for a phased loop test.
// I'm not sure how much that matters.
void SleepFor(::std::chrono::nanoseconds /*duration*/) override {}
void set_send_delay(std::chrono::nanoseconds send_delay) {
- event_loop_factory_.set_send_delay(send_delay);
+ MaybeMake();
+ event_loop_factory_->set_send_delay(send_delay);
}
private:
- SimulatedEventLoopFactory event_loop_factory_;
+ void MaybeMake() {
+ if (!event_loop_factory_) {
+ if (configuration()->has_nodes()) {
+ event_loop_factory_ = std::make_unique<SimulatedEventLoopFactory>(
+ configuration(), my_node());
+ } else {
+ event_loop_factory_ =
+ std::make_unique<SimulatedEventLoopFactory>(configuration());
+ }
+ }
+ }
+ std::unique_ptr<SimulatedEventLoopFactory> event_loop_factory_;
};
INSTANTIATE_TEST_CASE_P(SimulatedEventLoopDeathTest, AbstractEventLoopDeathTest,
diff --git a/aos/json_to_flatbuffer.cc b/aos/json_to_flatbuffer.cc
index 9549c1f..6508786 100644
--- a/aos/json_to_flatbuffer.cc
+++ b/aos/json_to_flatbuffer.cc
@@ -215,7 +215,7 @@
switch (token) {
case Tokenizer::TokenType::kEnd:
if (stack_.size() != 0) {
- printf("Failed to unwind stack all the way\n");
+ fprintf(stderr, "Failed to unwind stack all the way\n");
return false;
} else {
return true;
@@ -235,8 +235,8 @@
stack_.back().typetable->type_codes[field_index];
if (type_code.base_type != flatbuffers::ET_SEQUENCE) {
- printf("Field '%s' is not a sequence\n",
- stack_.back().field_name.c_str());
+ fprintf(stderr, "Field '%s' is not a sequence\n",
+ stack_.back().field_name.c_str());
return false;
}
@@ -249,7 +249,7 @@
case Tokenizer::TokenType::kEndObject: // }
if (stack_.size() == 0) {
// Somehow we popped more than we pushed. Error.
- printf("Empty stack\n");
+ fprintf(stderr, "Empty stack\n");
return false;
} else {
// End of a nested struct! Add it.
@@ -346,7 +346,8 @@
stack_.back().typetable, stack_.back().field_name.c_str());
if (stack_.back().field_index == -1) {
- printf("Invalid field name '%s'\n", stack_.back().field_name.c_str());
+ fprintf(stderr, "Invalid field name '%s'\n",
+ stack_.back().field_name.c_str());
return false;
}
} break;
@@ -360,7 +361,7 @@
stack_.back().typetable->type_codes[field_index];
if (type_code.is_vector != in_vector()) {
- printf("Type and json disagree on if we are in a vector or not\n");
+ fprintf(stderr, "Type and json disagree on if we are in a vector or not\n");
return false;
}
@@ -377,7 +378,7 @@
stack_.back().typetable->type_codes[field_index];
if (type_code.is_vector != in_vector()) {
- printf("Type and json disagree on if we are in a vector or not\n");
+ fprintf(stderr, "Type and json disagree on if we are in a vector or not\n");
return false;
}
@@ -394,7 +395,7 @@
stack_.back().typetable->type_codes[field_index];
if (type_code.is_vector != in_vector()) {
- printf("Type and json disagree on if we are in a vector or not\n");
+ fprintf(stderr, "Type and json disagree on if we are in a vector or not\n");
return false;
}
@@ -430,8 +431,8 @@
}
if (!found) {
- printf("Enum value '%s' not found for field '%s'\n", data.c_str(),
- type_table->names[field_index]);
+ fprintf(stderr, "Enum value '%s' not found for field '%s'\n",
+ data.c_str(), type_table->names[field_index]);
return false;
}
@@ -464,8 +465,8 @@
::std::vector<bool> *fields_in_use,
flatbuffers::FlatBufferBuilder *fbb) {
if ((*fields_in_use)[field_element.field_index]) {
- printf("Duplicate field: '%s'\n",
- typetable->names[field_element.field_index]);
+ fprintf(stderr, "Duplicate field: '%s'\n",
+ typetable->names[field_element.field_index]);
return false;
}
@@ -533,9 +534,9 @@
case flatbuffers::ET_STRING:
case flatbuffers::ET_UTYPE:
case flatbuffers::ET_SEQUENCE:
- printf("Mismatched type for field '%s'. Got: integer, expected %s\n",
- typetable->names[field_index],
- ElementaryTypeName(elementary_type));
+ fprintf(
+ stderr, "Mismatched type for field '%s'. Got: integer, expected %s\n",
+ typetable->names[field_index], ElementaryTypeName(elementary_type));
return false;
};
return false;
@@ -564,9 +565,9 @@
case flatbuffers::ET_ULONG:
case flatbuffers::ET_STRING:
case flatbuffers::ET_SEQUENCE:
- printf("Mismatched type for field '%s'. Got: double, expected %s\n",
- typetable->names[field_index],
- ElementaryTypeName(elementary_type));
+ fprintf(
+ stderr, "Mismatched type for field '%s'. Got: double, expected %s\n",
+ typetable->names[field_index], ElementaryTypeName(elementary_type));
return false;
case flatbuffers::ET_FLOAT:
fbb->AddElement<float>(field_offset, double_value, 0);
@@ -606,9 +607,9 @@
case flatbuffers::ET_BOOL:
case flatbuffers::ET_FLOAT:
case flatbuffers::ET_DOUBLE:
- printf("Mismatched type for field '%s'. Got: string, expected %s\n",
- typetable->names[field_index],
- ElementaryTypeName(elementary_type));
+ fprintf(
+ stderr, "Mismatched type for field '%s'. Got: string, expected %s\n",
+ typetable->names[field_index], ElementaryTypeName(elementary_type));
CHECK_EQ(type_code.sequence_ref, -1)
<< ": Field name " << typetable->names[field_index]
<< " Got string expected " << ElementaryTypeName(elementary_type);
@@ -699,9 +700,10 @@
case flatbuffers::ET_STRING:
case flatbuffers::ET_UTYPE:
case flatbuffers::ET_SEQUENCE:
- printf("Mismatched type for field '%s'. Got: integer, expected %s\n",
- stack_.back().field_name.c_str(),
- ElementaryTypeName(elementary_type));
+ fprintf(stderr,
+ "Mismatched type for field '%s'. Got: integer, expected %s\n",
+ stack_.back().field_name.c_str(),
+ ElementaryTypeName(elementary_type));
return false;
};
return false;
@@ -722,9 +724,10 @@
case flatbuffers::ET_ULONG:
case flatbuffers::ET_STRING:
case flatbuffers::ET_SEQUENCE:
- printf("Mismatched type for field '%s'. Got: double, expected %s\n",
- stack_.back().field_name.c_str(),
- ElementaryTypeName(elementary_type));
+ fprintf(stderr,
+ "Mismatched type for field '%s'. Got: double, expected %s\n",
+ stack_.back().field_name.c_str(),
+ ElementaryTypeName(elementary_type));
return false;
case flatbuffers::ET_FLOAT:
fbb_.PushElement<float>(double_value);
@@ -752,9 +755,10 @@
case flatbuffers::ET_ULONG:
case flatbuffers::ET_FLOAT:
case flatbuffers::ET_DOUBLE:
- printf("Mismatched type for field '%s'. Got: sequence, expected %s\n",
- stack_.back().field_name.c_str(),
- ElementaryTypeName(elementary_type));
+ fprintf(stderr,
+ "Mismatched type for field '%s'. Got: sequence, expected %s\n",
+ stack_.back().field_name.c_str(),
+ ElementaryTypeName(elementary_type));
return false;
case flatbuffers::ET_STRING:
case flatbuffers::ET_SEQUENCE:
diff --git a/aos/json_tokenizer.cc b/aos/json_tokenizer.cc
index e2e8f37..ac519c8 100644
--- a/aos/json_tokenizer.cc
+++ b/aos/json_tokenizer.cc
@@ -356,6 +356,7 @@
ConsumeWhitespace();
state_ = State::kExpectObjectEnd;
} else {
+ fprintf(stderr, "Error on line %d, expected } or ,\n", linenumber_);
return TokenType::kError;
}
break;
@@ -364,6 +365,7 @@
ConsumeWhitespace();
state_ = State::kExpectArrayEnd;
} else {
+ fprintf(stderr, "Error on line %d, expected ] or ,\n", linenumber_);
return TokenType::kError;
}
break;
@@ -393,6 +395,7 @@
ConsumeWhitespace();
state_ = State::kExpectObjectEnd;
} else {
+ fprintf(stderr, "Error on line %d, expected , or }\n", linenumber_);
return TokenType::kError;
}
} else if (object_type_.back() == ObjectType::kArray) {
@@ -405,6 +408,7 @@
ConsumeWhitespace();
state_ = State::kExpectArrayEnd;
} else {
+ fprintf(stderr, "Error on line %d, expected , or ]\n", linenumber_);
return TokenType::kError;
}
}
diff --git a/aos/network/BUILD b/aos/network/BUILD
index 08621d0..5d96183 100644
--- a/aos/network/BUILD
+++ b/aos/network/BUILD
@@ -9,7 +9,6 @@
"team_number.h",
],
deps = [
- "//aos:configuration",
"//aos/logging",
"//aos/util:string_to_num",
"@com_google_absl//absl/base",
diff --git a/aos/network/team_number.cc b/aos/network/team_number.cc
index a75cd4e..1b90079 100644
--- a/aos/network/team_number.cc
+++ b/aos/network/team_number.cc
@@ -7,10 +7,9 @@
#include <string>
+#include "absl/base/call_once.h"
#include "aos/logging/logging.h"
#include "aos/util/string_to_num.h"
-#include "aos/configuration.h"
-#include "absl/base/call_once.h"
namespace aos {
namespace network {
diff --git a/aos/testdata/config1_multinode.json b/aos/testdata/config1_multinode.json
new file mode 100644
index 0000000..32bce5c
--- /dev/null
+++ b/aos/testdata/config1_multinode.json
@@ -0,0 +1,42 @@
+{
+ "channels": [
+ {
+ "name": "/foo",
+ "type": ".aos.bar",
+ "max_size": 5,
+ "source_node": "pi2"
+ },
+ {
+ "name": "/foo2",
+ "type": ".aos.bar",
+ "source_node": "pi1"
+ }
+ ],
+ "applications": [
+ {
+ "name": "app1"
+ },
+ {
+ "name": "app2"
+ }
+ ],
+ "maps": [
+ {
+ "match": {
+ "name": "/batman"
+ },
+ "rename": {
+ "name": "/foo"
+ }
+ }
+ ],
+ "nodes": [
+ {
+ "name": "pi2",
+ "hostname": "raspberrypi2"
+ }
+ ],
+ "imports": [
+ "config2_multinode.json"
+ ]
+}
diff --git a/aos/testdata/config2_multinode.json b/aos/testdata/config2_multinode.json
new file mode 100644
index 0000000..d284ab2
--- /dev/null
+++ b/aos/testdata/config2_multinode.json
@@ -0,0 +1,51 @@
+{
+ "channels": [
+ {
+ "name": "/foo",
+ "type": ".aos.bar",
+ "max_size": 7,
+ "source_node": "pi1"
+ },
+ {
+ "name": "/foo3",
+ "type": ".aos.bar",
+ "max_size": 9,
+ "source_node": "pi1"
+ }
+ ],
+ "applications": [
+ {
+ "name": "app1",
+ "maps": [
+ {
+ "match": {
+ "name": "/bar"
+ },
+ "rename": {
+ "name": "/foo"
+ }
+ }
+ ]
+ }
+ ],
+ "maps": [
+ {
+ "match": {
+ "name": "/batman"
+ },
+ "rename": {
+ "name": "/bar"
+ }
+ }
+ ],
+ "nodes": [
+ {
+ "name": "pi1",
+ "hostname": "raspberrypi1"
+ },
+ {
+ "name": "pi2",
+ "hostname": "raspberrypi5"
+ }
+ ]
+}
diff --git a/aos/testdata/expected_multinode.json b/aos/testdata/expected_multinode.json
new file mode 100644
index 0000000..46d052e
--- /dev/null
+++ b/aos/testdata/expected_multinode.json
@@ -0,0 +1,67 @@
+{
+ "channels": [
+ {
+ "name": "/foo",
+ "type": ".aos.bar",
+ "max_size": 5,
+ "source_node": "pi2"
+ },
+ {
+ "name": "/foo2",
+ "type": ".aos.bar",
+ "source_node": "pi1"
+ },
+ {
+ "name": "/foo3",
+ "type": ".aos.bar",
+ "max_size": 9,
+ "source_node": "pi1"
+ }
+ ],
+ "maps": [
+ {
+ "match": {
+ "name": "/batman"
+ },
+ "rename": {
+ "name": "/bar"
+ }
+ },
+ {
+ "match": {
+ "name": "/batman"
+ },
+ "rename": {
+ "name": "/foo"
+ }
+ }
+ ],
+ "applications": [
+ {
+ "name": "app1",
+ "maps": [
+ {
+ "match": {
+ "name": "/bar"
+ },
+ "rename": {
+ "name": "/foo"
+ }
+ }
+ ]
+ },
+ {
+ "name": "app2"
+ }
+ ],
+ "nodes": [
+ {
+ "name": "pi1",
+ "hostname": "raspberrypi1"
+ },
+ {
+ "name": "pi2",
+ "hostname": "raspberrypi2"
+ }
+ ]
+}
diff --git a/aos/testdata/invalid_destination_node.json b/aos/testdata/invalid_destination_node.json
new file mode 100644
index 0000000..95dc6d3
--- /dev/null
+++ b/aos/testdata/invalid_destination_node.json
@@ -0,0 +1,23 @@
+{
+ "channels": [
+ {
+ "name": "/foo",
+ "type": ".aos.bar",
+ "source_node": "roborio",
+ "destination_nodes": [
+ "dest",
+ "trojan_horse"
+ ]
+ }
+ ],
+ "nodes": [
+ {
+ "name": "roborio",
+ "hostname": "roboRIO-971.local"
+ },
+ {
+ "name": "dest",
+ "hostname": "dest-971.local"
+ }
+ ]
+}
diff --git a/aos/testdata/invalid_nodes.json b/aos/testdata/invalid_nodes.json
new file mode 100644
index 0000000..52c32cf
--- /dev/null
+++ b/aos/testdata/invalid_nodes.json
@@ -0,0 +1,15 @@
+{
+ "channels": [
+ {
+ "name": "/foo",
+ "type": ".aos.bar",
+ "max_size": 5
+ }
+ ],
+ "nodes": [
+ {
+ "name": "roborio",
+ "hostname": "roboRIO-971.local"
+ }
+ ]
+}
diff --git a/aos/testdata/invalid_source_node.json b/aos/testdata/invalid_source_node.json
new file mode 100644
index 0000000..b73256d
--- /dev/null
+++ b/aos/testdata/invalid_source_node.json
@@ -0,0 +1,15 @@
+{
+ "channels": [
+ {
+ "name": "/foo",
+ "type": ".aos.bar",
+ "source_node": "whome?"
+ }
+ ],
+ "nodes": [
+ {
+ "name": "roborio",
+ "hostname": "roboRIO-971.local"
+ }
+ ]
+}
diff --git a/aos/testdata/self_forward.json b/aos/testdata/self_forward.json
new file mode 100644
index 0000000..101b018
--- /dev/null
+++ b/aos/testdata/self_forward.json
@@ -0,0 +1,18 @@
+{
+ "channels": [
+ {
+ "name": "/foo",
+ "type": ".aos.bar",
+ "source_node": "roborio",
+ "destination_nodes": [
+ "roborio"
+ ]
+ }
+ ],
+ "nodes": [
+ {
+ "name": "roborio",
+ "hostname": "roboRIO-971.local"
+ }
+ ]
+}