You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by bm...@apache.org on 2013/10/29 20:33:42 UTC
[1/3] git commit: Made Promise have a virtual destructor.
Updated Branches:
refs/heads/master 3c61c6ff6 -> 137f4e86e
Made Promise have a virtual destructor.
See summary.
From: Benjamin Hindman <be...@berkeley.edu>
Review: https://reviews.apache.org/r/14382
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/a8310ad8
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/a8310ad8
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/a8310ad8
Branch: refs/heads/master
Commit: a8310ad813d96168ae300ece72c76a14cbfaf73b
Parents: 3c61c6f
Author: Benjamin Mahler <bm...@twitter.com>
Authored: Tue Oct 29 11:51:52 2013 -0700
Committer: Benjamin Mahler <bm...@twitter.com>
Committed: Tue Oct 29 11:51:52 2013 -0700
----------------------------------------------------------------------
3rdparty/libprocess/include/process/future.hpp | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/a8310ad8/3rdparty/libprocess/include/process/future.hpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/include/process/future.hpp b/3rdparty/libprocess/include/process/future.hpp
index daf4b92..e03f8c9 100644
--- a/3rdparty/libprocess/include/process/future.hpp
+++ b/3rdparty/libprocess/include/process/future.hpp
@@ -253,7 +253,7 @@ class Promise
public:
Promise();
Promise(const T& t);
- ~Promise();
+ virtual ~Promise();
bool set(const T& _t);
bool set(const Future<T>& future); // Alias for associate.
[3/3] git commit: Injected Registrar into the Master and the tests.
Posted by bm...@apache.org.
Injected Registrar into the Master and the tests.
From: Benjamin Hindman <be...@berkeley.edu>
Review: https://reviews.apache.org/r/14382
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/137f4e86
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/137f4e86
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/137f4e86
Branch: refs/heads/master
Commit: 137f4e86ecb7f022b7fa4063431a044485904170
Parents: 7edb000
Author: Benjamin Mahler <bm...@twitter.com>
Authored: Tue Oct 29 12:10:27 2013 -0700
Committer: Benjamin Mahler <bm...@twitter.com>
Committed: Tue Oct 29 12:27:02 2013 -0700
----------------------------------------------------------------------
src/local/local.cpp | 41 ++++++++++++++++++++--
src/master/flags.hpp | 20 +++++++----
src/master/main.cpp | 32 +++++++++++++++++-
src/master/master.cpp | 29 ++++++----------
src/master/master.hpp | 8 ++++-
src/slave/slave.cpp | 7 ++--
src/tests/cluster.hpp | 60 +++++++++++++++++++++++++++++++--
src/tests/mesos.cpp | 2 ++
src/tests/slave_recovery_tests.cpp | 8 ++---
9 files changed, 169 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/local/local.cpp
----------------------------------------------------------------------
diff --git a/src/local/local.cpp b/src/local/local.cpp
index e4b5ec5..180756a 100644
--- a/src/local/local.cpp
+++ b/src/local/local.cpp
@@ -20,8 +20,10 @@
#include <sstream>
#include <vector>
-#include <stout/fatal.hpp>
+#include <stout/exit.hpp>
#include <stout/foreach.hpp>
+#include <stout/path.hpp>
+#include <stout/strings.hpp>
#include "local.hpp"
@@ -34,10 +36,15 @@
#include "master/drf_sorter.hpp"
#include "master/hierarchical_allocator_process.hpp"
#include "master/master.hpp"
+#include "master/registrar.hpp"
#include "slave/process_isolator.hpp"
#include "slave/slave.hpp"
+#include "state/leveldb.hpp"
+#include "state/protobuf.hpp"
+#include "state/storage.hpp"
+
using namespace mesos::internal;
using mesos::internal::master::allocator::Allocator;
@@ -46,6 +53,7 @@ using mesos::internal::master::allocator::DRFSorter;
using mesos::internal::master::allocator::HierarchicalDRFAllocatorProcess;
using mesos::internal::master::Master;
+using mesos::internal::master::Registrar;
using mesos::internal::slave::Slave;
using mesos::internal::slave::Isolator;
@@ -66,6 +74,9 @@ namespace local {
static Allocator* allocator = NULL;
static AllocatorProcess* allocatorProcess = NULL;
+static state::Storage* storage = NULL;
+static state::protobuf::State* state = NULL;
+static Registrar* registrar = NULL;
static Master* master = NULL;
static map<Isolator*, Slave*> slaves;
static MasterDetector* detector = NULL;
@@ -98,7 +109,24 @@ PID<Master> launch(const Flags& flags, Allocator* _allocator)
EXIT(1) << "Failed to start a local cluster while loading "
<< "master flags from the environment: " << load.error();
}
- master = new Master(_allocator, files, flags);
+
+ if (strings::startsWith(flags.registry, "zk://")) {
+ // TODO(benh):
+ EXIT(1) << "ZooKeeper based registry unimplemented";
+ } else if (flags.registry == "local") {
+ storage = new state::LevelDBStorage(
+ path::join(flags.work_dir, "registry"));
+ } else {
+ EXIT(1) << "'" << flags.registry << "' is not a supported"
+ << " option for registry persistence";
+ }
+
+ CHECK_NOTNULL(storage);
+
+ state = new state::protobuf::State(storage);
+ registrar = new Registrar(state);
+
+ master = new Master(_allocator, registrar, files, flags);
}
PID<Master> pid = process::spawn(master);
@@ -160,6 +188,15 @@ void shutdown()
delete files;
files = NULL;
+
+ delete registrar;
+ registrar = NULL;
+
+ delete state;
+ state = NULL;
+
+ delete storage;
+ storage = NULL;
}
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/master/flags.hpp
----------------------------------------------------------------------
diff --git a/src/master/flags.hpp b/src/master/flags.hpp
index 7b5f3de..16faf96 100644
--- a/src/master/flags.hpp
+++ b/src/master/flags.hpp
@@ -40,12 +40,17 @@ public:
"Can root submit frameworks?",
true);
- add(&Flags::slaves,
- "slaves",
- "Initial slaves that should be\n"
- "considered part of this cluster\n"
- "(or if using ZooKeeper a URL)",
- "*");
+ add(&Flags::work_dir,
+ "work_dir",
+ "Where to store master specific files\n",
+ "/tmp/mesos");
+
+ add(&Flags::registry,
+ "registry",
+ "Persistence strategy for the registry;\n"
+ "available options are 'local' or a ZooKeeper\n"
+ "URL (i.e., 'zk://host1:port1,host2:port2,.../path')",
+ "local");
add(&Flags::webui_dir,
"webui_dir",
@@ -111,7 +116,8 @@ public:
}
bool root_submissions;
- std::string slaves;
+ std::string work_dir;
+ std::string registry;
std::string webui_dir;
std::string whitelist;
std::string user_sorter;
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/master/main.cpp
----------------------------------------------------------------------
diff --git a/src/master/main.cpp b/src/master/main.cpp
index 19fcb9f..45caf9d 100644
--- a/src/master/main.cpp
+++ b/src/master/main.cpp
@@ -19,10 +19,13 @@
#include <mesos/mesos.hpp>
#include <stout/check.hpp>
+#include <stout/exit.hpp>
#include <stout/flags.hpp>
#include <stout/nothing.hpp>
#include <stout/os.hpp>
+#include <stout/path.hpp>
#include <stout/stringify.hpp>
+#include <stout/strings.hpp>
#include <stout/try.hpp>
#include "common/build.hpp"
@@ -36,6 +39,12 @@
#include "master/drf_sorter.hpp"
#include "master/hierarchical_allocator_process.hpp"
#include "master/master.hpp"
+#include "master/registrar.hpp"
+
+#include "state/leveldb.hpp"
+#include "state/protobuf.hpp"
+#include "state/storage.hpp"
+
using namespace mesos::internal;
using namespace mesos::internal::master;
@@ -119,8 +128,25 @@ int main(int argc, char** argv)
allocator::Allocator* allocator =
new allocator::Allocator(allocatorProcess);
+ state::Storage* storage = NULL;
+
+ if (strings::startsWith(flags.registry, "zk://")) {
+ // TODO(benh):
+ EXIT(1) << "ZooKeeper based registry unimplemented";
+ } else if (flags.registry == "local") {
+ storage = new state::LevelDBStorage(path::join(flags.work_dir, "registry"));
+ } else {
+ EXIT(1) << "'" << flags.registry << "' is not a supported"
+ << " option for registry persistence";
+ }
+
+ CHECK_NOTNULL(storage);
+
+ state::protobuf::State* state = new state::protobuf::State(storage);
+ Registrar* registrar = new Registrar(state);
+
Files files;
- Master* master = new Master(allocator, &files, flags);
+ Master* master = new Master(allocator, registrar, &files, flags);
process::spawn(master);
Try<MasterDetector*> detector =
@@ -133,6 +159,10 @@ int main(int argc, char** argv)
delete allocator;
delete allocatorProcess;
+ delete registrar;
+ delete state;
+ delete storage;
+
MasterDetector::destroy(detector.get());
return 0;
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/master/master.cpp
----------------------------------------------------------------------
diff --git a/src/master/master.cpp b/src/master/master.cpp
index 1147cc6..8e14a07 100644
--- a/src/master/master.cpp
+++ b/src/master/master.cpp
@@ -186,20 +186,29 @@ private:
};
-Master::Master(Allocator* _allocator, Files* _files)
+Master::Master(
+ Allocator* _allocator,
+ Registrar* _registrar,
+ Files* _files)
: ProcessBase("master"),
http(*this),
flags(),
allocator(_allocator),
+ registrar(_registrar),
files(_files),
completedFrameworks(MAX_COMPLETED_FRAMEWORKS) {}
-Master::Master(Allocator* _allocator, Files* _files, const Flags& _flags)
+Master::Master(
+ Allocator* _allocator,
+ Registrar* _registrar,
+ Files* _files,
+ const Flags& _flags)
: ProcessBase("master"),
http(*this),
flags(_flags),
allocator(_allocator),
+ registrar(_registrar),
files(_files),
completedFrameworks(MAX_COMPLETED_FRAMEWORKS) {}
@@ -1223,21 +1232,7 @@ void Master::registerSlave(const UPID& from, const SlaveInfo& slaveInfo)
LOG(INFO) << "Attempting to register slave on " << slave->info.hostname()
<< " at " << slave->pid;
- // TODO(benh): We assume all slaves can register for now.
- CHECK_EQ(flags.slaves, "*");
addSlave(slave);
-
-// // Checks if this slave, or if all slaves, can be accepted.
-// if (slaveHostnamePorts.contains(slaveInfo.hostname(), from.port)) {
-// run(&SlaveRegistrar::run, slave, self());
-// } else if (flags.slaves == "*") {
-// run(&SlaveRegistrar::run, slave, self(), slavesManager->self());
-// } else {
-// LOG(WARNING) << "Cannot register slave at "
-// << slaveInfo.hostname() << ":" << from.port
-// << " because not in allocated set of slaves!";
-// reply(ShutdownMessage());
-// }
}
@@ -1326,8 +1321,6 @@ void Master::reregisterSlave(
LOG(INFO) << "Attempting to re-register slave " << slave->id << " at "
<< slave->pid << " (" << slave->info.hostname() << ")";
- // TODO(benh): We assume all slaves can register for now.
- CHECK_EQ(flags.slaves, "*");
readdSlave(slave, executorInfos, tasks);
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/master/master.hpp
----------------------------------------------------------------------
diff --git a/src/master/master.hpp b/src/master/master.hpp
index 1eba03f..e377af8 100644
--- a/src/master/master.hpp
+++ b/src/master/master.hpp
@@ -47,6 +47,7 @@
#include "master/constants.hpp"
#include "master/flags.hpp"
+#include "master/registrar.hpp"
#include "messages/messages.hpp"
@@ -82,8 +83,12 @@ struct Role;
class Master : public ProtobufProcess<Master>
{
public:
- Master(allocator::Allocator* allocator, Files* files);
Master(allocator::Allocator* allocator,
+ Registrar* registrar,
+ Files* files);
+
+ Master(allocator::Allocator* allocator,
+ Registrar* registrar,
Files* files,
const Flags& flags);
@@ -302,6 +307,7 @@ private:
allocator::Allocator* allocator;
WhitelistWatcher* whitelistWatcher;
+ Registrar* registrar;
Files* files;
MasterInfo info;
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 06ef786..74bcbec 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -641,13 +641,14 @@ void Slave::doReliableRegistration()
// Slave started before master.
// (Vinod): Is the above comment true?
RegisterSlaveMessage message;
- message.mutable_slave()->MergeFrom(info);
+ message.mutable_slave()->CopyFrom(info);
send(master, message);
} else {
// Re-registering, so send tasks running.
ReregisterSlaveMessage message;
- message.mutable_slave_id()->MergeFrom(info.id());
- message.mutable_slave()->MergeFrom(info);
+ message.mutable_slave_id()->CopyFrom(info.id());
+ message.mutable_slave()->CopyFrom(info);
+ message.mutable_slave()->mutable_id()->CopyFrom(info.id());
foreachvalue (Framework* framework, frameworks){
foreachvalue (Executor* executor, framework->executors) {
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/tests/cluster.hpp
----------------------------------------------------------------------
diff --git a/src/tests/cluster.hpp b/src/tests/cluster.hpp
index 5621139..bea395a 100644
--- a/src/tests/cluster.hpp
+++ b/src/tests/cluster.hpp
@@ -30,6 +30,7 @@
#include <stout/nothing.hpp>
#include <stout/option.hpp>
#include <stout/owned.hpp>
+#include <stout/path.hpp>
#include <stout/try.hpp>
#include "detector/detector.hpp"
@@ -40,12 +41,17 @@
#include "master/hierarchical_allocator_process.hpp"
#include "master/flags.hpp"
#include "master/master.hpp"
+#include "master/registrar.hpp"
#include "slave/flags.hpp"
#include "slave/isolator.hpp"
#include "slave/process_isolator.hpp"
#include "slave/slave.hpp"
+#include "state/leveldb.hpp"
+#include "state/protobuf.hpp"
+#include "state/storage.hpp"
+
#include "zookeeper/url.hpp"
namespace mesos {
@@ -108,6 +114,9 @@ public:
master::Master* master;
master::allocator::Allocator* allocator;
master::allocator::AllocatorProcess* allocatorProcess;
+ state::Storage* storage;
+ state::protobuf::State* state;
+ master::Registrar* registrar;
MasterDetector* detector;
};
@@ -223,7 +232,28 @@ inline Try<process::PID<master::Master> > Cluster::Masters::start(
master.allocatorProcess = new master::allocator::HierarchicalDRFAllocatorProcess();
master.allocator = new master::allocator::Allocator(master.allocatorProcess);
- master.master = new master::Master(master.allocator, &cluster->files, flags);
+
+ if (strings::startsWith(flags.registry, "zk://")) {
+ // TODO(benh):
+ return Error("ZooKeeper based registry unimplemented");
+ } else if (flags.registry == "local") {
+ master.storage = new state::LevelDBStorage(
+ path::join(flags.work_dir, "registry"));
+ } else {
+ return Error("'" + flags.registry + "' is not a supported"
+ " option for registry persistence");
+ }
+
+ CHECK_NOTNULL(master.storage);
+
+ master.state = new state::protobuf::State(master.storage);
+ master.registrar = new master::Registrar(master.state);
+
+ master.master = new master::Master(
+ master.allocator,
+ master.registrar,
+ &cluster->files,
+ flags);
process::PID<master::Master> pid = process::spawn(master.master);
@@ -249,9 +279,31 @@ inline Try<process::PID<master::Master> > Cluster::Masters::start(
}
Master master;
+
master.allocatorProcess = NULL;
master.allocator = new master::allocator::Allocator(allocatorProcess);
- master.master = new master::Master(master.allocator, &cluster->files, flags);
+
+ if (strings::startsWith(flags.registry, "zk://")) {
+ // TODO(benh):
+ return Error("ZooKeeper based registry unimplemented");
+ } else if (flags.registry == "local") {
+ master.storage = new state::LevelDBStorage(
+ path::join(flags.work_dir, "registry"));
+ } else {
+ return Error("'" + flags.registry + "' is not a supported"
+ " option for registry persistence");
+ }
+
+ CHECK_NOTNULL(master.storage);
+
+ master.state = new state::protobuf::State(master.storage);
+ master.registrar = new master::Registrar(master.state);
+
+ master.master = new master::Master(
+ master.allocator,
+ master.registrar,
+ &cluster->files,
+ flags);
process::PID<master::Master> pid = process::spawn(master.master);
@@ -283,6 +335,10 @@ inline Try<Nothing> Cluster::Masters::stop(
delete master.allocator; // Terminates and waits for allocator process.
delete master.allocatorProcess; // May be NULL.
+ delete master.registrar;
+ delete master.state;
+ delete master.storage;
+
delete master.detector;
masters.erase(pid);
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/tests/mesos.cpp
----------------------------------------------------------------------
diff --git a/src/tests/mesos.cpp b/src/tests/mesos.cpp
index 10b3e53..351ffd6 100644
--- a/src/tests/mesos.cpp
+++ b/src/tests/mesos.cpp
@@ -32,6 +32,8 @@ master::Flags MesosTest::CreateMasterFlags()
const Try<std::string>& directory = environment->mkdtemp();
CHECK_SOME(directory) << "Failed to create temporary directory";
+ flags.work_dir = directory.get();
+
// Create a default credentials file.
const std::string& path = path::join(directory.get(), "credentials");
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/tests/slave_recovery_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/slave_recovery_tests.cpp b/src/tests/slave_recovery_tests.cpp
index 2812f11..37faf20 100644
--- a/src/tests/slave_recovery_tests.cpp
+++ b/src/tests/slave_recovery_tests.cpp
@@ -97,7 +97,7 @@ TEST_F(SlaveStateTest, CheckpointProtobuf)
expected.set_value("slave1");
const string& file = "slave.id";
- state::checkpoint(file, expected);
+ slave::state::checkpoint(file, expected);
const Result<SlaveID>& actual = ::protobuf::read<SlaveID>(file);
ASSERT_SOME(actual);
@@ -111,7 +111,7 @@ TEST_F(SlaveStateTest, CheckpointString)
// Checkpoint a test string.
const string expected = "test";
const string file = "test-file";
- state::checkpoint(file, expected);
+ slave::state::checkpoint(file, expected);
ASSERT_SOME_EQ(expected, os::read(file));
}
@@ -228,12 +228,12 @@ TYPED_TEST(SlaveRecoveryTest, RecoverSlaveState)
AWAIT_READY(_ack);
// Recover the state.
- Result<state::SlaveState> recover = state::recover(
+ Result<slave::state::SlaveState> recover = slave::state::recover(
paths::getMetaRootDir(flags.work_dir), true);
ASSERT_SOME(recover);
- state::SlaveState state = recover.get();
+ slave::state::SlaveState state = recover.get();
// Check slave id.
ASSERT_EQ(slaveId, state.id);
[2/3] git commit: Added the registrar.
Posted by bm...@apache.org.
Added the registrar.
From: Benjamin Hindman <be...@berkeley.edu>
Review: https://reviews.apache.org/r/14383
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/7edb0004
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/7edb0004
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/7edb0004
Branch: refs/heads/master
Commit: 7edb0004b974db8ac2c5e9be84e57e7349136de6
Parents: a8310ad
Author: Benjamin Mahler <bm...@twitter.com>
Authored: Tue Oct 29 12:08:23 2013 -0700
Committer: Benjamin Mahler <bm...@twitter.com>
Committed: Tue Oct 29 12:27:01 2013 -0700
----------------------------------------------------------------------
src/Makefile.am | 5 +-
src/master/registrar.cpp | 415 +++++++++++++++++++++++++++++++++++++
src/master/registrar.hpp | 53 +++++
src/master/registry.proto | 2 +-
src/tests/registrar_tests.cpp | 154 ++++++++++++++
src/tests/state_tests.cpp | 148 ++++++-------
6 files changed, 701 insertions(+), 76 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/7edb0004/src/Makefile.am
----------------------------------------------------------------------
diff --git a/src/Makefile.am b/src/Makefile.am
index 0b32d74..a11c76b 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -167,6 +167,7 @@ libmesos_no_3rdparty_la_SOURCES = \
master/master.cpp \
master/registry.hpp \
master/registry.proto \
+ master/registrar.cpp \
slave/constants.cpp \
slave/gc.cpp \
slave/monitor.cpp \
@@ -223,6 +224,7 @@ libmesos_no_3rdparty_la_SOURCES += common/attributes.hpp \
master/allocator.hpp \
master/constants.hpp master/drf_sorter.hpp master/flags.hpp \
master/hierarchical_allocator_process.hpp \
+ master/registrar.hpp \
master/master.hpp master/sorter.hpp \
messages/messages.hpp slave/constants.hpp \
slave/flags.hpp slave/gc.hpp slave/monitor.hpp \
@@ -298,7 +300,7 @@ libmesos_no_3rdparty_la_LIBADD += libstate.la
# The final result!
lib_LTLIBRARIES += libmesos.la
-libmesos_la_SOURCES = $(MESOS_PROTO) # Part of the distribution.
+libmesos_la_SOURCES = $(MESOS_PROTO) # Include as part of the distribution.
libmesos_la_LDFLAGS = -release $(PACKAGE_VERSION) -shared
@@ -781,6 +783,7 @@ mesos_tests_SOURCES = \
tests/paths_tests.cpp \
tests/protobuf_io_tests.cpp \
tests/reaper_tests.cpp \
+ tests/registrar_tests.cpp \
tests/resource_offers_tests.cpp \
tests/resources_tests.cpp \
tests/sasl_tests.cpp \
http://git-wip-us.apache.org/repos/asf/mesos/blob/7edb0004/src/master/registrar.cpp
----------------------------------------------------------------------
diff --git a/src/master/registrar.cpp b/src/master/registrar.cpp
new file mode 100644
index 0000000..42fe30e
--- /dev/null
+++ b/src/master/registrar.cpp
@@ -0,0 +1,415 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <deque>
+#include <string>
+
+#include <process/defer.hpp>
+#include <process/dispatch.hpp>
+#include <process/future.hpp>
+#include <process/process.hpp>
+
+#include <stout/lambda.hpp>
+#include <stout/none.hpp>
+#include <stout/nothing.hpp>
+#include <stout/option.hpp>
+
+#include "common/type_utils.hpp"
+
+#include "master/registrar.hpp"
+#include "master/registry.hpp"
+
+#include "state/protobuf.hpp"
+
+using mesos::internal::state::protobuf::State;
+using mesos::internal::state::protobuf::Variable;
+
+using process::dispatch;
+using process::Future;
+using process::Process;
+using process::Promise;
+using process::spawn;
+using process::terminate;
+using process::wait; // Necessary on some OS's to disambiguate.
+
+namespace mesos {
+namespace internal {
+namespace master {
+
+class RegistrarProcess : public Process<RegistrarProcess>
+{
+public:
+ RegistrarProcess(State* _state)
+ : ProcessBase("registrar"),
+ state(_state)
+ {
+ slaves.variable = None();
+ slaves.updating = false;
+ }
+
+ virtual ~RegistrarProcess() {}
+
+ // Registrar implementation.
+ Future<bool> admit(const SlaveID& id, const SlaveInfo& info);
+ Future<bool> readmit(const SlaveInfo& info);
+ Future<bool> remove(const SlaveInfo& info);
+
+private:
+ template <typename T>
+ struct Mutation : process::Promise<bool>
+ {
+ virtual Try<T> apply(T t) = 0;
+ };
+
+ struct Admit : Mutation<registry::Slaves>
+ {
+ Admit(const SlaveID& _id, const SlaveInfo& _info)
+ : id(_id), info(_info) {}
+
+ virtual Try<registry::Slaves> apply(registry::Slaves slaves)
+ {
+ // Check and see if this slave already exists.
+ foreach (const registry::Slave& slave, slaves.slaves()) {
+ if (slave.info().id() == id) {
+ set(false);
+ return slaves; // No mutation.
+ }
+ }
+
+ // Okay, add the slave!
+ registry::Slave* slave = slaves.add_slaves();
+ slave->mutable_info()->CopyFrom(info);
+ slave->mutable_info()->mutable_id()->MergeFrom(id);
+ return slaves;
+ }
+
+ const SlaveID id;
+ const SlaveInfo info;
+ };
+
+ // NOTE: even thought readmission does not mutate the state we model
+ // it as a mutation so that it is performed in sequence with other
+ // mutations.
+ struct Readmit : Mutation<registry::Slaves>
+ {
+ Readmit(const SlaveInfo& _info) : info(_info) { CHECK(info.has_id()); }
+
+ virtual Try<registry::Slaves> apply(registry::Slaves slaves)
+ {
+ bool found = false;
+ foreach (const registry::Slave& slave, slaves.slaves()) {
+ if (slave.info().id() == info.id()) {
+ set(true);
+ found = true;
+ }
+ }
+ if (!found) {
+ set(false);
+ }
+ return slaves;
+ }
+
+ const SlaveInfo info;
+ };
+
+ struct Remove : Mutation<registry::Slaves>
+ {
+ Remove(const SlaveInfo& _info) : info(_info) { CHECK(info.has_id()); }
+
+ virtual Try<registry::Slaves> apply(registry::Slaves slaves)
+ {
+ bool removed = false;
+ for (int i = 0; i < slaves.slaves().size(); i++) {
+ const registry::Slave& slave = slaves.slaves(i);
+ if (slave.info().id() == info.id()) {
+ for (int j = i + 1; j < slaves.slaves().size(); j++) {
+ slaves.mutable_slaves()->SwapElements(i, j);
+ }
+ slaves.mutable_slaves()->RemoveLast();
+ removed = true;
+ break;
+ }
+ }
+ if (!removed) {
+ set(false);
+ }
+ return slaves; // May or may not have been mutated.
+ }
+
+ const SlaveInfo info;
+ };
+
+ struct {
+ Option<Variable<registry::Slaves> > variable;
+ std::deque<Mutation<registry::Slaves>*> mutations;
+ bool updating; // Used to signify fetching (recovering) or storing.
+ } slaves;
+
+ // Continuations.
+ Future<bool> _admit(const SlaveID& id, const SlaveInfo& info);
+ Future<bool> _readmit(const SlaveInfo& info);
+ Future<bool> _remove(const SlaveInfo& info);
+
+ // Helper for recovering state (performing fetch).
+ Future<Nothing> recover();
+ void _recover(const Future<Variable<registry::Slaves> >& recovery);
+
+ // Helper for updating state (performing store).
+ void update();
+ Future<bool> _update(const Option<Variable<registry::Slaves> >& variable);
+ void __update();
+
+ State* state;
+
+ // Used to compose our operations with recovery.
+ Promise<Nothing> recovered;
+};
+
+
+Future<Nothing> RegistrarProcess::recover()
+{
+ LOG(INFO) << "Recovering registrar";
+
+ // "Recover" the 'slaves' variable by fetching it from the state.
+ if (slaves.variable.isNone() && !slaves.updating) {
+ state->fetch<registry::Slaves>("slaves")
+ .onAny(defer(self(), &Self::_recover, lambda::_1));
+
+ // TODO(benh): Don't wait forever to recover?
+ }
+
+ // TODO(benh): Recover other variables too.
+
+ return recovered.future();
+}
+
+
+void RegistrarProcess::_recover(
+ const Future<Variable<registry::Slaves> >& recovery)
+{
+ slaves.updating = false;
+
+ CHECK(!recovery.isPending());
+
+ if (recovery.isFailed() || recovery.isDiscarded()) {
+ LOG(WARNING) << "Failed to recover registrar: " << recovery.isFailed()
+ ? recovery.failure() : "future discarded";
+ recover(); // Retry! TODO(benh): Don't retry forever?
+ } else {
+ LOG(INFO) << "Successfully recovered registrar";
+
+ // Save the slaves variable.
+ slaves.variable = recovery.get();
+
+ // Signal the recovery is complete.
+ recovered.set(Nothing());
+ }
+}
+
+
+Future<bool> RegistrarProcess::admit(
+ const SlaveID& id,
+ const SlaveInfo& info)
+{
+ return recover()
+ .then(defer(self(), &Self::_admit, id, info));
+}
+
+
+Future<bool> RegistrarProcess::_admit(
+ const SlaveID& id,
+ const SlaveInfo& info)
+{
+ CHECK_SOME(slaves.variable);
+ Mutation<registry::Slaves>* mutation = new Admit(id, info);
+ slaves.mutations.push_back(mutation);
+ Future<bool> future = mutation->future();
+ if (!slaves.updating) {
+ update();
+ }
+ return future;
+}
+
+
+Future<bool> RegistrarProcess::readmit(const SlaveInfo& info)
+{
+ return recover()
+ .then(defer(self(), &Self::_readmit, info));
+}
+
+
+Future<bool> RegistrarProcess::_readmit(
+ const SlaveInfo& info)
+{
+ CHECK_SOME(slaves.variable);
+
+ if (!info.has_id()) {
+ return Future<bool>::failed("Expecting SlaveInfo to have a SlaveID");
+ }
+
+ Mutation<registry::Slaves>* mutation = new Readmit(info);
+ slaves.mutations.push_back(mutation);
+ Future<bool> future = mutation->future();
+ if (!slaves.updating) {
+ update();
+ }
+ return future;
+}
+
+
+Future<bool> RegistrarProcess::remove(const SlaveInfo& info)
+{
+ return recover()
+ .then(defer(self(), &Self::_remove, info));
+}
+
+
+Future<bool> RegistrarProcess::_remove(
+ const SlaveInfo& info)
+{
+ CHECK_SOME(slaves.variable);
+
+ if (!info.has_id()) {
+ return Future<bool>::failed("Expecting SlaveInfo to have a SlaveID");
+ }
+
+ Mutation<registry::Slaves>* mutation = new Remove(info);
+ slaves.mutations.push_back(mutation);
+ Future<bool> future = mutation->future();
+ if (!slaves.updating) {
+ update();
+ }
+ return future;
+}
+
+
+void RegistrarProcess::update()
+{
+ if (!slaves.mutations.empty()) {
+ CHECK(!slaves.updating);
+
+ slaves.updating = true;
+
+ LOG(INFO) << "Attempting to update 'slaves'";
+
+ CHECK_SOME(slaves.variable);
+
+ Variable<registry::Slaves> variable = slaves.variable.get();
+
+ foreach (Mutation<registry::Slaves>* mutation, slaves.mutations) {
+ Try<registry::Slaves> slaves = mutation->apply(variable.get());
+ if (slaves.isError()) {
+ mutation->fail("Failed to mutate 'slaves': " + slaves.error());
+ } else {
+ Try<Variable<registry::Slaves> > v = variable.mutate(slaves.get());
+ if (v.isError()) {
+ mutation->fail("Failed to mutate 'slaves': " + v.error());
+ } else {
+ variable = v.get();
+ }
+ }
+ }
+
+ // Perform the store! Save the future so we can associate it with
+ // the mutations that are part of this update.
+ Future<bool> future =
+ state->store(variable).then(defer(self(), &Self::_update, lambda::_1));
+
+ // TODO(benh): Add a timeout so we don't wait forever.
+
+ // Toggle 'updating' if the store fails or is discarded.
+ future
+ .onDiscarded(defer(self(), &Self::__update))
+ .onFailed(defer(self(), &Self::__update));
+
+ // Now associate the store with all the mutations.
+ while (!slaves.mutations.empty()) {
+ Mutation<registry::Slaves>* mutation = slaves.mutations.front();
+ slaves.mutations.pop_front();
+ mutation->associate(future); // No-op if already failed above.
+ delete mutation;
+ }
+ }
+}
+
+
+Future<bool> RegistrarProcess::_update(
+ const Option<Variable<registry::Slaves> >& variable)
+{
+ slaves.updating = false;
+
+ if (variable.isNone()) {
+ LOG(WARNING) << "Failed to update 'slaves': version mismatch";
+ return Future<bool>::failed("Failed to update 'slaves': version mismatch");
+ }
+
+ LOG(INFO) << "Successfully updated 'slaves'";
+
+ slaves.variable = variable.get();
+
+ if (!slaves.mutations.empty()) {
+ update();
+ }
+
+ return true;
+}
+
+
+void RegistrarProcess::__update()
+{
+ LOG(WARNING) << "Failed to update 'slaves'";
+ slaves.updating = false;
+}
+
+
+Registrar::Registrar(State* state)
+{
+ process = new RegistrarProcess(state);
+ spawn(process);
+}
+
+
+Registrar::~Registrar()
+{
+ terminate(process);
+ wait(process);
+}
+
+
+Future<bool> Registrar::admit(
+ const SlaveID& id,
+ const SlaveInfo& info)
+{
+ return dispatch(process, &RegistrarProcess::admit, id, info);
+}
+
+
+Future<bool> Registrar::readmit(const SlaveInfo& info)
+{
+ return dispatch(process, &RegistrarProcess::readmit, info);
+}
+
+
+Future<bool> Registrar::remove(const SlaveInfo& info)
+{
+ return dispatch(process, &RegistrarProcess::remove, info);
+}
+
+} // namespace master {
+} // namespace internal {
+} // namespace mesos {
http://git-wip-us.apache.org/repos/asf/mesos/blob/7edb0004/src/master/registrar.hpp
----------------------------------------------------------------------
diff --git a/src/master/registrar.hpp b/src/master/registrar.hpp
new file mode 100644
index 0000000..5742c36
--- /dev/null
+++ b/src/master/registrar.hpp
@@ -0,0 +1,53 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef __MASTER_REGISTRAR_HPP__
+#define __MASTER_REGISTRAR_HPP__
+
+#include <mesos/mesos.hpp>
+
+#include <process/future.hpp>
+
+#include "state/protobuf.hpp"
+
+namespace mesos {
+namespace internal {
+namespace master {
+
+// Forward declaration.
+class RegistrarProcess;
+
+class Registrar
+{
+public:
+ Registrar(state::protobuf::State* state);
+ ~Registrar();
+
+ process::Future<bool> admit(const SlaveID& id, const SlaveInfo& info);
+ process::Future<bool> readmit(const SlaveInfo& info);
+ process::Future<bool> remove(const SlaveInfo& info);
+
+private:
+ RegistrarProcess* process;
+};
+
+} // namespace master {
+} // namespace internal {
+} // namespace mesos {
+
+#endif // __MASTER_REGISTRAR_HPP__
http://git-wip-us.apache.org/repos/asf/mesos/blob/7edb0004/src/master/registry.proto
----------------------------------------------------------------------
diff --git a/src/master/registry.proto b/src/master/registry.proto
index 877bfa1..bd85099 100644
--- a/src/master/registry.proto
+++ b/src/master/registry.proto
@@ -27,6 +27,6 @@ message Slave {
}
-message Registry {
+message Slaves {
repeated Slave slaves = 1;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/7edb0004/src/tests/registrar_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/registrar_tests.cpp b/src/tests/registrar_tests.cpp
new file mode 100644
index 0000000..51975f5
--- /dev/null
+++ b/src/tests/registrar_tests.cpp
@@ -0,0 +1,154 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <map>
+#include <string>
+
+#include <process/gmock.hpp>
+#include <process/gtest.hpp>
+#include <process/pid.hpp>
+#include <process/process.hpp>
+
+#include "master/registrar.hpp"
+
+#include "state/leveldb.hpp"
+#include "state/protobuf.hpp"
+#include "state/storage.hpp"
+
+using namespace mesos;
+using namespace mesos::internal;
+
+using namespace process;
+
+using std::map;
+using std::string;
+
+using testing::_;
+using testing::Eq;
+
+namespace mesos {
+namespace internal {
+namespace master {
+
+class RegistrarTest : public ::testing::Test
+{
+public:
+ RegistrarTest()
+ : storage(NULL),
+ state(NULL),
+ path(os::getcwd() + "/.state") {}
+
+protected:
+ virtual void SetUp()
+ {
+ os::rmdir(path);
+ storage = new state::LevelDBStorage(path);
+ state = new state::protobuf::State(storage);
+ }
+
+ virtual void TearDown()
+ {
+ delete state;
+ delete storage;
+ os::rmdir(path);
+ }
+
+ state::Storage* storage;
+ state::protobuf::State* state;
+
+private:
+ const std::string path;
+};
+
+
+TEST_F(RegistrarTest, admit)
+{
+ Registrar registrar(state);
+
+ SlaveID id1;
+ id1.set_value("1");
+
+ SlaveInfo info1;
+ info1.set_hostname("localhost");
+ info1.mutable_id()->CopyFrom(id1);
+
+ AWAIT_EQ(true, registrar.admit(id1, info1));
+ AWAIT_EQ(false, registrar.admit(id1, info1));
+}
+
+
+TEST_F(RegistrarTest, readmit)
+{
+ Registrar registrar(state);
+
+ SlaveID id1;
+ id1.set_value("1");
+
+ SlaveInfo info1;
+ info1.set_hostname("localhost");
+ info1.mutable_id()->CopyFrom(id1);
+
+ SlaveID id2;
+ id2.set_value("2");
+
+ SlaveInfo info2;
+ info2.set_hostname("localhost");
+ info2.mutable_id()->CopyFrom(id2);
+
+ AWAIT_EQ(true, registrar.admit(id1, info1));
+
+ AWAIT_EQ(true, registrar.readmit(info1));
+
+ AWAIT_EQ(false, registrar.readmit(info2));
+}
+
+
+TEST_F(RegistrarTest, remove)
+{
+ Registrar registrar(state);
+
+ SlaveID id1;
+ id1.set_value("1");
+
+ SlaveInfo info1;
+ info1.set_hostname("localhost");
+ info1.mutable_id()->CopyFrom(id1);
+
+ SlaveID id2;
+ id2.set_value("2");
+
+ SlaveInfo info2;
+ info2.set_hostname("localhost");
+ info2.mutable_id()->CopyFrom(id2);
+
+ AWAIT_EQ(true, registrar.admit(id1, info1));
+
+ AWAIT_EQ(true, registrar.admit(id2, info2));
+
+ AWAIT_EQ(true, registrar.remove(info1));
+
+ AWAIT_EQ(false, registrar.remove(info1));
+
+ AWAIT_EQ(true, registrar.admit(id1, info1));
+
+ AWAIT_EQ(true, registrar.remove(info2));
+}
+
+} // namespace master {
+} // namespace internal {
+} // namespace mesos {
http://git-wip-us.apache.org/repos/asf/mesos/blob/7edb0004/src/tests/state_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/state_tests.cpp b/src/tests/state_tests.cpp
index efc7315..03c5388 100644
--- a/src/tests/state_tests.cpp
+++ b/src/tests/state_tests.cpp
@@ -51,7 +51,7 @@ using namespace mesos::internal;
using namespace process;
-using mesos::internal::registry::Registry;
+using mesos::internal::registry::Slaves;
using mesos::internal::registry::Slave;
using state::LevelDBStorage;
@@ -66,50 +66,50 @@ using state::protobuf::Variable;
void FetchAndStoreAndFetch(State* state)
{
- Future<Variable<Registry> > future1 = state->fetch<Registry>("registry");
+ Future<Variable<Slaves> > future1 = state->fetch<Slaves>("slaves");
AWAIT_READY(future1);
- Variable<Registry> variable = future1.get();
+ Variable<Slaves> variable = future1.get();
- Registry registry1 = variable.get();
- EXPECT_TRUE(registry1.slaves().size() == 0);
+ Slaves slaves1 = variable.get();
+ EXPECT_TRUE(slaves1.slaves().size() == 0);
- Slave* slave = registry1.add_slaves();
+ Slave* slave = slaves1.add_slaves();
slave->mutable_info()->set_hostname("localhost");
- variable = variable.mutate(registry1);
+ variable = variable.mutate(slaves1);
- Future<Option<Variable<Registry> > > future2 = state->store(variable);
+ Future<Option<Variable<Slaves> > > future2 = state->store(variable);
AWAIT_READY(future2);
ASSERT_SOME(future2.get());
- future1 = state->fetch<Registry>("registry");
+ future1 = state->fetch<Slaves>("slaves");
AWAIT_READY(future1);
variable = future1.get();
- Registry registry2 = variable.get();
- ASSERT_TRUE(registry2.slaves().size() == 1);
- EXPECT_EQ("localhost", registry2.slaves(0).info().hostname());
+ Slaves slaves2 = variable.get();
+ ASSERT_TRUE(slaves2.slaves().size() == 1);
+ EXPECT_EQ("localhost", slaves2.slaves(0).info().hostname());
}
void FetchAndStoreAndStoreAndFetch(State* state)
{
- Future<Variable<Registry> > future1 = state->fetch<Registry>("registry");
+ Future<Variable<Slaves> > future1 = state->fetch<Slaves>("slaves");
AWAIT_READY(future1);
- Variable<Registry> variable = future1.get();
+ Variable<Slaves> variable = future1.get();
- Registry registry1 = variable.get();
- EXPECT_TRUE(registry1.slaves().size() == 0);
+ Slaves slaves1 = variable.get();
+ EXPECT_TRUE(slaves1.slaves().size() == 0);
- Slave* slave = registry1.add_slaves();
+ Slave* slave = slaves1.add_slaves();
slave->mutable_info()->set_hostname("localhost");
- variable = variable.mutate(registry1);
+ variable = variable.mutate(slaves1);
- Future<Option<Variable<Registry> > > future2 = state->store(variable);
+ Future<Option<Variable<Slaves> > > future2 = state->store(variable);
AWAIT_READY(future2);
ASSERT_SOME(future2.get());
@@ -119,75 +119,75 @@ void FetchAndStoreAndStoreAndFetch(State* state)
AWAIT_READY(future2);
ASSERT_SOME(future2.get());
- future1 = state->fetch<Registry>("registry");
+ future1 = state->fetch<Slaves>("slaves");
AWAIT_READY(future1);
variable = future1.get();
- Registry registry2 = variable.get();
- ASSERT_TRUE(registry2.slaves().size() == 1);
- EXPECT_EQ("localhost", registry2.slaves(0).info().hostname());
+ Slaves slaves2 = variable.get();
+ ASSERT_TRUE(slaves2.slaves().size() == 1);
+ EXPECT_EQ("localhost", slaves2.slaves(0).info().hostname());
}
void FetchAndStoreAndStoreFailAndFetch(State* state)
{
- Future<Variable<Registry> > future1 = state->fetch<Registry>("registry");
+ Future<Variable<Slaves> > future1 = state->fetch<Slaves>("slaves");
AWAIT_READY(future1);
- Variable<Registry> variable1 = future1.get();
+ Variable<Slaves> variable1 = future1.get();
- Registry registry1 = variable1.get();
- EXPECT_TRUE(registry1.slaves().size() == 0);
+ Slaves slaves1 = variable1.get();
+ EXPECT_TRUE(slaves1.slaves().size() == 0);
- Slave* slave1 = registry1.add_slaves();
+ Slave* slave1 = slaves1.add_slaves();
slave1->mutable_info()->set_hostname("localhost1");
- Variable<Registry> variable2 = variable1.mutate(registry1);
+ Variable<Slaves> variable2 = variable1.mutate(slaves1);
- Future<Option<Variable<Registry> > > future2 = state->store(variable2);
+ Future<Option<Variable<Slaves> > > future2 = state->store(variable2);
AWAIT_READY(future2);
ASSERT_SOME(future2.get());
- Registry registry2 = variable1.get();
- EXPECT_TRUE(registry2.slaves().size() == 0);
+ Slaves slaves2 = variable1.get();
+ EXPECT_TRUE(slaves2.slaves().size() == 0);
- Slave* slave2 = registry2.add_slaves();
+ Slave* slave2 = slaves2.add_slaves();
slave2->mutable_info()->set_hostname("localhost2");
- variable2 = variable1.mutate(registry2);
+ variable2 = variable1.mutate(slaves2);
future2 = state->store(variable2);
AWAIT_READY(future2);
EXPECT_TRUE(future2.get().isNone());
- future1 = state->fetch<Registry>("registry");
+ future1 = state->fetch<Slaves>("slaves");
AWAIT_READY(future1);
variable1 = future1.get();
- registry1 = variable1.get();
- ASSERT_TRUE(registry1.slaves().size() == 1);
- EXPECT_EQ("localhost1", registry1.slaves(0).info().hostname());
+ slaves1 = variable1.get();
+ ASSERT_TRUE(slaves1.slaves().size() == 1);
+ EXPECT_EQ("localhost1", slaves1.slaves(0).info().hostname());
}
void FetchAndStoreAndExpungeAndFetch(State* state)
{
- Future<Variable<Registry> > future1 = state->fetch<Registry>("registry");
+ Future<Variable<Slaves> > future1 = state->fetch<Slaves>("slaves");
AWAIT_READY(future1);
- Variable<Registry> variable = future1.get();
+ Variable<Slaves> variable = future1.get();
- Registry registry1 = variable.get();
- EXPECT_TRUE(registry1.slaves().size() == 0);
+ Slaves slaves1 = variable.get();
+ EXPECT_TRUE(slaves1.slaves().size() == 0);
- Slave* slave = registry1.add_slaves();
+ Slave* slave = slaves1.add_slaves();
slave->mutable_info()->set_hostname("localhost");
- variable = variable.mutate(registry1);
+ variable = variable.mutate(slaves1);
- Future<Option<Variable<Registry> > > future2 = state->store(variable);
+ Future<Option<Variable<Slaves> > > future2 = state->store(variable);
AWAIT_READY(future2);
ASSERT_SOME(future2.get());
@@ -197,32 +197,32 @@ void FetchAndStoreAndExpungeAndFetch(State* state)
AWAIT_READY(future3);
ASSERT_TRUE(future3.get());
- future1 = state->fetch<Registry>("registry");
+ future1 = state->fetch<Slaves>("slaves");
AWAIT_READY(future1);
variable = future1.get();
- Registry registry2 = variable.get();
- ASSERT_EQ(0, registry2.slaves().size());
+ Slaves slaves2 = variable.get();
+ ASSERT_EQ(0, slaves2.slaves().size());
}
void FetchAndStoreAndExpungeAndExpunge(State* state)
{
- Future<Variable<Registry> > future1 = state->fetch<Registry>("registry");
+ Future<Variable<Slaves> > future1 = state->fetch<Slaves>("slaves");
AWAIT_READY(future1);
- Variable<Registry> variable = future1.get();
+ Variable<Slaves> variable = future1.get();
- Registry registry1 = variable.get();
- EXPECT_TRUE(registry1.slaves().size() == 0);
+ Slaves slaves1 = variable.get();
+ EXPECT_TRUE(slaves1.slaves().size() == 0);
- Slave* slave = registry1.add_slaves();
+ Slave* slave = slaves1.add_slaves();
slave->mutable_info()->set_hostname("localhost");
- variable = variable.mutate(registry1);
+ variable = variable.mutate(slaves1);
- Future<Option<Variable<Registry> > > future2 = state->store(variable);
+ Future<Option<Variable<Slaves> > > future2 = state->store(variable);
AWAIT_READY(future2);
ASSERT_SOME(future2.get());
@@ -240,20 +240,20 @@ void FetchAndStoreAndExpungeAndExpunge(State* state)
void FetchAndStoreAndExpungeAndStoreAndFetch(State* state)
{
- Future<Variable<Registry> > future1 = state->fetch<Registry>("registry");
+ Future<Variable<Slaves> > future1 = state->fetch<Slaves>("slaves");
AWAIT_READY(future1);
- Variable<Registry> variable = future1.get();
+ Variable<Slaves> variable = future1.get();
- Registry registry1 = variable.get();
- EXPECT_TRUE(registry1.slaves().size() == 0);
+ Slaves slaves1 = variable.get();
+ EXPECT_TRUE(slaves1.slaves().size() == 0);
- Slave* slave = registry1.add_slaves();
+ Slave* slave = slaves1.add_slaves();
slave->mutable_info()->set_hostname("localhost");
- variable = variable.mutate(registry1);
+ variable = variable.mutate(slaves1);
- Future<Option<Variable<Registry> > > future2 = state->store(variable);
+ Future<Option<Variable<Slaves> > > future2 = state->store(variable);
AWAIT_READY(future2);
ASSERT_SOME(future2.get());
@@ -267,40 +267,40 @@ void FetchAndStoreAndExpungeAndStoreAndFetch(State* state)
AWAIT_READY(future2);
ASSERT_SOME(future2.get());
- future1 = state->fetch<Registry>("registry");
+ future1 = state->fetch<Slaves>("slaves");
AWAIT_READY(future1);
variable = future1.get();
- Registry registry2 = variable.get();
- ASSERT_TRUE(registry2.slaves().size() == 1);
- EXPECT_EQ("localhost", registry2.slaves(0).info().hostname());
+ Slaves slaves2 = variable.get();
+ ASSERT_TRUE(slaves2.slaves().size() == 1);
+ EXPECT_EQ("localhost", slaves2.slaves(0).info().hostname());
}
void Names(State* state)
{
- Future<Variable<Registry> > future1 = state->fetch<Registry>("registry");
+ Future<Variable<Slaves> > future1 = state->fetch<Slaves>("slaves");
AWAIT_READY(future1);
- Variable<Registry> variable = future1.get();
+ Variable<Slaves> variable = future1.get();
- Registry registry1 = variable.get();
- EXPECT_TRUE(registry1.slaves().size() == 0);
+ Slaves slaves1 = variable.get();
+ EXPECT_TRUE(slaves1.slaves().size() == 0);
- Slave* slave = registry1.add_slaves();
+ Slave* slave = slaves1.add_slaves();
slave->mutable_info()->set_hostname("localhost");
- variable = variable.mutate(registry1);
+ variable = variable.mutate(slaves1);
- Future<Option<Variable<Registry> > > future2 = state->store(variable);
+ Future<Option<Variable<Slaves> > > future2 = state->store(variable);
AWAIT_READY(future2);
ASSERT_SOME(future2.get());
Future<std::vector<std::string> > names = state->names();
AWAIT_READY(names);
ASSERT_TRUE(names.get().size() == 1);
- EXPECT_EQ("registry", names.get()[0]);
+ EXPECT_EQ("slaves", names.get()[0]);
}