Revert "remove all multi-version client code related to grv cache"

This reverts commit 6d05ecffc2.

bindings/c/CMakeLists.txt

@@ -1,6 +1,8 @@
 set(FDB_C_SRCS
   fdb_c.cpp
-  foundationdb/fdb_c.h)
+  foundationdb/fdb_c.h
+  foundationdb/fdb_c_internal.h
+  foundationdb/fdb_c_types.h)
 
 file(MAKE_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/foundationdb)
 

bindings/c/fdb_c.cpp

@@ -26,6 +26,7 @@
 #include "fdbclient/MultiVersionTransaction.h"
 #include "fdbclient/MultiVersionAssignmentVars.h"
 #include "foundationdb/fdb_c.h"
+#include "foundationdb/fdb_c_internal.h"
 
 int g_api_version = 0;
 
@@ -416,6 +417,14 @@ extern "C" DLLEXPORT FDBFuture* fdb_database_create_snapshot(FDBDatabase* db,
 	                        .extractPtr());
 }
 
+extern "C" DLLEXPORT DatabaseSharedState* fdb_database_create_shared_state(FDBDatabase* db) {
+	return (DatabaseSharedState*)(DB(db)->createSharedState());
+}
+
+extern "C" DLLEXPORT void fdb_database_set_shared_state(FDBDatabase* db, DatabaseSharedState* p) {
+	(DB(db)->setSharedState(p));
+}
+
 // Get network thread busyness (updated every 1s)
 // A value of 0 indicates that the client is more or less idle
 // A value of 1 (or more) indicates that the client is saturated

bindings/c/foundationdb/fdb_c.h

@@ -58,20 +58,12 @@
 #include <stdint.h>
 
 #include "fdb_c_options.g.h"
+#include "fdb_c_types.h"
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-/* Pointers to these opaque types represent objects in the FDB API */
-typedef struct FDB_future FDBFuture;
-typedef struct FDB_result FDBResult;
-typedef struct FDB_database FDBDatabase;
-typedef struct FDB_transaction FDBTransaction;
-
-typedef int fdb_error_t;
-typedef int fdb_bool_t;
-
 DLLEXPORT const char* fdb_get_error(fdb_error_t code);
 
 DLLEXPORT fdb_bool_t fdb_error_predicate(int predicate_test, fdb_error_t code);

bindings/c/foundationdb/fdb_c_internal.h

@@ -0,0 +1,45 @@
+/*
+ * fdb_c_internal.h
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2013-2022 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.
+ */
+
+#ifndef FDB_C_INTERNAL_H
+#define FDB_C_INTERNAL_H
+#pragma once
+
+#ifndef DLLEXPORT
+#define DLLEXPORT
+#endif
+
+#include "fdb_c_types.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+// forward declaration and typedef
+typedef struct DatabaseSharedState DatabaseSharedState;
+
+DLLEXPORT DatabaseSharedState* fdb_database_create_shared_state(FDBDatabase* db);
+
+DLLEXPORT void fdb_database_set_shared_state(FDBDatabase* db, DatabaseSharedState* p);
+
+#ifdef __cplusplus
+}
+#endif
+#endif

bindings/c/foundationdb/fdb_c_types.h

@@ -0,0 +1,45 @@
+/*
+ * fdb_c_types.h
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2013-2022 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.
+ */
+
+#ifndef FDB_C_TYPES_H
+#define FDB_C_TYPES_H
+#pragma once
+
+#ifndef DLLEXPORT
+#define DLLEXPORT
+#endif
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/* Pointers to these opaque types represent objects in the FDB API */
+typedef struct FDB_future FDBFuture;
+typedef struct FDB_result FDBResult;
+typedef struct FDB_database FDBDatabase;
+typedef struct FDB_transaction FDBTransaction;
+
+typedef int fdb_error_t;
+typedef int fdb_bool_t;
+
+#ifdef __cplusplus
+}
+#endif
+#endif

fdbclient/DatabaseContext.h

@@ -29,6 +29,7 @@
 #include <unordered_map>
 #pragma once
 
+#include "fdbclient/FDBTypes.h"
 #include "fdbclient/NativeAPI.actor.h"
 #include "fdbclient/KeyRangeMap.h"
 #include "fdbclient/CommitProxyInterface.h"
@@ -480,6 +481,11 @@ public:
 	int outstandingWatches;
 	int maxOutstandingWatches;
 
+	// Manage any shared state that may be used by MVC
+	DatabaseSharedState* sharedStatePtr;
+	DatabaseSharedState* initSharedState();
+	void setSharedState(DatabaseSharedState* p);
+
 	// GRV Cache
 	// Database-level read version cache storing the most recent successful GRV as well as the time it was requested.
 	double lastGrvTime;

