Create configuration database node files on demand

fdbserver/CMakeLists.txt

@@ -70,6 +70,8 @@ set(FDBSERVER_SRCS
   OldTLogServer_4_6.actor.cpp
   OldTLogServer_6_0.actor.cpp
   OldTLogServer_6_2.actor.cpp
+  OnDemandStore.actor.cpp
+  OnDemandStore.h
   Orderer.actor.h
   PaxosConfigConsumer.actor.cpp
   PaxosConfigConsumer.h

fdbserver/ConfigDatabaseUnitTests.actor.cpp

@@ -59,18 +59,10 @@ class WriteToTransactionEnvironment {
 	ConfigTransactionInterface cti;
 	ConfigFollowerInterface cfi;
 	Reference<IConfigDatabaseNode> node;
-	UID nodeID;
 	Future<Void> ctiServer;
 	Future<Void> cfiServer;
 	Version lastWrittenVersion{ 0 };
 
-	ACTOR static Future<Void> setupNode(WriteToTransactionEnvironment* self) {
-		wait(self->node->initialize("./", self->nodeID));
-		self->ctiServer = self->node->serve(self->cti);
-		self->cfiServer = self->node->serve(self->cfi);
-		return Void();
-	}
-
 	ACTOR static Future<Void> set(WriteToTransactionEnvironment* self, Optional<KeyRef> configClass, int64_t value) {
 		state Reference<IConfigTransaction> tr = IConfigTransaction::createSimple(self->cti);
 		auto configKey = encodeConfigKey(configClass, "test_long"_sr);
@@ -89,11 +81,13 @@ class WriteToTransactionEnvironment {
 		return Void();
 	}
 
-public:
-	WriteToTransactionEnvironment()
-	  : node(IConfigDatabaseNode::createSimple()), nodeID(deterministicRandom()->randomUniqueID()) {}
+	void setup() {
+		ctiServer = node->serve(cti);
+		cfiServer = node->serve(cfi);
+	}
 
-	Future<Void> setup() { return setupNode(this); }
+public:
+	WriteToTransactionEnvironment() : node(IConfigDatabaseNode::createSimple("./")) { setup(); }
 
 	Future<Void> set(Optional<KeyRef> configClass, int64_t value) { return set(this, configClass, value); }
 
@@ -101,11 +95,11 @@ public:
 
 	Future<Void> compact() { return cfi.compact.getReply(ConfigFollowerCompactRequest{ lastWrittenVersion }); }
 
-	Future<Void> restartNode() {
+	void restartNode() {
 		cfiServer.cancel();
 		ctiServer.cancel();
-		node = IConfigDatabaseNode::createSimple();
-		return setupNode(this);
+		node = IConfigDatabaseNode::createSimple("./");
+		setup();
 	}
 
 	ConfigTransactionInterface getTransactionInterface() const { return cti; }
@@ -268,10 +262,10 @@ class TransactionEnvironment {
 	}
 
 public:
-	Future<Void> setup() { return writeTo.setup(); }
-
-	Future<Void> restartNode() { return writeTo.restartNode(); }
+	// TODO: Remove this?
+	Future<Void> setup() { return Void(); }
 
+	void restartNode() { writeTo.restartNode(); }
 	Future<Void> set(Optional<KeyRef> configClass, int64_t value) { return writeTo.set(configClass, value); }
 	Future<Void> clear(Optional<KeyRef> configClass) { return writeTo.clear(configClass); }
 	Future<Void> check(Optional<KeyRef> configClass, Optional<int64_t> expected) {
@@ -289,7 +283,6 @@ class TransactionToLocalConfigEnvironment {
 	Future<Void> broadcastServer;
 
 	ACTOR static Future<Void> setup(TransactionToLocalConfigEnvironment* self) {
-		wait(self->writeTo.setup());
 		wait(self->readFrom.setup());
 		self->readFrom.connectToBroadcaster(IDependentAsyncVar<ConfigBroadcastFollowerInterface>::create(self->cbfi));
 		self->broadcastServer = self->broadcaster.serve(self->cbfi->get());
@@ -303,7 +296,7 @@ public:
 
 	Future<Void> setup() { return setup(this); }
 
-	Future<Void> restartNode() { return writeTo.restartNode(); }
+	void restartNode() { writeTo.restartNode(); }
 
 	void changeBroadcaster() {
 		broadcastServer.cancel();
@@ -554,7 +547,7 @@ TEST_CASE("/fdbserver/ConfigDB/TransactionToLocalConfig/RestartNode") {
 	state TransactionToLocalConfigEnvironment env("class-A");
 	wait(env.setup());
 	wait(set(env, "class-A"_sr, 1));
-	wait(env.restartNode());
+	env.restartNode();
 	wait(check(env, 1));
 	return Void();
 }
@@ -593,7 +586,7 @@ TEST_CASE("/fdbserver/ConfigDB/Transaction/Restart") {
 	state TransactionEnvironment env;
 	wait(env.setup());
 	wait(set(env, "class-A"_sr, 1));
-	wait(env.restartNode());
+	env.restartNode();
 	wait(check(env, "class-A"_sr, 1));
 	return Void();
 }

fdbserver/Coordination.actor.cpp

@@ -23,6 +23,7 @@
 #include "fdbserver/IConfigDatabaseNode.h"
 #include "fdbserver/IKeyValueStore.h"
 #include "fdbserver/Knobs.h"
+#include "fdbserver/OnDemandStore.h"
 #include "fdbserver/WorkerInterface.actor.h"
 #include "fdbserver/Status.h"
 #include "flow/ActorCollection.h"
@@ -78,51 +79,6 @@ ServerCoordinators::ServerCoordinators(Reference<ClusterConnectionFile> cf) : Cl
 	}
 }
 
-// The coordination server wants to create its key value store only if it is actually used
-struct OnDemandStore {
-public:
-	OnDemandStore(std::string folder, UID myID) : folder(folder), store(nullptr), myID(myID) {}
-	~OnDemandStore() {
-		if (store)
-			store->close();
-	}
-
-	IKeyValueStore* get() {
-		if (!store)
-			open();
-		return store;
-	}
-
-	bool exists() const {
-		if (store)
-			return true;
-		return fileExists(joinPath(folder, "coordination-0.fdq")) ||
-		       fileExists(joinPath(folder, "coordination-1.fdq")) || fileExists(joinPath(folder, "coordination.fdb"));
-	}
-
-	IKeyValueStore* operator->() { return get(); }
-
-	Future<Void> getError() const { return onErr(err.getFuture()); }
-
-private:
-	std::string folder;
-	UID myID;
-	IKeyValueStore* store;
-	Promise<Future<Void>> err;
-
-	ACTOR static Future<Void> onErr(Future<Future<Void>> e) {
-		Future<Void> f = wait(e);
-		wait(f);
-		return Void();
-	}
-
-	void open() {
-		platform::createDirectory(folder);
-		store = keyValueStoreMemory(joinPath(folder, "coordination-"), myID, 500e6);
-		err.send(store->getError());
-	}
-};
-
 ACTOR Future<Void> localGenerationReg(GenerationRegInterface interf, OnDemandStore* pstore) {
 	state GenerationRegVal v;
 	state OnDemandStore& store = *pstore;
@@ -180,7 +136,8 @@ ACTOR Future<Void> localGenerationReg(GenerationRegInterface interf, OnDemandSto
 TEST_CASE("/fdbserver/Coordination/localGenerationReg/simple") {
 	state GenerationRegInterface reg;
 	state OnDemandStore store("simfdb/unittests/", //< FIXME
-	                          deterministicRandom()->randomUniqueID());
+	                          deterministicRandom()->randomUniqueID(),
+	                          "coordination");
 	state Future<Void> actor = localGenerationReg(reg, &store);
 	state Key the_key(deterministicRandom()->randomAlphaNumeric(deterministicRandom()->randomInt(0, 10)));
 
@@ -618,7 +575,7 @@ ACTOR Future<Void> coordinationServer(std::string dataFolder, Optional<bool> use
 	state UID myID = deterministicRandom()->randomUniqueID();
 	state LeaderElectionRegInterface myLeaderInterface(g_network);
 	state GenerationRegInterface myInterface(g_network);
-	state OnDemandStore store(dataFolder, myID);
+	state OnDemandStore store(dataFolder, myID, "coordination");
 	state ConfigTransactionInterface configTransactionInterface;
 	state ConfigFollowerInterface configFollowerInterface;
 	state Reference<IConfigDatabaseNode> configDatabaseNode;
@@ -631,11 +588,10 @@ ACTOR Future<Void> coordinationServer(std::string dataFolder, Optional<bool> use
 		configTransactionInterface.setupWellKnownEndpoints();
 		configFollowerInterface.setupWellKnownEndpoints();
 		if (useTestConfigDB.get()) {
-			configDatabaseNode = IConfigDatabaseNode::createSimple();
+			configDatabaseNode = IConfigDatabaseNode::createSimple(dataFolder);
 		} else {
-			configDatabaseNode = IConfigDatabaseNode::createPaxos();
+			configDatabaseNode = IConfigDatabaseNode::createPaxos(dataFolder);
 		}
-		wait(configDatabaseNode->initialize(dataFolder, UID{}));
 		configDatabaseServer =
 		    configDatabaseNode->serve(configTransactionInterface) || configDatabaseNode->serve(configFollowerInterface);
 	}

fdbserver/IConfigDatabaseNode.cpp

@@ -22,10 +22,10 @@
 #include "fdbserver/PaxosConfigDatabaseNode.h"
 #include "fdbserver/SimpleConfigDatabaseNode.h"
 
-Reference<IConfigDatabaseNode> IConfigDatabaseNode::createSimple() {
-	return makeReference<SimpleConfigDatabaseNode>();
+Reference<IConfigDatabaseNode> IConfigDatabaseNode::createSimple(std::string const& folder) {
+	return makeReference<SimpleConfigDatabaseNode>(folder);
 }
 
-Reference<IConfigDatabaseNode> IConfigDatabaseNode::createPaxos() {
-	return makeReference<PaxosConfigDatabaseNode>();
+Reference<IConfigDatabaseNode> IConfigDatabaseNode::createPaxos(std::string const& folder) {
+	return makeReference<PaxosConfigDatabaseNode>(folder);
 }

fdbserver/IConfigDatabaseNode.h

@@ -31,8 +31,7 @@ class IConfigDatabaseNode : public ReferenceCounted<IConfigDatabaseNode> {
 public:
 	virtual Future<Void> serve(ConfigTransactionInterface const&) = 0;
 	virtual Future<Void> serve(ConfigFollowerInterface const&) = 0;
-	virtual Future<Void> initialize(std::string const& dataFolder, UID id) = 0;
 
-	static Reference<IConfigDatabaseNode> createSimple();
-	static Reference<IConfigDatabaseNode> createPaxos();
+	static Reference<IConfigDatabaseNode> createSimple(std::string const& folder);
+	static Reference<IConfigDatabaseNode> createPaxos(std::string const& folder);
 };

fdbserver/OnDemandStore.actor.cpp

@@ -0,0 +1,63 @@
+/*
+ * OnDemandStore.actor.cpp
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2013-2018 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed 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 "fdbserver/OnDemandStore.h"
+#include "flow/actorcompiler.h" // must be last include
+
+ACTOR static Future<Void> onErr(Future<Future<Void>> e) {
+	Future<Void> f = wait(e);
+	wait(f);
+	return Void();
+}
+
+void OnDemandStore::open() {
+	platform::createDirectory(folder);
+	store = keyValueStoreMemory(joinPath(folder, prefix), myID, 500e6);
+	err.send(store->getError());
+}
+
+OnDemandStore::OnDemandStore(std::string const& folder, UID myID, std::string const& prefix)
+  : folder(folder), prefix(prefix), store(nullptr), myID(myID) {}
+
+OnDemandStore::~OnDemandStore() {
+	if (store) {
+		store->close();
+	}
+}
+
+IKeyValueStore* OnDemandStore::get() {
+	if (!store) {
+		open();
+	}
+	return store;
+}
+
+bool OnDemandStore::exists() const {
+	return store || fileExists(joinPath(folder, prefix + "-0.fdq")) ||
+	       fileExists(joinPath(folder, prefix + "-1.fdq")) || fileExists(joinPath(folder, prefix + ".fdb"));
+}
+
+IKeyValueStore* OnDemandStore::operator->() {
+	return get();
+}
+
+Future<Void> OnDemandStore::getError() const {
+	return onErr(err.getFuture());
+}

fdbserver/OnDemandStore.h

@@ -0,0 +1,44 @@
+/*
+ * OnDemandStore.h
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2013-2018 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed 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.
+ */
+
+#pragma once
+
+#include "flow/Arena.h"
+#include "flow/IRandom.h"
+#include "flow/Platform.h"
+#include "fdbserver/IKeyValueStore.h"
+
+// Create a key value store if and only if it is actually used
+class OnDemandStore : NonCopyable {
+	std::string folder;
+	UID myID;
+	IKeyValueStore* store;
+	Promise<Future<Void>> err;
+	std::string prefix;
+	void open();
+
+public:
+	OnDemandStore(std::string const& folder, UID myID, std::string const& prefix);
+	~OnDemandStore();
+	IKeyValueStore* get();
+	bool exists() const;
+	IKeyValueStore* operator->();
+	Future<Void> getError() const;
+};

fdbserver/PaxosConfigDatabaseNode.actor.cpp

@@ -22,7 +22,10 @@
 
 class PaxosConfigDatabaseNodeImpl {};
 
-PaxosConfigDatabaseNode::PaxosConfigDatabaseNode() : impl(std::make_unique<PaxosConfigDatabaseNodeImpl>()) {}
+PaxosConfigDatabaseNode::PaxosConfigDatabaseNode(std::string const& folder) {
+	// TODO: Implement
+	ASSERT(false);
+}
 
 PaxosConfigDatabaseNode::~PaxosConfigDatabaseNode() = default;
 
@@ -37,9 +40,3 @@ Future<Void> PaxosConfigDatabaseNode::serve(ConfigFollowerInterface const& cfi)
 	ASSERT(false);
 	return Void();
 }
-
-Future<Void> PaxosConfigDatabaseNode::initialize(std::string const& dataFolder, UID id) {
-	// TODO: Implement
-	ASSERT(false);
-	return Void();
-}

fdbserver/PaxosConfigDatabaseNode.h

@@ -26,9 +26,8 @@ class PaxosConfigDatabaseNode : public IConfigDatabaseNode {
 	std::unique_ptr<class PaxosConfigDatabaseNodeImpl> impl;
 
 public:
-	PaxosConfigDatabaseNode();
+	PaxosConfigDatabaseNode(std::string const& folder);
 	~PaxosConfigDatabaseNode();
 	Future<Void> serve(ConfigTransactionInterface const&) override;
 	Future<Void> serve(ConfigFollowerInterface const&) override;
-	Future<Void> initialize(std::string const& dataFolder, UID id) override;
 };

fdbserver/SimpleConfigDatabaseNode.actor.cpp

@@ -22,6 +22,7 @@
 
 #include "fdbserver/SimpleConfigDatabaseNode.h"
 #include "fdbserver/IKeyValueStore.h"
+#include "fdbserver/OnDemandStore.h"
 #include "flow/Arena.h"
 #include "flow/genericactors.actor.h"
 #include "flow/UnitTest.h"
@@ -82,11 +83,11 @@ TEST_CASE("/fdbserver/ConfigDB/SimpleConfigDatabaseNode/Internal/versionedMutati
 }
 
 class SimpleConfigDatabaseNodeImpl {
-	IKeyValueStore* kvStore; // FIXME: Prevent leak
-	std::map<std::string, std::string> config;
-	Future<Void> initFuture;
-
 	UID id;
+	// TODO: Listen for errors
+	OnDemandStore kvStore;
+	std::map<std::string, std::string> config;
+
 	CounterCollection cc;
 
 	// Follower counters
@@ -282,7 +283,6 @@ class SimpleConfigDatabaseNodeImpl {
 	}
 
 	ACTOR static Future<Void> serve(SimpleConfigDatabaseNodeImpl* self, ConfigTransactionInterface const* cti) {
-		ASSERT(self->initFuture.isValid() && self->initFuture.isReady());
 		loop {
 			//wait(traceQueuedMutations(self));
 			choose {
@@ -367,7 +367,6 @@ class SimpleConfigDatabaseNodeImpl {
 	}
 
 	ACTOR static Future<Void> serve(SimpleConfigDatabaseNodeImpl* self, ConfigFollowerInterface const* cfi) {
-		ASSERT(self->initFuture.isValid() && self->initFuture.isReady());
 		loop {
 			choose {
 				when(ConfigFollowerGetSnapshotAndChangesRequest req =
@@ -387,35 +386,24 @@ class SimpleConfigDatabaseNodeImpl {
 	}
 
 public:
-	SimpleConfigDatabaseNodeImpl()
-	  : cc("ConfigDatabaseNode"), compactRequests("CompactRequests", cc),
-	    successfulChangeRequests("SuccessfulChangeRequests", cc), failedChangeRequests("FailedChangeRequests", cc),
-	    snapshotRequests("SnapshotRequests", cc), successfulCommits("SuccessfulCommits", cc),
-	    failedCommits("FailedCommits", cc), setMutations("SetMutations", cc), clearMutations("ClearMutations", cc),
-	    getValueRequests("GetValueRequests", cc), newVersionRequests("NewVersionRequests", cc) {}
-
-	~SimpleConfigDatabaseNodeImpl() {
-		if (kvStore) {
-			kvStore->close();
-		}
+	SimpleConfigDatabaseNodeImpl(std::string const& folder)
+	  : id(deterministicRandom()->randomUniqueID()), kvStore(folder, id, "global-conf"), cc("ConfigDatabaseNode"),
+	    compactRequests("CompactRequests", cc), successfulChangeRequests("SuccessfulChangeRequests", cc),
+	    failedChangeRequests("FailedChangeRequests", cc), snapshotRequests("SnapshotRequests", cc),
+	    successfulCommits("SuccessfulCommits", cc), failedCommits("FailedCommits", cc),
+	    setMutations("SetMutations", cc), clearMutations("ClearMutations", cc),
+	    getValueRequests("GetValueRequests", cc), newVersionRequests("NewVersionRequests", cc) {
+		logger = traceCounters(
+		    "ConfigDatabaseNodeMetrics", id, SERVER_KNOBS->WORKER_LOGGING_INTERVAL, &cc, "ConfigDatabaseNode");
 	}
 
 	Future<Void> serve(ConfigTransactionInterface const& cti) { return serve(this, &cti); }
 
 	Future<Void> serve(ConfigFollowerInterface const& cfi) { return serve(this, &cfi); }
-
-	Future<Void> initialize(std::string const& dataFolder, UID id) {
-		platform::createDirectory(dataFolder);
-		this->id = id;
-		kvStore = keyValueStoreMemory(joinPath(dataFolder, "globalconf-" + id.toString()), id, 500e6);
-		logger = traceCounters(
-		    "ConfigDatabaseNodeMetrics", id, SERVER_KNOBS->WORKER_LOGGING_INTERVAL, &cc, "ConfigDatabaseNode");
-		initFuture = kvStore->init();
-		return initFuture;
-	}
 };
 
-SimpleConfigDatabaseNode::SimpleConfigDatabaseNode() : impl(std::make_unique<SimpleConfigDatabaseNodeImpl>()) {}
+SimpleConfigDatabaseNode::SimpleConfigDatabaseNode(std::string const& folder)
+  : impl(std::make_unique<SimpleConfigDatabaseNodeImpl>(folder)) {}
 
 SimpleConfigDatabaseNode::~SimpleConfigDatabaseNode() = default;
 
@@ -426,7 +414,3 @@ Future<Void> SimpleConfigDatabaseNode::serve(ConfigTransactionInterface const& c
 Future<Void> SimpleConfigDatabaseNode::serve(ConfigFollowerInterface const& cfi) {
 	return impl->serve(cfi);
 }
-
-Future<Void> SimpleConfigDatabaseNode::initialize(std::string const& dataFolder, UID id) {
-	return impl->initialize(dataFolder, id);
-}

fdbserver/SimpleConfigDatabaseNode.h

@@ -26,9 +26,8 @@ class SimpleConfigDatabaseNode : public IConfigDatabaseNode {
 	std::unique_ptr<class SimpleConfigDatabaseNodeImpl> impl;
 
 public:
-	SimpleConfigDatabaseNode();
+	SimpleConfigDatabaseNode(std::string const& folder);
 	~SimpleConfigDatabaseNode();
 	Future<Void> serve(ConfigTransactionInterface const&) override;
 	Future<Void> serve(ConfigFollowerInterface const&) override;
-	Future<Void> initialize(std::string const& dataFolder, UID id) override;
 };