Merge pull request apple#56 from atn34/switch-cluster-files

Switch cluster files - multi version client changes
atn34 · Oct 28, 2019 · 24d8336 · 24d8336
2 parents 75ec5fd + 7309754
commit 24d8336
Show file tree

Hide file tree

Showing 5 changed files with 63 additions and 7 deletions.
diff --git a/fdbclient/DatabaseContext.h b/fdbclient/DatabaseContext.h
@@ -106,8 +106,9 @@ class DatabaseContext : public ReferenceCounted<DatabaseContext>, public FastAll
 	Future<Void> switchConnectionFile(Reference<ClusterConnectionFile> standby);
 	Future<Void> connectionFileChanged();
 	bool switchable = false;
+	Future<Void> switchActor = Never();
 
-//private: 
+	// private:
 	explicit DatabaseContext( Reference<AsyncVar<Reference<ClusterConnectionFile>>> connectionFile, Reference<AsyncVar<ClientDBInfo>> clientDBInfo,
 		Future<Void> clientInfoMonitor, TaskPriority taskID, LocalityData const& clientLocality, 
 		bool enableLocalityLoadBalance, bool lockAware, bool internal = true, int apiVersion = Database::API_VERSION_LATEST, bool switchable = false );

diff --git a/fdbclient/MultiVersionTransaction.actor.cpp b/fdbclient/MultiVersionTransaction.actor.cpp
@@ -548,13 +548,61 @@ void MultiVersionTransaction::addReadConflictRange(const KeyRangeRef& keys) {
 }
 
 void MultiVersionTransaction::atomicOp(const KeyRef& key, const ValueRef& value, uint32_t operationType) {
-	throw client_invalid_operation();
+	auto tr = getTransaction();
+	if (tr.transaction) {
+		tr.transaction->atomicOp(key, value, operationType);
+	}
 }
 
+struct OnDbReady : ThreadCallback, ThreadSafeReferenceCounted<OnDbReady> {
+	bool canFire(int notMadeActive) override { return true; }
+	void fire(const Void& unused, int& userParam) override {
+		onMainThreadVoid(
+		    [this]() {
+			    auto tr = db->createTransaction();
+			    tr->set(LiteralStringRef("\xff\xff/cluster_file_path"), value);
+			    delref();
+		    },
+		    nullptr);
+	}
+	void error(const Error& e, int& userParam) override {
+		if (e.code() != error_code_operation_cancelled) {
+			TraceEvent(SevError, "OnDbReadyError").error(e).detail("ClientLibrary", this->client->libPath);
+		}
+		delref();
+	}
+	OnDbReady() = default;
+	Reference<ClientInfo> client;
+	Reference<IDatabase> db;
+	Value value;
+};
+
 void MultiVersionTransaction::set(const KeyRef& key, const ValueRef& value) {
-	auto tr = getTransaction();
-	if(tr.transaction) {
-		tr.transaction->set(key, value);
+	if (key == LiteralStringRef("\xff\xff/cluster_file_path")) {
+		onMainThreadVoid(
+		    [db = this->db, value = Value(value)] {
+			    for (const auto& connector : db->dbState->connectionAttempts) {
+				    connector->clusterFilePath = value.toString();
+				    ThreadFuture<Void> onReady;
+				    if (connector->client->external) {
+					    onReady = connector->candidateDatabase.castTo<DLDatabase>()->onReady();
+				    } else {
+					    onReady = ThreadFuture<Void>(Void());
+				    }
+				    Reference<OnDbReady> cb = Reference<OnDbReady>(new OnDbReady);
+				    cb->client = connector->client;
+				    cb->db = connector->candidateDatabase;
+				    cb->value = value;
+				    int userParam;
+				    onReady.callOrSetAsCallback(cb.extractPtr(), userParam, 0);
+			    }
+		    },
+		    nullptr);
+	} else {
+		auto tr = getTransaction();
+		if (tr.transaction) {
+			tr.transaction->set(key, value);
+		}
 	}
 }
 

diff --git a/fdbclient/MultiVersionTransaction.h b/fdbclient/MultiVersionTransaction.h
@@ -314,7 +314,7 @@ class MultiVersionDatabase : public IDatabase, ThreadSafeReferenceCounted<MultiV
 		void error(const Error& e, int& userParam);
 
 		const Reference<ClientInfo> client;
-		const std::string clusterFilePath;
+		std::string clusterFilePath;
 
 		const Reference<DatabaseState> dbState;
 

diff --git a/fdbclient/NativeAPI.actor.cpp b/fdbclient/NativeAPI.actor.cpp
@@ -767,7 +767,9 @@ Reference<ClusterConnectionFile> DatabaseContext::getConnectionFile() {
 
 Future<Void> DatabaseContext::switchConnectionFile(Reference<ClusterConnectionFile> standby) {
 	ASSERT(switchable);
-	return switchConnectionFileImpl(standby, this);
+	switchActor.cancel();
+	switchActor = switchConnectionFileImpl(standby, this);
+	return switchActor;
 }
 
 Future<Void> DatabaseContext::connectionFileChanged() {

diff --git a/fdbclient/ReadYourWrites.actor.cpp b/fdbclient/ReadYourWrites.actor.cpp
@@ -1614,6 +1614,11 @@ void ReadYourWritesTransaction::set( const KeyRef& key, const ValueRef& value )
 		BinaryReader::fromStringRef<ClientWorkerInterface>(value, IncludeVersion()).reboot.send( RebootRequest(false, true) );
 		return;
 	}
+	if (key == LiteralStringRef("\xff\xff/cluster_file_path")) {
+		uncancellable(holdWhile(getDatabase(), getDatabase()->switchConnectionFile(Reference<ClusterConnectionFile>(
+		                                           new ClusterConnectionFile(value.toString())))));
+		return;
+	}
 	if (key == metadataVersionKey) {
 		throw client_invalid_operation();
 	}