fdbclient/FDBTypes.h

@@ -1329,6 +1329,21 @@ struct TenantMode {
 
 	uint32_t mode;
 };
+struct GRVCacheSpace {
+	Version cachedReadVersion;
+	double lastGrvTime;
+
+	GRVCacheSpace() : cachedReadVersion(Version(0)), lastGrvTime(0.0) {}
+};
+
+// This structure can be extended in the future to include additional features that required a shared state
+struct DatabaseSharedState {
+	Mutex mutexLock;
+	GRVCacheSpace grvCacheSpace;
+	int refCount;
+
+	DatabaseSharedState() : mutexLock(Mutex()), grvCacheSpace(GRVCacheSpace()), refCount(0) {}
+};
 
 inline bool isValidPerpetualStorageWiggleLocality(std::string locality) {
 	int pos = locality.find(':');

fdbclient/IClientApi.h

@@ -137,6 +137,10 @@ public:
 	// Management API, create snapshot
 	virtual ThreadFuture<Void> createSnapshot(const StringRef& uid, const StringRef& snapshot_command) = 0;
 
+	// Interface to manage shared state across multiple connections to the same Database
+	virtual DatabaseSharedState* createSharedState() = 0;
+	virtual void setSharedState(DatabaseSharedState* p) = 0;
+
 	// used in template functions as the Transaction type that can be created through createTransaction()
 	using TransactionT = ITransaction;
 };

fdbclient/MultiVersionTransaction.actor.cpp

@@ -447,6 +447,19 @@ ThreadFuture<Void> DLDatabase::createSnapshot(const StringRef& uid, const String
 	return toThreadFuture<Void>(api, f, [](FdbCApi::FDBFuture* f, FdbCApi* api) { return Void(); });
 }
 
