apple/foundationdb
1
0
Fork 0
mirror of https://github.com/apple/foundationdb.git synced 2025-05-21 22:33:17 +08:00
Code Issues Packages Projects Releases Wiki Activity

Added *BOOLEAN_PARAM macros to enforce documentation of boolean parameters

Browse Source
This commit is contained in:
sfc-gh-tclinkenbeard 2021-07-02 15:04:42 -07:00
parent 49d3b0c853
commit 79ff07a071
30 changed files with 250 additions and 163 deletions

2
fdbbackup/FileConverter.actor.cpp

@ -598,7 +598,7 @@ int main(int argc, char** argv) {
Error::init(); Error::init();
StringRef url(param.container_url); StringRef url(param.container_url);
setupNetwork(0, true); setupNetwork(0, UseMetrics::TRUE);
TraceEvent::setNetworkThread(); TraceEvent::setNetworkThread();
openTraceFile(NetworkAddress(), 10 << 20, 10 << 20, param.log_dir, "convert", param.trace_log_group); openTraceFile(NetworkAddress(), 10 << 20, 10 << 20, param.log_dir, "convert", param.trace_log_group);

2
fdbbackup/FileDecoder.actor.cpp

@ -579,7 +579,7 @@ int main(int argc, char** argv) {
Error::init(); Error::init();
StringRef url(param.container_url); StringRef url(param.container_url);
setupNetwork(0, true); setupNetwork(0, UseMetrics::TRUE);
TraceEvent::setNetworkThread(); TraceEvent::setNetworkThread();
openTraceFile(NetworkAddress(), 10 << 20, 10 << 20, param.log_dir, "decode", param.trace_log_group); openTraceFile(NetworkAddress(), 10 << 20, 10 << 20, param.log_dir, "decode", param.trace_log_group);

6
fdbbackup/backup.actor.cpp

@ -3769,7 +3769,7 @@ int main(int argc, char* argv[]) {
Reference<IBackupContainer> c; Reference<IBackupContainer> c;
try { try {
setupNetwork(0, true); setupNetwork(0, UseMetrics::TRUE);
} catch (Error& e) { } catch (Error& e) {
fprintf(stderr, "ERROR: %s\n", e.what()); fprintf(stderr, "ERROR: %s\n", e.what());
return FDB_EXIT_ERROR; return FDB_EXIT_ERROR;
@ -3813,7 +3813,7 @@ int main(int argc, char* argv[]) {
} }
try { try {
db = Database::createDatabase(ccf, -1, true, localities); db = Database::createDatabase(ccf, -1, IsInternal::TRUE, localities);
} catch (Error& e) { } catch (Error& e) {
fprintf(stderr, "ERROR: %s\n", e.what()); fprintf(stderr, "ERROR: %s\n", e.what());
fprintf(stderr, "ERROR: Unable to connect to cluster from `%s'\n", ccf->getFilename().c_str()); fprintf(stderr, "ERROR: Unable to connect to cluster from `%s'\n", ccf->getFilename().c_str());
@ -3833,7 +3833,7 @@ int main(int argc, char* argv[]) {
} }
try { try {
sourceDb = Database::createDatabase(sourceCcf, -1, true, localities); sourceDb = Database::createDatabase(sourceCcf, -1, IsInternal::TRUE, localities);
} catch (Error& e) { } catch (Error& e) {
fprintf(stderr, "ERROR: %s\n", e.what()); fprintf(stderr, "ERROR: %s\n", e.what());
fprintf(stderr, "ERROR: Unable to connect to cluster from `%s'\n", sourceCcf->getFilename().c_str()); fprintf(stderr, "ERROR: Unable to connect to cluster from `%s'\n", sourceCcf->getFilename().c_str());

2
fdbcli/fdbcli.actor.cpp

@ -3322,7 +3322,7 @@ ACTOR Future<int> cli(CLIOptions opt, LineNoise* plinenoise) {
TraceEvent::setNetworkThread(); TraceEvent::setNetworkThread();
try { try {
db = Database::createDatabase(ccf, -1, false); db = Database::createDatabase(ccf, -1, IsInternal::FALSE);
if (!opt.exec.present()) { if (!opt.exec.present()) {
printf("Using cluster file `%s'.\n", ccf->getFilename().c_str()); printf("Using cluster file `%s'.\n", ccf->getFilename().c_str());
} }

29
fdbclient/DatabaseContext.h

@ -146,6 +146,13 @@ public:
WatchMetadata(Key key, Optional<Value> value, Version version, TransactionInfo info, TagSet tags); WatchMetadata(Key key, Optional<Value> value, Version version, TransactionInfo info, TagSet tags);
}; };
#ifndef __DATABASE_BOOLEAN_PARAMS__
#define __DATABASE_BOOLEAN_PARAMS__
DECLARE_BOOLEAN_PARAM(LockAware);
DECLARE_BOOLEAN_PARAM(EnableLocalityLoadBalance);
#endif
DECLARE_BOOLEAN_PARAM(IsSwitchable);
class DatabaseContext : public ReferenceCounted<DatabaseContext>, public FastAllocated<DatabaseContext>, NonCopyable { class DatabaseContext : public ReferenceCounted<DatabaseContext>, public FastAllocated<DatabaseContext>, NonCopyable {
public: public:
static DatabaseContext* allocateOnForeignThread() { static DatabaseContext* allocateOnForeignThread() {
@ -157,11 +164,11 @@ public:
static Database create(Reference<AsyncVar<ClientDBInfo>> clientInfo, static Database create(Reference<AsyncVar<ClientDBInfo>> clientInfo,
Future<Void> clientInfoMonitor, Future<Void> clientInfoMonitor,
LocalityData clientLocality, LocalityData clientLocality,
bool enableLocalityLoadBalance, EnableLocalityLoadBalance,
TaskPriority taskID = TaskPriority::DefaultEndpoint, TaskPriority taskID = TaskPriority::DefaultEndpoint,
bool lockAware = false, LockAware lockAware = LockAware::FALSE,
int apiVersion = Database::API_VERSION_LATEST, int apiVersion = Database::API_VERSION_LATEST,
bool switchable = false); IsSwitchable switchable = IsSwitchable::FALSE);
~DatabaseContext(); ~DatabaseContext();
@ -217,7 +224,7 @@ public:
void setOption(FDBDatabaseOptions::Option option, Optional<StringRef> value); void setOption(FDBDatabaseOptions::Option option, Optional<StringRef> value);
Error deferredError; Error deferredError;
bool lockAware; LockAware lockAware{ LockAware::FALSE };
bool isError() const { return deferredError.code() != invalid_error_code; } bool isError() const { return deferredError.code() != invalid_error_code; }
@ -242,7 +249,7 @@ public:
// new cluster. // new cluster.
Future<Void> switchConnectionFile(Reference<ClusterConnectionFile> standby); Future<Void> switchConnectionFile(Reference<ClusterConnectionFile> standby);
Future<Void> connectionFileChanged(); Future<Void> connectionFileChanged();
bool switchable = false; IsSwitchable switchable{ IsSwitchable::FALSE };
// Management API, Attempt to kill or suspend a process, return 1 for request sent out, 0 for failure // Management API, Attempt to kill or suspend a process, return 1 for request sent out, 0 for failure
Future<int64_t> rebootWorker(StringRef address, bool check = false, int duration = 0); Future<int64_t> rebootWorker(StringRef address, bool check = false, int duration = 0);
@ -259,11 +266,11 @@ public:
Future<Void> clientInfoMonitor, Future<Void> clientInfoMonitor,
TaskPriority taskID, TaskPriority taskID,
LocalityData const& clientLocality, LocalityData const& clientLocality,
bool enableLocalityLoadBalance, EnableLocalityLoadBalance,
bool lockAware, LockAware,
bool internal = true, IsInternal internal = IsInternal::TRUE,
int apiVersion = Database::API_VERSION_LATEST, int apiVersion = Database::API_VERSION_LATEST,
bool switchable = false); IsSwitchable switchable = IsSwitchable::FALSE);
explicit DatabaseContext(const Error& err); explicit DatabaseContext(const Error& err);
@ -282,7 +289,7 @@ public:
UID proxiesLastChange; UID proxiesLastChange;
LocalityData clientLocality; LocalityData clientLocality;
QueueModel queueModel; QueueModel queueModel;
bool enableLocalityLoadBalance; EnableLocalityLoadBalance enableLocalityLoadBalance{ EnableLocalityLoadBalance::FALSE };
struct VersionRequest { struct VersionRequest {
SpanID spanContext; SpanID spanContext;
@ -329,7 +336,7 @@ public:
std::unordered_map<UID, Reference<TSSMetrics>> tssMetrics; std::unordered_map<UID, Reference<TSSMetrics>> tssMetrics;
UID dbId; UID dbId;
bool internal; // Only contexts created through the C client and fdbcli are non-internal IsInternal internal; // Only contexts created through the C client and fdbcli are non-internal
PrioritizedTransactionTagMap<ClientTagThrottleData> throttledTags; PrioritizedTransactionTagMap<ClientTagThrottleData> throttledTags;

168
fdbclient/NativeAPI.actor.cpp

@ -85,6 +85,12 @@ using std::max;
using std::min; using std::min;
using std::pair; using std::pair;
DEFINE_BOOLEAN_PARAM(LockAware);
DEFINE_BOOLEAN_PARAM(EnableLocalityLoadBalance);
DEFINE_BOOLEAN_PARAM(IsSwitchable);
DEFINE_BOOLEAN_PARAM(UseMetrics);
DEFINE_BOOLEAN_PARAM(IsInternal);
namespace { namespace {
template <class Interface, class Request> template <class Interface, class Request>
@ -94,7 +100,8 @@ Future<REPLY_TYPE(Request)> loadBalance(
RequestStream<Request> Interface::*channel, RequestStream<Request> Interface::*channel,
const Request& request = Request(), const Request& request = Request(),
TaskPriority taskID = TaskPriority::DefaultPromiseEndpoint, TaskPriority taskID = TaskPriority::DefaultPromiseEndpoint,
bool atMostOnce = false, // if true, throws request_maybe_delivered() instead of retrying automatically AtMostOnce atMostOnce =
AtMostOnce::FALSE, // if true, throws request_maybe_delivered() instead of retrying automatically
QueueModel* model = nullptr) { QueueModel* model = nullptr) {
if (alternatives->hasCaches) { if (alternatives->hasCaches) {
return loadBalance(alternatives->locations(), channel, request, taskID, atMostOnce, model); return loadBalance(alternatives->locations(), channel, request, taskID, atMostOnce, model);
@ -296,11 +303,16 @@ void DatabaseContext::validateVersion(Version version) {
ASSERT(version > 0 || version == latestVersion); ASSERT(version > 0 || version == latestVersion);
} }
void validateOptionValue(Optional<StringRef> value, bool shouldBePresent) { void validateOptionValuePresent(Optional<StringRef> value) {
if (shouldBePresent && !value.present()) if (!value.present()) {
throw invalid_option_value(); throw invalid_option_value();
if (!shouldBePresent && value.present() && value.get().size() > 0) }
}
void validateOptionValueNotPresent(Optional<StringRef> value) {
if (value.present() && value.get().size() > 0) {
throw invalid_option_value(); throw invalid_option_value();
}
} }
void dumpMutations(const MutationListRef& mutations) { void dumpMutations(const MutationListRef& mutations) {
@ -1080,11 +1092,11 @@ DatabaseContext::DatabaseContext(Reference<AsyncVar<Reference<ClusterConnectionF
Future<Void> clientInfoMonitor, Future<Void> clientInfoMonitor,
TaskPriority taskID, TaskPriority taskID,
LocalityData const& clientLocality, LocalityData const& clientLocality,
bool enableLocalityLoadBalance, EnableLocalityLoadBalance enableLocalityLoadBalance,
bool lockAware, LockAware lockAware,
bool internal, IsInternal internal,
int apiVersion, int apiVersion,
bool switchable) IsSwitchable switchable)
: connectionFile(connectionFile), clientInfo(clientInfo), coordinator(coordinator), : connectionFile(connectionFile), clientInfo(clientInfo), coordinator(coordinator),
clientInfoMonitor(clientInfoMonitor), taskID(taskID), clientLocality(clientLocality), clientInfoMonitor(clientInfoMonitor), taskID(taskID), clientLocality(clientLocality),
enableLocalityLoadBalance(enableLocalityLoadBalance), lockAware(lockAware), apiVersion(apiVersion), enableLocalityLoadBalance(enableLocalityLoadBalance), lockAware(lockAware), apiVersion(apiVersion),
@ -1360,7 +1372,7 @@ DatabaseContext::DatabaseContext(const Error& err)
transactionsThrottled("Throttled", cc), transactionsProcessBehind("ProcessBehind", cc), latencies(1000), transactionsThrottled("Throttled", cc), transactionsProcessBehind("ProcessBehind", cc), latencies(1000),
readLatencies(1000), commitLatencies(1000), GRVLatencies(1000), mutationsPerCommit(1000), bytesPerCommit(1000), readLatencies(1000), commitLatencies(1000), GRVLatencies(1000), mutationsPerCommit(1000), bytesPerCommit(1000),
smoothMidShardSize(CLIENT_KNOBS->SHARD_STAT_SMOOTH_AMOUNT), smoothMidShardSize(CLIENT_KNOBS->SHARD_STAT_SMOOTH_AMOUNT),
transactionsExpensiveClearCostEstCount("ExpensiveClearCostEstCount", cc), internal(false), transactionsExpensiveClearCostEstCount("ExpensiveClearCostEstCount", cc), internal(IsInternal::FALSE),
transactionTracingEnabled(true) {} transactionTracingEnabled(true) {}
// Static constructor used by server processes to create a DatabaseContext // Static constructor used by server processes to create a DatabaseContext
@ -1368,11 +1380,11 @@ DatabaseContext::DatabaseContext(const Error& err)
Database DatabaseContext::create(Reference<AsyncVar<ClientDBInfo>> clientInfo, Database DatabaseContext::create(Reference<AsyncVar<ClientDBInfo>> clientInfo,
Future<Void> clientInfoMonitor, Future<Void> clientInfoMonitor,
LocalityData clientLocality, LocalityData clientLocality,
bool enableLocalityLoadBalance, EnableLocalityLoadBalance enableLocalityLoadBalance,
TaskPriority taskID, TaskPriority taskID,
bool lockAware, LockAware lockAware,
int apiVersion, int apiVersion,
bool switchable) { IsSwitchable switchable) {
return Database(new DatabaseContext(Reference<AsyncVar<Reference<ClusterConnectionFile>>>(), return Database(new DatabaseContext(Reference<AsyncVar<Reference<ClusterConnectionFile>>>(),
clientInfo, clientInfo,
makeReference<AsyncVar<Optional<ClientLeaderRegInterface>>>(), makeReference<AsyncVar<Optional<ClientLeaderRegInterface>>>(),
@ -1381,7 +1393,7 @@ Database DatabaseContext::create(Reference<AsyncVar<ClientDBInfo>> clientInfo,
clientLocality, clientLocality,
enableLocalityLoadBalance, enableLocalityLoadBalance,
lockAware, lockAware,
true, IsInternal::TRUE,
apiVersion, apiVersion,
switchable)); switchable));
} }
@ -1491,7 +1503,7 @@ bool DatabaseContext::sampleOnCost(uint64_t cost) const {
} }
int64_t extractIntOption(Optional<StringRef> value, int64_t minValue, int64_t maxValue) { int64_t extractIntOption(Optional<StringRef> value, int64_t minValue, int64_t maxValue) {
validateOptionValue(value, true); validateOptionValuePresent(value);
if (value.get().size() != 8) { if (value.get().size() != 8) {
throw invalid_option_value(); throw invalid_option_value();
} }
@ -1553,23 +1565,23 @@ void DatabaseContext::setOption(FDBDatabaseOptions::Option option, Optional<Stri
locationCache.insert(allKeys, Reference<LocationInfo>()); locationCache.insert(allKeys, Reference<LocationInfo>());
break; break;
case FDBDatabaseOptions::SNAPSHOT_RYW_ENABLE: case FDBDatabaseOptions::SNAPSHOT_RYW_ENABLE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
snapshotRywEnabled++; snapshotRywEnabled++;
break; break;
case FDBDatabaseOptions::SNAPSHOT_RYW_DISABLE: case FDBDatabaseOptions::SNAPSHOT_RYW_DISABLE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
snapshotRywEnabled--; snapshotRywEnabled--;
break; break;
case FDBDatabaseOptions::DISTRIBUTED_TRANSACTION_TRACE_ENABLE: case FDBDatabaseOptions::DISTRIBUTED_TRANSACTION_TRACE_ENABLE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
transactionTracingEnabled++; transactionTracingEnabled++;
break; break;
case FDBDatabaseOptions::DISTRIBUTED_TRANSACTION_TRACE_DISABLE: case FDBDatabaseOptions::DISTRIBUTED_TRANSACTION_TRACE_DISABLE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
transactionTracingEnabled--; transactionTracingEnabled--;
break; break;
case FDBDatabaseOptions::USE_CONFIG_DATABASE: case FDBDatabaseOptions::USE_CONFIG_DATABASE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
useConfigDatabase = true; useConfigDatabase = true;
break; break;
default: default:
@ -1669,7 +1681,7 @@ extern IPAddress determinePublicIPAutomatically(ClusterConnectionString const& c
// on another thread // on another thread
Database Database::createDatabase(Reference<ClusterConnectionFile> connFile, Database Database::createDatabase(Reference<ClusterConnectionFile> connFile,
int apiVersion, int apiVersion,
bool internal, IsInternal internal,
LocalityData const& clientLocality, LocalityData const& clientLocality,
DatabaseContext* preallocatedDb) { DatabaseContext* preallocatedDb) {
if (!g_network) if (!g_network)
@ -1730,11 +1742,11 @@ Database Database::createDatabase(Reference<ClusterConnectionFile> connFile,
clientInfoMonitor, clientInfoMonitor,
TaskPriority::DefaultEndpoint, TaskPriority::DefaultEndpoint,
clientLocality, clientLocality,
true, EnableLocalityLoadBalance::TRUE,
false, LockAware::FALSE,
internal, internal,
apiVersion, apiVersion,
/*switchable*/ true); IsSwitchable::TRUE);
} else { } else {
db = new DatabaseContext(connectionFile, db = new DatabaseContext(connectionFile,
clientInfo, clientInfo,
@ -1742,11 +1754,11 @@ Database Database::createDatabase(Reference<ClusterConnectionFile> connFile,
clientInfoMonitor, clientInfoMonitor,
TaskPriority::DefaultEndpoint, TaskPriority::DefaultEndpoint,
clientLocality, clientLocality,
true, EnableLocalityLoadBalance::TRUE,
false, LockAware::FALSE,
internal, internal,
apiVersion, apiVersion,
/*switchable*/ true); IsSwitchable::TRUE);
} }
auto database = Database(db); auto database = Database(db);
@ -1756,7 +1768,7 @@ Database Database::createDatabase(Reference<ClusterConnectionFile> connFile,
Database Database::createDatabase(std::string connFileName, Database Database::createDatabase(std::string connFileName,
int apiVersion, int apiVersion,
bool internal, IsInternal internal,
LocalityData const& clientLocality) { LocalityData const& clientLocality) {
Reference<ClusterConnectionFile> rccf = Reference<ClusterConnectionFile>( Reference<ClusterConnectionFile> rccf = Reference<ClusterConnectionFile>(
new ClusterConnectionFile(ClusterConnectionFile::lookupClusterFileName(connFileName).first)); new ClusterConnectionFile(ClusterConnectionFile::lookupClusterFileName(connFileName).first));
@ -1803,15 +1815,15 @@ void setNetworkOption(FDBNetworkOptions::Option option, Optional<StringRef> valu
networkOptions.traceDirectory = value.present() ? value.get().toString() : ""; networkOptions.traceDirectory = value.present() ? value.get().toString() : "";
break; break;
case FDBNetworkOptions::TRACE_ROLL_SIZE: case FDBNetworkOptions::TRACE_ROLL_SIZE:
validateOptionValue(value, true); validateOptionValuePresent(value);
networkOptions.traceRollSize = extractIntOption(value, 0, std::numeric_limits<int64_t>::max()); networkOptions.traceRollSize = extractIntOption(value, 0, std::numeric_limits<int64_t>::max());
break; break;
case FDBNetworkOptions::TRACE_MAX_LOGS_SIZE: case FDBNetworkOptions::TRACE_MAX_LOGS_SIZE:
validateOptionValue(value, true); validateOptionValuePresent(value);
networkOptions.traceMaxLogsSize = extractIntOption(value, 0, std::numeric_limits<int64_t>::max()); networkOptions.traceMaxLogsSize = extractIntOption(value, 0, std::numeric_limits<int64_t>::max());
break; break;
case FDBNetworkOptions::TRACE_FORMAT: case FDBNetworkOptions::TRACE_FORMAT:
validateOptionValue(value, true); validateOptionValuePresent(value);
networkOptions.traceFormat = value.get().toString(); networkOptions.traceFormat = value.get().toString();
if (!validateTraceFormat(networkOptions.traceFormat)) { if (!validateTraceFormat(networkOptions.traceFormat)) {
fprintf(stderr, "Unrecognized trace format: `%s'\n", networkOptions.traceFormat.c_str()); fprintf(stderr, "Unrecognized trace format: `%s'\n", networkOptions.traceFormat.c_str());
@ -1819,7 +1831,7 @@ void setNetworkOption(FDBNetworkOptions::Option option, Optional<StringRef> valu
} }
break; break;
case FDBNetworkOptions::TRACE_FILE_IDENTIFIER: case FDBNetworkOptions::TRACE_FILE_IDENTIFIER:
validateOptionValue(value, true); validateOptionValuePresent(value);
networkOptions.traceFileIdentifier = value.get().toString(); networkOptions.traceFileIdentifier = value.get().toString();
if (networkOptions.traceFileIdentifier.length() > CLIENT_KNOBS->TRACE_LOG_FILE_IDENTIFIER_MAX_LENGTH) { if (networkOptions.traceFileIdentifier.length() > CLIENT_KNOBS->TRACE_LOG_FILE_IDENTIFIER_MAX_LENGTH) {
fprintf(stderr, "Trace file identifier provided is too long.\n"); fprintf(stderr, "Trace file identifier provided is too long.\n");
@ -1840,7 +1852,7 @@ void setNetworkOption(FDBNetworkOptions::Option option, Optional<StringRef> valu
} }
break; break;
case FDBNetworkOptions::TRACE_CLOCK_SOURCE: case FDBNetworkOptions::TRACE_CLOCK_SOURCE:
validateOptionValue(value, true); validateOptionValuePresent(value);
networkOptions.traceClockSource = value.get().toString(); networkOptions.traceClockSource = value.get().toString();
if (!validateTraceClockSource(networkOptions.traceClockSource)) { if (!validateTraceClockSource(networkOptions.traceClockSource)) {
fprintf(stderr, "Unrecognized trace clock source: `%s'\n", networkOptions.traceClockSource.c_str()); fprintf(stderr, "Unrecognized trace clock source: `%s'\n", networkOptions.traceClockSource.c_str());
@ -1848,7 +1860,7 @@ void setNetworkOption(FDBNetworkOptions::Option option, Optional<StringRef> valu
} }
break; break;
case FDBNetworkOptions::KNOB: { case FDBNetworkOptions::KNOB: {
validateOptionValue(value, true); validateOptionValuePresent(value);
std::string optionValue = value.get().toString(); std::string optionValue = value.get().toString();
TraceEvent("SetKnob").detail("KnobString", optionValue); TraceEvent("SetKnob").detail("KnobString", optionValue);
@ -1872,42 +1884,42 @@ void setNetworkOption(FDBNetworkOptions::Option option, Optional<StringRef> valu
break; break;
} }
case FDBNetworkOptions::TLS_PLUGIN: case FDBNetworkOptions::TLS_PLUGIN:
validateOptionValue(value, true); validateOptionValuePresent(value);
break; break;
case FDBNetworkOptions::TLS_CERT_PATH: case FDBNetworkOptions::TLS_CERT_PATH:
validateOptionValue(value, true); validateOptionValuePresent(value);
tlsConfig.setCertificatePath(value.get().toString()); tlsConfig.setCertificatePath(value.get().toString());
break; break;
case FDBNetworkOptions::TLS_CERT_BYTES: { case FDBNetworkOptions::TLS_CERT_BYTES: {
validateOptionValue(value, true); validateOptionValuePresent(value);
tlsConfig.setCertificateBytes(value.get().toString()); tlsConfig.setCertificateBytes(value.get().toString());
break; break;
} }
case FDBNetworkOptions::TLS_CA_PATH: { case FDBNetworkOptions::TLS_CA_PATH: {
validateOptionValue(value, true); validateOptionValuePresent(value);
tlsConfig.setCAPath(value.get().toString()); tlsConfig.setCAPath(value.get().toString());
break; break;
} }
case FDBNetworkOptions::TLS_CA_BYTES: { case FDBNetworkOptions::TLS_CA_BYTES: {
validateOptionValue(value, true); validateOptionValuePresent(value);
tlsConfig.setCABytes(value.get().toString()); tlsConfig.setCABytes(value.get().toString());
break; break;
} }
case FDBNetworkOptions::TLS_PASSWORD: case FDBNetworkOptions::TLS_PASSWORD:
validateOptionValue(value, true); validateOptionValuePresent(value);
tlsConfig.setPassword(value.get().toString()); tlsConfig.setPassword(value.get().toString());
break; break;
case FDBNetworkOptions::TLS_KEY_PATH: case FDBNetworkOptions::TLS_KEY_PATH:
validateOptionValue(value, true); validateOptionValuePresent(value);
tlsConfig.setKeyPath(value.get().toString()); tlsConfig.setKeyPath(value.get().toString());
break; break;
case FDBNetworkOptions::TLS_KEY_BYTES: { case FDBNetworkOptions::TLS_KEY_BYTES: {
validateOptionValue(value, true); validateOptionValuePresent(value);
tlsConfig.setKeyBytes(value.get().toString()); tlsConfig.setKeyBytes(value.get().toString());
break; break;
} }
case FDBNetworkOptions::TLS_VERIFY_PEERS: case FDBNetworkOptions::TLS_VERIFY_PEERS:
validateOptionValue(value, true); validateOptionValuePresent(value);
tlsConfig.clearVerifyPeers(); tlsConfig.clearVerifyPeers();
tlsConfig.addVerifyPeers(value.get().toString()); tlsConfig.addVerifyPeers(value.get().toString());
break; break;
@ -1918,16 +1930,16 @@ void setNetworkOption(FDBNetworkOptions::Option option, Optional<StringRef> valu
enableBuggify(false, BuggifyType::Client); enableBuggify(false, BuggifyType::Client);
break; break;
case FDBNetworkOptions::CLIENT_BUGGIFY_SECTION_ACTIVATED_PROBABILITY: case FDBNetworkOptions::CLIENT_BUGGIFY_SECTION_ACTIVATED_PROBABILITY:
validateOptionValue(value, true); validateOptionValuePresent(value);
clearBuggifySections(BuggifyType::Client); clearBuggifySections(BuggifyType::Client);
P_BUGGIFIED_SECTION_ACTIVATED[int(BuggifyType::Client)] = double(extractIntOption(value, 0, 100)) / 100.0; P_BUGGIFIED_SECTION_ACTIVATED[int(BuggifyType::Client)] = double(extractIntOption(value, 0, 100)) / 100.0;
break; break;
case FDBNetworkOptions::CLIENT_BUGGIFY_SECTION_FIRED_PROBABILITY: case FDBNetworkOptions::CLIENT_BUGGIFY_SECTION_FIRED_PROBABILITY:
validateOptionValue(value, true); validateOptionValuePresent(value);
P_BUGGIFIED_SECTION_FIRES[int(BuggifyType::Client)] = double(extractIntOption(value, 0, 100)) / 100.0; P_BUGGIFIED_SECTION_FIRES[int(BuggifyType::Client)] = double(extractIntOption(value, 0, 100)) / 100.0;
break; break;
case FDBNetworkOptions::DISABLE_CLIENT_STATISTICS_LOGGING: case FDBNetworkOptions::DISABLE_CLIENT_STATISTICS_LOGGING:
validateOptionValue(value, false); validateOptionValuePresent(value);
networkOptions.logClientInfo = false; networkOptions.logClientInfo = false;
break; break;
case FDBNetworkOptions::SUPPORTED_CLIENT_VERSIONS: { case FDBNetworkOptions::SUPPORTED_CLIENT_VERSIONS: {
@ -1947,11 +1959,11 @@ void setNetworkOption(FDBNetworkOptions::Option option, Optional<StringRef> valu
break; break;
} }
case FDBNetworkOptions::ENABLE_RUN_LOOP_PROFILING: // Same as ENABLE_SLOW_TASK_PROFILING case FDBNetworkOptions::ENABLE_RUN_LOOP_PROFILING: // Same as ENABLE_SLOW_TASK_PROFILING
validateOptionValue(value, false); validateOptionValuePresent(value);
networkOptions.runLoopProfilingEnabled = true; networkOptions.runLoopProfilingEnabled = true;
break; break;
case FDBNetworkOptions::DISTRIBUTED_CLIENT_TRACER: { case FDBNetworkOptions::DISTRIBUTED_CLIENT_TRACER: {
validateOptionValue(value, true); validateOptionValuePresent(value);
std::string tracer = value.get().toString(); std::string tracer = value.get().toString();
if (tracer == "none" || tracer == "disabled") { if (tracer == "none" || tracer == "disabled") {
openTracer(TracerType::DISABLED); openTracer(TracerType::DISABLED);
@ -1994,7 +2006,7 @@ ACTOR Future<Void> monitorNetworkBusyness() {
} }
// Setup g_network and start monitoring for network busyness // Setup g_network and start monitoring for network busyness
void setupNetwork(uint64_t transportId, bool useMetrics) { void setupNetwork(uint64_t transportId, UseMetrics useMetrics) {
if (g_network) if (g_network)
throw network_already_setup(); throw network_already_setup();
@ -2469,7 +2481,7 @@ ACTOR Future<Optional<Value>> getValue(Future<Version> version,
GetValueRequest( GetValueRequest(
span.context, key, ver, cx->sampleReadTags() ? tags : Optional<TagSet>(), getValueID), span.context, key, ver, cx->sampleReadTags() ? tags : Optional<TagSet>(), getValueID),
TaskPriority::DefaultPromiseEndpoint, TaskPriority::DefaultPromiseEndpoint,
false, AtMostOnce::FALSE,
cx->enableLocalityLoadBalance ? &cx->queueModel : nullptr))) { cx->enableLocalityLoadBalance ? &cx->queueModel : nullptr))) {
reply = _reply; reply = _reply;
} }
@ -2581,7 +2593,7 @@ ACTOR Future<Key> getKey(Database cx, KeySelector k, Future<Version> version, Tr
&StorageServerInterface::getKey, &StorageServerInterface::getKey,
req, req,
TaskPriority::DefaultPromiseEndpoint, TaskPriority::DefaultPromiseEndpoint,
false, AtMostOnce::FALSE,
cx->enableLocalityLoadBalance ? &cx->queueModel : nullptr))) { cx->enableLocalityLoadBalance ? &cx->queueModel : nullptr))) {
reply = _reply; reply = _reply;
} }
@ -2974,7 +2986,7 @@ ACTOR Future<RangeResult> getExactRange(Database cx,
&StorageServerInterface::getKeyValues, &StorageServerInterface::getKeyValues,
req, req,
TaskPriority::DefaultPromiseEndpoint, TaskPriority::DefaultPromiseEndpoint,
false, AtMostOnce::FALSE,
cx->enableLocalityLoadBalance ? &cx->queueModel : nullptr))) { cx->enableLocalityLoadBalance ? &cx->queueModel : nullptr))) {
rep = _rep; rep = _rep;
} }
@ -3325,7 +3337,7 @@ ACTOR Future<RangeResult> getRange(Database cx,
&StorageServerInterface::getKeyValues, &StorageServerInterface::getKeyValues,
req, req,
TaskPriority::DefaultPromiseEndpoint, TaskPriority::DefaultPromiseEndpoint,
false, AtMostOnce::FALSE,
cx->enableLocalityLoadBalance ? &cx->queueModel : nullptr)); cx->enableLocalityLoadBalance ? &cx->queueModel : nullptr));
rep = _rep; rep = _rep;
++cx->transactionPhysicalReadsCompleted; ++cx->transactionPhysicalReadsCompleted;
@ -4797,7 +4809,7 @@ ACTOR static Future<Void> tryCommit(Database cx,
&CommitProxyInterface::commit, &CommitProxyInterface::commit,
req, req,
TaskPriority::DefaultPromiseEndpoint, TaskPriority::DefaultPromiseEndpoint,
true); AtMostOnce::TRUE);
} }
choose { choose {
@ -5047,7 +5059,7 @@ Future<Void> Transaction::commit() {
void Transaction::setOption(FDBTransactionOptions::Option option, Optional<StringRef> value) { void Transaction::setOption(FDBTransactionOptions::Option option, Optional<StringRef> value) {
switch (option) { switch (option) {
case FDBTransactionOptions::INITIALIZE_NEW_DATABASE: case FDBTransactionOptions::INITIALIZE_NEW_DATABASE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
if (readVersion.isValid()) if (readVersion.isValid())
throw read_version_already_set(); throw read_version_already_set();
readVersion = Version(0); readVersion = Version(0);
@ -5055,37 +5067,37 @@ void Transaction::setOption(FDBTransactionOptions::Option option, Optional<Strin
break; break;
case FDBTransactionOptions::CAUSAL_READ_RISKY: case FDBTransactionOptions::CAUSAL_READ_RISKY:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.getReadVersionFlags |= GetReadVersionRequest::FLAG_CAUSAL_READ_RISKY; options.getReadVersionFlags |= GetReadVersionRequest::FLAG_CAUSAL_READ_RISKY;
break; break;
case FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE: case FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.priority = TransactionPriority::IMMEDIATE; options.priority = TransactionPriority::IMMEDIATE;
break; break;
case FDBTransactionOptions::PRIORITY_BATCH: case FDBTransactionOptions::PRIORITY_BATCH:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.priority = TransactionPriority::BATCH; options.priority = TransactionPriority::BATCH;
break; break;
case FDBTransactionOptions::CAUSAL_WRITE_RISKY: case FDBTransactionOptions::CAUSAL_WRITE_RISKY:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.causalWriteRisky = true; options.causalWriteRisky = true;
break; break;
case FDBTransactionOptions::COMMIT_ON_FIRST_PROXY: case FDBTransactionOptions::COMMIT_ON_FIRST_PROXY:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.commitOnFirstProxy = true; options.commitOnFirstProxy = true;
break; break;
case FDBTransactionOptions::CHECK_WRITES_ENABLE: case FDBTransactionOptions::CHECK_WRITES_ENABLE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.checkWritesEnabled = true; options.checkWritesEnabled = true;
break; break;
case FDBTransactionOptions::DEBUG_DUMP: case FDBTransactionOptions::DEBUG_DUMP:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.debugDump = true; options.debugDump = true;
break; break;
@ -5095,7 +5107,7 @@ void Transaction::setOption(FDBTransactionOptions::Option option, Optional<Strin
break; break;
case FDBTransactionOptions::DEBUG_TRANSACTION_IDENTIFIER: case FDBTransactionOptions::DEBUG_TRANSACTION_IDENTIFIER:
validateOptionValue(value, true); validateOptionValuePresent(value);
if (value.get().size() > 100 || value.get().size() == 0) { if (value.get().size() > 100 || value.get().size() == 0) {
throw invalid_option_value(); throw invalid_option_value();
@ -5122,7 +5134,7 @@ void Transaction::setOption(FDBTransactionOptions::Option option, Optional<Strin
break; break;
case FDBTransactionOptions::LOG_TRANSACTION: case FDBTransactionOptions::LOG_TRANSACTION:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
if (trLogInfo && !trLogInfo->identifier.empty()) { if (trLogInfo && !trLogInfo->identifier.empty()) {
trLogInfo->logTo(TransactionLogInfo::TRACE_LOG); trLogInfo->logTo(TransactionLogInfo::TRACE_LOG);
} else { } else {
@ -5133,7 +5145,7 @@ void Transaction::setOption(FDBTransactionOptions::Option option, Optional<Strin
break; break;
case FDBTransactionOptions::TRANSACTION_LOGGING_MAX_FIELD_LENGTH: case FDBTransactionOptions::TRANSACTION_LOGGING_MAX_FIELD_LENGTH:
validateOptionValue(value, true); validateOptionValuePresent(value);
{ {
int maxFieldLength = extractIntOption(value, -1, std::numeric_limits<int32_t>::max()); int maxFieldLength = extractIntOption(value, -1, std::numeric_limits<int32_t>::max());
if (maxFieldLength == 0) { if (maxFieldLength == 0) {
@ -5147,7 +5159,7 @@ void Transaction::setOption(FDBTransactionOptions::Option option, Optional<Strin
break; break;
case FDBTransactionOptions::SERVER_REQUEST_TRACING: case FDBTransactionOptions::SERVER_REQUEST_TRACING:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
debugTransaction(deterministicRandom()->randomUniqueID()); debugTransaction(deterministicRandom()->randomUniqueID());
if (trLogInfo && !trLogInfo->identifier.empty()) { if (trLogInfo && !trLogInfo->identifier.empty()) {
TraceEvent(SevInfo, "TransactionBeingTraced") TraceEvent(SevInfo, "TransactionBeingTraced")
@ -5157,23 +5169,23 @@ void Transaction::setOption(FDBTransactionOptions::Option option, Optional<Strin
break; break;
case FDBTransactionOptions::MAX_RETRY_DELAY: case FDBTransactionOptions::MAX_RETRY_DELAY:
validateOptionValue(value, true); validateOptionValuePresent(value);
options.maxBackoff = extractIntOption(value, 0, std::numeric_limits<int32_t>::max()) / 1000.0; options.maxBackoff = extractIntOption(value, 0, std::numeric_limits<int32_t>::max()) / 1000.0;
break; break;
case FDBTransactionOptions::SIZE_LIMIT: case FDBTransactionOptions::SIZE_LIMIT:
validateOptionValue(value, true); validateOptionValuePresent(value);
options.sizeLimit = extractIntOption(value, 32, CLIENT_KNOBS->TRANSACTION_SIZE_LIMIT); options.sizeLimit = extractIntOption(value, 32, CLIENT_KNOBS->TRANSACTION_SIZE_LIMIT);
break; break;
case FDBTransactionOptions::LOCK_AWARE: case FDBTransactionOptions::LOCK_AWARE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.lockAware = true; options.lockAware = true;
options.readOnly = false; options.readOnly = false;
break; break;
case FDBTransactionOptions::READ_LOCK_AWARE: case FDBTransactionOptions::READ_LOCK_AWARE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
if (!options.lockAware) { if (!options.lockAware) {
options.lockAware = true; options.lockAware = true;
options.readOnly = true; options.readOnly = true;
@ -5181,34 +5193,34 @@ void Transaction::setOption(FDBTransactionOptions::Option option, Optional<Strin
break; break;
case FDBTransactionOptions::FIRST_IN_BATCH: case FDBTransactionOptions::FIRST_IN_BATCH:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.firstInBatch = true; options.firstInBatch = true;
break; break;
case FDBTransactionOptions::USE_PROVISIONAL_PROXIES: case FDBTransactionOptions::USE_PROVISIONAL_PROXIES:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.getReadVersionFlags |= GetReadVersionRequest::FLAG_USE_PROVISIONAL_PROXIES; options.getReadVersionFlags |= GetReadVersionRequest::FLAG_USE_PROVISIONAL_PROXIES;
info.useProvisionalProxies = true; info.useProvisionalProxies = true;
break; break;
case FDBTransactionOptions::INCLUDE_PORT_IN_ADDRESS: case FDBTransactionOptions::INCLUDE_PORT_IN_ADDRESS:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.includePort = true; options.includePort = true;
break; break;
case FDBTransactionOptions::TAG: case FDBTransactionOptions::TAG:
validateOptionValue(value, true); validateOptionValuePresent(value);
options.tags.addTag(value.get()); options.tags.addTag(value.get());
break; break;
case FDBTransactionOptions::AUTO_THROTTLE_TAG: case FDBTransactionOptions::AUTO_THROTTLE_TAG:
validateOptionValue(value, true); validateOptionValuePresent(value);
options.tags.addTag(value.get()); options.tags.addTag(value.get());
options.readTags.addTag(value.get()); options.readTags.addTag(value.get());
break; break;
case FDBTransactionOptions::SPAN_PARENT: case FDBTransactionOptions::SPAN_PARENT:
validateOptionValue(value, true); validateOptionValuePresent(value);
if (value.get().size() != 16) { if (value.get().size() != 16) {
throw invalid_option_value(); throw invalid_option_value();
} }
@ -5216,12 +5228,12 @@ void Transaction::setOption(FDBTransactionOptions::Option option, Optional<Strin
break; break;
case FDBTransactionOptions::REPORT_CONFLICTING_KEYS: case FDBTransactionOptions::REPORT_CONFLICTING_KEYS:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.reportConflictingKeys = true; options.reportConflictingKeys = true;
break; break;
case FDBTransactionOptions::EXPENSIVE_CLEAR_COST_ESTIMATION_ENABLE: case FDBTransactionOptions::EXPENSIVE_CLEAR_COST_ESTIMATION_ENABLE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.expensiveClearCostEstimation = true; options.expensiveClearCostEstimation = true;
break; break;
@ -6117,7 +6129,7 @@ ACTOR Future<Void> snapCreate(Database cx, Standalone<StringRef> snapCmd, UID sn
&CommitProxyInterface::proxySnapReq, &CommitProxyInterface::proxySnapReq,
ProxySnapRequest(snapCmd, snapUID, snapUID), ProxySnapRequest(snapCmd, snapUID, snapUID),
cx->taskID, cx->taskID,
true /*atmostOnce*/))) { AtMostOnce::TRUE))) {
TraceEvent("SnapCreateExit").detail("SnapCmd", snapCmd.toString()).detail("UID", snapUID); TraceEvent("SnapCreateExit").detail("SnapCmd", snapCmd.toString()).detail("UID", snapUID);
return Void(); return Void();
} }

15
fdbclient/NativeAPI.actor.h

@ -27,6 +27,7 @@
#elif !defined(FDBCLIENT_NATIVEAPI_ACTOR_H) #elif !defined(FDBCLIENT_NATIVEAPI_ACTOR_H)
#define FDBCLIENT_NATIVEAPI_ACTOR_H #define FDBCLIENT_NATIVEAPI_ACTOR_H
#include "flow/BooleanParam.h"
#include "flow/flow.h" #include "flow/flow.h"
#include "flow/TDMetric.actor.h" #include "flow/TDMetric.actor.h"
#include "fdbclient/FDBTypes.h" #include "fdbclient/FDBTypes.h"
@ -51,7 +52,8 @@ void addref(DatabaseContext* ptr);
template <> template <>
void delref(DatabaseContext* ptr); void delref(DatabaseContext* ptr);
void validateOptionValue(Optional<StringRef> value, bool shouldBePresent); void validateOptionValuePresent(Optional<StringRef> value);
void validateOptionValueNotPresent(Optional<StringRef> value);
void enableClientInfoLogging(); void enableClientInfoLogging();
@ -72,6 +74,9 @@ struct NetworkOptions {
NetworkOptions(); NetworkOptions();
}; };
// TODO: Reduce scope?
DECLARE_BOOLEAN_PARAM(IsInternal);
class Database { class Database {
public: public:
enum { API_VERSION_LATEST = -1 }; enum { API_VERSION_LATEST = -1 };
@ -81,13 +86,13 @@ public:
// on another thread // on another thread
static Database createDatabase(Reference<ClusterConnectionFile> connFile, static Database createDatabase(Reference<ClusterConnectionFile> connFile,
int apiVersion, int apiVersion,
bool internal = true, IsInternal internal = IsInternal::TRUE,
LocalityData const& clientLocality = LocalityData(), LocalityData const& clientLocality = LocalityData(),
DatabaseContext* preallocatedDb = nullptr); DatabaseContext* preallocatedDb = nullptr);
static Database createDatabase(std::string connFileName, static Database createDatabase(std::string connFileName,
int apiVersion, int apiVersion,
bool internal = true, IsInternal internal = IsInternal::TRUE,
LocalityData const& clientLocality = LocalityData()); LocalityData const& clientLocality = LocalityData());
Database() {} // an uninitialized database can be destructed or reassigned safely; that's it Database() {} // an uninitialized database can be destructed or reassigned safely; that's it
@ -111,8 +116,10 @@ private:
void setNetworkOption(FDBNetworkOptions::Option option, Optional<StringRef> value = Optional<StringRef>()); void setNetworkOption(FDBNetworkOptions::Option option, Optional<StringRef> value = Optional<StringRef>());
DECLARE_BOOLEAN_PARAM(UseMetrics);
// Configures the global networking machinery // Configures the global networking machinery
void setupNetwork(uint64_t transportId = 0, bool useMetrics = false); void setupNetwork(uint64_t transportId = 0, UseMetrics useMetrics = UseMetrics::FALSE);
// This call blocks while the network is running. To use the API in a single-threaded // This call blocks while the network is running. To use the API in a single-threaded
// environment, the calling program must have ACTORs already launched that are waiting // environment, the calling program must have ACTORs already launched that are waiting

22
fdbclient/ReadYourWrites.actor.cpp

@ -2165,7 +2165,7 @@ void ReadYourWritesTransaction::setOption(FDBTransactionOptions::Option option,
void ReadYourWritesTransaction::setOptionImpl(FDBTransactionOptions::Option option, Optional<StringRef> value) { void ReadYourWritesTransaction::setOptionImpl(FDBTransactionOptions::Option option, Optional<StringRef> value) {
switch (option) { switch (option) {
case FDBTransactionOptions::READ_YOUR_WRITES_DISABLE: case FDBTransactionOptions::READ_YOUR_WRITES_DISABLE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
if (!reading.isReady() || !cache.empty() || !writes.empty()) if (!reading.isReady() || !cache.empty() || !writes.empty())
throw client_invalid_operation(); throw client_invalid_operation();
@ -2174,26 +2174,26 @@ void ReadYourWritesTransaction::setOptionImpl(FDBTransactionOptions::Option opti
break; break;
case FDBTransactionOptions::READ_AHEAD_DISABLE: case FDBTransactionOptions::READ_AHEAD_DISABLE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.readAheadDisabled = true; options.readAheadDisabled = true;
break; break;
case FDBTransactionOptions::NEXT_WRITE_NO_WRITE_CONFLICT_RANGE: case FDBTransactionOptions::NEXT_WRITE_NO_WRITE_CONFLICT_RANGE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.nextWriteDisableConflictRange = true; options.nextWriteDisableConflictRange = true;
break; break;
case FDBTransactionOptions::ACCESS_SYSTEM_KEYS: case FDBTransactionOptions::ACCESS_SYSTEM_KEYS:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.readSystemKeys = true; options.readSystemKeys = true;
options.writeSystemKeys = true; options.writeSystemKeys = true;
break; break;
case FDBTransactionOptions::READ_SYSTEM_KEYS: case FDBTransactionOptions::READ_SYSTEM_KEYS:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.readSystemKeys = true; options.readSystemKeys = true;
break; break;
@ -2217,30 +2217,30 @@ void ReadYourWritesTransaction::setOptionImpl(FDBTransactionOptions::Option opti
transactionDebugInfo->transactionName = value.present() ? value.get().toString() : ""; transactionDebugInfo->transactionName = value.present() ? value.get().toString() : "";
break; break;
case FDBTransactionOptions::SNAPSHOT_RYW_ENABLE: case FDBTransactionOptions::SNAPSHOT_RYW_ENABLE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.snapshotRywEnabled++; options.snapshotRywEnabled++;
break; break;
case FDBTransactionOptions::SNAPSHOT_RYW_DISABLE: case FDBTransactionOptions::SNAPSHOT_RYW_DISABLE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.snapshotRywEnabled--; options.snapshotRywEnabled--;
break; break;
case FDBTransactionOptions::USED_DURING_COMMIT_PROTECTION_DISABLE: case FDBTransactionOptions::USED_DURING_COMMIT_PROTECTION_DISABLE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.disableUsedDuringCommitProtection = true; options.disableUsedDuringCommitProtection = true;
break; break;
case FDBTransactionOptions::SPECIAL_KEY_SPACE_RELAXED: case FDBTransactionOptions::SPECIAL_KEY_SPACE_RELAXED:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.specialKeySpaceRelaxed = true; options.specialKeySpaceRelaxed = true;
break; break;
case FDBTransactionOptions::SPECIAL_KEY_SPACE_ENABLE_WRITES: case FDBTransactionOptions::SPECIAL_KEY_SPACE_ENABLE_WRITES:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.specialKeySpaceChangeConfiguration = true; options.specialKeySpaceChangeConfiguration = true;
break; break;
case FDBTransactionOptions::BYPASS_UNREADABLE: case FDBTransactionOptions::BYPASS_UNREADABLE:
validateOptionValue(value, false); validateOptionValueNotPresent(value);
options.bypassUnreadable = true; options.bypassUnreadable = true;
break; break;
default: default:

2
fdbclient/ThreadSafeTransaction.cpp

@ -122,7 +122,7 @@ ThreadSafeDatabase::ThreadSafeDatabase(std::string connFilename, int apiVersion)
[db, connFile, apiVersion]() { [db, connFile, apiVersion]() {
try { try {
Database::createDatabase( Database::createDatabase(
Reference<ClusterConnectionFile>(connFile), apiVersion, false, LocalityData(), db) Reference<ClusterConnectionFile>(connFile), apiVersion, IsInternal::FALSE, LocalityData(), db)
.extractPtr(); .extractPtr();
} catch (Error& e) { } catch (Error& e) {
new (db) DatabaseContext(e); new (db) DatabaseContext(e);

4
fdbrpc/LoadBalance.actor.cpp

@ -18,9 +18,13 @@
* limitations under the License. * limitations under the License.
*/ */
#include "fdbrpc/LoadBalance.actor.h"
#include "flow/flow.h" #include "flow/flow.h"
#include "flow/actorcompiler.h" // This must be the last #include. #include "flow/actorcompiler.h" // This must be the last #include.
DEFINE_BOOLEAN_PARAM(AtMostOnce);
DEFINE_BOOLEAN_PARAM(TriedAllOptions);
// Throwing all_alternatives_failed will cause the client to issue a GetKeyLocationRequest to the proxy, so this actor // Throwing all_alternatives_failed will cause the client to issue a GetKeyLocationRequest to the proxy, so this actor
// attempts to limit the number of these errors thrown by a single client to prevent it from saturating the proxies with // attempts to limit the number of these errors thrown by a single client to prevent it from saturating the proxies with
// these requests // these requests

30
fdbrpc/LoadBalance.actor.h

@ -28,6 +28,7 @@
#elif !defined(FLOW_LOADBALANCE_ACTOR_H) #elif !defined(FLOW_LOADBALANCE_ACTOR_H)
#define FLOW_LOADBALANCE_ACTOR_H #define FLOW_LOADBALANCE_ACTOR_H
#include "flow/BooleanParam.h"
#include "flow/flow.h" #include "flow/flow.h"
#include "flow/Knobs.h" #include "flow/Knobs.h"
@ -238,6 +239,9 @@ Future<Void> tssComparison(Req req,
return Void(); return Void();
} }
DECLARE_BOOLEAN_PARAM(AtMostOnce);
DECLARE_BOOLEAN_PARAM(TriedAllOptions);
// Stores state for a request made by the load balancer // Stores state for a request made by the load balancer
template <class Request, class Interface, class Multi> template <class Request, class Interface, class Multi>
struct RequestData : NonCopyable { struct RequestData : NonCopyable {
@ -245,7 +249,7 @@ struct RequestData : NonCopyable {
Future<Reply> response; Future<Reply> response;
Reference<ModelHolder> modelHolder; Reference<ModelHolder> modelHolder;
bool triedAllOptions = false; TriedAllOptions triedAllOptions{ TriedAllOptions::FALSE };
bool requestStarted = false; // true once the request has been sent to an alternative bool requestStarted = false; // true once the request has been sent to an alternative
bool requestProcessed = false; // true once a response has been received and handled by checkAndProcessResult bool requestProcessed = false; // true once a response has been received and handled by checkAndProcessResult
@ -284,7 +288,7 @@ struct RequestData : NonCopyable {
// Initializes the request state and starts it, possibly after a backoff delay // Initializes the request state and starts it, possibly after a backoff delay
void startRequest( void startRequest(
double backoff, double backoff,
bool triedAllOptions, TriedAllOptions triedAllOptions,
RequestStream<Request> const* stream, RequestStream<Request> const* stream,
Request& request, Request& request,
QueueModel* model, QueueModel* model,
@ -320,8 +324,8 @@ struct RequestData : NonCopyable {
// A return value with an error means that the error should be thrown back to original caller // A return value with an error means that the error should be thrown back to original caller
static ErrorOr<bool> checkAndProcessResultImpl(Reply const& result, static ErrorOr<bool> checkAndProcessResultImpl(Reply const& result,
Reference<ModelHolder> modelHolder, Reference<ModelHolder> modelHolder,
bool atMostOnce, AtMostOnce atMostOnce,
bool triedAllOptions) { TriedAllOptions triedAllOptions) {
ASSERT(modelHolder); ASSERT(modelHolder);
Optional<LoadBalancedReply> loadBalancedReply; Optional<LoadBalancedReply> loadBalancedReply;
@ -377,7 +381,7 @@ struct RequestData : NonCopyable {
// A return value of true means that the request completed successfully // A return value of true means that the request completed successfully
// A return value of false means that the request failed but should be retried // A return value of false means that the request failed but should be retried
// In the event of a non-retryable failure, an error is thrown indicating the failure // In the event of a non-retryable failure, an error is thrown indicating the failure
bool checkAndProcessResult(bool atMostOnce) { bool checkAndProcessResult(AtMostOnce atMostOnce) {
ASSERT(response.isReady()); ASSERT(response.isReady());
requestProcessed = true; requestProcessed = true;
@ -412,9 +416,9 @@ struct RequestData : NonCopyable {
// We need to process the lagging request in order to update the queue model // We need to process the lagging request in order to update the queue model
Reference<ModelHolder> holderCapture = std::move(modelHolder); Reference<ModelHolder> holderCapture = std::move(modelHolder);
bool triedAllOptionsCapture = triedAllOptions; auto triedAllOptionsCapture = triedAllOptions;
Future<Void> updateModel = map(response, [holderCapture, triedAllOptionsCapture](Reply result) { Future<Void> updateModel = map(response, [holderCapture, triedAllOptionsCapture](Reply result) {
checkAndProcessResultImpl(result, holderCapture, false, triedAllOptionsCapture); checkAndProcessResultImpl(result, holderCapture, AtMostOnce::FALSE, triedAllOptionsCapture);
return Void(); return Void();
}); });
model->addActor.send(updateModel); model->addActor.send(updateModel);
@ -441,7 +445,8 @@ Future<REPLY_TYPE(Request)> loadBalance(
RequestStream<Request> Interface::*channel, RequestStream<Request> Interface::*channel,
Request request = Request(), Request request = Request(),
TaskPriority taskID = TaskPriority::DefaultPromiseEndpoint, TaskPriority taskID = TaskPriority::DefaultPromiseEndpoint,
bool atMostOnce = false, // if true, throws request_maybe_delivered() instead of retrying automatically AtMostOnce atMostOnce =
AtMostOnce::FALSE, // if true, throws request_maybe_delivered() instead of retrying automatically
QueueModel* model = nullptr) { QueueModel* model = nullptr) {
state RequestData<Request, Interface, Multi> firstRequestData; state RequestData<Request, Interface, Multi> firstRequestData;
@ -453,6 +458,8 @@ Future<REPLY_TYPE(Request)> loadBalance(
state Promise<Void> requestFinished; state Promise<Void> requestFinished;
state double startTime = now(); state double startTime = now();
state TriedAllOptions triedAllOptions = TriedAllOptions::FALSE;
setReplyPriority(request, taskID); setReplyPriority(request, taskID);
if (!alternatives) if (!alternatives)
return Never(); return Never();
@ -556,7 +563,6 @@ Future<REPLY_TYPE(Request)> loadBalance(
state int numAttempts = 0; state int numAttempts = 0;
state double backoff = 0; state double backoff = 0;
state bool triedAllOptions = false;
// Issue requests to selected servers. // Issue requests to selected servers.
loop { loop {
if (now() - startTime > (g_network->isSimulated() ? 30.0 : 600.0)) { if (now() - startTime > (g_network->isSimulated() ? 30.0 : 600.0)) {
@ -595,7 +601,7 @@ Future<REPLY_TYPE(Request)> loadBalance(
break; break;
nextAlt = (nextAlt + 1) % alternatives->size(); nextAlt = (nextAlt + 1) % alternatives->size();
if (nextAlt == startAlt) if (nextAlt == startAlt)
triedAllOptions = true; triedAllOptions = TriedAllOptions::TRUE;
stream = nullptr; stream = nullptr;
} }
@ -702,7 +708,7 @@ Future<REPLY_TYPE(Request)> loadBalance(
nextAlt = (nextAlt + 1) % alternatives->size(); nextAlt = (nextAlt + 1) % alternatives->size();
if (nextAlt == startAlt) if (nextAlt == startAlt)
triedAllOptions = true; triedAllOptions = TriedAllOptions::TRUE;
resetReply(request, taskID); resetReply(request, taskID);
secondDelay = Never(); secondDelay = Never();
} }
@ -724,7 +730,7 @@ Future<REPLY_TYPE(Request)> basicLoadBalance(Reference<ModelInterface<Multi>> al
RequestStream<Request> Interface::*channel, RequestStream<Request> Interface::*channel,
Request request = Request(), Request request = Request(),
TaskPriority taskID = TaskPriority::DefaultPromiseEndpoint, TaskPriority taskID = TaskPriority::DefaultPromiseEndpoint,
bool atMostOnce = false) { AtMostOnce atMostOnce = AtMostOnce::FALSE) {
setReplyPriority(request, taskID); setReplyPriority(request, taskID);
if (!alternatives) if (!alternatives)
return Never(); return Never();

2
fdbserver/BackupWorker.actor.cpp

@ -243,7 +243,7 @@ struct BackupData {
minKnownCommittedVersion(invalidVersion), savedVersion(req.startVersion - 1), popVersion(req.startVersion - 1), minKnownCommittedVersion(invalidVersion), savedVersion(req.startVersion - 1), popVersion(req.startVersion - 1),
cc("BackupWorker", myId.toString()), pulledVersion(0), paused(false), cc("BackupWorker", myId.toString()), pulledVersion(0), paused(false),
lock(new FlowLock(SERVER_KNOBS->BACKUP_LOCK_BYTES)) { lock(new FlowLock(SERVER_KNOBS->BACKUP_LOCK_BYTES)) {
cx = openDBOnServer(db, TaskPriority::DefaultEndpoint, true, true); cx = openDBOnServer(db, TaskPriority::DefaultEndpoint, LockAware::TRUE);
specialCounter(cc, "SavedVersion", [this]() { return this->savedVersion; }); specialCounter(cc, "SavedVersion", [this]() { return this->savedVersion; });
specialCounter(cc, "MinKnownCommittedVersion", [this]() { return this->minKnownCommittedVersion; }); specialCounter(cc, "MinKnownCommittedVersion", [this]() { return this->minKnownCommittedVersion; });

6
fdbserver/ClusterController.actor.cpp

@ -133,9 +133,9 @@ public:
serverInfo(new AsyncVar<ServerDBInfo>()), db(DatabaseContext::create(clientInfo, serverInfo(new AsyncVar<ServerDBInfo>()), db(DatabaseContext::create(clientInfo,
Future<Void>(), Future<Void>(),
LocalityData(), LocalityData(),
true, EnableLocalityLoadBalance::TRUE,
TaskPriority::DefaultEndpoint, TaskPriority::DefaultEndpoint,
true)) // SOMEDAY: Locality! LockAware::TRUE)) // SOMEDAY: Locality!
{} {}
void setDistributor(const DataDistributorInterface& interf) { void setDistributor(const DataDistributorInterface& interf) {
@ -2860,7 +2860,7 @@ public:
serverInfo.clusterInterface = ccInterface; serverInfo.clusterInterface = ccInterface;
serverInfo.myLocality = locality; serverInfo.myLocality = locality;
db.serverInfo->set(serverInfo); db.serverInfo->set(serverInfo);
cx = openDBOnServer(db.serverInfo, TaskPriority::DefaultEndpoint, true, true); cx = openDBOnServer(db.serverInfo, TaskPriority::DefaultEndpoint, LockAware::TRUE);
} }
~ClusterControllerData() { ~ClusterControllerData() {

14
fdbserver/DataDistribution.actor.cpp

@ -5765,7 +5765,7 @@ ACTOR Future<Void> dataDistribution(Reference<DataDistributorData> self,
state double lastLimited = 0; state double lastLimited = 0;
self->addActor.send(monitorBatchLimitedTime(self->dbInfo, &lastLimited)); self->addActor.send(monitorBatchLimitedTime(self->dbInfo, &lastLimited));
state Database cx = openDBOnServer(self->dbInfo, TaskPriority::DataDistributionLaunch, true, true); state Database cx = openDBOnServer(self->dbInfo, TaskPriority::DataDistributionLaunch, LockAware::TRUE);
cx->locationCacheSize = SERVER_KNOBS->DD_LOCATION_CACHE_SIZE; cx->locationCacheSize = SERVER_KNOBS->DD_LOCATION_CACHE_SIZE;
// cx->setOption( FDBDatabaseOptions::LOCATION_CACHE_SIZE, StringRef((uint8_t*) // cx->setOption( FDBDatabaseOptions::LOCATION_CACHE_SIZE, StringRef((uint8_t*)
@ -6106,7 +6106,7 @@ static std::set<int> const& normalDataDistributorErrors() {
} }
ACTOR Future<Void> ddSnapCreateCore(DistributorSnapRequest snapReq, Reference<AsyncVar<struct ServerDBInfo>> db) { ACTOR Future<Void> ddSnapCreateCore(DistributorSnapRequest snapReq, Reference<AsyncVar<struct ServerDBInfo>> db) {
state Database cx = openDBOnServer(db, TaskPriority::DefaultDelay, true, true); state Database cx = openDBOnServer(db, TaskPriority::DefaultDelay, LockAware::TRUE);
state ReadYourWritesTransaction tr(cx); state ReadYourWritesTransaction tr(cx);
loop { loop {
try { try {
@ -6447,7 +6447,7 @@ ACTOR Future<Void> dataDistributor(DataDistributorInterface di, Reference<AsyncV
state Reference<DataDistributorData> self(new DataDistributorData(db, di.id())); state Reference<DataDistributorData> self(new DataDistributorData(db, di.id()));
state Future<Void> collection = actorCollection(self->addActor.getFuture()); state Future<Void> collection = actorCollection(self->addActor.getFuture());
state PromiseStream<GetMetricsListRequest> getShardMetricsList; state PromiseStream<GetMetricsListRequest> getShardMetricsList;
state Database cx = openDBOnServer(db, TaskPriority::DefaultDelay, true, true); state Database cx = openDBOnServer(db, TaskPriority::DefaultDelay, LockAware::TRUE);
state ActorCollection actors(false); state ActorCollection actors(false);
state DDEnabledState ddEnabledState; state DDEnabledState ddEnabledState;
self->addActor.send(actors.getResult()); self->addActor.send(actors.getResult());
@ -6498,8 +6498,8 @@ ACTOR Future<Void> dataDistributor(DataDistributorInterface di, Reference<AsyncV
std::unique_ptr<DDTeamCollection> testTeamCollection(int teamSize, std::unique_ptr<DDTeamCollection> testTeamCollection(int teamSize,
Reference<IReplicationPolicy> policy, Reference<IReplicationPolicy> policy,
int processCount) { int processCount) {
Database database = Database database = DatabaseContext::create(
DatabaseContext::create(makeReference<AsyncVar<ClientDBInfo>>(), Never(), LocalityData(), false); makeReference<AsyncVar<ClientDBInfo>>(), Never(), LocalityData(), EnableLocalityLoadBalance::FALSE);
DatabaseConfiguration conf; DatabaseConfiguration conf;
conf.storageTeamSize = teamSize; conf.storageTeamSize = teamSize;
@ -6541,8 +6541,8 @@ std::unique_ptr<DDTeamCollection> testTeamCollection(int teamSize,
std::unique_ptr<DDTeamCollection> testMachineTeamCollection(int teamSize, std::unique_ptr<DDTeamCollection> testMachineTeamCollection(int teamSize,
Reference<IReplicationPolicy> policy, Reference<IReplicationPolicy> policy,
int processCount) { int processCount) {
Database database = Database database = DatabaseContext::create(
DatabaseContext::create(makeReference<AsyncVar<ClientDBInfo>>(), Never(), LocalityData(), false); makeReference<AsyncVar<ClientDBInfo>>(), Never(), LocalityData(), EnableLocalityLoadBalance::FALSE);
DatabaseConfiguration conf; DatabaseConfiguration conf;
conf.storageTeamSize = teamSize; conf.storageTeamSize = teamSize;

2
fdbserver/GrvProxyServer.actor.cpp

@ -253,7 +253,7 @@ struct GrvProxyData {
RequestStream<GetReadVersionRequest> getConsistentReadVersion, RequestStream<GetReadVersionRequest> getConsistentReadVersion,
Reference<AsyncVar<ServerDBInfo>> db) Reference<AsyncVar<ServerDBInfo>> db)
: dbgid(dbgid), stats(dbgid), master(master), getConsistentReadVersion(getConsistentReadVersion), : dbgid(dbgid), stats(dbgid), master(master), getConsistentReadVersion(getConsistentReadVersion),
cx(openDBOnServer(db, TaskPriority::DefaultEndpoint, true, true)), db(db), lastStartCommit(0), cx(openDBOnServer(db, TaskPriority::DefaultEndpoint, LockAware::TRUE)), db(db), lastStartCommit(0),
lastCommitLatency(SERVER_KNOBS->REQUIRED_MIN_RECOVERY_DURATION), updateCommitRequests(0), lastCommitTime(0), lastCommitLatency(SERVER_KNOBS->REQUIRED_MIN_RECOVERY_DURATION), updateCommitRequests(0), lastCommitTime(0),
minKnownCommittedVersion(invalidVersion) {} minKnownCommittedVersion(invalidVersion) {}
}; };

2
fdbserver/OldTLogServer_6_0.actor.cpp

@ -311,7 +311,7 @@ struct TLogData : NonCopyable {
targetVolatileBytes(SERVER_KNOBS->TLOG_SPILL_THRESHOLD), overheadBytesInput(0), overheadBytesDurable(0), targetVolatileBytes(SERVER_KNOBS->TLOG_SPILL_THRESHOLD), overheadBytesInput(0), overheadBytesDurable(0),
concurrentLogRouterReads(SERVER_KNOBS->CONCURRENT_LOG_ROUTER_READS), ignorePopRequest(false), concurrentLogRouterReads(SERVER_KNOBS->CONCURRENT_LOG_ROUTER_READS), ignorePopRequest(false),
ignorePopDeadline(), ignorePopUid(), dataFolder(folder), toBePopped() { ignorePopDeadline(), ignorePopUid(), dataFolder(folder), toBePopped() {
cx = openDBOnServer(dbInfo, TaskPriority::DefaultEndpoint, true, true); cx = openDBOnServer(dbInfo, TaskPriority::DefaultEndpoint, LockAware::TRUE);
} }
}; };

2
fdbserver/OldTLogServer_6_2.actor.cpp

@ -375,7 +375,7 @@ struct TLogData : NonCopyable {
peekMemoryLimiter(SERVER_KNOBS->TLOG_SPILL_REFERENCE_MAX_PEEK_MEMORY_BYTES), peekMemoryLimiter(SERVER_KNOBS->TLOG_SPILL_REFERENCE_MAX_PEEK_MEMORY_BYTES),
concurrentLogRouterReads(SERVER_KNOBS->CONCURRENT_LOG_ROUTER_READS), ignorePopRequest(false), concurrentLogRouterReads(SERVER_KNOBS->CONCURRENT_LOG_ROUTER_READS), ignorePopRequest(false),
ignorePopDeadline(), ignorePopUid(), dataFolder(folder), toBePopped() { ignorePopDeadline(), ignorePopUid(), dataFolder(folder), toBePopped() {
cx = openDBOnServer(dbInfo, TaskPriority::DefaultEndpoint, true, true); cx = openDBOnServer(dbInfo, TaskPriority::DefaultEndpoint, LockAware::TRUE);
} }
}; };

2
fdbserver/ProxyCommitData.actor.h

@ -247,7 +247,7 @@ struct ProxyCommitData {
mostRecentProcessedRequestNumber(0), getConsistentReadVersion(getConsistentReadVersion), commit(commit), mostRecentProcessedRequestNumber(0), getConsistentReadVersion(getConsistentReadVersion), commit(commit),
lastCoalesceTime(0), localCommitBatchesStarted(0), locked(false), lastCoalesceTime(0), localCommitBatchesStarted(0), locked(false),
commitBatchInterval(SERVER_KNOBS->COMMIT_TRANSACTION_BATCH_INTERVAL_MIN), firstProxy(firstProxy), commitBatchInterval(SERVER_KNOBS->COMMIT_TRANSACTION_BATCH_INTERVAL_MIN), firstProxy(firstProxy),
cx(openDBOnServer(db, TaskPriority::DefaultEndpoint, true, true)), db(db), cx(openDBOnServer(db, TaskPriority::DefaultEndpoint, LockAware::TRUE)), db(db),
singleKeyMutationEvent(LiteralStringRef("SingleKeyMutation")), commitBatchesMemBytesCount(0), lastTxsPop(0), singleKeyMutationEvent(LiteralStringRef("SingleKeyMutation")), commitBatchesMemBytesCount(0), lastTxsPop(0),
lastStartCommit(0), lastCommitLatency(SERVER_KNOBS->REQUIRED_MIN_RECOVERY_DURATION), lastCommitTime(0), lastStartCommit(0), lastCommitLatency(SERVER_KNOBS->REQUIRED_MIN_RECOVERY_DURATION), lastCommitTime(0),
lastMasterReset(now()), lastResolverReset(now()) { lastMasterReset(now()), lastResolverReset(now()) {

2
fdbserver/Ratekeeper.actor.cpp

@ -1409,7 +1409,7 @@ ACTOR Future<Void> configurationMonitor(RatekeeperData* self) {
} }
ACTOR Future<Void> ratekeeper(RatekeeperInterface rkInterf, Reference<AsyncVar<ServerDBInfo>> dbInfo) { ACTOR Future<Void> ratekeeper(RatekeeperInterface rkInterf, Reference<AsyncVar<ServerDBInfo>> dbInfo) {
state RatekeeperData self(rkInterf.id(), openDBOnServer(dbInfo, TaskPriority::DefaultEndpoint, true, true)); state RatekeeperData self(rkInterf.id(), openDBOnServer(dbInfo, TaskPriority::DefaultEndpoint, LockAware::TRUE));
state Future<Void> timeout = Void(); state Future<Void> timeout = Void();
state std::vector<Future<Void>> tlogTrackers; state std::vector<Future<Void>> tlogTrackers;
state std::vector<TLogInterface> tlogInterfs; state std::vector<TLogInterface> tlogInterfs;

2
fdbserver/RestoreWorker.actor.cpp

@ -410,7 +410,7 @@ ACTOR Future<Void> restoreWorker(Reference<ClusterConnectionFile> connFile,
LocalityData locality, LocalityData locality,
std::string coordFolder) { std::string coordFolder) {
try { try {
Database cx = Database::createDatabase(connFile, Database::API_VERSION_LATEST, true, locality); Database cx = Database::createDatabase(connFile, Database::API_VERSION_LATEST, IsInternal::TRUE, locality);
wait(reportErrors(_restoreWorker(cx, locality), "RestoreWorker")); wait(reportErrors(_restoreWorker(cx, locality), "RestoreWorker"));
} catch (Error& e) { } catch (Error& e) {
TraceEvent("FastRestoreWorker").detail("Error", e.what()); TraceEvent("FastRestoreWorker").detail("Error", e.what());

7
fdbserver/Status.actor.cpp

@ -2605,10 +2605,9 @@ ACTOR Future<JsonBuilderObject> lockedStatusFetcher(Reference<AsyncVar<ServerDBI
std::set<std::string>* incomplete_reasons) { std::set<std::string>* incomplete_reasons) {
state JsonBuilderObject statusObj; state JsonBuilderObject statusObj;
state Database cx = openDBOnServer(db, state Database cx =
TaskPriority::DefaultEndpoint, openDBOnServer(db,
true, TaskPriority::DefaultEndpoint); // Open a new database connection that isn't lock-aware
false); // Open a new database connection that isn't lock-aware
state Transaction tr(cx); state Transaction tr(cx);
state int timeoutSeconds = 5; state int timeoutSeconds = 5;
state Future<Void> getTimeout = delay(timeoutSeconds); state Future<Void> getTimeout = delay(timeoutSeconds);

4
fdbserver/StorageCache.actor.cpp

@ -251,7 +251,7 @@ public:
newestAvailableVersion.insert(allKeys, invalidVersion); newestAvailableVersion.insert(allKeys, invalidVersion);
newestDirtyVersion.insert(allKeys, invalidVersion); newestDirtyVersion.insert(allKeys, invalidVersion);
addCacheRange(CacheRangeInfo::newNotAssigned(allKeys)); addCacheRange(CacheRangeInfo::newNotAssigned(allKeys));
cx = openDBOnServer(db, TaskPriority::DefaultEndpoint, true, true); cx = openDBOnServer(db, TaskPriority::DefaultEndpoint, LockAware::TRUE);
} }
// Puts the given cacheRange into cachedRangeMap. The caller is responsible for adding cacheRanges // Puts the given cacheRange into cachedRangeMap. The caller is responsible for adding cacheRanges
@ -1392,7 +1392,7 @@ ACTOR Future<Void> fetchKeys(StorageCacheData* data, AddingCacheRange* cacheRang
// TODO: NEELAM: what's this for? // TODO: NEELAM: what's this for?
// FIXME: remove when we no longer support upgrades from 5.X // FIXME: remove when we no longer support upgrades from 5.X
if (debug_getRangeRetries >= 100) { if (debug_getRangeRetries >= 100) {
data->cx->enableLocalityLoadBalance = false; data->cx->enableLocalityLoadBalance = EnableLocalityLoadBalance::FALSE;
} }
debug_getRangeRetries++; debug_getRangeRetries++;

2
fdbserver/TLogServer.actor.cpp

@ -386,7 +386,7 @@ struct TLogData : NonCopyable {
commitLatencyDist(Histogram::getHistogram(LiteralStringRef("tLog"), commitLatencyDist(Histogram::getHistogram(LiteralStringRef("tLog"),
LiteralStringRef("commit"), LiteralStringRef("commit"),
Histogram::Unit::microseconds)) { Histogram::Unit::microseconds)) {
cx = openDBOnServer(dbInfo, TaskPriority::DefaultEndpoint, true, true); cx = openDBOnServer(dbInfo, TaskPriority::DefaultEndpoint, LockAware::TRUE);
} }
}; };

10
fdbserver/WorkerInterface.actor.h

@ -830,10 +830,16 @@ ACTOR Future<Void> traceRole(Role role, UID roleId);
struct ServerDBInfo; struct ServerDBInfo;
#ifndef __DATABASE_BOOLEAN_PARAMS__
#define __DATABASE_BOOLEAN_PARAMS__
DECLARE_BOOLEAN_PARAM(EnableLocalityLoadBalance);
DECLARE_BOOLEAN_PARAM(LockAware);
#endif
class Database openDBOnServer(Reference<AsyncVar<ServerDBInfo>> const& db, class Database openDBOnServer(Reference<AsyncVar<ServerDBInfo>> const& db,
TaskPriority taskID = TaskPriority::DefaultEndpoint, TaskPriority taskID = TaskPriority::DefaultEndpoint,
bool enableLocalityLoadBalance = true, LockAware lockAware = LockAware::FALSE,
bool lockAware = false); EnableLocalityLoadBalance enableLocalityLoadBalance = EnableLocalityLoadBalance::TRUE);
ACTOR Future<Void> extractClusterInterface(Reference<AsyncVar<Optional<struct ClusterControllerFullInterface>>> a, ACTOR Future<Void> extractClusterInterface(Reference<AsyncVar<Optional<struct ClusterControllerFullInterface>>> a,
Reference<AsyncVar<Optional<struct ClusterInterface>>> b); Reference<AsyncVar<Optional<struct ClusterInterface>>> b);

4
fdbserver/masterserver.actor.cpp

@ -577,7 +577,7 @@ Future<Void> sendMasterRegistration(MasterData* self,
} }
ACTOR Future<Void> updateRegistration(Reference<MasterData> self, Reference<ILogSystem> logSystem) { ACTOR Future<Void> updateRegistration(Reference<MasterData> self, Reference<ILogSystem> logSystem) {
state Database cx = openDBOnServer(self->dbInfo, TaskPriority::DefaultEndpoint, true, true); state Database cx = openDBOnServer(self->dbInfo, TaskPriority::DefaultEndpoint, LockAware::TRUE);
state Future<Void> trigger = self->registrationTrigger.onTrigger(); state Future<Void> trigger = self->registrationTrigger.onTrigger();
state Future<Void> updateLogsKey; state Future<Void> updateLogsKey;
@ -1965,7 +1965,7 @@ ACTOR Future<Void> masterCore(Reference<MasterData> self) {
self->addActor.send(resolutionBalancing(self)); self->addActor.send(resolutionBalancing(self));
self->addActor.send(changeCoordinators(self)); self->addActor.send(changeCoordinators(self));
Database cx = openDBOnServer(self->dbInfo, TaskPriority::DefaultEndpoint, true, true); Database cx = openDBOnServer(self->dbInfo, TaskPriority::DefaultEndpoint, LockAware::TRUE);
self->addActor.send(configurationMonitor(self, cx)); self->addActor.send(configurationMonitor(self, cx));
if (self->configuration.backupWorkerEnabled) { if (self->configuration.backupWorkerEnabled) {
self->addActor.send(recruitBackupWorkers(self, cx)); self->addActor.send(recruitBackupWorkers(self, cx));

6
fdbserver/storageserver.actor.cpp

@ -852,7 +852,7 @@ public:
newestDirtyVersion.insert(allKeys, invalidVersion); newestDirtyVersion.insert(allKeys, invalidVersion);
addShard(ShardInfo::newNotAssigned(allKeys)); addShard(ShardInfo::newNotAssigned(allKeys));
cx = openDBOnServer(db, TaskPriority::DefaultEndpoint, true, true); cx = openDBOnServer(db, TaskPriority::DefaultEndpoint, LockAware::TRUE);
} }
//~StorageServer() { fclose(log); } //~StorageServer() { fclose(log); }
@ -2970,7 +2970,7 @@ ACTOR Future<Void> fetchKeys(StorageServer* data, AddingShard* shard) {
// FIXME: remove when we no longer support upgrades from 5.X // FIXME: remove when we no longer support upgrades from 5.X
if (debug_getRangeRetries >= 100) { if (debug_getRangeRetries >= 100) {
data->cx->enableLocalityLoadBalance = false; data->cx->enableLocalityLoadBalance = EnableLocalityLoadBalance::FALSE;
TraceEvent(SevWarnAlways, "FKDisableLB").detail("FKID", fetchKeysID); TraceEvent(SevWarnAlways, "FKDisableLB").detail("FKID", fetchKeysID);
} }
@ -3018,7 +3018,7 @@ ACTOR Future<Void> fetchKeys(StorageServer* data, AddingShard* shard) {
} }
// FIXME: remove when we no longer support upgrades from 5.X // FIXME: remove when we no longer support upgrades from 5.X
data->cx->enableLocalityLoadBalance = true; data->cx->enableLocalityLoadBalance = EnableLocalityLoadBalance::TRUE;
TraceEvent(SevWarnAlways, "FKReenableLB").detail("FKID", fetchKeysID); TraceEvent(SevWarnAlways, "FKReenableLB").detail("FKID", fetchKeysID);
// We have completed the fetch and write of the data, now we wait for MVCC window to pass. // We have completed the fetch and write of the data, now we wait for MVCC window to pass.

2
fdbserver/tester.actor.cpp

@ -607,7 +607,7 @@ ACTOR Future<Void> testerServerWorkload(WorkloadRequest work,
startRole(Role::TESTER, workIface.id(), UID(), details); startRole(Role::TESTER, workIface.id(), UID(), details);
if (work.useDatabase) { if (work.useDatabase) {
cx = Database::createDatabase(ccf, -1, true, locality); cx = Database::createDatabase(ccf, -1, IsInternal::TRUE, locality);
wait(delay(1.0)); wait(delay(1.0));
} }

12
fdbserver/worker.actor.cpp

@ -138,8 +138,8 @@ ACTOR static Future<Void> extractClientInfo(Reference<AsyncVar<ServerDBInfo>> db
Database openDBOnServer(Reference<AsyncVar<ServerDBInfo>> const& db, Database openDBOnServer(Reference<AsyncVar<ServerDBInfo>> const& db,
TaskPriority taskID, TaskPriority taskID,
bool enableLocalityLoadBalance, LockAware lockAware,
bool lockAware) { EnableLocalityLoadBalance enableLocalityLoadBalance) {
auto info = makeReference<AsyncVar<ClientDBInfo>>(); auto info = makeReference<AsyncVar<ClientDBInfo>>();
auto cx = DatabaseContext::create(info, auto cx = DatabaseContext::create(info,
extractClientInfo(db, info), extractClientInfo(db, info),
@ -1215,15 +1215,15 @@ ACTOR Future<Void> workerServer(Reference<ClusterConnectionFile> connFile,
if (metricsConnFile.size() > 0) { if (metricsConnFile.size() > 0) {
try { try {
state Database db = state Database db =
Database::createDatabase(metricsConnFile, Database::API_VERSION_LATEST, true, locality); Database::createDatabase(metricsConnFile, Database::API_VERSION_LATEST, IsInternal::TRUE, locality);
metricsLogger = runMetrics(db, KeyRef(metricsPrefix)); metricsLogger = runMetrics(db, KeyRef(metricsPrefix));
} catch (Error& e) { } catch (Error& e) {
TraceEvent(SevWarnAlways, "TDMetricsBadClusterFile").error(e).detail("ConnFile", metricsConnFile); TraceEvent(SevWarnAlways, "TDMetricsBadClusterFile").error(e).detail("ConnFile", metricsConnFile);
} }
} else { } else {
bool lockAware = metricsPrefix.size() && metricsPrefix[0] == '\xff'; auto lockAware = metricsPrefix.size() && metricsPrefix[0] == '\xff' ? LockAware::TRUE : LockAware::FALSE;
metricsLogger = runMetrics(openDBOnServer(dbInfo, TaskPriority::DefaultEndpoint, true, lockAware), metricsLogger =
KeyRef(metricsPrefix)); runMetrics(openDBOnServer(dbInfo, TaskPriority::DefaultEndpoint, lockAware), KeyRef(metricsPrefix));
} }
} }

45
flow/BooleanParam.h Normal file

@ -0,0 +1,45 @@
/*
* Arena.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/Trace.h"
#define DECLARE_BOOLEAN_PARAM(ParamName) \
class ParamName { \
bool value; \
explicit constexpr ParamName(bool value) : value(value) {} \
\
public: \
operator bool() const { return value; } \
static ParamName const TRUE, FALSE; \
}; \
template <> \
struct Traceable<ParamName> : std::true_type { \
static std::string toString(ParamName const& value) { return Traceable<bool>::toString(value); } \
};
#define DEFINE_BOOLEAN_PARAM(ParamName) \
ParamName const ParamName::TRUE = ParamName(true); \
ParamName const ParamName::FALSE = ParamName(false);
#define BOOLEAN_PARAM(ParamName) \
DECLARE_BOOLEAN_PARAM(ParamName) \
DEFINE_BOOLEAN_PARAM(ParamName)\

1
flow/CMakeLists.txt

@ -6,6 +6,7 @@ set(FLOW_SRCS
Arena.cpp Arena.cpp
Arena.h Arena.h
AsioReactor.h AsioReactor.h
BooleanParam.h
CompressedInt.actor.cpp CompressedInt.actor.cpp
CompressedInt.h CompressedInt.h
Deque.cpp Deque.cpp