+DatabaseSharedState* DLDatabase::createSharedState() {
+	if (!api->databaseCreateSharedState) {
+		return nullptr;
+	}
+	return api->databaseCreateSharedState(db);
+}
+
+void DLDatabase::setSharedState(DatabaseSharedState* p) {
+	if (api->databaseSetSharedState) {
+		api->databaseSetSharedState(db, p);
+	}
+}
+
 // Get network thread busyness
 double DLDatabase::getMainThreadBusyness() {
 	if (api->databaseGetMainThreadBusyness != nullptr) {
@@ -723,6 +736,7 @@ Reference<IDatabase> DLApi::createDatabase609(const char* clusterFilePath) {
 Reference<IDatabase> DLApi::createDatabase(const char* clusterFilePath) {
 	if (headerVersion >= 610) {
 		FdbCApi::FDBDatabase* db;
+		// can the FdbCApi wrapper signature be changed to add this ptr?
 		throwIfError(api->createDatabase(clusterFilePath, &db));
 		return Reference<IDatabase>(new DLDatabase(api, db));
 	} else {
@@ -1177,7 +1191,13 @@ MultiVersionDatabase::MultiVersionDatabase(MultiVersionApi* api,
   : dbState(new DatabaseState(clusterFilePath, versionMonitorDb)) {
 	dbState->db = db;
 	dbState->dbVar->set(db);
-
+	auto stateMapKey = clusterFilePath;
+	if (api->clusterSharedStateMap.find(stateMapKey) == api->clusterSharedStateMap.end()) {
+		DatabaseSharedState* p = db->createSharedState();
+		api->clusterSharedStateMap[stateMapKey] = p;
+	} else {
+		db->setSharedState(api->clusterSharedStateMap[stateMapKey]);
+	}
 	if (openConnectors) {
 		if (!api->localClientDisabled) {
 			dbState->addClient(api->getLocalClient());
@@ -1287,6 +1307,19 @@ ThreadFuture<Void> MultiVersionDatabase::createSnapshot(const StringRef& uid, co
 	return abortableFuture(f, dbState->dbVar->get().onChange);
 }
 
+DatabaseSharedState* MultiVersionDatabase::createSharedState() {
+	if (dbState->db) {
+		return dbState->db->createSharedState();
+	}
+	return nullptr;
+}
+
+void MultiVersionDatabase::setSharedState(DatabaseSharedState* p) {
+	if (dbState->db) {
+		dbState->db->setSharedState(p);
+	}
+}
+
 // Get network thread busyness
 // Return the busyness for the main thread. When using external clients, take the larger of the local client
 // and the external client's busyness.
@@ -1504,6 +1537,14 @@ void MultiVersionDatabase::DatabaseState::updateDatabase(Reference<IDatabase> ne
 		}
 	}
 
+	auto stateMapKey = clusterFilePath;
+	if (MultiVersionApi::api->clusterSharedStateMap.find(stateMapKey) ==
+	    MultiVersionApi::api->clusterSharedStateMap.end()) {
+		DatabaseSharedState* p = db->createSharedState();
+		MultiVersionApi::api->clusterSharedStateMap[stateMapKey] = p;
+	} else {
+		db->setSharedState(MultiVersionApi::api->clusterSharedStateMap[stateMapKey]);
+	}
 	dbVar->set(db);
 
 	ASSERT(protocolVersionMonitor.isValid());

fdbclient/MultiVersionTransaction.h

@@ -137,6 +137,9 @@ struct FdbCApi : public ThreadSafeReferenceCounted<FdbCApi> {
 	                                     int uidLength,
 	                                     uint8_t const* snapshotCommmand,
 	                                     int snapshotCommandLength);
+	DatabaseSharedState* (*databaseCreateSharedState)(FDBDatabase* database);
+	void (*databaseSetSharedState)(FDBDatabase* database, DatabaseSharedState* p);
+
 	double (*databaseGetMainThreadBusyness)(FDBDatabase* database);
 	FDBFuture* (*databaseGetServerProtocol)(FDBDatabase* database, uint64_t expectedVersion);
 
@@ -392,6 +395,9 @@ public:
 	ThreadFuture<Void> forceRecoveryWithDataLoss(const StringRef& dcid) override;
 	ThreadFuture<Void> createSnapshot(const StringRef& uid, const StringRef& snapshot_command) override;
 
+	DatabaseSharedState* createSharedState() override;
+	void setSharedState(DatabaseSharedState* p) override;
+
 private:
 	const Reference<FdbCApi> api;
 	FdbCApi::FDBDatabase*
@@ -620,6 +626,9 @@ public:
 	ThreadFuture<Void> forceRecoveryWithDataLoss(const StringRef& dcid) override;
 	ThreadFuture<Void> createSnapshot(const StringRef& uid, const StringRef& snapshot_command) override;
 
+	DatabaseSharedState* createSharedState() override;
+	void setSharedState(DatabaseSharedState* p) override;
+
 	// private:
 
 	struct LegacyVersionMonitor;
@@ -742,6 +751,8 @@ public:
 
 	bool callbackOnMainThread;
 	bool localClientDisabled;
+	// Map of (clusterFilePath + protocolVersion) -> DatabaseSharedState pointer
+	std::map<std::string, DatabaseSharedState*> clusterSharedStateMap;
 
 	static bool apiVersionAtLeast(int minVersion);
 

fdbclient/NativeAPI.actor.cpp

@@ -206,7 +206,23 @@ void DatabaseContext::removeTssMapping(StorageServerInterface const& ssi) {
 	}
 }
 
+void updateCachedReadVersionShared(double t, Version v, DatabaseSharedState* p) {
+	MutexHolder mutex(p->mutexLock);
+	TraceEvent("CheckpointCacheUpdateShared")
+	    .detail("Version", v)
+	    .detail("CurTime", t)
+	    .detail("LastVersion", p->grvCacheSpace.cachedReadVersion)
+	    .detail("LastTime", p->grvCacheSpace.lastGrvTime);
+	p->grvCacheSpace.cachedReadVersion = v;
+	if (t > p->grvCacheSpace.lastGrvTime) {
+		p->grvCacheSpace.lastGrvTime = t;
+	}
+}
+
 void DatabaseContext::updateCachedReadVersion(double t, Version v) {
+	if (sharedStatePtr) {
+		return updateCachedReadVersionShared(t, v, sharedStatePtr);
+	}
 	if (v >= cachedReadVersion) {
 		TraceEvent(SevDebug, "CachedReadVersionUpdate")
 		    .detail("Version", v)
@@ -225,10 +241,18 @@ void DatabaseContext::updateCachedReadVersion(double t, Version v) {
 }
 
 Version DatabaseContext::getCachedReadVersion() {
+	if (sharedStatePtr) {
+		MutexHolder mutex(sharedStatePtr->mutexLock);
+		return sharedStatePtr->grvCacheSpace.cachedReadVersion;
+	}
 	return cachedReadVersion;
 }
 
 double DatabaseContext::getLastGrvTime() {
+	if (sharedStatePtr) {
+		MutexHolder mutex(sharedStatePtr->mutexLock);
+		return sharedStatePtr->grvCacheSpace.lastGrvTime;
+	}
 	return lastGrvTime;
 }
 
@@ -1331,11 +1355,11 @@ DatabaseContext::DatabaseContext(Reference<AsyncVar<Reference<IClusterConnection
     transactionsExpensiveClearCostEstCount("ExpensiveClearCostEstCount", cc),
     transactionGrvFullBatches("NumGrvFullBatches", cc), transactionGrvTimedOutBatches("NumGrvTimedOutBatches", cc),
     latencies(1000), readLatencies(1000), commitLatencies(1000), GRVLatencies(1000), mutationsPerCommit(1000),
-    bytesPerCommit(1000), outstandingWatches(0), lastGrvTime(0.0), cachedReadVersion(0), lastRkBatchThrottleTime(0.0),
-    lastRkDefaultThrottleTime(0.0), lastProxyRequestTime(0.0), transactionTracingSample(false), taskID(taskID),
-    clientInfo(clientInfo), clientInfoMonitor(clientInfoMonitor), coordinator(coordinator), apiVersion(apiVersion),
-    mvCacheInsertLocation(0), healthMetricsLastUpdated(0), detailedHealthMetricsLastUpdated(0),
-    smoothMidShardSize(CLIENT_KNOBS->SHARD_STAT_SMOOTH_AMOUNT),
+    bytesPerCommit(1000), outstandingWatches(0), sharedStatePtr(nullptr), lastGrvTime(0.0), cachedReadVersion(0),
+    lastRkBatchThrottleTime(0.0), lastRkDefaultThrottleTime(0.0), lastProxyRequestTime(0.0),
+    transactionTracingSample(false), taskID(taskID), clientInfo(clientInfo), clientInfoMonitor(clientInfoMonitor),
+    coordinator(coordinator), apiVersion(apiVersion), mvCacheInsertLocation(0), healthMetricsLastUpdated(0),
+    detailedHealthMetricsLastUpdated(0), smoothMidShardSize(CLIENT_KNOBS->SHARD_STAT_SMOOTH_AMOUNT),
     specialKeySpace(std::make_unique<SpecialKeySpace>(specialKeys.begin, specialKeys.end, /* test */ false)),
     connectToDatabaseEventCacheHolder(format("ConnectToDatabase/%s", dbId.toString().c_str())) {
 	dbId = deterministicRandom()->randomUniqueID();
@@ -1623,6 +1647,13 @@ DatabaseContext::~DatabaseContext() {
 	if (grvUpdateHandler.isValid()) {
 		grvUpdateHandler.cancel();
 	}
+	if (sharedStatePtr) {
+		sharedStatePtr->refCount--;
+		if (sharedStatePtr->refCount <= 0) {
+			delete sharedStatePtr;
+			sharedStatePtr = nullptr;
+		}
+	}
 	for (auto it = server_interf.begin(); it != server_interf.end(); it = server_interf.erase(it))
 		it->second->notifyContextDestroyed();
 	ASSERT_ABORT(server_interf.empty());
@@ -7332,6 +7363,17 @@ Future<Void> DatabaseContext::createSnapshot(StringRef uid, StringRef snapshot_c
 	return createSnapshotActor(this, UID::fromString(uid_str), snapshot_command);
 }
 
+DatabaseSharedState* DatabaseContext::initSharedState() {
+	DatabaseSharedState* newState = new DatabaseSharedState();
+	setSharedState(newState);
+	return newState;
+}
+
+void DatabaseContext::setSharedState(DatabaseSharedState* p) {
+	sharedStatePtr = p;
+	sharedStatePtr->refCount++;
+}
+
 ACTOR Future<Void> storageFeedVersionUpdater(StorageServerInterface interf, ChangeFeedStorageData* self) {
 	state Promise<Void> destroyed = self->destroyed;
 	loop {

fdbclient/ThreadSafeTransaction.cpp

@@ -96,6 +96,14 @@ ThreadFuture<Void> ThreadSafeDatabase::createSnapshot(const StringRef& uid, cons
 	return onMainThread([db, snapUID, cmd]() -> Future<Void> { return db->createSnapshot(snapUID, cmd); });
 }
 
+DatabaseSharedState* ThreadSafeDatabase::createSharedState() {
+	return db->initSharedState();
+}
+
+void ThreadSafeDatabase::setSharedState(DatabaseSharedState* p) {
+	db->setSharedState(p);
+}
+
 // Return the main network thread busyness
 double ThreadSafeDatabase::getMainThreadBusyness() {
 	ASSERT(g_network);

fdbclient/ThreadSafeTransaction.h

@@ -57,6 +57,9 @@ public:
 	ThreadFuture<Void> forceRecoveryWithDataLoss(const StringRef& dcid) override;
 	ThreadFuture<Void> createSnapshot(const StringRef& uid, const StringRef& snapshot_command) override;
 
+	DatabaseSharedState* createSharedState() override;
+	void setSharedState(DatabaseSharedState* p) override;
+
 private:
 	friend class ThreadSafeTransaction;
 	bool isConfigDB{ false };