Merge pull request #6602 from apple/blob_integration

Blob integration
This commit is contained in:
Evan Tschannen 2022-03-23 12:02:43 -07:00 committed by GitHub
commit 4a085fc844
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
56 changed files with 7895 additions and 2250 deletions

View File

@ -0,0 +1,100 @@
#!/usr/bin/env python3
#
# alloc_instrumentation_traces.py
#
# 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.
#
"""
Example trace:
{ "Severity": "10", "Time": "194.878474", "DateTime": "2022-02-01T16:28:27Z", "Type": "MemSample", "Machine": "2.1.1.0:2", "ID": "0000000000000000", "Count": "943", "TotalSize": "540000000", "SampleCount": "54", "Hash": "980074757", "Bt": "addr2line -e fdbserver.debug -p -C -f -i 0x1919b72 0x3751d43 0x37518cc 0x19930f8 0x199dac3 0x1999e7c 0x21a1061 0x31e8fc5 0x31e784a 0x10ab3a8 0x36bf4c6 0x36bf304 0x36beea4 0x36bf352 0x36bfa1c 0x10ab3a8 0x37b22fe 0x37a16ee 0x368c754 0x19202d5 0x7fb3fe2d6555 0x1077029", "ThreadID": "10074331651862410074", "LogGroup": "default" }
"""
# This program analyzes MemSample trace events produced by setting ALLOC_INSTRUMENTATION in FastAlloc.h
# It outputs the top memory users by total size as well as number of allocations.
# Example usage: cat trace.* | ./alloc_instrumentation_traces.py
import sys
import json
byCnt = []
bySize = []
totalSize = 0
lastTimestamp = ""
for line in sys.stdin:
ev = json.loads(line.rstrip())
type = ev["Type"]
if (type != 'MemSample'):
continue
bt = ev["Bt"]
if (bt == "na"):
continue
timestamp = ev["Time"]
cnt = int(ev["Count"])
scnt = int(ev["SampleCount"])
size = int(ev["TotalSize"])
h = ev["Hash"]
if (timestamp != lastTimestamp):
byCnt = []
bySize = []
totalSize = 0
lastTimestamp = timestamp
# print(str(cnt) + " " + str(scnt) + " " + str(size) + " " + h)
byCnt.append( (cnt, scnt, size, h, bt) )
bySize.append( (size, cnt, size, h, bt) )
totalSize += size
byCnt.sort(reverse=True)
bySize.sort(reverse=True)
btByHash = {}
byte_suffix = ["Bytes", "KB", "MB", "GB", "TB"]
def byte_str(bytes):
suffix_idx = 0
while (bytes >= 1024 * 10):
suffix_idx += 1
bytes //= 1024
return str(bytes) + ' ' + byte_suffix[suffix_idx]
print("By Size")
print("-------\r\n")
for x in bySize[:10]:
# print(str(x[0]) + ": " + x[3])
print(str(x[1]) + " / " + byte_str(x[0]) + " (" + byte_str(x[0] // x[1]) + " per alloc):\r\n" + x[4] + "\r\n")
btByHash[x[3]] = x[4]
print()
print("By Count")
print("--------\r\n")
for x in byCnt[:5]:
# print(str(x[0]) + ": " + x[3])
print(str(x[0]) + " / " + byte_str(x[2]) + " (" + byte_str(x[2] // x[0]) + " per alloc):\r\n" + x[4] + "\r\n")
btByHash[x[3]] = x[4]

View File

@ -265,7 +265,7 @@ CommandFactory configureFactory(
"commit_proxies=<COMMIT_PROXIES>|grv_proxies=<GRV_PROXIES>|logs=<LOGS>|resolvers=<RESOLVERS>>*|"
"count=<TSS_COUNT>|perpetual_storage_wiggle=<WIGGLE_SPEED>|perpetual_storage_wiggle_locality="
"<<LOCALITY_KEY>:<LOCALITY_VALUE>|0>|storage_migration_type={disabled|gradual|aggressive}"
"|tenant_mode={disabled|optional_experimental|required_experimental}",
"|tenant_mode={disabled|optional_experimental|required_experimental}|blob_granules_enabled={0|1}",
"change the database configuration",
"The `new' option, if present, initializes a new database with the given configuration rather than changing "
"the configuration of an existing one. When used, both a redundancy mode and a storage engine must be "

View File

@ -790,6 +790,7 @@ void configureGenerator(const char* text, const char* line, std::vector<std::str
"perpetual_storage_wiggle_locality=",
"storage_migration_type="
"tenant_mode=",
"blob_granules_enabled=",
nullptr };
arrayGenerator(text, line, opts, lc);
}

View File

@ -77,17 +77,18 @@ struct BlobGranuleChunkRef {
constexpr static FileIdentifier file_identifier = 865198;
KeyRangeRef keyRange;
Version includedVersion;
Version snapshotVersion;
Optional<BlobFilePointerRef> snapshotFile; // not set if it's an incremental read
VectorRef<BlobFilePointerRef> deltaFiles;
GranuleDeltas newDeltas;
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, keyRange, includedVersion, snapshotFile, deltaFiles, newDeltas);
serializer(ar, keyRange, includedVersion, snapshotVersion, snapshotFile, deltaFiles, newDeltas);
}
};
enum BlobGranuleSplitState { Unknown = 0, Started = 1, Assigned = 2, Done = 3 };
enum BlobGranuleSplitState { Unknown = 0, Initialized = 1, Assigned = 2, Done = 3 };
struct BlobGranuleHistoryValue {
constexpr static FileIdentifier file_identifier = 991434;

View File

@ -37,6 +37,7 @@ struct BlobWorkerStats {
Counter readReqDeltaBytesReturned;
Counter commitVersionChecks;
Counter granuleUpdateErrors;
Counter granuleRequestTimeouts;
int numRangesAssigned;
int mutationBytesBuffered;
@ -57,7 +58,8 @@ struct BlobWorkerStats {
wrongShardServer("WrongShardServer", cc), changeFeedInputBytes("RangeFeedInputBytes", cc),
readReqTotalFilesReturned("ReadReqTotalFilesReturned", cc),
readReqDeltaBytesReturned("ReadReqDeltaBytesReturned", cc), commitVersionChecks("CommitVersionChecks", cc),
granuleUpdateErrors("GranuleUpdateErrors", cc), numRangesAssigned(0), mutationBytesBuffered(0) {
granuleUpdateErrors("GranuleUpdateErrors", cc), granuleRequestTimeouts("GranuleRequestTimeouts", cc),
numRangesAssigned(0), mutationBytesBuffered(0), activeReadRequests(0) {
specialCounter(cc, "NumRangesAssigned", [this]() { return this->numRangesAssigned; });
specialCounter(cc, "MutationBytesBuffered", [this]() { return this->mutationBytesBuffered; });
specialCounter(cc, "ActiveReadRequests", [this]() { return this->activeReadRequests; });

View File

@ -34,6 +34,7 @@ struct BlobWorkerInterface {
RequestStream<struct BlobGranuleFileRequest> blobGranuleFileRequest;
RequestStream<struct AssignBlobRangeRequest> assignBlobRangeRequest;
RequestStream<struct RevokeBlobRangeRequest> revokeBlobRangeRequest;
RequestStream<struct GetGranuleAssignmentsRequest> granuleAssignmentsRequest;
RequestStream<struct GranuleStatusStreamRequest> granuleStatusStreamRequest;
RequestStream<struct HaltBlobWorkerRequest> haltBlobWorker;
@ -58,6 +59,7 @@ struct BlobWorkerInterface {
blobGranuleFileRequest,
assignBlobRangeRequest,
revokeBlobRangeRequest,
granuleAssignmentsRequest,
granuleStatusStreamRequest,
haltBlobWorker,
locality,
@ -94,19 +96,6 @@ struct BlobGranuleFileRequest {
}
};
struct AssignBlobRangeReply {
constexpr static FileIdentifier file_identifier = 6431923;
bool epochOk; // false if the worker has seen a new manager
AssignBlobRangeReply() {}
explicit AssignBlobRangeReply(bool epochOk) : epochOk(epochOk) {}
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, epochOk);
}
};
struct RevokeBlobRangeRequest {
constexpr static FileIdentifier file_identifier = 4844288;
Arena arena;
@ -114,7 +103,7 @@ struct RevokeBlobRangeRequest {
int64_t managerEpoch;
int64_t managerSeqno;
bool dispose;
ReplyPromise<AssignBlobRangeReply> reply;
ReplyPromise<Void> reply;
RevokeBlobRangeRequest() {}
@ -124,6 +113,12 @@ struct RevokeBlobRangeRequest {
}
};
/*
* Continue: Blob worker should continue handling a granule that was evaluated for a split
* Normal: Blob worker should open the granule and start processing it
*/
enum AssignRequestType { Normal = 0, Continue = 1 };
struct AssignBlobRangeRequest {
constexpr static FileIdentifier file_identifier = 905381;
Arena arena;
@ -133,16 +128,15 @@ struct AssignBlobRangeRequest {
// If continueAssignment is true, this is just to instruct the worker that it *still* owns the range, so it should
// re-snapshot it and continue.
// For an initial assignment, reassignent, split, or merge, continueAssignment==false.
bool continueAssignment;
AssignRequestType type;
ReplyPromise<AssignBlobRangeReply> reply;
ReplyPromise<Void> reply;
AssignBlobRangeRequest() {}
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, keyRange, managerEpoch, managerSeqno, continueAssignment, reply, arena);
serializer(ar, keyRange, managerEpoch, managerSeqno, type, reply, arena);
}
};
@ -153,22 +147,22 @@ struct GranuleStatusReply : public ReplyPromiseStreamReply {
KeyRange granuleRange;
bool doSplit;
bool writeHotSplit;
int64_t epoch;
int64_t seqno;
UID granuleID;
Version startVersion;
Version latestVersion;
GranuleStatusReply() {}
explicit GranuleStatusReply(KeyRange range,
bool doSplit,
bool writeHotSplit,
int64_t epoch,
int64_t seqno,
UID granuleID,
Version startVersion,
Version latestVersion)
: granuleRange(range), doSplit(doSplit), epoch(epoch), seqno(seqno), granuleID(granuleID),
startVersion(startVersion), latestVersion(latestVersion) {}
Version startVersion)
: granuleRange(range), doSplit(doSplit), writeHotSplit(writeHotSplit), epoch(epoch), seqno(seqno),
granuleID(granuleID), startVersion(startVersion) {}
int expectedSize() const { return sizeof(GranuleStatusReply) + granuleRange.expectedSize(); }
@ -179,11 +173,11 @@ struct GranuleStatusReply : public ReplyPromiseStreamReply {
ReplyPromiseStreamReply::sequence,
granuleRange,
doSplit,
writeHotSplit,
epoch,
seqno,
granuleID,
startVersion,
latestVersion);
startVersion);
}
};
@ -220,4 +214,42 @@ struct HaltBlobWorkerRequest {
}
};
struct GranuleAssignmentRef {
KeyRangeRef range;
int64_t epochAssigned;
int64_t seqnoAssigned;
GranuleAssignmentRef() {}
explicit GranuleAssignmentRef(KeyRangeRef range, int64_t epochAssigned, int64_t seqnoAssigned)
: range(range), epochAssigned(epochAssigned), seqnoAssigned(seqnoAssigned) {}
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, range, epochAssigned, seqnoAssigned);
}
};
struct GetGranuleAssignmentsReply {
constexpr static FileIdentifier file_identifier = 9191718;
Arena arena;
VectorRef<GranuleAssignmentRef> assignments;
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, assignments, arena);
}
};
struct GetGranuleAssignmentsRequest {
constexpr static FileIdentifier file_identifier = 4121494;
int64_t managerEpoch;
ReplyPromise<GetGranuleAssignmentsReply> reply;
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, managerEpoch, reply);
}
};
#endif

View File

@ -80,6 +80,7 @@ void ClientKnobs::initialize(Randomize randomize) {
init( CHANGE_FEED_LOCATION_LIMIT, 10000 );
init( CHANGE_FEED_CACHE_SIZE, 100000 ); if( randomize && BUGGIFY ) CHANGE_FEED_CACHE_SIZE = 1;
init( CHANGE_FEED_POP_TIMEOUT, 5.0 );
init( CHANGE_FEED_STREAM_MIN_BYTES, 1e4 ); if( randomize && BUGGIFY ) CHANGE_FEED_STREAM_MIN_BYTES = 1;
init( MAX_BATCH_SIZE, 1000 ); if( randomize && BUGGIFY ) MAX_BATCH_SIZE = 1;
init( GRV_BATCH_TIMEOUT, 0.005 ); if( randomize && BUGGIFY ) GRV_BATCH_TIMEOUT = 0.1;
@ -275,8 +276,9 @@ void ClientKnobs::initialize(Randomize randomize) {
init( BUSYNESS_SPIKE_START_THRESHOLD, 0.100 );
init( BUSYNESS_SPIKE_SATURATED_THRESHOLD, 0.500 );
// blob granules
init( ENABLE_BLOB_GRANULES, false );
// multi-version client control
init( MVC_CLIENTLIB_CHUNK_SIZE, 8*1024 );
init( MVC_CLIENTLIB_CHUNKS_PER_TRANSACTION, 32 );
// clang-format on
}

View File

@ -79,6 +79,7 @@ public:
int64_t CHANGE_FEED_LOCATION_LIMIT;
int64_t CHANGE_FEED_CACHE_SIZE;
double CHANGE_FEED_POP_TIMEOUT;
int64_t CHANGE_FEED_STREAM_MIN_BYTES;
int MAX_BATCH_SIZE;
double GRV_BATCH_TIMEOUT;
@ -267,8 +268,9 @@ public:
double BUSYNESS_SPIKE_START_THRESHOLD;
double BUSYNESS_SPIKE_SATURATED_THRESHOLD;
// blob granules
bool ENABLE_BLOB_GRANULES;
// multi-version client control
int MVC_CLIENTLIB_CHUNK_SIZE;
int MVC_CLIENTLIB_CHUNKS_PER_TRANSACTION;
ClientKnobs(Randomize randomize);
void initialize(Randomize randomize);

View File

@ -50,6 +50,7 @@ void DatabaseConfiguration::resetInternal() {
perpetualStorageWiggleSpeed = 0;
perpetualStorageWiggleLocality = "0";
storageMigrationType = StorageMigrationType::DEFAULT;
blobGranulesEnabled = false;
tenantMode = TenantMode::DISABLED;
}
@ -404,6 +405,7 @@ StatusObject DatabaseConfiguration::toJSON(bool noPolicies) const {
result["perpetual_storage_wiggle"] = perpetualStorageWiggleSpeed;
result["perpetual_storage_wiggle_locality"] = perpetualStorageWiggleLocality;
result["storage_migration_type"] = storageMigrationType.toString();
result["blob_granules_enabled"] = (int32_t)blobGranulesEnabled;
result["tenant_mode"] = tenantMode.toString();
return result;
}
@ -633,6 +635,9 @@ bool DatabaseConfiguration::setInternal(KeyRef key, ValueRef value) {
tenantMode = (TenantMode::Mode)type;
} else if (ck == LiteralStringRef("proxies")) {
overwriteProxiesCount();
} else if (ck == LiteralStringRef("blob_granules_enabled")) {
parse((&type), value);
blobGranulesEnabled = (type != 0);
} else {
return false;
}

View File

@ -250,6 +250,8 @@ struct DatabaseConfiguration {
// Storage Migration Type
StorageMigrationType storageMigrationType;
// Blob Granules
bool blobGranulesEnabled;
TenantMode tenantMode;
// Excluded servers (no state should be here)

View File

@ -181,6 +181,7 @@ struct ChangeFeedStorageData : ReferenceCounted<ChangeFeedStorageData> {
NotifiedVersion version;
NotifiedVersion desired;
Promise<Void> destroyed;
UID interfToken;
~ChangeFeedStorageData() { destroyed.send(Void()); }
};
@ -196,6 +197,10 @@ struct ChangeFeedData : ReferenceCounted<ChangeFeedData> {
std::vector<Reference<ChangeFeedStorageData>> storageData;
AsyncVar<int> notAtLatest;
Promise<Void> refresh;
Version maxSeenVersion;
Version endVersion = invalidVersion;
Version popVersion =
invalidVersion; // like TLog pop version, set by SS and client can check it to see if they missed data
ChangeFeedData() : notAtLatest(1) {}
};
@ -292,6 +297,10 @@ public:
StorageMetrics const& permittedError,
int shardLimit,
int expectedShardCount);
Future<Void> splitStorageMetricsStream(PromiseStream<Key> const& resultsStream,
KeyRange const& keys,
StorageMetrics const& limit,
StorageMetrics const& estimated);
Future<Standalone<VectorRef<KeyRef>>> splitStorageMetrics(KeyRange const& keys,
StorageMetrics const& limit,
StorageMetrics const& estimated);
@ -355,7 +364,9 @@ public:
Key rangeID,
Version begin = 0,
Version end = std::numeric_limits<Version>::max(),
KeyRange range = allKeys);
KeyRange range = allKeys,
int replyBufferSize = -1,
bool canReadPopped = true);
Future<std::vector<OverlappingChangeFeedEntry>> getOverlappingChangeFeeds(KeyRangeRef ranges, Version minVersion);
Future<Void> popChangeFeedMutations(Key rangeID, Version version);

View File

@ -175,6 +175,17 @@ std::map<std::string, std::string> configForToken(std::string const& mode) {
}
out[p + key] = format("%d", type);
}
if (key == "blob_granules_enabled") {
int enabled = std::stoi(value);
if (enabled != 0 && enabled != 1) {
printf("Error: Only 0 or 1 are valid values for blob_granules_enabled. "
"1 enables blob granules and 0 disables them.\n");
return out;
}
out[p + key] = value;
}
if (key == "tenant_mode") {
TenantMode tenantMode;
if (value == "disabled") {

File diff suppressed because it is too large Load Diff

View File

@ -80,6 +80,8 @@ struct Notified {
val = std::move(r.val);
}
int numWaiting() { return waiting.size(); }
private:
using Item = std::pair<ValueType, Promise<Void>>;
struct ItemCompare {

View File

@ -810,6 +810,7 @@ const KeyRef JSONSchemas::statusSchema = LiteralStringRef(R"statusSchema(
"aggressive",
"gradual"
]},
"blob_granules_enabled":0,
"tenant_mode": {
"$enum":[
"disabled",

View File

@ -292,7 +292,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
init( SQLITE_CHUNK_SIZE_PAGES_SIM, 1024 ); // 4MB
init( SQLITE_READER_THREADS, 64 ); // number of read threads
init( SQLITE_WRITE_WINDOW_SECONDS, -1 );
init( SQLITE_CURSOR_MAX_LIFETIME_BYTES, 1e6 ); if( randomize && BUGGIFY ) SQLITE_CURSOR_MAX_LIFETIME_BYTES = 0;
init( SQLITE_CURSOR_MAX_LIFETIME_BYTES, 1e6 ); if (buggifySmallShards || simulationMediumShards) SQLITE_CURSOR_MAX_LIFETIME_BYTES = MIN_SHARD_BYTES; if( randomize && BUGGIFY ) SQLITE_CURSOR_MAX_LIFETIME_BYTES = 0;
init( SQLITE_WRITE_WINDOW_LIMIT, -1 );
if( randomize && BUGGIFY ) {
// Choose an window between .01 and 1.01 seconds.
@ -538,6 +538,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
init( POLICY_GENERATIONS, 100 ); if( randomize && BUGGIFY ) POLICY_GENERATIONS = 10;
init( DBINFO_SEND_AMOUNT, 5 );
init( DBINFO_BATCH_DELAY, 0.1 );
init( SINGLETON_RECRUIT_BME_DELAY, 10.0 );
//Move Keys
init( SHARD_READY_DELAY, 0.25 );
@ -650,6 +651,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
init( FETCH_KEYS_PARALLELISM_BYTES, 4e6 ); if( randomize && BUGGIFY ) FETCH_KEYS_PARALLELISM_BYTES = 3e6;
init( FETCH_KEYS_PARALLELISM, 2 );
init( FETCH_KEYS_LOWER_PRIORITY, 0 );
init( FETCH_CHANGEFEED_PARALLELISM, 2 );
init( BUGGIFY_BLOCK_BYTES, 10000 );
init( STORAGE_COMMIT_BYTES, 10000000 ); if( randomize && BUGGIFY ) STORAGE_COMMIT_BYTES = 2000000;
init( STORAGE_FETCH_BYTES, 2500000 ); if( randomize && BUGGIFY ) STORAGE_FETCH_BYTES = 500000;
@ -680,6 +682,8 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
init( FETCH_KEYS_TOO_LONG_TIME_CRITERIA, 300.0 );
init( MAX_STORAGE_COMMIT_TIME, 120.0 ); //The max fsync stall time on the storage server and tlog before marking a disk as failed
init( RANGESTREAM_LIMIT_BYTES, 2e6 ); if( randomize && BUGGIFY ) RANGESTREAM_LIMIT_BYTES = 1;
init( CHANGEFEEDSTREAM_LIMIT_BYTES, 1e6 ); if( randomize && BUGGIFY ) CHANGEFEEDSTREAM_LIMIT_BYTES = 1;
init( BLOBWORKERSTATUSSTREAM_LIMIT_BYTES, 1e4 ); if( randomize && BUGGIFY ) BLOBWORKERSTATUSSTREAM_LIMIT_BYTES = 1;
init( ENABLE_CLEAR_RANGE_EAGER_READS, true );
init( CHECKPOINT_TRANSFER_BLOCK_BYTES, 40e6 );
init( QUICK_GET_VALUE_FALLBACK, true );
@ -826,12 +830,23 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi
init( ENABLE_ENCRYPTION, false );
// Blob granlues
init( BG_URL, "" ); // TODO: store in system key space, eventually
init( BG_SNAPSHOT_FILE_TARGET_BYTES, 10000000 ); if( randomize && BUGGIFY ) { deterministicRandom()->random01() < 0.1 ? BG_SNAPSHOT_FILE_TARGET_BYTES /= 100 : BG_SNAPSHOT_FILE_TARGET_BYTES /= 10; }
init( BG_URL, isSimulated ? "file://fdbblob/" : "" ); // TODO: store in system key space or something, eventually
init( BG_SNAPSHOT_FILE_TARGET_BYTES, 10000000 ); if( buggifySmallShards ) BG_SNAPSHOT_FILE_TARGET_BYTES = 100000; else if (simulationMediumShards || (randomize && BUGGIFY) ) BG_SNAPSHOT_FILE_TARGET_BYTES = 1000000;
init( BG_DELTA_BYTES_BEFORE_COMPACT, BG_SNAPSHOT_FILE_TARGET_BYTES/2 );
init( BG_DELTA_FILE_TARGET_BYTES, BG_DELTA_BYTES_BEFORE_COMPACT/10 );
init( BG_MAX_SPLIT_FANOUT, 10 ); if( randomize && BUGGIFY ) BG_MAX_SPLIT_FANOUT = deterministicRandom()->randomInt(5, 15);
init( BG_HOT_SNAPSHOT_VERSIONS, 5000000 );
init( BLOB_WORKER_INITIAL_SNAPSHOT_PARALLELISM, 8 ); if( randomize && BUGGIFY ) BLOB_WORKER_INITIAL_SNAPSHOT_PARALLELISM = 1;
init( BLOB_WORKER_TIMEOUT, 10.0 ); if( randomize && BUGGIFY ) BLOB_WORKER_TIMEOUT = 1.0;
init( BLOB_WORKER_REQUEST_TIMEOUT, 5.0 ); if( randomize && BUGGIFY ) BLOB_WORKER_REQUEST_TIMEOUT = 1.0;
init( BLOB_WORKERLIST_FETCH_INTERVAL, 1.0 );
init( BLOB_WORKER_BATCH_GRV_INTERVAL, 0.1 );
init( BLOB_MANAGER_STATUS_EXP_BACKOFF_MIN, 0.1 );
init( BLOB_MANAGER_STATUS_EXP_BACKOFF_MAX, 5.0 );
init( BLOB_MANAGER_STATUS_EXP_BACKOFF_EXPONENT, 1.5 );
// clang-format on

View File

@ -470,6 +470,7 @@ public:
double RECRUITMENT_TIMEOUT;
int DBINFO_SEND_AMOUNT;
double DBINFO_BATCH_DELAY;
double SINGLETON_RECRUIT_BME_DELAY;
// Move Keys
double SHARD_READY_DELAY;
@ -586,6 +587,7 @@ public:
int FETCH_KEYS_PARALLELISM_BYTES;
int FETCH_KEYS_PARALLELISM;
int FETCH_KEYS_LOWER_PRIORITY;
int FETCH_CHANGEFEED_PARALLELISM;
int BUGGIFY_BLOCK_BYTES;
double STORAGE_DURABILITY_LAG_REJECT_THRESHOLD;
double STORAGE_DURABILITY_LAG_MIN_RATE;
@ -616,6 +618,8 @@ public:
double FETCH_KEYS_TOO_LONG_TIME_CRITERIA;
double MAX_STORAGE_COMMIT_TIME;
int64_t RANGESTREAM_LIMIT_BYTES;
int64_t CHANGEFEEDSTREAM_LIMIT_BYTES;
int64_t BLOBWORKERSTATUSSTREAM_LIMIT_BYTES;
bool ENABLE_CLEAR_RANGE_EAGER_READS;
bool QUICK_GET_VALUE_FALLBACK;
bool QUICK_GET_KEY_VALUES_FALLBACK;
@ -781,8 +785,18 @@ public:
int BG_SNAPSHOT_FILE_TARGET_BYTES;
int BG_DELTA_FILE_TARGET_BYTES;
int BG_DELTA_BYTES_BEFORE_COMPACT;
int BG_MAX_SPLIT_FANOUT;
int BG_HOT_SNAPSHOT_VERSIONS;
int BLOB_WORKER_INITIAL_SNAPSHOT_PARALLELISM;
double BLOB_WORKER_TIMEOUT; // Blob Manager's reaction time to a blob worker failure
double BLOB_WORKER_REQUEST_TIMEOUT; // Blob Worker's server-side request timeout
double BLOB_WORKERLIST_FETCH_INTERVAL;
double BLOB_WORKER_BATCH_GRV_INTERVAL;
double BLOB_MANAGER_STATUS_EXP_BACKOFF_MIN;
double BLOB_MANAGER_STATUS_EXP_BACKOFF_MAX;
double BLOB_MANAGER_STATUS_EXP_BACKOFF_EXPONENT;
ServerKnobs(Randomize, ClientKnobs*, IsSimulated);
void initialize(Randomize, ClientKnobs*, IsSimulated);

View File

@ -773,6 +773,7 @@ struct ChangeFeedStreamReply : public ReplyPromiseStreamReply {
VectorRef<MutationsAndVersionRef> mutations;
bool atLatestVersion = false;
Version minStreamVersion = invalidVersion;
Version popVersion = invalidVersion;
ChangeFeedStreamReply() {}
@ -786,6 +787,7 @@ struct ChangeFeedStreamReply : public ReplyPromiseStreamReply {
mutations,
atLatestVersion,
minStreamVersion,
popVersion,
arena);
}
};
@ -798,12 +800,18 @@ struct ChangeFeedStreamRequest {
Version begin = 0;
Version end = 0;
KeyRange range;
int replyBufferSize = -1;
bool canReadPopped = true;
UID debugUID; // This is only used for debugging and tracing, but being able to link a client + server side stream
// is so useful for testing, and this is such small overhead compared to streaming large amounts of
// change feed data, it is left in the interface
ReplyPromiseStream<ChangeFeedStreamReply> reply;
ChangeFeedStreamRequest() {}
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, rangeID, begin, end, range, reply, spanContext, arena);
serializer(ar, rangeID, begin, end, range, reply, spanContext, replyBufferSize, canReadPopped, debugUID, arena);
}
};
@ -881,19 +889,21 @@ struct FetchCheckpointRequest {
struct OverlappingChangeFeedEntry {
Key rangeId;
KeyRange range;
bool stopped = false;
Version emptyVersion;
Version stopVersion;
bool operator==(const OverlappingChangeFeedEntry& r) const {
return rangeId == r.rangeId && range == r.range && stopped == r.stopped;
return rangeId == r.rangeId && range == r.range && emptyVersion == r.emptyVersion &&
stopVersion == r.stopVersion;
}
OverlappingChangeFeedEntry() {}
OverlappingChangeFeedEntry(Key const& rangeId, KeyRange const& range, bool stopped)
: rangeId(rangeId), range(range), stopped(stopped) {}
OverlappingChangeFeedEntry(Key const& rangeId, KeyRange const& range, Version emptyVersion, Version stopVersion)
: rangeId(rangeId), range(range), emptyVersion(emptyVersion), stopVersion(stopVersion) {}
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, rangeId, range, stopped);
serializer(ar, rangeId, range, emptyVersion, stopVersion);
}
};

View File

@ -1153,30 +1153,33 @@ const KeyRangeRef blobGranuleMappingKeys(LiteralStringRef("\xff\x02/bgm/"), Lite
const KeyRangeRef blobGranuleLockKeys(LiteralStringRef("\xff\x02/bgl/"), LiteralStringRef("\xff\x02/bgl0"));
const KeyRangeRef blobGranuleSplitKeys(LiteralStringRef("\xff\x02/bgs/"), LiteralStringRef("\xff\x02/bgs0"));
const KeyRangeRef blobGranuleHistoryKeys(LiteralStringRef("\xff\x02/bgh/"), LiteralStringRef("\xff\x02/bgh0"));
const KeyRangeRef blobGranulePruneKeys(LiteralStringRef("\xff\x02/bgp/"), LiteralStringRef("\xff\x02/bgp0"));
const KeyRangeRef blobGranuleVersionKeys(LiteralStringRef("\xff\x02/bgv/"), LiteralStringRef("\xff\x02/bgv0"));
const KeyRef blobGranulePruneChangeKey = LiteralStringRef("\xff\x02/bgpChange");
const uint8_t BG_FILE_TYPE_DELTA = 'D';
const uint8_t BG_FILE_TYPE_SNAPSHOT = 'S';
const Key blobGranuleFileKeyFor(UID granuleID, uint8_t fileType, Version fileVersion) {
const Key blobGranuleFileKeyFor(UID granuleID, Version fileVersion, uint8_t fileType) {
ASSERT(fileType == 'D' || fileType == 'S');
BinaryWriter wr(AssumeVersion(ProtocolVersion::withBlobGranule()));
wr.serializeBytes(blobGranuleFileKeys.begin);
wr << granuleID;
wr << fileType;
wr << bigEndian64(fileVersion);
wr << fileType;
return wr.toValue();
}
std::tuple<UID, uint8_t, Version> decodeBlobGranuleFileKey(KeyRef const& key) {
std::tuple<UID, Version, uint8_t> decodeBlobGranuleFileKey(KeyRef const& key) {
UID granuleID;
uint8_t fileType;
Version fileVersion;
uint8_t fileType;
BinaryReader reader(key.removePrefix(blobGranuleFileKeys.begin), AssumeVersion(ProtocolVersion::withBlobGranule()));
reader >> granuleID;
reader >> fileType;
reader >> fileVersion;
reader >> fileType;
ASSERT(fileType == 'D' || fileType == 'S');
return std::tuple(granuleID, fileType, bigEndian64(fileVersion));
return std::tuple(granuleID, bigEndian64(fileVersion), fileType);
}
const KeyRange blobGranuleFileKeyRangeFor(UID granuleID) {
@ -1206,6 +1209,25 @@ std::tuple<Standalone<StringRef>, int64_t, int64_t> decodeBlobGranuleFileValue(V
return std::tuple(filename, offset, length);
}
const Value blobGranulePruneValueFor(Version version, KeyRange range, bool force) {
BinaryWriter wr(IncludeVersion(ProtocolVersion::withBlobGranule()));
wr << version;
wr << range;
wr << force;
return wr.toValue();
}
std::tuple<Version, KeyRange, bool> decodeBlobGranulePruneValue(ValueRef const& value) {
Version version;
KeyRange range;
bool force;
BinaryReader reader(value, IncludeVersion());
reader >> version;
reader >> range;
reader >> force;
return std::tuple(version, range, force);
}
const Value blobGranuleMappingValueFor(UID const& workerID) {
BinaryWriter wr(IncludeVersion(ProtocolVersion::withBlobGranule()));
wr << workerID;
@ -1284,7 +1306,8 @@ std::pair<BlobGranuleSplitState, Version> decodeBlobGranuleSplitValue(const Valu
BinaryReader reader(value, IncludeVersion());
reader >> st;
reader >> v;
return std::pair(st, v);
return std::pair(st, bigEndian64(v));
}
const Key blobGranuleHistoryKeyFor(KeyRangeRef const& range, Version version) {

View File

@ -563,13 +563,21 @@ extern const KeyRangeRef blobGranuleSplitKeys;
// \xff\x02/bgh/(beginKey,endKey,startVersion) = { granuleUID, [parentGranuleHistoryKeys] }
extern const KeyRangeRef blobGranuleHistoryKeys;
const Key blobGranuleFileKeyFor(UID granuleID, uint8_t fileType, Version fileVersion);
std::tuple<UID, uint8_t, Version> decodeBlobGranuleFileKey(ValueRef const& value);
// \xff\x02/bgp/(start,end) = (version, force)
extern const KeyRangeRef blobGranulePruneKeys;
extern const KeyRangeRef blobGranuleVersionKeys;
extern const KeyRef blobGranulePruneChangeKey;
const Key blobGranuleFileKeyFor(UID granuleID, Version fileVersion, uint8_t fileType);
std::tuple<UID, Version, uint8_t> decodeBlobGranuleFileKey(KeyRef const& key);
const KeyRange blobGranuleFileKeyRangeFor(UID granuleID);
const Value blobGranuleFileValueFor(StringRef const& filename, int64_t offset, int64_t length);
std::tuple<Standalone<StringRef>, int64_t, int64_t> decodeBlobGranuleFileValue(ValueRef const& value);
const Value blobGranulePruneValueFor(Version version, KeyRange range, bool force);
std::tuple<Version, KeyRange, bool> decodeBlobGranulePruneValue(ValueRef const& value);
const Value blobGranuleMappingValueFor(UID const& workerID);
UID decodeBlobGranuleMappingValue(ValueRef const& value);
@ -587,7 +595,7 @@ const Value blobGranuleSplitValueFor(BlobGranuleSplitState st);
std::pair<BlobGranuleSplitState, Version> decodeBlobGranuleSplitValue(ValueRef const& value);
const Key blobGranuleHistoryKeyFor(KeyRangeRef const& range, Version version);
std::pair<KeyRange, Version> decodeBlobGranuleHistoryKey(KeyRef const& value);
std::pair<KeyRange, Version> decodeBlobGranuleHistoryKey(KeyRef const& key);
const KeyRange blobGranuleHistoryKeyRangeFor(KeyRangeRef const& range);
const Value blobGranuleHistoryValueFor(Standalone<BlobGranuleHistoryValue> const& historyValue);

View File

@ -45,6 +45,7 @@
#include "flow/actorcompiler.h" // This must be the last #include.
static NetworkAddressList g_currentDeliveryPeerAddress = NetworkAddressList();
static Future<Void> g_currentDeliveryPeerDisconnect;
constexpr int PACKET_LEN_WIDTH = sizeof(uint32_t);
const uint64_t TOKEN_STREAM_FLAG = 1;
@ -545,28 +546,20 @@ ACTOR Future<Void> connectionWriter(Reference<Peer> self, Reference<IConnection>
}
}
ACTOR Future<Void> delayedHealthUpdate(NetworkAddress address) {
ACTOR Future<Void> delayedHealthUpdate(NetworkAddress address, bool* tooManyConnectionsClosed) {
state double start = now();
state bool delayed = false;
loop {
if (FLOW_KNOBS->HEALTH_MONITOR_MARK_FAILED_UNSTABLE_CONNECTIONS &&
FlowTransport::transport().healthMonitor()->tooManyConnectionsClosed(address) && address.isPublic()) {
if (!delayed) {
TraceEvent("TooManyConnectionsClosedMarkFailed")
.detail("Dest", address)
.detail("StartTime", start)
.detail("ClosedCount", FlowTransport::transport().healthMonitor()->closedConnectionsCount(address));
IFailureMonitor::failureMonitor().setStatus(address, FailureStatus(true));
}
delayed = true;
wait(delayJittered(FLOW_KNOBS->MAX_RECONNECTION_TIME * 2.0));
} else {
if (delayed) {
if (*tooManyConnectionsClosed) {
TraceEvent("TooManyConnectionsClosedMarkAvailable")
.detail("Dest", address)
.detail("StartTime", start)
.detail("TimeElapsed", now() - start)
.detail("ClosedCount", FlowTransport::transport().healthMonitor()->closedConnectionsCount(address));
*tooManyConnectionsClosed = false;
}
IFailureMonitor::failureMonitor().setStatus(address, FailureStatus(false));
break;
@ -586,6 +579,7 @@ ACTOR Future<Void> connectionKeeper(Reference<Peer> self,
state Future<Void> delayedHealthUpdateF;
state Optional<double> firstConnFailedTime = Optional<double>();
state int retryConnect = false;
state bool tooManyConnectionsClosed = false;
loop {
try {
@ -635,7 +629,8 @@ ACTOR Future<Void> connectionKeeper(Reference<Peer> self,
IFailureMonitor::failureMonitor().setStatus(self->destination, FailureStatus(false));
}
if (self->unsent.empty()) {
delayedHealthUpdateF = delayedHealthUpdate(self->destination);
delayedHealthUpdateF =
delayedHealthUpdate(self->destination, &tooManyConnectionsClosed);
choose {
when(wait(delayedHealthUpdateF)) {
conn->close();
@ -675,7 +670,7 @@ ACTOR Future<Void> connectionKeeper(Reference<Peer> self,
try {
self->transport->countConnEstablished++;
if (!delayedHealthUpdateF.isValid())
delayedHealthUpdateF = delayedHealthUpdate(self->destination);
delayedHealthUpdateF = delayedHealthUpdate(self->destination, &tooManyConnectionsClosed);
wait(connectionWriter(self, conn) || reader || connectionMonitor(self) ||
self->resetConnection.onTrigger());
TraceEvent("ConnectionReset", conn ? conn->getDebugID() : UID())
@ -761,6 +756,17 @@ ACTOR Future<Void> connectionKeeper(Reference<Peer> self,
if (conn) {
if (self->destination.isPublic() && e.code() == error_code_connection_failed) {
FlowTransport::transport().healthMonitor()->reportPeerClosed(self->destination);
if (FLOW_KNOBS->HEALTH_MONITOR_MARK_FAILED_UNSTABLE_CONNECTIONS &&
FlowTransport::transport().healthMonitor()->tooManyConnectionsClosed(self->destination) &&
self->destination.isPublic()) {
TraceEvent("TooManyConnectionsClosedMarkFailed")
.detail("Dest", self->destination)
.detail(
"ClosedCount",
FlowTransport::transport().healthMonitor()->closedConnectionsCount(self->destination));
tooManyConnectionsClosed = true;
IFailureMonitor::failureMonitor().setStatus(self->destination, FailureStatus(true));
}
}
conn->close();
@ -776,6 +782,9 @@ ACTOR Future<Void> connectionKeeper(Reference<Peer> self,
// Clients might send more packets in response, which needs to go out on the next connection
IFailureMonitor::failureMonitor().notifyDisconnect(self->destination);
Promise<Void> disconnect = self->disconnect;
self->disconnect = Promise<Void>();
disconnect.send(Void());
if (e.code() == error_code_actor_cancelled)
throw;
@ -918,7 +927,8 @@ ACTOR static void deliver(TransportData* self,
Endpoint destination,
TaskPriority priority,
ArenaReader reader,
bool inReadSocket) {
bool inReadSocket,
Future<Void> disconnect) {
// We want to run the task at the right priority. If the priority is higher than the current priority (which is
// ReadSocket) we can just upgrade. Otherwise we'll context switch so that we don't block other tasks that might run
// with a higher priority. ReplyPromiseStream needs to guarantee that messages are received in the order they were
@ -937,13 +947,16 @@ ACTOR static void deliver(TransportData* self,
}
try {
g_currentDeliveryPeerAddress = destination.addresses;
g_currentDeliveryPeerDisconnect = disconnect;
StringRef data = reader.arenaReadAll();
ASSERT(data.size() > 8);
ArenaObjectReader objReader(reader.arena(), reader.arenaReadAll(), AssumeVersion(reader.protocolVersion()));
receiver->receive(objReader);
g_currentDeliveryPeerAddress = { NetworkAddress() };
g_currentDeliveryPeerDisconnect = Future<Void>();
} catch (Error& e) {
g_currentDeliveryPeerAddress = { NetworkAddress() };
g_currentDeliveryPeerDisconnect = Future<Void>();
TraceEvent(SevError, "ReceiverError")
.error(e)
.detail("Token", destination.token.toString())
@ -977,7 +990,8 @@ static void scanPackets(TransportData* transport,
const uint8_t* e,
Arena& arena,
NetworkAddress const& peerAddress,
ProtocolVersion peerProtocolVersion) {
ProtocolVersion peerProtocolVersion,
Future<Void> disconnect) {
// Find each complete packet in the given byte range and queue a ready task to deliver it.
// Remove the complete packets from the range by increasing unprocessed_begin.
// There won't be more than 64K of data plus one packet, so this shouldn't take a long time.
@ -1090,7 +1104,7 @@ static void scanPackets(TransportData* transport,
// we have many messages to UnknownEndpoint we want to optimize earlier. As deliver is an actor it
// will allocate some state on the heap and this prevents it from doing that.
if (priority != TaskPriority::UnknownEndpoint || (token.first() & TOKEN_STREAM_FLAG) != 0) {
deliver(transport, Endpoint({ peerAddress }, token), priority, std::move(reader), true);
deliver(transport, Endpoint({ peerAddress }, token), priority, std::move(reader), true, disconnect);
}
unprocessed_begin = p = p + packetLen;
@ -1285,8 +1299,13 @@ ACTOR static Future<Void> connectionReader(TransportData* transport,
if (!expectConnectPacket) {
if (compatible || peerProtocolVersion.hasStableInterfaces()) {
scanPackets(
transport, unprocessed_begin, unprocessed_end, arena, peerAddress, peerProtocolVersion);
scanPackets(transport,
unprocessed_begin,
unprocessed_end,
arena,
peerAddress,
peerProtocolVersion,
peer->disconnect.getFuture());
} else {
unprocessed_begin = unprocessed_end;
peer->resetPing.trigger();
@ -1483,6 +1502,10 @@ Endpoint FlowTransport::loadedEndpoint(const UID& token) {
return Endpoint(g_currentDeliveryPeerAddress, token);
}
Future<Void> FlowTransport::loadedDisconnect() {
return g_currentDeliveryPeerDisconnect;
}
void FlowTransport::addPeerReference(const Endpoint& endpoint, bool isStream) {
if (!isStream || !endpoint.getPrimaryAddress().isValid() || !endpoint.getPrimaryAddress().isPublic())
return;
@ -1556,8 +1579,12 @@ static void sendLocal(TransportData* self, ISerializeSource const& what, const E
ASSERT(copy.size() > 0);
TaskPriority priority = self->endpoints.getPriority(destination.token);
if (priority != TaskPriority::UnknownEndpoint || (destination.token.first() & TOKEN_STREAM_FLAG) != 0) {
deliver(
self, destination, priority, ArenaReader(copy.arena(), copy, AssumeVersion(currentProtocolVersion)), false);
deliver(self,
destination,
priority,
ArenaReader(copy.arena(), copy, AssumeVersion(currentProtocolVersion)),
false,
Never());
}
}

View File

@ -169,6 +169,7 @@ struct Peer : public ReferenceCounted<Peer> {
int connectIncomingCount;
int connectFailedCount;
ContinuousSample<double> connectLatencies;
Promise<Void> disconnect;
explicit Peer(TransportData* transport, NetworkAddress const& destination);
@ -269,6 +270,7 @@ public:
static NetworkAddressList getGlobalLocalAddresses() { return transport().getLocalAddresses(); }
Endpoint loadedEndpoint(const UID& token);
Future<Void> loadedDisconnect();
HealthMonitor* healthMonitor();

View File

@ -326,14 +326,14 @@ struct NetNotifiedQueueWithAcknowledgements final : NotifiedQueue<T>,
AcknowledgementReceiver acknowledgements;
Endpoint requestStreamEndpoint;
bool sentError = false;
Promise<Void> onConnect;
NetNotifiedQueueWithAcknowledgements(int futures, int promises) : NotifiedQueue<T>(futures, promises) {}
NetNotifiedQueueWithAcknowledgements(int futures, int promises)
: NotifiedQueue<T>(futures, promises), onConnect(nullptr) {}
NetNotifiedQueueWithAcknowledgements(int futures, int promises, const Endpoint& remoteEndpoint)
: NotifiedQueue<T>(futures, promises), FlowReceiver(remoteEndpoint, true) {
: NotifiedQueue<T>(futures, promises), FlowReceiver(remoteEndpoint, true), onConnect(nullptr) {
// A ReplyPromiseStream will be terminated on the server side if the network connection with the client breaks
acknowledgements.failures = tagError<Void>(
makeDependent<T>(IFailureMonitor::failureMonitor()).onDisconnect(remoteEndpoint.getPrimaryAddress()),
operation_obsolete());
acknowledgements.failures = tagError<Void>(FlowTransport::transport().loadedDisconnect(), operation_obsolete());
}
void destroy() override { delete this; }
@ -350,11 +350,17 @@ struct NetNotifiedQueueWithAcknowledgements final : NotifiedQueue<T>,
// GetKeyValuesStream requests on the same endpoint will fail
IFailureMonitor::failureMonitor().endpointNotFound(requestStreamEndpoint);
}
if (onConnect.isValid() && onConnect.canBeSet()) {
onConnect.send(Void());
}
this->sendError(message.getError());
} else {
if (message.get().asUnderlyingType().acknowledgeToken.present()) {
acknowledgements = AcknowledgementReceiver(
FlowTransport::transport().loadedEndpoint(message.get().asUnderlyingType().acknowledgeToken.get()));
if (onConnect.isValid() && onConnect.canBeSet()) {
onConnect.send(Void());
}
}
if (acknowledgements.sequence != message.get().asUnderlyingType().sequence) {
TraceEvent(SevError, "StreamSequenceMismatch")
@ -487,6 +493,18 @@ public:
void setRequestStreamEndpoint(const Endpoint& endpoint) { queue->requestStreamEndpoint = endpoint; }
bool connected() { return queue->acknowledgements.getRawEndpoint().isValid() || queue->error.isValid(); }
Future<Void> onConnected() {
if (connected()) {
return Void();
}
if (!queue->onConnect.isValid()) {
queue->onConnect = Promise<Void>();
}
return queue->onConnect.getFuture();
}
~ReplyPromiseStream() {
if (queue)
queue->delPromiseRef();
@ -513,6 +531,19 @@ public:
return queue->onEmpty.getFuture();
}
bool isError() const { return !queue->isError(); }
// throws, used to short circuit waiting on the queue if there has been an unexpected error
Future<Void> onError() {
if (queue->hasError() && queue->error.code() != error_code_end_of_stream) {
throw queue->error;
}
if (!queue->onError.isValid()) {
queue->onError = Promise<Void>();
}
return queue->onError.getFuture();
}
uint32_t size() const { return queue->size(); }
// Must be called on the server before sending results on the stream to ratelimit the amount of data outstanding to
@ -729,10 +760,13 @@ public:
Future<Void> disc =
makeDependent<T>(IFailureMonitor::failureMonitor()).onDisconnectOrFailure(getEndpoint());
auto& p = getReplyPromiseStream(value);
Reference<Peer> peer =
FlowTransport::transport().sendUnreliable(SerializeSource<T>(value), getEndpoint(), true);
// FIXME: defer sending the message until we know the connection is established
endStreamOnDisconnect(disc, p, getEndpoint(), peer);
if (disc.isReady()) {
p.sendError(request_maybe_delivered());
} else {
Reference<Peer> peer =
FlowTransport::transport().sendUnreliable(SerializeSource<T>(value), getEndpoint(), true);
endStreamOnDisconnect(disc, p, getEndpoint(), peer);
}
return p;
} else {
send(value);

View File

@ -210,9 +210,21 @@ void endStreamOnDisconnect(Future<Void> signal,
Reference<Peer> peer = Reference<Peer>()) {
state PeerHolder holder = PeerHolder(peer);
stream.setRequestStreamEndpoint(endpoint);
choose {
when(wait(signal)) { stream.sendError(connection_failed()); }
when(wait(stream.getErrorFutureAndDelPromiseRef())) {}
try {
choose {
when(wait(signal)) { stream.sendError(connection_failed()); }
when(wait(peer.isValid() ? peer->disconnect.getFuture() : Never())) {
stream.sendError(connection_failed());
}
when(wait(stream.getErrorFutureAndDelPromiseRef())) {}
}
} catch (Error& e) {
if (e.code() == error_code_broken_promise) {
// getErrorFutureAndDelPromiseRef returned, wait on stream connect or error
if (!stream.connected()) {
wait(signal || stream.onConnected());
}
}
}
}

View File

@ -0,0 +1,104 @@
/*
* BlobGranuleServerCommon.actor.cpp
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2021 Apple Inc. and the FoundationDB project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include "fdbclient/SystemData.h"
#include "fdbclient/BlobGranuleCommon.h"
#include "fdbserver/BlobGranuleServerCommon.actor.h"
#include "fdbclient/CommitTransaction.h"
#include "fdbclient/FDBTypes.h"
#include "fdbclient/ReadYourWrites.h"
#include "flow/Arena.h"
#include "flow/actorcompiler.h" // has to be last include
// Gets the latest granule history node for range that was persisted
ACTOR Future<Optional<GranuleHistory>> getLatestGranuleHistory(Transaction* tr, KeyRange range) {
state KeyRange historyRange = blobGranuleHistoryKeyRangeFor(range);
state RangeResult result = wait(tr->getRange(historyRange, 1, Snapshot::False, Reverse::True));
ASSERT(result.size() <= 1);
Optional<GranuleHistory> history;
if (!result.empty()) {
std::pair<KeyRange, Version> decodedKey = decodeBlobGranuleHistoryKey(result[0].key);
ASSERT(range == decodedKey.first);
history = GranuleHistory(range, decodedKey.second, decodeBlobGranuleHistoryValue(result[0].value));
}
return history;
}
// Gets the files based on the file key range [startKey, endKey)
// and populates the files object accordingly
ACTOR Future<Void> readGranuleFiles(Transaction* tr, Key* startKey, Key endKey, GranuleFiles* files, UID granuleID) {
loop {
int lim = BUGGIFY ? 2 : 1000;
RangeResult res = wait(tr->getRange(KeyRangeRef(*startKey, endKey), lim));
for (auto& it : res) {
UID gid;
uint8_t fileType;
Version version;
Standalone<StringRef> filename;
int64_t offset;
int64_t length;
std::tie(gid, version, fileType) = decodeBlobGranuleFileKey(it.key);
ASSERT(gid == granuleID);
std::tie(filename, offset, length) = decodeBlobGranuleFileValue(it.value);
BlobFileIndex idx(version, filename.toString(), offset, length);
if (fileType == 'S') {
ASSERT(files->snapshotFiles.empty() || files->snapshotFiles.back().version < idx.version);
files->snapshotFiles.push_back(idx);
} else {
ASSERT(fileType == 'D');
ASSERT(files->deltaFiles.empty() || files->deltaFiles.back().version < idx.version);
files->deltaFiles.push_back(idx);
}
}
if (res.more) {
*startKey = keyAfter(res.back().key);
} else {
break;
}
}
return Void();
}
// Wrapper around readGranuleFiles
// Gets all files belonging to the granule with id granule ID
ACTOR Future<GranuleFiles> loadHistoryFiles(Database cx, UID granuleID) {
state KeyRange range = blobGranuleFileKeyRangeFor(granuleID);
state Key startKey = range.begin;
state GranuleFiles files;
state Transaction tr(cx);
loop {
try {
tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE);
wait(readGranuleFiles(&tr, &startKey, range.end, &files, granuleID));
return files;
} catch (Error& e) {
wait(tr.onError(e));
}
}
}

View File

@ -0,0 +1,70 @@
/*
* BlobGranuleServerCommon.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.
*/
#if defined(NO_INTELLISENSE) && !defined(FDBSERVER_BLOBGRANULESERVERCOMMON_ACTOR_G_H)
#define FDBSERVER_BLOBGRANULESERVERCOMMON_ACTOR_G_H
#include "fdbserver/BlobGranuleServerCommon.actor.g.h"
#elif !defined(FDBSERVER_BLOBGRANULESERVERCOMMON_ACTOR_H)
#define FDBSERVER_BLOBGRANULESERVERCOMMON_ACTOR_H
#pragma once
#include "flow/flow.h"
#include "fdbclient/CommitTransaction.h"
#include "fdbclient/FDBTypes.h"
#include "fdbclient/BlobGranuleCommon.h"
#include "flow/actorcompiler.h" // has to be last include
struct GranuleHistory {
KeyRange range;
Version version;
Standalone<BlobGranuleHistoryValue> value;
GranuleHistory() {}
GranuleHistory(KeyRange range, Version version, Standalone<BlobGranuleHistoryValue> value)
: range(range), version(version), value(value) {}
};
// Stores info about a file in blob storage
struct BlobFileIndex {
Version version;
std::string filename;
int64_t offset;
int64_t length;
BlobFileIndex() {}
BlobFileIndex(Version version, std::string filename, int64_t offset, int64_t length)
: version(version), filename(filename), offset(offset), length(length) {}
};
// Stores the files that comprise a blob granule
struct GranuleFiles {
std::deque<BlobFileIndex> snapshotFiles;
std::deque<BlobFileIndex> deltaFiles;
};
class Transaction;
ACTOR Future<Optional<GranuleHistory>> getLatestGranuleHistory(Transaction* tr, KeyRange range);
ACTOR Future<Void> readGranuleFiles(Transaction* tr, Key* startKey, Key endKey, GranuleFiles* files, UID granuleID);
ACTOR Future<GranuleFiles> loadHistoryFiles(Database cx, UID granuleID);
#endif

File diff suppressed because it is too large Load Diff

View File

@ -30,6 +30,8 @@ struct BlobManagerInterface {
constexpr static FileIdentifier file_identifier = 369169;
RequestStream<ReplyPromise<Void>> waitFailure;
RequestStream<struct HaltBlobManagerRequest> haltBlobManager;
RequestStream<struct HaltBlobGranulesRequest> haltBlobGranules;
RequestStream<struct BlobManagerExclusionSafetyCheckRequest> blobManagerExclCheckReq;
struct LocalityData locality;
UID myId;
@ -44,7 +46,7 @@ struct BlobManagerInterface {
template <class Archive>
void serialize(Archive& ar) {
serializer(ar, waitFailure, haltBlobManager, locality, myId);
serializer(ar, waitFailure, haltBlobManager, haltBlobGranules, blobManagerExclCheckReq, locality, myId);
}
};
@ -62,4 +64,46 @@ struct HaltBlobManagerRequest {
}
};
struct HaltBlobGranulesRequest {
constexpr static FileIdentifier file_identifier = 904267;
UID requesterID;
ReplyPromise<Void> reply;
HaltBlobGranulesRequest() {}
explicit HaltBlobGranulesRequest(UID uid) : requesterID(uid) {}
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, requesterID, reply);
}
};
struct BlobManagerExclusionSafetyCheckReply {
constexpr static FileIdentifier file_identifier = 8068627;
bool safe;
BlobManagerExclusionSafetyCheckReply() : safe(false) {}
explicit BlobManagerExclusionSafetyCheckReply(bool safe) : safe(safe) {}
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, safe);
}
};
struct BlobManagerExclusionSafetyCheckRequest {
constexpr static FileIdentifier file_identifier = 1996387;
std::vector<AddressExclusion> exclusions;
ReplyPromise<BlobManagerExclusionSafetyCheckReply> reply;
BlobManagerExclusionSafetyCheckRequest() {}
explicit BlobManagerExclusionSafetyCheckRequest(std::vector<AddressExclusion> exclusions)
: exclusions(exclusions) {}
template <class Ar>
void serialize(Ar& ar) {
serializer(ar, exclusions, reply);
}
};
#endif

File diff suppressed because it is too large Load Diff

View File

@ -5,6 +5,8 @@ set(FDBSERVER_SRCS
BackupProgress.actor.cpp
BackupProgress.actor.h
BackupWorker.actor.cpp
BlobGranuleServerCommon.actor.cpp
BlobGranuleServerCommon.actor.h
BlobManager.actor.cpp
BlobManagerInterface.h
BlobWorker.actor.cpp
@ -178,6 +180,7 @@ set(FDBSERVER_SRCS
workloads/BackupToDBAbort.actor.cpp
workloads/BackupToDBCorrectness.actor.cpp
workloads/BackupToDBUpgrade.actor.cpp
workloads/BlobGranuleCorrectnessWorkload.actor.cpp
workloads/BlobGranuleVerifier.actor.cpp
workloads/BlobStoreWorkload.h
workloads/BulkLoad.actor.cpp

View File

@ -25,6 +25,7 @@
#include <set>
#include <vector>
#include "fdbclient/SystemData.h"
#include "fdbrpc/FailureMonitor.h"
#include "fdbserver/EncryptKeyProxyInterface.h"
#include "flow/ActorCollection.h"
@ -105,7 +106,10 @@ struct RatekeeperSingleton : Singleton<RatekeeperInterface> {
brokenPromiseToNever(interface.get().haltRatekeeper.getReply(HaltRatekeeperRequest(cc->id)));
}
}
void recruit(ClusterControllerData* cc) const { cc->recruitRatekeeper.set(true); }
void recruit(ClusterControllerData* cc) const {
cc->lastRecruitTime = now();
cc->recruitRatekeeper.set(true);
}
};
struct DataDistributorSingleton : Singleton<DataDistributorInterface> {
@ -127,7 +131,10 @@ struct DataDistributorSingleton : Singleton<DataDistributorInterface> {
brokenPromiseToNever(interface.get().haltDataDistributor.getReply(HaltDataDistributorRequest(cc->id)));
}
}
void recruit(ClusterControllerData* cc) const { cc->recruitDistributor.set(true); }
void recruit(ClusterControllerData* cc) const {
cc->lastRecruitTime = now();
cc->recruitDistributor.set(true);
}
};
struct BlobManagerSingleton : Singleton<BlobManagerInterface> {
@ -149,7 +156,17 @@ struct BlobManagerSingleton : Singleton<BlobManagerInterface> {
brokenPromiseToNever(interface.get().haltBlobManager.getReply(HaltBlobManagerRequest(cc->id)));
}
}
void recruit(ClusterControllerData* cc) const { cc->recruitBlobManager.set(true); }
void recruit(ClusterControllerData* cc) const {
cc->lastRecruitTime = now();
cc->recruitBlobManager.set(true);
}
void haltBlobGranules(ClusterControllerData* cc, Optional<Standalone<StringRef>> pid) const {
if (interface.present()) {
cc->id_worker[pid].haltBlobManager =
brokenPromiseToNever(interface.get().haltBlobGranules.getReply(HaltBlobGranulesRequest(cc->id)));
}
}
};
struct EncryptKeyProxySingleton : Singleton<EncryptKeyProxyInterface> {
@ -171,7 +188,10 @@ struct EncryptKeyProxySingleton : Singleton<EncryptKeyProxyInterface> {
brokenPromiseToNever(interface.get().haltEncryptKeyProxy.getReply(HaltEncryptKeyProxyRequest(cc->id)));
}
}
void recruit(ClusterControllerData* cc) const { cc->recruitEncryptKeyProxy.set(true); }
void recruit(ClusterControllerData* cc) const {
cc->lastRecruitTime = now();
cc->recruitEncryptKeyProxy.set(true);
}
};
ACTOR Future<Void> handleLeaderReplacement(Reference<ClusterRecoveryData> self, Future<Void> leaderFail) {
@ -594,7 +614,7 @@ void checkBetterSingletons(ClusterControllerData* self) {
WorkerDetails newDDWorker = findNewProcessForSingleton(self, ProcessClass::DataDistributor, id_used);
WorkerDetails newBMWorker;
if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
if (self->db.blobGranulesEnabled.get()) {
newBMWorker = findNewProcessForSingleton(self, ProcessClass::BlobManager, id_used);
}
@ -608,7 +628,7 @@ void checkBetterSingletons(ClusterControllerData* self) {
auto bestFitnessForDD = findBestFitnessForSingleton(self, newDDWorker, ProcessClass::DataDistributor);
ProcessClass::Fitness bestFitnessForBM;
if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
if (self->db.blobGranulesEnabled.get()) {
bestFitnessForBM = findBestFitnessForSingleton(self, newBMWorker, ProcessClass::BlobManager);
}
@ -632,7 +652,7 @@ void checkBetterSingletons(ClusterControllerData* self) {
self, newDDWorker, ddSingleton, bestFitnessForDD, self->recruitingDistributorID);
bool bmHealthy = true;
if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
if (self->db.blobGranulesEnabled.get()) {
bmHealthy = isHealthySingleton<BlobManagerInterface>(
self, newBMWorker, bmSingleton, bestFitnessForBM, self->recruitingBlobManagerID);
}
@ -656,7 +676,7 @@ void checkBetterSingletons(ClusterControllerData* self) {
Optional<Standalone<StringRef>> newDDProcessId = newDDWorker.interf.locality.processId();
Optional<Standalone<StringRef>> currBMProcessId, newBMProcessId;
if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
if (self->db.blobGranulesEnabled.get()) {
currBMProcessId = bmSingleton.interface.get().locality.processId();
newBMProcessId = newBMWorker.interf.locality.processId();
}
@ -669,7 +689,7 @@ void checkBetterSingletons(ClusterControllerData* self) {
std::vector<Optional<Standalone<StringRef>>> currPids = { currRKProcessId, currDDProcessId };
std::vector<Optional<Standalone<StringRef>>> newPids = { newRKProcessId, newDDProcessId };
if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
if (self->db.blobGranulesEnabled.get()) {
currPids.emplace_back(currBMProcessId);
newPids.emplace_back(newBMProcessId);
}
@ -683,7 +703,7 @@ void checkBetterSingletons(ClusterControllerData* self) {
auto newColocMap = getColocCounts(newPids);
// if the knob is disabled, the BM coloc counts should have no affect on the coloc counts check below
if (!CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
if (!self->db.blobGranulesEnabled.get()) {
ASSERT(currColocMap[currBMProcessId] == 0);
ASSERT(newColocMap[newBMProcessId] == 0);
}
@ -704,7 +724,7 @@ void checkBetterSingletons(ClusterControllerData* self) {
rkSingleton.recruit(self);
} else if (newColocMap[newDDProcessId] < currColocMap[currDDProcessId]) {
ddSingleton.recruit(self);
} else if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES && newColocMap[newBMProcessId] < currColocMap[currBMProcessId]) {
} else if (self->db.blobGranulesEnabled.get() && newColocMap[newBMProcessId] < currColocMap[currBMProcessId]) {
bmSingleton.recruit(self);
} else if (SERVER_KNOBS->ENABLE_ENCRYPTION && newColocMap[newEKPProcessId] < currColocMap[currEKPProcessId]) {
ekpSingleton.recruit(self);
@ -715,14 +735,20 @@ void checkBetterSingletons(ClusterControllerData* self) {
ACTOR Future<Void> doCheckOutstandingRequests(ClusterControllerData* self) {
try {
wait(delay(SERVER_KNOBS->CHECK_OUTSTANDING_INTERVAL));
while (!self->goodRecruitmentTime.isReady()) {
wait(self->goodRecruitmentTime);
while (now() - self->lastRecruitTime < SERVER_KNOBS->SINGLETON_RECRUIT_BME_DELAY ||
!self->goodRecruitmentTime.isReady()) {
if (now() - self->lastRecruitTime < SERVER_KNOBS->SINGLETON_RECRUIT_BME_DELAY) {
wait(delay(SERVER_KNOBS->SINGLETON_RECRUIT_BME_DELAY + 0.001 - (now() - self->lastRecruitTime)));
}
if (!self->goodRecruitmentTime.isReady()) {
wait(self->goodRecruitmentTime);
}
}
checkOutstandingRecruitmentRequests(self);
checkOutstandingStorageRequests(self);
if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
if (self->db.blobGranulesEnabled.get()) {
checkOutstandingBlobWorkerRequests(self);
}
checkBetterSingletons(self);
@ -1232,7 +1258,7 @@ void registerWorker(RegisterWorkerRequest req,
self, w, currSingleton, registeringSingleton, self->recruitingRatekeeperID);
}
if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES && req.blobManagerInterf.present()) {
if (self->db.blobGranulesEnabled.get() && req.blobManagerInterf.present()) {
auto currSingleton = BlobManagerSingleton(self->db.serverInfo->get().blobManager);
auto registeringSingleton = BlobManagerSingleton(req.blobManagerInterf);
haltRegisteringOrCurrentSingleton<BlobManagerInterface>(
@ -2096,9 +2122,9 @@ ACTOR Future<int64_t> getNextBMEpoch(ClusterControllerData* self) {
tr->set(blobManagerEpochKey, blobManagerEpochValueFor(newEpoch));
wait(tr->commit());
TraceEvent(SevDebug, "CCNextBlobManagerEpoch", self->id).detail("Epoch", newEpoch);
return newEpoch;
} catch (Error& e) {
printf("Acquiring blob manager lock got error %s\n", e.name());
wait(tr->onError(e));
}
}
@ -2219,6 +2245,11 @@ ACTOR Future<Void> startBlobManager(ClusterControllerData* self) {
id_used);
int64_t nextEpoch = wait(getNextBMEpoch(self));
if (!self->masterProcessId.present() ||
self->masterProcessId != self->db.serverInfo->get().master.locality.processId() ||
self->db.serverInfo->get().recoveryState < RecoveryState::ACCEPTING_COMMITS) {
continue;
}
InitializeBlobManagerRequest req(deterministicRandom()->randomUniqueID(), nextEpoch);
state WorkerDetails worker = bmWorker.worker;
if (self->onMasterIsBetter(worker, ProcessClass::BlobManager)) {
@ -2262,6 +2293,30 @@ ACTOR Future<Void> startBlobManager(ClusterControllerData* self) {
}
}
ACTOR Future<Void> watchBlobGranulesConfigKey(ClusterControllerData* self) {
state Reference<ReadYourWritesTransaction> tr = makeReference<ReadYourWritesTransaction>(self->cx);
state Key blobGranuleConfigKey = configKeysPrefix.withSuffix("blob_granules_enabled"_sr);
loop {
try {
tr->reset();
tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr->setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE);
Optional<Value> blobConfig = wait(tr->get(blobGranuleConfigKey));
if (blobConfig.present()) {
self->db.blobGranulesEnabled.set(blobConfig.get() == LiteralStringRef("1"));
}
state Future<Void> watch = tr->watch(blobGranuleConfigKey);
wait(tr->commit());
wait(watch);
} catch (Error& e) {
wait(tr->onError(e));
}
}
}
ACTOR Future<Void> monitorBlobManager(ClusterControllerData* self) {
while (self->db.serverInfo->get().recoveryState < RecoveryState::ACCEPTING_COMMITS) {
wait(self->db.serverInfo->onChange());
@ -2269,17 +2324,34 @@ ACTOR Future<Void> monitorBlobManager(ClusterControllerData* self) {
loop {
if (self->db.serverInfo->get().blobManager.present() && !self->recruitBlobManager.get()) {
choose {
when(wait(waitFailureClient(self->db.serverInfo->get().blobManager.get().waitFailure,
SERVER_KNOBS->BLOB_MANAGER_FAILURE_TIME))) {
TraceEvent("CCBlobManagerDied", self->id)
.detail("BMID", self->db.serverInfo->get().blobManager.get().id());
self->db.clearInterf(ProcessClass::BlobManagerClass);
state Future<Void> wfClient = waitFailureClient(self->db.serverInfo->get().blobManager.get().waitFailure,
SERVER_KNOBS->BLOB_MANAGER_FAILURE_TIME);
loop {
choose {
when(wait(wfClient)) {
TraceEvent("CCBlobManagerDied", self->id)
.detail("BMID", self->db.serverInfo->get().blobManager.get().id());
self->db.clearInterf(ProcessClass::BlobManagerClass);
break;
}
when(wait(self->recruitBlobManager.onChange())) { break; }
when(wait(self->db.blobGranulesEnabled.onChange())) {
// if there is a blob manager present but blob granules are now disabled, stop the BM
if (!self->db.blobGranulesEnabled.get()) {
const auto& blobManager = self->db.serverInfo->get().blobManager;
BlobManagerSingleton(blobManager)
.haltBlobGranules(self, blobManager.get().locality.processId());
break;
}
}
}
when(wait(self->recruitBlobManager.onChange())) {}
}
} else {
} else if (self->db.blobGranulesEnabled.get()) {
// if there is no blob manager present but blob granules are now enabled, recruit a BM
wait(startBlobManager(self));
} else {
// if there is no blob manager present and blob granules are disabled, wait for a config change
wait(self->db.blobGranulesEnabled.onChange());
}
}
}
@ -2434,10 +2506,8 @@ ACTOR Future<Void> clusterControllerCore(ClusterControllerFullInterface interf,
self.addActor.send(handleForcedRecoveries(&self, interf));
self.addActor.send(monitorDataDistributor(&self));
self.addActor.send(monitorRatekeeper(&self));
if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
self.addActor.send(monitorBlobManager(&self));
}
// self.addActor.send(monitorTSSMapping(&self));
self.addActor.send(monitorBlobManager(&self));
self.addActor.send(watchBlobGranulesConfigKey(&self));
self.addActor.send(dbInfoUpdater(&self));
self.addActor.send(traceCounters("ClusterControllerMetrics",
self.id,
@ -3055,4 +3125,4 @@ TEST_CASE("/fdbserver/clustercontroller/shouldTriggerFailoverDueToDegradedServer
return Void();
}
} // namespace
} // namespace

View File

@ -137,6 +137,7 @@ public:
std::map<NetworkAddress, std::pair<double, OpenDatabaseRequest>> clientStatus;
Future<Void> clientCounter;
int clientCount;
AsyncVar<bool> blobGranulesEnabled;
DBInfo()
: clientInfo(new AsyncVar<ClientDBInfo>()), serverInfo(new AsyncVar<ServerDBInfo>()),
@ -147,7 +148,8 @@ public:
EnableLocalityLoadBalance::True,
TaskPriority::DefaultEndpoint,
LockAware::True)), // SOMEDAY: Locality!
unfinishedRecoveries(0), logGenerations(0), cachePopulated(false), clientCount(0) {
unfinishedRecoveries(0), logGenerations(0), cachePopulated(false), clientCount(0),
blobGranulesEnabled(config.blobGranulesEnabled) {
clientCounter = countClients(this);
}
@ -3225,6 +3227,7 @@ public:
// recruitX is used to signal when role X needs to be (re)recruited.
// recruitingXID is used to track the ID of X's interface which is being recruited.
// We use AsyncVars to kill (i.e. halt) singletons that have been replaced.
double lastRecruitTime = 0;
AsyncVar<bool> recruitDistributor;
Optional<UID> recruitingDistributorID;
AsyncVar<bool> recruitRatekeeper;

View File

@ -1887,11 +1887,21 @@ ACTOR Future<Void> proxyCheckSafeExclusion(Reference<AsyncVar<ServerDBInfo> cons
return Void();
}
try {
state Future<ErrorOr<DistributorExclusionSafetyCheckReply>> safeFuture =
state Future<ErrorOr<DistributorExclusionSafetyCheckReply>> ddSafeFuture =
db->get().distributor.get().distributorExclCheckReq.tryGetReply(
DistributorExclusionSafetyCheckRequest(req.exclusions));
DistributorExclusionSafetyCheckReply _reply = wait(throwErrorOr(safeFuture));
DistributorExclusionSafetyCheckReply _reply = wait(throwErrorOr(ddSafeFuture));
reply.safe = _reply.safe;
if (db->get().blobManager.present()) {
TraceEvent("SafetyCheckCommitProxyBM").detail("BMID", db->get().blobManager.get().id());
state Future<ErrorOr<BlobManagerExclusionSafetyCheckReply>> bmSafeFuture =
db->get().blobManager.get().blobManagerExclCheckReq.tryGetReply(
BlobManagerExclusionSafetyCheckRequest(req.exclusions));
BlobManagerExclusionSafetyCheckReply _reply = wait(throwErrorOr(bmSafeFuture));
reply.safe &= _reply.safe;
} else {
TraceEvent("SafetyCheckCommitProxyNoBM");
}
} catch (Error& e) {
TraceEvent("SafetyCheckCommitProxyResponseError").error(e);
if (e.code() != error_code_operation_cancelled) {

View File

@ -19,6 +19,8 @@
*/
#include <cinttypes>
#include <vector>
#include "fdbclient/FDBOptions.g.h"
#include "fdbclient/SystemData.h"
#include "flow/ActorCollection.h"

View File

@ -70,6 +70,10 @@ namespace {
const int MACHINE_REBOOT_TIME = 10;
// The max number of extra blob worker machines we might (i.e. randomly) add to the simulated cluster.
// Note that this is in addition to the two we always have.
const int NUM_EXTRA_BW_MACHINES = 5;
bool destructed = false;
// Configuration details specified in workload test files that change the simulation
@ -262,6 +266,9 @@ class TestConfig {
configDBType = configDBTypeFromString(value);
}
}
if (attrib == "blobGranulesEnabled") {
blobGranulesEnabled = strcmp(value.c_str(), "true") == 0;
}
}
ifs.close();
@ -298,6 +305,7 @@ public:
Optional<bool> generateFearless, buggify;
Optional<int> datacenters, desiredTLogCount, commitProxyCount, grvProxyCount, resolverCount, storageEngineType,
stderrSeverity, machineCount, processesPerMachine, coordinators;
bool blobGranulesEnabled = false;
Optional<std::string> config;
bool allowDefaultTenant = true;
@ -354,6 +362,7 @@ public:
.add("coordinators", &coordinators)
.add("configDB", &configDBType)
.add("extraMachineCountDC", &extraMachineCountDC)
.add("blobGranulesEnabled", &blobGranulesEnabled)
.add("allowDefaultTenant", &allowDefaultTenant)
.add("allowDisablingTenants", &allowDisablingTenants);
try {
@ -2085,16 +2094,20 @@ void setupSimulatedSystem(std::vector<Future<Void>>* systemActors,
coordinatorCount);
ASSERT_LE(dcCoordinators, machines);
// FIXME: temporarily code to test storage cache
// FIXME: we hardcode some machines to specifically test storage cache and blob workers
// TODO: caching disabled for this merge
if (dc == 0) {
machines++;
int storageCacheMachines = dc == 0 ? 1 : 0;
int blobWorkerMachines = 0;
if (testConfig.blobGranulesEnabled) {
int blobWorkerProcesses = 1 + deterministicRandom()->randomInt(0, NUM_EXTRA_BW_MACHINES + 1);
blobWorkerMachines = std::max(1, blobWorkerProcesses / processesPerMachine);
}
int useSeedForMachine = deterministicRandom()->randomInt(0, machines);
int totalMachines = machines + storageCacheMachines + blobWorkerMachines;
int useSeedForMachine = deterministicRandom()->randomInt(0, totalMachines);
Standalone<StringRef> zoneId;
Standalone<StringRef> newZoneId;
for (int machine = 0; machine < machines; machine++) {
for (int machine = 0; machine < totalMachines; machine++) {
Standalone<StringRef> machineId(deterministicRandom()->randomUniqueID().toString());
if (machine == 0 || machineCount - dataCenters <= 4 || assignedMachines != 4 ||
simconfig.db.regions.size() || deterministicRandom()->random01() < 0.5) {
@ -2124,11 +2137,19 @@ void setupSimulatedSystem(std::vector<Future<Void>>* systemActors,
}
}
// FIXME: temporarily code to test storage cache
// FIXME: hack to add machines specifically to test storage cache and blob workers
// TODO: caching disabled for this merge
if (machine == machines - 1 && dc == 0) {
processClass = ProcessClass(ProcessClass::StorageCacheClass, ProcessClass::CommandLineSource);
nonVersatileMachines++;
// `machines` here is the normal (non-temporary) machines that totalMachines comprises of
if (machine >= machines) {
if (storageCacheMachines > 0 && dc == 0) {
processClass = ProcessClass(ProcessClass::StorageCacheClass, ProcessClass::CommandLineSource);
nonVersatileMachines++;
storageCacheMachines--;
} else if (blobWorkerMachines > 0) { // add blob workers to every DC
processClass = ProcessClass(ProcessClass::BlobWorkerClass, ProcessClass::CommandLineSource);
nonVersatileMachines++;
blobWorkerMachines--;
}
}
std::vector<IPAddress> ips;
@ -2303,8 +2324,9 @@ ACTOR void setupAndRun(std::string dataFolder,
// Disable the default tenant in backup and DR tests for now. This is because backup does not currently duplicate
// the tenant map and related state.
// TODO: reenable when backup/DR supports tenants.
if (std::string_view(testFile).find("Backup") != std::string_view::npos || testConfig.extraDB != 0) {
// TODO: reenable when backup/DR or BlobGranule supports tenants.
if (std::string_view(testFile).find("Backup") != std::string_view::npos ||
std::string_view(testFile).find("BlobGranule") != std::string_view::npos || testConfig.extraDB != 0) {
allowDefaultTenant = false;
}

View File

@ -804,7 +804,7 @@ ACTOR static Future<JsonBuilderObject> processStatusFetcher(
roles.addRole("ratekeeper", db->get().ratekeeper.get());
}
if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES && db->get().blobManager.present()) {
if (configuration.present() && configuration.get().blobGranulesEnabled && db->get().blobManager.present()) {
roles.addRole("blob_manager", db->get().blobManager.get());
}
@ -875,7 +875,7 @@ ACTOR static Future<JsonBuilderObject> processStatusFetcher(
wait(yield());
}
if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
if (configuration.present() && configuration.get().blobGranulesEnabled) {
for (auto blobWorker : blobWorkers) {
roles.addRole("blob_worker", blobWorker);
wait(yield());
@ -2983,7 +2983,7 @@ ACTOR Future<StatusReply> clusterGetStatus(
errorOr(getGrvProxiesAndMetrics(db, address_workers));
state Future<ErrorOr<std::vector<BlobWorkerInterface>>> blobWorkersFuture;
if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
if (configuration.present() && configuration.get().blobGranulesEnabled) {
blobWorkersFuture = errorOr(timeoutError(getBlobWorkers(cx, true), 5.0));
}
@ -3121,7 +3121,7 @@ ACTOR Future<StatusReply> clusterGetStatus(
}
// ...also blob workers
if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
if (configuration.present() && configuration.get().blobGranulesEnabled) {
ErrorOr<std::vector<BlobWorkerInterface>> _blobWorkers = wait(blobWorkersFuture);
if (_blobWorkers.present()) {
blobWorkers = _blobWorkers.get();

File diff suppressed because it is too large Load Diff

View File

@ -524,20 +524,21 @@ std::vector<DiskStore> getDiskStores(std::string folder) {
// Register the worker interf to cluster controller (cc) and
// re-register the worker when key roles interface, e.g., cc, dd, ratekeeper, change.
ACTOR Future<Void> registrationClient(Reference<AsyncVar<Optional<ClusterControllerFullInterface>> const> ccInterface,
WorkerInterface interf,
Reference<AsyncVar<ClusterControllerPriorityInfo>> asyncPriorityInfo,
ProcessClass initialClass,
Reference<AsyncVar<Optional<DataDistributorInterface>> const> ddInterf,
Reference<AsyncVar<Optional<RatekeeperInterface>> const> rkInterf,
Reference<AsyncVar<Optional<BlobManagerInterface>> const> bmInterf,
Reference<AsyncVar<Optional<EncryptKeyProxyInterface>> const> ekpInterf,
Reference<AsyncVar<bool> const> degraded,
Reference<IClusterConnectionRecord> connRecord,
Reference<AsyncVar<std::set<std::string>> const> issues,
Reference<ConfigNode> configNode,
Reference<LocalConfiguration> localConfig,
Reference<AsyncVar<ServerDBInfo>> dbInfo) {
ACTOR Future<Void> registrationClient(
Reference<AsyncVar<Optional<ClusterControllerFullInterface>> const> ccInterface,
WorkerInterface interf,
Reference<AsyncVar<ClusterControllerPriorityInfo>> asyncPriorityInfo,
ProcessClass initialClass,
Reference<AsyncVar<Optional<DataDistributorInterface>> const> ddInterf,
Reference<AsyncVar<Optional<RatekeeperInterface>> const> rkInterf,
Reference<AsyncVar<Optional<std::pair<int64_t, BlobManagerInterface>>> const> bmInterf,
Reference<AsyncVar<Optional<EncryptKeyProxyInterface>> const> ekpInterf,
Reference<AsyncVar<bool> const> degraded,
Reference<IClusterConnectionRecord> connRecord,
Reference<AsyncVar<std::set<std::string>> const> issues,
Reference<ConfigNode> configNode,
Reference<LocalConfiguration> localConfig,
Reference<AsyncVar<ServerDBInfo>> dbInfo) {
// Keeps the cluster controller (as it may be re-elected) informed that this worker exists
// The cluster controller uses waitFailureClient to find out if we die, and returns from registrationReply
// (requiring us to re-register) The registration request piggybacks optional distributor interface if it exists.
@ -567,7 +568,8 @@ ACTOR Future<Void> registrationClient(Reference<AsyncVar<Optional<ClusterControl
requestGeneration++,
ddInterf->get(),
rkInterf->get(),
bmInterf->get(),
bmInterf->get().present() ? bmInterf->get().get().second
: Optional<BlobManagerInterface>(),
ekpInterf->get(),
degraded->get(),
localConfig->lastSeenVersion(),
@ -1138,7 +1140,9 @@ ACTOR Future<Void> storageServerRollbackRebooter(std::set<std::pair<UID, KeyValu
DUMPTOKEN(recruited.getKeyValueStoreType);
DUMPTOKEN(recruited.watchValue);
DUMPTOKEN(recruited.getKeyValuesStream);
DUMPTOKEN(recruited.getMappedKeyValues);
DUMPTOKEN(recruited.changeFeedStream);
DUMPTOKEN(recruited.changeFeedPop);
DUMPTOKEN(recruited.changeFeedVersionUpdate);
prevStorageServer =
storageServer(store, recruited, db, folder, Promise<Void>(), Reference<IClusterConnectionRecord>(nullptr));
@ -1375,6 +1379,24 @@ ACTOR Future<Void> chaosMetricsLogger() {
}
}
// like genericactors setWhenDoneOrError, but we need to take into account the bm epoch. We don't want to reset it if
// this manager was replaced by a later manager (with a higher epoch) on this worker
ACTOR Future<Void> resetBlobManagerWhenDoneOrError(
Future<Void> blobManagerProcess,
Reference<AsyncVar<Optional<std::pair<int64_t, BlobManagerInterface>>>> var,
int64_t epoch) {
try {
wait(blobManagerProcess);
} catch (Error& e) {
if (e.code() == error_code_actor_cancelled)
throw;
}
if (var->get().present() && var->get().get().first == epoch) {
var->set(Optional<std::pair<int64_t, BlobManagerInterface>>());
}
return Void();
}
ACTOR Future<Void> workerServer(Reference<IClusterConnectionRecord> connRecord,
Reference<AsyncVar<Optional<ClusterControllerFullInterface>> const> ccInterface,
LocalityData locality,
@ -1396,7 +1418,8 @@ ACTOR Future<Void> workerServer(Reference<IClusterConnectionRecord> connRecord,
state Reference<AsyncVar<Optional<DataDistributorInterface>>> ddInterf(
new AsyncVar<Optional<DataDistributorInterface>>());
state Reference<AsyncVar<Optional<RatekeeperInterface>>> rkInterf(new AsyncVar<Optional<RatekeeperInterface>>());
state Reference<AsyncVar<Optional<BlobManagerInterface>>> bmInterf(new AsyncVar<Optional<BlobManagerInterface>>());
state Reference<AsyncVar<Optional<std::pair<int64_t, BlobManagerInterface>>>> bmEpochAndInterf(
new AsyncVar<Optional<std::pair<int64_t, BlobManagerInterface>>>());
state Reference<AsyncVar<Optional<EncryptKeyProxyInterface>>> ekpInterf(
new AsyncVar<Optional<EncryptKeyProxyInterface>>());
state Future<Void> handleErrors = workerHandleErrors(errors.getFuture()); // Needs to be stopped last
@ -1417,6 +1440,7 @@ ACTOR Future<Void> workerServer(Reference<IClusterConnectionRecord> connRecord,
state std::map<SharedLogsKey, SharedLogsValue> sharedLogs;
state Reference<AsyncVar<UID>> activeSharedTLog(new AsyncVar<UID>());
state WorkerCache<InitializeBackupReply> backupWorkerCache;
state WorkerCache<InitializeBlobWorkerReply> blobWorkerCache;
state std::string coordFolder = abspath(_coordFolder);
@ -1529,6 +1553,7 @@ ACTOR Future<Void> workerServer(Reference<IClusterConnectionRecord> connRecord,
DUMPTOKEN(recruited.getValue);
DUMPTOKEN(recruited.getKey);
DUMPTOKEN(recruited.getKeyValues);
DUMPTOKEN(recruited.getMappedKeyValues);
DUMPTOKEN(recruited.getShardState);
DUMPTOKEN(recruited.waitMetrics);
DUMPTOKEN(recruited.splitMetrics);
@ -1540,7 +1565,9 @@ ACTOR Future<Void> workerServer(Reference<IClusterConnectionRecord> connRecord,
DUMPTOKEN(recruited.getKeyValueStoreType);
DUMPTOKEN(recruited.watchValue);
DUMPTOKEN(recruited.getKeyValuesStream);
DUMPTOKEN(recruited.getMappedKeyValues);
DUMPTOKEN(recruited.changeFeedStream);
DUMPTOKEN(recruited.changeFeedPop);
DUMPTOKEN(recruited.changeFeedVersionUpdate);
Promise<Void> recovery;
Future<Void> f = storageServer(kv, recruited, dbInfo, folder, recovery, connRecord);
@ -1668,7 +1695,7 @@ ACTOR Future<Void> workerServer(Reference<IClusterConnectionRecord> connRecord,
initialClass,
ddInterf,
rkInterf,
bmInterf,
bmEpochAndInterf,
ekpInterf,
degraded,
connRecord,
@ -1870,21 +1897,30 @@ ACTOR Future<Void> workerServer(Reference<IClusterConnectionRecord> connRecord,
BlobManagerInterface recruited(locality, req.reqId);
recruited.initEndpoints();
if (bmInterf->get().present()) {
recruited = bmInterf->get().get();
if (bmEpochAndInterf->get().present() && bmEpochAndInterf->get().get().first == req.epoch) {
recruited = bmEpochAndInterf->get().get().second;
TEST(true); // Recruited while already a blob manager.
} else {
// TODO: it'd be more optimal to halt the last manager if present here, but it will figure it out
// via the epoch check
// Also, not halting lets us handle the case here where the last BM had a higher
// epoch and somehow the epochs got out of order by a delayed initialize request. The one we start
// here will just halt on the lock check.
startRole(Role::BLOB_MANAGER, recruited.id(), interf.id());
DUMPTOKEN(recruited.waitFailure);
DUMPTOKEN(recruited.haltBlobManager);
DUMPTOKEN(recruited.haltBlobGranules);
DUMPTOKEN(recruited.blobManagerExclCheckReq);
Future<Void> blobManagerProcess = blobManager(recruited, dbInfo, req.epoch);
errorForwarders.add(forwardError(
errors,
Role::BLOB_MANAGER,
recruited.id(),
setWhenDoneOrError(blobManagerProcess, bmInterf, Optional<BlobManagerInterface>())));
bmInterf->set(Optional<BlobManagerInterface>(recruited));
errorForwarders.add(
forwardError(errors,
Role::BLOB_MANAGER,
recruited.id(),
resetBlobManagerWhenDoneOrError(blobManagerProcess, bmEpochAndInterf, req.epoch)));
bmEpochAndInterf->set(
Optional<std::pair<int64_t, BlobManagerInterface>>(std::pair(req.epoch, recruited)));
}
TraceEvent("BlobManagerReceived", req.reqId).detail("BlobManagerId", recruited.id());
req.reply.send(recruited);
@ -2028,6 +2064,7 @@ ACTOR Future<Void> workerServer(Reference<IClusterConnectionRecord> connRecord,
DUMPTOKEN(recruited.getValue);
DUMPTOKEN(recruited.getKey);
DUMPTOKEN(recruited.getKeyValues);
DUMPTOKEN(recruited.getMappedKeyValues);
DUMPTOKEN(recruited.getShardState);
DUMPTOKEN(recruited.waitMetrics);
DUMPTOKEN(recruited.splitMetrics);
@ -2039,7 +2076,9 @@ ACTOR Future<Void> workerServer(Reference<IClusterConnectionRecord> connRecord,
DUMPTOKEN(recruited.getKeyValueStoreType);
DUMPTOKEN(recruited.watchValue);
DUMPTOKEN(recruited.getKeyValuesStream);
DUMPTOKEN(recruited.getMappedKeyValues);
DUMPTOKEN(recruited.changeFeedStream);
DUMPTOKEN(recruited.changeFeedPop);
DUMPTOKEN(recruited.changeFeedVersionUpdate);
// printf("Recruited as storageServer\n");
std::string filename =
@ -2086,13 +2125,26 @@ ACTOR Future<Void> workerServer(Reference<IClusterConnectionRecord> connRecord,
}
}
when(InitializeBlobWorkerRequest req = waitNext(interf.blobWorker.getFuture())) {
BlobWorkerInterface recruited(locality, req.interfaceId);
recruited.initEndpoints();
startRole(Role::BLOB_WORKER, recruited.id(), interf.id());
if (!blobWorkerCache.exists(req.reqId)) {
BlobWorkerInterface recruited(locality, req.interfaceId);
recruited.initEndpoints();
startRole(Role::BLOB_WORKER, recruited.id(), interf.id());
ReplyPromise<InitializeBlobWorkerReply> blobWorkerReady = req.reply;
Future<Void> bw = blobWorker(recruited, blobWorkerReady, dbInfo);
errorForwarders.add(forwardError(errors, Role::BLOB_WORKER, recruited.id(), bw));
DUMPTOKEN(recruited.waitFailure);
DUMPTOKEN(recruited.blobGranuleFileRequest);
DUMPTOKEN(recruited.assignBlobRangeRequest);
DUMPTOKEN(recruited.revokeBlobRangeRequest);
DUMPTOKEN(recruited.granuleAssignmentsRequest);
DUMPTOKEN(recruited.granuleStatusStreamRequest);
DUMPTOKEN(recruited.haltBlobWorker);
ReplyPromise<InitializeBlobWorkerReply> blobWorkerReady = req.reply;
Future<Void> bw = blobWorker(recruited, blobWorkerReady, dbInfo);
errorForwarders.add(forwardError(errors, Role::BLOB_WORKER, recruited.id(), bw));
} else {
forwardPromise(req.reply, blobWorkerCache.get(req.reqId));
}
}
when(InitializeCommitProxyRequest req = waitNext(interf.commitProxy.getFuture())) {
LocalLineage _;

View File

@ -0,0 +1,884 @@
/*
* BlobGranuleCorrectnessWorkload.actor.cpp
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2018 Apple Inc. and the FoundationDB project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <cmath>
#include <map>
#include <utility>
#include <vector>
#include "contrib/fmt-8.1.1/include/fmt/format.h"
#include "fdbclient/BlobGranuleReader.actor.h"
#include "fdbclient/ManagementAPI.actor.h"
#include "fdbclient/NativeAPI.actor.h"
#include "fdbclient/ReadYourWrites.h"
#include "fdbclient/SystemData.h"
#include "fdbserver/Knobs.h"
#include "fdbserver/TesterInterface.actor.h"
#include "fdbserver/workloads/workloads.actor.h"
#include "flow/Arena.h"
#include "flow/IRandom.h"
#include "flow/genericactors.actor.h"
#include "flow/actorcompiler.h" // This must be the last #include.
#define BGW_DEBUG true
struct WriteData {
Version writeVersion;
Version clearVersion;
int32_t val;
int16_t valLength;
// start as MAX_VERSION while uncommitted/uncleared so that they're ignored by concurrent readers
explicit WriteData(int32_t val, int16_t valLength)
: writeVersion(MAX_VERSION), clearVersion(MAX_VERSION), val(val), valLength(valLength) {}
};
struct KeyData {
int nextClearIdx;
std::vector<WriteData> writes;
};
static std::vector<int> targetValSizes = { 40, 100, 500 };
struct ThreadData : ReferenceCounted<ThreadData>, NonCopyable {
// directory info
int32_t directoryID;
KeyRange directoryRange;
// key + value gen data
// in vector for efficient random selection
std::vector<uint32_t> usedKeys;
// by key for tracking data
std::map<uint32_t, KeyData> keyData;
std::deque<Version> writeVersions;
// randomized parameters that can be different per directory
int targetByteRate;
bool nextKeySequential;
int16_t targetValLength;
double reuseKeyProb;
int targetIDsPerKey;
// communication between workers
Promise<Void> firstWriteSuccessful;
Version minSuccessfulReadVersion = MAX_VERSION;
// stats
int64_t errors = 0;
int64_t mismatches = 0;
int64_t reads = 0;
int64_t timeTravelReads = 0;
int64_t timeTravelTooOld = 0;
int64_t rowsRead = 0;
int64_t bytesRead = 0;
int64_t rowsWritten = 0;
int64_t bytesWritten = 0;
ThreadData(uint32_t directoryID, int64_t targetByteRate)
: directoryID(directoryID), targetByteRate(targetByteRate) {
directoryRange =
KeyRangeRef(StringRef(format("%08x", directoryID)), StringRef(format("%08x", directoryID + 1)));
targetByteRate *= (0.5 + deterministicRandom()->random01());
targetValLength = deterministicRandom()->randomChoice(targetValSizes);
targetValLength *= (0.5 + deterministicRandom()->random01());
nextKeySequential = deterministicRandom()->random01() < 0.5;
reuseKeyProb = 0.1 + (deterministicRandom()->random01() * 0.8);
targetIDsPerKey = 1 + deterministicRandom()->randomInt(1, 10);
if (BGW_DEBUG) {
fmt::print("Directory {0} initialized with the following parameters:\n", directoryID);
fmt::print(" targetByteRate={0}\n", targetByteRate);
fmt::print(" targetValLength={0}\n", targetValLength);
fmt::print(" nextKeySequential={0}\n", nextKeySequential);
fmt::print(" reuseKeyProb={0}\n", reuseKeyProb);
fmt::print(" targetIDsPerKey={0}\n", targetIDsPerKey);
}
}
// TODO could make keys variable length?
Key getKey(uint32_t key, uint32_t id) { return StringRef(format("%08x/%08x/%08x", directoryID, key, id)); }
};
// For debugging mismatches on what data should be and why
// set mismatch to true, dir id and key id to the directory and key id that are wrong, and rv to read version that read
// the wrong value
#define DEBUG_MISMATCH false
#define DEBUG_DIR_ID 0
#define DEBUG_KEY_ID 0
#define DEBUG_RV invalidVersion
#define DEBUG_KEY_OP(dirId, keyId) BGW_DEBUG&& DEBUG_MISMATCH&& dirId == DEBUG_DIR_ID&& DEBUG_KEY_ID == keyId
#define DEBUG_READ_OP(dirId, rv) BGW_DEBUG&& DEBUG_MISMATCH&& dirId == DEBUG_DIR_ID&& rv == DEBUG_RV
/*
* This is a stand-alone workload designed to validate blob granule correctness.
* By enabling distinct ranges and writing to those parts of the key space, we can control what parts of the key space
* are written to blob, and can validate that the granule data is correct at any desired version.
*/
struct BlobGranuleCorrectnessWorkload : TestWorkload {
bool doSetup;
double testDuration;
// parameters global across all clients
int64_t targetByteRate;
std::vector<Reference<ThreadData>> directories;
std::vector<Future<Void>> clients;
DatabaseConfiguration config;
Reference<BackupContainerFileSystem> bstore;
BlobGranuleCorrectnessWorkload(WorkloadContext const& wcx) : TestWorkload(wcx) {
doSetup = !clientId; // only do this on the "first" client
testDuration = getOption(options, LiteralStringRef("testDuration"), 120.0);
// randomize global test settings based on shared parameter to get similar workload across tests, but then vary
// different parameters within those constraints
int64_t randomness = sharedRandomNumber;
// randomize between low and high directory count
int64_t targetDirectories = 1 + (randomness % 8);
randomness /= 8;
int64_t targetMyDirectories =
(targetDirectories / clientCount) + ((targetDirectories % clientCount > clientId) ? 1 : 0);
if (targetMyDirectories > 0) {
int myDirectories = 1;
if (targetMyDirectories > 1) {
myDirectories = deterministicRandom()->randomInt(1, 2 * targetMyDirectories + 1);
}
// anywhere from 2 delta files per second to 1 delta file every 2 seconds, spread across all directories
int denom = std::min(clientCount, (int)targetDirectories);
targetByteRate = 2 * SERVER_KNOBS->BG_DELTA_FILE_TARGET_BYTES / (1 + (randomness % 4)) / denom;
randomness /= 4;
// either do equal across all of my directories, or skewed
bool skewed = myDirectories > 1 && deterministicRandom()->random01() < 0.4;
int skewMultiplier;
if (skewed) {
// first directory has 1/2, second has 1/4, third has 1/8, etc...
skewMultiplier = 2;
targetByteRate /= 2;
} else {
skewMultiplier = 1;
targetByteRate /= myDirectories;
}
for (int i = 0; i < myDirectories; i++) {
// set up directory with its own randomness
uint32_t dirId = i * clientCount + clientId;
if (BGW_DEBUG) {
fmt::print("Client {0}/{1} creating directory {2}\n", clientId, clientCount, dirId);
}
directories.push_back(makeReference<ThreadData>(dirId, targetByteRate));
targetByteRate /= skewMultiplier;
}
}
}
ACTOR Future<Void> setUpBlobRange(Database cx, KeyRange range) {
if (BGW_DEBUG) {
fmt::print(
"Setting up blob granule range for [{0} - {1})\n", range.begin.printable(), range.end.printable());
}
state Reference<ReadYourWritesTransaction> tr = makeReference<ReadYourWritesTransaction>(cx);
loop {
try {
tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr->setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE);
tr->set(blobRangeChangeKey, deterministicRandom()->randomUniqueID().toString());
wait(krmSetRange(tr, blobRangeKeys.begin, range, LiteralStringRef("1")));
wait(tr->commit());
if (BGW_DEBUG) {
fmt::print("Successfully set up blob granule range for [{0} - {1})\n",
range.begin.printable(),
range.end.printable());
}
return Void();
} catch (Error& e) {
wait(tr->onError(e));
}
}
}
std::string description() const override { return "BlobGranuleCorrectnessWorkload"; }
Future<Void> setup(Database const& cx) override { return _setup(cx, this); }
ACTOR Future<Void> _setup(Database cx, BlobGranuleCorrectnessWorkload* self) {
if (self->doSetup) {
// FIXME: run the actual FDBCLI command instead of copy/pasting its implementation
wait(success(ManagementAPI::changeConfig(cx.getReference(), "blob_granules_enabled=1", true)));
}
if (self->directories.empty()) {
return Void();
}
state int directoryIdx = 0;
for (; directoryIdx < self->directories.size(); directoryIdx++) {
// Set up the blob range first
wait(self->setUpBlobRange(cx, self->directories[directoryIdx]->directoryRange));
}
if (BGW_DEBUG) {
printf("Initializing Blob Granule Correctness s3 stuff\n");
}
try {
if (g_network->isSimulated()) {
if (BGW_DEBUG) {
printf("Blob Granule Correctness constructing simulated backup container\n");
}
self->bstore = BackupContainerFileSystem::openContainerFS("file://fdbblob/");
} else {
if (BGW_DEBUG) {
printf("Blob Granule Correctness constructing backup container from %s\n",
SERVER_KNOBS->BG_URL.c_str());
}
self->bstore = BackupContainerFileSystem::openContainerFS(SERVER_KNOBS->BG_URL);
if (BGW_DEBUG) {
printf("Blob Granule Correctness constructed backup container\n");
}
}
} catch (Error& e) {
if (BGW_DEBUG) {
printf("Blob Granule Correctness got backup container init error %s\n", e.name());
}
throw e;
}
return Void();
}
// FIXME: typedef this pair type and/or chunk list
ACTOR Future<std::pair<RangeResult, Standalone<VectorRef<BlobGranuleChunkRef>>>>
readFromBlob(Database cx, BlobGranuleCorrectnessWorkload* self, KeyRange range, Version version) {
state RangeResult out;
state Standalone<VectorRef<BlobGranuleChunkRef>> chunks;
state Transaction tr(cx);
loop {
try {
Standalone<VectorRef<BlobGranuleChunkRef>> chunks_ = wait(tr.readBlobGranules(range, 0, version));
chunks = chunks_;
break;
} catch (Error& e) {
wait(tr.onError(e));
}
}
for (const BlobGranuleChunkRef& chunk : chunks) {
RangeResult chunkRows = wait(readBlobGranule(chunk, range, version, self->bstore));
out.arena().dependsOn(chunkRows.arena());
out.append(out.arena(), chunkRows.begin(), chunkRows.size());
}
return std::pair(out, chunks);
}
// handle retries + errors
// It's ok to reset the transaction here because its read version is only used for reading the granule mapping from
// the system keyspace
ACTOR Future<Version> doGrv(Transaction* tr) {
loop {
try {
Version readVersion = wait(tr->getReadVersion());
return readVersion;
} catch (Error& e) {
wait(tr->onError(e));
}
}
}
ACTOR Future<Void> waitFirstSnapshot(BlobGranuleCorrectnessWorkload* self,
Database cx,
Reference<ThreadData> threadData,
bool doSetup) {
// read entire keyspace at the start until granules for the entire thing are available
loop {
state Transaction tr(cx);
try {
Version rv = wait(self->doGrv(&tr));
state Version readVersion = rv;
std::pair<RangeResult, Standalone<VectorRef<BlobGranuleChunkRef>>> blob =
wait(self->readFromBlob(cx, self, threadData->directoryRange, readVersion));
fmt::print("Directory {0} got {1} RV {2}\n",
threadData->directoryID,
doSetup ? "initial" : "final",
readVersion);
threadData->minSuccessfulReadVersion = readVersion;
return Void();
} catch (Error& e) {
if (e.code() == error_code_operation_cancelled) {
throw e;
}
if (e.code() != error_code_blob_granule_transaction_too_old) {
wait(tr.onError(e));
} else {
wait(delay(1.0));
}
}
}
}
void logMismatch(Reference<ThreadData> threadData,
const Optional<Key>& lastMatching,
const Optional<Key>& expectedKey,
const Optional<Key>& blobKey,
const Optional<Value>& expectedValue,
const Optional<Value>& blobValue,
uint32_t startKey,
uint32_t endKey,
Version readVersion,
const std::pair<RangeResult, Standalone<VectorRef<BlobGranuleChunkRef>>>& blob) {
threadData->mismatches++;
if (!BGW_DEBUG) {
return;
}
TraceEvent ev(SevError, "BGMismatch");
ev.detail("DirectoryID", format("%08x", threadData->directoryID))
.detail("RangeStart", format("%08x", startKey))
.detail("RangeEnd", format("%08x", endKey))
.detail("Version", readVersion);
fmt::print("Found mismatch! Request for dir {0} [{1} - {2}) @ {3}\n",
format("%08x", threadData->directoryID),
format("%08x", startKey),
format("%08x", endKey),
readVersion);
if (lastMatching.present()) {
fmt::print(" last correct: {}\n", lastMatching.get().printable());
}
if (expectedValue.present() || blobValue.present()) {
// value mismatch
ASSERT(blobKey.present());
ASSERT(blobKey == expectedKey);
fmt::print(" Value mismatch for {0}.\n Expected={1}\n Actual={2}\n",
blobKey.get().printable(),
expectedValue.get().printable(),
blobValue.get().printable());
} else {
// key mismatch
fmt::print(" Expected Key: {0}\n", expectedKey.present() ? expectedKey.get().printable() : "<missing>");
fmt::print(" Actual Key: {0}\n", blobKey.present() ? blobKey.get().printable() : "<missing>");
}
fmt::print("Chunks:\n");
for (auto& chunk : blob.second) {
fmt::print("[{0} - {1})\n", chunk.keyRange.begin.printable(), chunk.keyRange.end.printable());
fmt::print(" SnapshotFile:\n {}\n",
chunk.snapshotFile.present() ? chunk.snapshotFile.get().toString().c_str() : "<none>");
fmt::print(" DeltaFiles:\n");
for (auto& df : chunk.deltaFiles) {
fmt::print(" {}\n", df.toString());
}
fmt::print(" Deltas: ({})", chunk.newDeltas.size());
if (chunk.newDeltas.size() > 0) {
fmt::print(" with version [{0} - {1}]",
chunk.newDeltas[0].version,
chunk.newDeltas[chunk.newDeltas.size() - 1].version);
}
fmt::print(" IncludedVersion: {}\n", chunk.includedVersion);
}
printf("\n");
}
Value genVal(uint32_t val, uint16_t valLen) {
std::string v(valLen, 'x');
auto valFormatted = format("%08x", val);
ASSERT(valFormatted.size() <= v.size());
for (int i = 0; i < valFormatted.size(); i++) {
v[i] = valFormatted[i];
}
// copy into an arena
// TODO do this in original arena? a bit more efficient that way
Arena a;
return Standalone<StringRef>(StringRef(a, v), a);
}
bool validateValue(const Value& v, uint32_t val, uint16_t valLen) {
if (v.size() != valLen) {
return false;
}
// check for correct value portion
auto valFormatted = format("%08x", val);
ASSERT(valFormatted.size() <= v.size());
if (v.substr(0, valFormatted.size()) != valFormatted) {
return false;
}
// check for corruption
for (int i = valFormatted.size(); i < v.size(); i++) {
if (v[i] != 'x') {
return false;
}
}
return true;
}
bool validateResult(Reference<ThreadData> threadData,
std::pair<RangeResult, Standalone<VectorRef<BlobGranuleChunkRef>>> blob,
int startKeyInclusive,
int endKeyExclusive,
Version beginVersion,
Version readVersion) {
auto checkIt = threadData->keyData.lower_bound(startKeyInclusive);
if (checkIt != threadData->keyData.end() && checkIt->first < startKeyInclusive) {
checkIt++;
}
int resultIdx = 0;
Optional<Key> lastMatching;
if (DEBUG_READ_OP(threadData->directoryID, readVersion)) {
fmt::print("DBG READ: [{0} - {1}) @ {2}\n",
format("%08x", startKeyInclusive),
format("%08x", endKeyExclusive),
readVersion);
}
while (checkIt != threadData->keyData.end() && checkIt->first < endKeyExclusive) {
uint32_t key = checkIt->first;
if (DEBUG_READ_OP(threadData->directoryID, readVersion)) {
fmt::print("DBG READ: Key {0}\n", format("%08x", key));
}
// TODO could binary search this to find clearVersion if it gets long
int idIdx = 0;
for (; idIdx < checkIt->second.writes.size() && checkIt->second.writes[idIdx].clearVersion <= readVersion;
idIdx++) {
// iterate until we find the oldest tag that should have not been cleared
/*if (DEBUG_READ_OP(threadData->directoryID, readVersion)) {
fmt::print(
"DBG READ: Skip ID {0} cleared @ {1}\n", idIdx, checkIt->second.writes[idIdx].clearVersion);
}*/
}
for (; idIdx < checkIt->second.writes.size() && checkIt->second.writes[idIdx].writeVersion <= readVersion;
idIdx++) {
Key nextKeyShouldBe = threadData->getKey(key, idIdx);
if (DEBUG_READ_OP(threadData->directoryID, readVersion)) {
fmt::print("DBG READ: Checking ID {0} ({1}) written @ {2}\n",
format("%08x", idIdx),
idIdx,
checkIt->second.writes[idIdx].writeVersion);
}
if (resultIdx >= blob.first.size()) {
// missing at end!!
logMismatch(threadData,
lastMatching,
nextKeyShouldBe,
Optional<Key>(),
Optional<Value>(),
Optional<Value>(),
startKeyInclusive,
endKeyExclusive,
readVersion,
blob);
return false;
}
if (nextKeyShouldBe != blob.first[resultIdx].key) {
// key mismatch!
if (DEBUG_READ_OP(threadData->directoryID, readVersion)) {
printf("key mismatch!\n");
}
logMismatch(threadData,
lastMatching,
nextKeyShouldBe,
blob.first[resultIdx].key,
Optional<Value>(),
Optional<Value>(),
startKeyInclusive,
endKeyExclusive,
readVersion,
blob);
return false;
} else if (!validateValue(blob.first[resultIdx].value,
checkIt->second.writes[idIdx].val,
checkIt->second.writes[idIdx].valLength)) {
logMismatch(threadData,
lastMatching,
nextKeyShouldBe,
blob.first[resultIdx].key,
genVal(checkIt->second.writes[idIdx].val, checkIt->second.writes[idIdx].valLength),
blob.first[resultIdx].value,
startKeyInclusive,
endKeyExclusive,
readVersion,
blob);
return false;
// value mismatch for same key
} else {
lastMatching = nextKeyShouldBe;
}
resultIdx++;
}
checkIt++;
}
if (resultIdx < blob.first.size()) {
// blob has extra stuff!!
logMismatch(threadData,
lastMatching,
Optional<Key>(),
blob.first[resultIdx].key,
Optional<Value>(),
Optional<Value>(),
startKeyInclusive,
endKeyExclusive,
readVersion,
blob);
return false;
}
return true;
}
ACTOR Future<Void> readWorker(BlobGranuleCorrectnessWorkload* self,
Future<Void> firstSnapshot,
Database cx,
Reference<ThreadData> threadData) {
state double last = now();
state double targetBytesReadPerQuery =
SERVER_KNOBS->BG_SNAPSHOT_FILE_TARGET_BYTES * 2.0 / deterministicRandom()->randomInt(1, 11);
// read at higher read rate than write rate to validate data
state double targetReadBytesPerSec = threadData->targetByteRate * 4;
ASSERT(targetReadBytesPerSec > 0);
state Version readVersion;
TraceEvent("BlobGranuleCorrectnessReaderStart").log();
if (BGW_DEBUG) {
printf("BGW read thread starting\n");
}
// wait for data to read
wait(firstSnapshot);
wait(threadData->firstWriteSuccessful.getFuture());
TraceEvent("BlobGranuleCorrectnessReaderReady").log();
if (BGW_DEBUG) {
printf("BGW read thread ready\n");
}
loop {
try {
// Do 1 read
// pick key range by doing random start key, and then picking the end key based on that
int startKeyIdx = deterministicRandom()->randomInt(0, threadData->usedKeys.size());
state uint32_t startKey = threadData->usedKeys[startKeyIdx];
auto endKeyIt = threadData->keyData.find(startKey);
ASSERT(endKeyIt != threadData->keyData.end());
int targetQueryBytes = (deterministicRandom()->randomInt(1, 20) * targetBytesReadPerQuery) / 10;
int estimatedQueryBytes = 0;
for (int i = 0; estimatedQueryBytes < targetQueryBytes && endKeyIt != threadData->keyData.end();
i++, endKeyIt++) {
// iterate forward until end or target keys have passed
estimatedQueryBytes += (1 + endKeyIt->second.writes.size() - endKeyIt->second.nextClearIdx) *
threadData->targetValLength;
}
state uint32_t endKey;
if (endKeyIt == threadData->keyData.end()) {
endKey = std::numeric_limits<uint32_t>::max();
} else {
endKey = endKeyIt->first;
}
state KeyRange range = KeyRangeRef(threadData->getKey(startKey, 0), threadData->getKey(endKey, 0));
// pick read version
// TODO could also pick begin version here
ASSERT(threadData->writeVersions.back() >= threadData->minSuccessfulReadVersion);
// randomly choose up to date vs time travel read
if (deterministicRandom()->random01() < 0.5) {
threadData->reads++;
readVersion = threadData->writeVersions.back();
} else {
threadData->timeTravelReads++;
loop {
int readVersionIdx = deterministicRandom()->randomInt(0, threadData->writeVersions.size());
readVersion = threadData->writeVersions[readVersionIdx];
if (readVersion >= threadData->minSuccessfulReadVersion) {
break;
}
}
}
std::pair<RangeResult, Standalone<VectorRef<BlobGranuleChunkRef>>> blob =
wait(self->readFromBlob(cx, self, range, readVersion));
self->validateResult(threadData, blob, startKey, endKey, 0, readVersion);
int resultBytes = blob.first.expectedSize();
threadData->rowsRead += blob.first.size();
threadData->bytesRead += resultBytes;
wait(poisson(&last, (resultBytes + 1) / targetReadBytesPerSec));
} catch (Error& e) {
if (e.code() == error_code_operation_cancelled) {
throw;
}
if (e.code() == error_code_blob_granule_transaction_too_old) {
threadData->timeTravelTooOld++;
} else {
threadData->errors++;
if (BGW_DEBUG) {
printf("BGWorkload got unexpected error %s\n", e.name());
}
}
}
}
}
ACTOR Future<Void> writeWorker(BlobGranuleCorrectnessWorkload* self,
Future<Void> firstSnapshot,
Database cx,
Reference<ThreadData> threadData) {
state double last = now();
state int keysPerQuery = 100;
// state int targetBytesPerQuery = threadData->targetValLength * keysPerQuery;
// state double targetTps = (1.0 * threadData->targetByteRate) / targetBytesPerQuery;
state uint32_t nextVal = 0;
TraceEvent("BlobGranuleCorrectnessWriterStart").log();
wait(firstSnapshot);
TraceEvent("BlobGranuleCorrectnessWriterReady").log();
loop {
state Transaction tr(cx);
// pick rows to write and clear, generate values for writes
state std::vector<std::tuple<uint32_t, uint32_t, uint32_t, uint16_t>> keyAndIdToWrite;
state std::vector<std::pair<uint32_t, uint32_t>> keyAndIdToClear;
state int queryKeys =
keysPerQuery * (0.1 + deterministicRandom()->random01() * 1.8); // 10% to 190% of target keys per query
for (int i = 0; i < queryKeys; i++) {
uint32_t key;
if (threadData->keyData.empty() || deterministicRandom()->random01() > threadData->reuseKeyProb) {
// new key
if (threadData->nextKeySequential) {
key = threadData->usedKeys.size();
} else {
key = std::numeric_limits<uint32_t>::max();
while (key == std::numeric_limits<uint32_t>::max() ||
threadData->keyData.find(key) != threadData->keyData.end()) {
key = deterministicRandom()->randomUInt32();
}
}
// add new key to data structures
threadData->usedKeys.push_back(key);
threadData->keyData.insert({ key, KeyData() });
} else {
int keyIdx = deterministicRandom()->randomInt(0, threadData->usedKeys.size());
key = threadData->usedKeys[keyIdx];
}
auto keyIt = threadData->keyData.find(key);
ASSERT(keyIt != threadData->keyData.end());
int unclearedIds = keyIt->second.writes.size() - keyIt->second.nextClearIdx;
// if we are at targetIDs, 50% chance of adding one or clearing. If we are closer to 0, higher chance of
// adding one, if we are closer to 2x target IDs, higher chance of clearing one
double probAddId = (threadData->targetIDsPerKey * 2.0 - unclearedIds) / threadData->targetIDsPerKey;
if (deterministicRandom()->random01() < probAddId ||
keyIt->second.nextClearIdx == keyIt->second.writes.size()) {
int32_t val = nextVal++;
int16_t valLen = (0.5 + deterministicRandom()->random01()) * threadData->targetValLength;
if (valLen < 10) {
valLen = 10;
}
uint32_t nextId = keyIt->second.writes.size();
keyIt->second.writes.push_back(WriteData(val, valLen));
keyAndIdToWrite.push_back(std::tuple(key, nextId, val, valLen));
} else {
uint32_t idToClear = keyIt->second.nextClearIdx++;
keyAndIdToClear.push_back(std::pair(key, idToClear));
}
}
state int64_t txnBytes;
loop {
try {
// write rows in txn
for (auto& it : keyAndIdToWrite) {
Value v = self->genVal(std::get<2>(it), std::get<3>(it));
tr.set(threadData->getKey(std::get<0>(it), std::get<1>(it)), v);
}
for (auto& it : keyAndIdToClear) {
tr.clear(singleKeyRange(threadData->getKey(it.first, it.second)));
}
txnBytes = tr.getSize();
wait(tr.commit());
break;
} catch (Error& e) {
wait(tr.onError(e));
}
}
Version commitVersion = tr.getCommittedVersion();
// once txn is committed, update write map
for (auto& it : keyAndIdToWrite) {
uint32_t key = std::get<0>(it);
uint32_t id = std::get<1>(it);
auto keyIt = threadData->keyData.find(key);
ASSERT(keyIt != threadData->keyData.end());
keyIt->second.writes[id].writeVersion = commitVersion;
if (DEBUG_KEY_OP(threadData->directoryID, key)) {
fmt::print("DBG: {0} WRITE {1} = {2}:{3}\n",
commitVersion,
format("%08x/%08x/%08x", threadData->directoryID, key, id),
std::get<2>(it),
std::get<3>(it));
}
}
for (auto& it : keyAndIdToClear) {
auto keyIt = threadData->keyData.find(it.first);
ASSERT(keyIt != threadData->keyData.end());
keyIt->second.writes[it.second].clearVersion = commitVersion;
if (DEBUG_KEY_OP(threadData->directoryID, it.first)) {
fmt::print("DBG: {0} CLEAR {1}\n",
commitVersion,
format("%08x/%08x/%08x", threadData->directoryID, it.first, it.second));
}
}
threadData->writeVersions.push_back(commitVersion);
if (threadData->firstWriteSuccessful.canBeSet()) {
threadData->firstWriteSuccessful.send(Void());
}
threadData->rowsWritten += queryKeys;
threadData->bytesWritten += txnBytes;
// wait
wait(poisson(&last, (txnBytes + 1.0) / threadData->targetByteRate));
}
}
Future<Void> start(Database const& cx) override {
clients.reserve(3 * directories.size());
for (auto& it : directories) {
// Wait for blob worker to initialize snapshot before starting test for that range
Future<Void> start = waitFirstSnapshot(this, cx, it, true);
clients.push_back(timeout(writeWorker(this, start, cx, it), testDuration, Void()));
clients.push_back(timeout(readWorker(this, start, cx, it), testDuration, Void()));
}
return delay(testDuration);
}
ACTOR Future<bool> checkDirectory(Database cx,
BlobGranuleCorrectnessWorkload* self,
Reference<ThreadData> threadData) {
state bool result = true;
state int finalRowsValidated;
if (threadData->writeVersions.empty()) {
// never had a successful write during the test, likely due to many chaos events. Just wait for granules to
// become available and call that a pass, since writer is stopped and will never guarantee anything is
// written
if (BGW_DEBUG) {
fmt::print("Directory {0} doing final availability check\n", threadData->directoryID);
}
wait(self->waitFirstSnapshot(self, cx, threadData, false));
} else {
// otherwise, read at last write version and ensure everything becomes available and matches
// it's possible that waitFirstSnapshot finished but then writer never wrote anything before test timed out
state Version readVersion = threadData->writeVersions.back();
if (BGW_DEBUG) {
fmt::print("Directory {0} doing final data check @ {1}\n", threadData->directoryID, readVersion);
}
std::pair<RangeResult, Standalone<VectorRef<BlobGranuleChunkRef>>> blob =
wait(self->readFromBlob(cx, self, threadData->directoryRange, readVersion));
result = self->validateResult(threadData, blob, 0, std::numeric_limits<uint32_t>::max(), 0, readVersion);
finalRowsValidated = blob.first.size();
// then if we are still good, do another check at a higher version (not checking data) to ensure availabiity
// of empty versions
if (result) {
if (BGW_DEBUG) {
fmt::print("Directory {0} doing final availability check after data check\n",
threadData->directoryID);
}
wait(self->waitFirstSnapshot(self, cx, threadData, false));
}
}
bool initialCheck = result;
result &= threadData->mismatches == 0 && (threadData->timeTravelTooOld == 0);
fmt::print("Blob Granule Workload Directory {0} {1}:\n", threadData->directoryID, result ? "passed" : "failed");
fmt::print(" Final granule check {0}successful\n", initialCheck ? "" : "un");
fmt::print(" {} Rows read in final check\n", finalRowsValidated);
fmt::print(" {} mismatches\n", threadData->mismatches);
fmt::print(" {} time travel too old\n", threadData->timeTravelTooOld);
fmt::print(" {} errors\n", threadData->errors);
fmt::print(" {} rows written\n", threadData->rowsWritten);
fmt::print(" {} bytes written\n", threadData->bytesWritten);
fmt::print(" {} unique keys\n", threadData->usedKeys.size());
fmt::print(" {} real-time reads\n", threadData->reads);
fmt::print(" {} time travel reads\n", threadData->timeTravelReads);
fmt::print(" {} rows read\n", threadData->rowsRead);
fmt::print(" {} bytes read\n", threadData->bytesRead);
// FIXME: add above as details to trace event
TraceEvent("BlobGranuleWorkloadChecked").detail("Directory", threadData->directoryID).detail("Result", result);
// For some reason simulation is still passing when this fails?.. so assert for now
ASSERT(result);
return result;
}
ACTOR Future<bool> _check(Database cx, BlobGranuleCorrectnessWorkload* self) {
// check error counts, and do an availability check at the end
state std::vector<Future<bool>> results;
for (auto& it : self->directories) {
results.push_back(self->checkDirectory(cx, self, it));
}
state bool allSuccessful = true;
for (auto& f : results) {
bool dirSuccess = wait(f);
allSuccessful &= dirSuccess;
}
return allSuccessful;
}
Future<bool> check(Database const& cx) override { return _check(cx, this); }
void getMetrics(std::vector<PerfMetric>& m) override {}
};
WorkloadFactory<BlobGranuleCorrectnessWorkload> BlobGranuleCorrectnessWorkloadFactory("BlobGranuleCorrectnessWorkload");

View File

@ -24,17 +24,20 @@
#include "contrib/fmt-8.1.1/include/fmt/format.h"
#include "fdbclient/BlobGranuleReader.actor.h"
#include "fdbclient/ManagementAPI.actor.h"
#include "fdbclient/NativeAPI.actor.h"
#include "fdbclient/ReadYourWrites.h"
#include "fdbclient/SystemData.h"
#include "fdbserver/Knobs.h"
#include "fdbserver/TesterInterface.actor.h"
#include "fdbserver/workloads/workloads.actor.h"
#include "flow/Error.h"
#include "flow/IRandom.h"
#include "flow/genericactors.actor.h"
#include "flow/actorcompiler.h" // This must be the last #include.
#define BGV_DEBUG false
#define BGV_DEBUG true
/*
* This workload is designed to verify the correctness of the blob data produced by the blob workers.
@ -59,6 +62,9 @@ struct BlobGranuleVerifierWorkload : TestWorkload {
int64_t rowsRead = 0;
int64_t bytesRead = 0;
std::vector<Future<Void>> clients;
bool enablePruning;
DatabaseConfiguration config;
Reference<BackupContainerFileSystem> bstore;
AsyncVar<Standalone<VectorRef<KeyRangeRef>>> granuleRanges;
@ -72,6 +78,7 @@ struct BlobGranuleVerifierWorkload : TestWorkload {
timeTravelLimit = getOption(options, LiteralStringRef("timeTravelLimit"), testDuration);
timeTravelBufferSize = getOption(options, LiteralStringRef("timeTravelBufferSize"), 100000000);
threads = getOption(options, LiteralStringRef("threads"), 1);
enablePruning = getOption(options, LiteralStringRef("enablePruning"), false /*sharedRandomNumber % 2 == 0*/);
ASSERT(threads >= 1);
if (BGV_DEBUG) {
@ -126,19 +133,22 @@ struct BlobGranuleVerifierWorkload : TestWorkload {
}
std::string description() const override { return "BlobGranuleVerifier"; }
Future<Void> setup(Database const& cx) override {
if (!CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
Future<Void> setup(Database const& cx) override { return _setup(cx, this); }
ACTOR Future<Void> _setup(Database cx, BlobGranuleVerifierWorkload* self) {
if (!self->doSetup) {
wait(delay(0));
return Void();
}
if (doSetup) {
double initialDelay = deterministicRandom()->random01() * (maxDelay - minDelay) + minDelay;
if (BGV_DEBUG) {
printf("BGW setup initial delay of %.3f\n", initialDelay);
}
return setUpBlobRange(cx, delay(initialDelay));
wait(success(ManagementAPI::changeConfig(cx.getReference(), "blob_granules_enabled=1", true)));
double initialDelay = deterministicRandom()->random01() * (self->maxDelay - self->minDelay) + self->minDelay;
if (BGV_DEBUG) {
printf("BGW setup initial delay of %.3f\n", initialDelay);
}
return delay(0);
wait(self->setUpBlobRange(cx, delay(initialDelay)));
return Void();
}
ACTOR Future<Void> findGranules(Database cx, BlobGranuleVerifierWorkload* self) {
@ -159,20 +169,35 @@ struct BlobGranuleVerifierWorkload : TestWorkload {
// assumes we can read the whole range in one transaction at a single version
ACTOR Future<std::pair<RangeResult, Version>> readFromFDB(Database cx, KeyRange range) {
state bool first = true;
state Version v;
state RangeResult out;
state Transaction tr(cx);
state KeyRange currentRange = range;
loop {
try {
RangeResult r = wait(tr.getRange(currentRange, CLIENT_KNOBS->TOO_MANY));
state RangeResult r = wait(tr.getRange(currentRange, CLIENT_KNOBS->TOO_MANY));
Version grv = wait(tr.getReadVersion());
// need consistent version snapshot of range
if (first) {
v = grv;
first = false;
} else if (v != grv) {
// reset the range and restart the read at a higher version
TraceEvent(SevDebug, "BGVFDBReadReset").detail("ReadVersion", v);
TEST(true); // BGV transaction reset
fmt::print("Resetting BGV GRV {0} -> {1}\n", v, grv);
first = true;
out = RangeResult();
currentRange = range;
tr.reset();
continue;
}
out.arena().dependsOn(r.arena());
out.append(out.arena(), r.begin(), r.size());
if (r.more) {
currentRange = KeyRangeRef(keyAfter(r.back().key), currentRange.end);
} else {
Version _v = wait(tr.getReadVersion());
v = _v;
break;
}
} catch (Error& e) {
@ -299,11 +324,108 @@ struct BlobGranuleVerifierWorkload : TestWorkload {
OldRead(KeyRange range, Version v, RangeResult oldResult) : range(range), v(v), oldResult(oldResult) {}
};
ACTOR Future<Void> verifyGranules(Database cx, BlobGranuleVerifierWorkload* self) {
// utility to prune <range> at pruneVersion=<version> with the <force> flag
ACTOR Future<Void> pruneAtVersion(Database cx, KeyRange range, Version version, bool force) {
state Reference<ReadYourWritesTransaction> tr = makeReference<ReadYourWritesTransaction>(cx);
state Key pruneKey;
loop {
try {
tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr->setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE);
Value pruneValue = blobGranulePruneValueFor(version, range, force);
tr->atomicOp(
addVersionStampAtEnd(blobGranulePruneKeys.begin), pruneValue, MutationRef::SetVersionstampedKey);
tr->set(blobGranulePruneChangeKey, deterministicRandom()->randomUniqueID().toString());
state Future<Standalone<StringRef>> fTrVs = tr->getVersionstamp();
wait(tr->commit());
Standalone<StringRef> vs = wait(fTrVs);
pruneKey = blobGranulePruneKeys.begin.withSuffix(vs);
if (BGV_DEBUG) {
fmt::print("pruneAtVersion for range [{0} - {1}) at version {2} succeeded\n",
range.begin.printable(),
range.end.printable(),
version);
}
break;
} catch (Error& e) {
if (BGV_DEBUG) {
fmt::print("pruneAtVersion for range [{0} - {1}) at version {2} encountered error {3}\n",
range.begin.printable(),
range.end.printable(),
version,
e.name());
}
wait(tr->onError(e));
}
}
tr->reset();
loop {
try {
tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr->setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE);
Optional<Value> pruneVal = wait(tr->get(pruneKey));
if (!pruneVal.present()) {
return Void();
}
state Future<Void> watchFuture = tr->watch(pruneKey);
wait(tr->commit());
wait(watchFuture);
} catch (Error& e) {
wait(tr->onError(e));
}
}
}
ACTOR Future<Void> killBlobWorkers(Database cx, BlobGranuleVerifierWorkload* self) {
state Transaction tr(cx);
state std::set<UID> knownWorkers;
state bool first = true;
loop {
try {
RangeResult r = wait(tr.getRange(blobWorkerListKeys, CLIENT_KNOBS->TOO_MANY));
state std::vector<UID> haltIds;
state std::vector<Future<ErrorOr<Void>>> haltRequests;
for (auto& it : r) {
BlobWorkerInterface interf = decodeBlobWorkerListValue(it.value);
if (first) {
knownWorkers.insert(interf.id());
}
if (knownWorkers.count(interf.id())) {
haltIds.push_back(interf.id());
haltRequests.push_back(interf.haltBlobWorker.tryGetReply(HaltBlobWorkerRequest(1e6, UID())));
}
}
first = false;
wait(waitForAll(haltRequests));
bool allPresent = true;
for (int i = 0; i < haltRequests.size(); i++) {
if (haltRequests[i].get().present()) {
knownWorkers.erase(haltIds[i]);
} else {
allPresent = false;
}
}
if (allPresent) {
return Void();
} else {
wait(delay(1.0));
}
} catch (Error& e) {
wait(tr.onError(e));
}
}
}
ACTOR Future<Void> verifyGranules(Database cx, BlobGranuleVerifierWorkload* self, bool allowPruning) {
state double last = now();
state double endTime = last + self->testDuration;
state std::map<double, OldRead> timeTravelChecks;
state int64_t timeTravelChecksMemory = 0;
state Version prevPruneVersion = -1;
state UID dbgId = debugRandom()->randomUniqueID();
TraceEvent("BlobGranuleVerifierStart");
if (BGV_DEBUG) {
@ -325,15 +447,53 @@ struct BlobGranuleVerifierWorkload : TestWorkload {
state OldRead oldRead = timeTravelIt->second;
timeTravelChecksMemory -= oldRead.oldResult.expectedSize();
timeTravelIt = timeTravelChecks.erase(timeTravelIt);
if (prevPruneVersion == -1) {
prevPruneVersion = oldRead.v;
}
// advance iterator before doing read, so if it gets error we don't retry it
try {
state Version newPruneVersion = 0;
state bool doPruning = allowPruning && deterministicRandom()->random01() < 0.5;
if (doPruning) {
Version maxPruneVersion = oldRead.v;
for (auto& it : timeTravelChecks) {
maxPruneVersion = std::min(it.second.v, maxPruneVersion);
}
if (prevPruneVersion < maxPruneVersion) {
newPruneVersion = deterministicRandom()->randomInt64(prevPruneVersion, maxPruneVersion);
prevPruneVersion = std::max(prevPruneVersion, newPruneVersion);
wait(self->pruneAtVersion(cx, normalKeys, newPruneVersion, false));
} else {
doPruning = false;
}
}
std::pair<RangeResult, Standalone<VectorRef<BlobGranuleChunkRef>>> reReadResult =
wait(self->readFromBlob(cx, self, oldRead.range, oldRead.v));
self->compareResult(oldRead.oldResult, reReadResult, oldRead.range, oldRead.v, false);
self->timeTravelReads++;
if (doPruning) {
wait(self->killBlobWorkers(cx, self));
std::pair<RangeResult, Standalone<VectorRef<BlobGranuleChunkRef>>> versionRead =
wait(self->readFromBlob(cx, self, oldRead.range, prevPruneVersion));
try {
Version minSnapshotVersion = newPruneVersion;
for (auto& it : versionRead.second) {
minSnapshotVersion = std::min(minSnapshotVersion, it.snapshotVersion);
}
std::pair<RangeResult, Standalone<VectorRef<BlobGranuleChunkRef>>> versionRead =
wait(self->readFromBlob(cx, self, oldRead.range, minSnapshotVersion - 1));
ASSERT(false);
} catch (Error& e) {
if (e.code() == error_code_actor_cancelled) {
throw;
}
ASSERT(e.code() == error_code_blob_granule_transaction_too_old);
}
}
} catch (Error& e) {
if (e.code() == error_code_transaction_too_old) {
if (e.code() == error_code_blob_granule_transaction_too_old) {
self->timeTravelTooOld++;
// TODO: add debugging info for when this is a failure
}
@ -365,8 +525,7 @@ struct BlobGranuleVerifierWorkload : TestWorkload {
if (e.code() == error_code_operation_cancelled) {
throw;
}
if (e.code() != error_code_transaction_too_old && e.code() != error_code_wrong_shard_server &&
BGV_DEBUG) {
if (e.code() != error_code_blob_granule_transaction_too_old && BGV_DEBUG) {
printf("BGVerifier got unexpected error %s\n", e.name());
}
self->errors++;
@ -377,32 +536,65 @@ struct BlobGranuleVerifierWorkload : TestWorkload {
}
Future<Void> start(Database const& cx) override {
if (!CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
return Void();
}
clients.reserve(threads + 1);
clients.push_back(timeout(findGranules(cx, this), testDuration, Void()));
for (int i = 0; i < threads; i++) {
if (enablePruning && clientId == 0) {
clients.push_back(
timeout(reportErrors(verifyGranules(cx, this), "BlobGranuleVerifier"), testDuration, Void()));
timeout(reportErrors(verifyGranules(cx, this, true), "BlobGranuleVerifier"), testDuration, Void()));
} else if (!enablePruning) {
for (int i = 0; i < threads; i++) {
clients.push_back(timeout(
reportErrors(verifyGranules(cx, this, false), "BlobGranuleVerifier"), testDuration, Void()));
}
}
return delay(testDuration);
}
// handle retries + errors
// It's ok to reset the transaction here because its read version is only used for reading the granule mapping from
// the system keyspace
ACTOR Future<Version> doGrv(Transaction* tr) {
loop {
try {
Version readVersion = wait(tr->getReadVersion());
return readVersion;
} catch (Error& e) {
wait(tr->onError(e));
}
}
}
ACTOR Future<bool> _check(Database cx, BlobGranuleVerifierWorkload* self) {
// check error counts, and do an availability check at the end
state Transaction tr(cx);
state Version readVersion = wait(tr.getReadVersion());
state Version readVersion = wait(self->doGrv(&tr));
state Version startReadVersion = readVersion;
state int checks = 0;
state KeyRange last;
state bool availabilityPassed = true;
state Standalone<VectorRef<KeyRangeRef>> allRanges = self->granuleRanges.get();
state Standalone<VectorRef<KeyRangeRef>> allRanges;
if (self->granuleRanges.get().empty()) {
if (BGV_DEBUG) {
fmt::print("Waiting to get granule ranges for check\n");
}
state Future<Void> rangeFetcher = self->findGranules(cx, self);
loop {
wait(self->granuleRanges.onChange());
if (!self->granuleRanges.get().empty()) {
break;
}
}
rangeFetcher.cancel();
if (BGV_DEBUG) {
fmt::print("Got granule ranges for check\n");
}
}
allRanges = self->granuleRanges.get();
for (auto& range : allRanges) {
state KeyRange r = range;
state PromiseStream<Standalone<BlobGranuleChunkRef>> chunkStream;
if (BGV_DEBUG) {
fmt::print("Final availability check [{0} - {1}) @ {2}\n",
r.begin.printable(),
@ -412,18 +604,32 @@ struct BlobGranuleVerifierWorkload : TestWorkload {
try {
loop {
tr.reset();
try {
Standalone<VectorRef<BlobGranuleChunkRef>> chunks =
wait(tr.readBlobGranules(r, 0, readVersion));
ASSERT(chunks.size() > 0);
last = chunks.back().keyRange;
checks += chunks.size();
break;
} catch (Error& e) {
wait(tr.onError(e));
// it's possible for blob granules to never get opened for the entire test due to fault
// injection. If we get blob_granule_transaction_too_old, for the latest read version, the
// granule still needs to open. Wait for that to happen at a higher read version.
if (e.code() == error_code_blob_granule_transaction_too_old) {
wait(delay(1.0));
tr.reset();
Version rv = wait(self->doGrv(&tr));
readVersion = rv;
} else {
wait(tr.onError(e));
}
}
}
} catch (Error& e) {
if (e.code() == error_code_actor_cancelled) {
throw;
}
if (e.code() == error_code_end_of_stream) {
break;
}
@ -441,7 +647,11 @@ struct BlobGranuleVerifierWorkload : TestWorkload {
break;
}
}
fmt::print("Blob Granule Verifier finished with:\n");
if (BGV_DEBUG && startReadVersion != readVersion) {
fmt::print("Availability check updated read version from {0} to {1}\n", startReadVersion, readVersion);
}
bool result = availabilityPassed && self->mismatches == 0 && (checks > 0) && (self->timeTravelTooOld == 0);
fmt::print("Blob Granule Verifier {0} {1}:\n", self->clientId, result ? "passed" : "failed");
fmt::print(" {} successful final granule checks\n", checks);
fmt::print(" {} failed final granule checks\n", availabilityPassed ? 0 : 1);
fmt::print(" {} mismatches\n", self->mismatches);
@ -451,18 +661,17 @@ struct BlobGranuleVerifierWorkload : TestWorkload {
fmt::print(" {} time travel reads\n", self->timeTravelReads);
fmt::print(" {} rows\n", self->rowsRead);
fmt::print(" {} bytes\n", self->bytesRead);
// FIXME: add above as details
TraceEvent("BlobGranuleVerifierChecked");
return availabilityPassed && self->mismatches == 0 && checks > 0 && self->timeTravelTooOld == 0;
// FIXME: add above as details to trace event
TraceEvent("BlobGranuleVerifierChecked").detail("Result", result);
// For some reason simulation is still passing when this fails?.. so assert for now
ASSERT(result);
return result;
}
Future<bool> check(Database const& cx) override {
if (!CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
return true;
}
return _check(cx, this);
}
Future<bool> check(Database const& cx) override { return _check(cx, this); }
void getMetrics(std::vector<PerfMetric>& m) override {}
};

View File

@ -297,7 +297,7 @@ struct ConsistencyCheckWorkload : TestWorkload {
wait(::success(self->checkForExtraDataStores(cx, self)));
// Check blob workers are operating as expected
if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
if (configuration.blobGranulesEnabled) {
bool blobWorkersCorrect = wait(self->checkBlobWorkers(cx, configuration, self));
if (!blobWorkersCorrect)
self->testFailure("Blob workers incorrect");
@ -2003,25 +2003,24 @@ struct ConsistencyCheckWorkload : TestWorkload {
int numBlobWorkerProcesses = 0;
for (const auto& worker : workers) {
NetworkAddress addr = worker.interf.stableAddress();
bool inCCDc = worker.interf.locality.dcId() == ccDcId;
if (!configuration.isExcludedServer(worker.interf.addresses())) {
if (worker.processClass == ProcessClass::BlobWorkerClass) {
numBlobWorkerProcesses++;
// this is a worker with processClass == BWClass, so should have exactly one blob worker
if (blobWorkersByAddr[addr] == 0) {
TraceEvent("ConsistencyCheck_NoBWsOnBWClass")
// this is a worker with processClass == BWClass, so should have exactly one blob worker if it's in
// the same DC
int desiredBlobWorkersOnAddr = inCCDc ? 1 : 0;
if (blobWorkersByAddr[addr] != desiredBlobWorkersOnAddr) {
TraceEvent("ConsistencyCheck_WrongBWCountOnBWClass")
.detail("Address", addr)
.detail("NumBlobWorkersOnAddr", blobWorkersByAddr[addr]);
.detail("NumBlobWorkersOnAddr", blobWorkersByAddr[addr])
.detail("DesiredBlobWorkersOnAddr", desiredBlobWorkersOnAddr)
.detail("BwDcId", worker.interf.locality.dcId())
.detail("CcDcId", ccDcId);
return false;
}
/* TODO: replace above code with this once blob manager recovery is handled
if (blobWorkersByAddr[addr] != 1) {
TraceEvent("ConsistencyCheck_NoBWOrManyBWsOnBWClass")
.detail("Address", addr)
.detail("NumBlobWorkersOnAddr", blobWorkersByAddr[addr]);
return false;
}
*/
} else {
// this is a worker with processClass != BWClass, so there should be no BWs on it
if (blobWorkersByAddr[addr] > 0) {
@ -2359,7 +2358,7 @@ struct ConsistencyCheckWorkload : TestWorkload {
}
// Check BlobManager
if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES && db.blobManager.present() &&
if (config.blobGranulesEnabled && db.blobManager.present() &&
(!nonExcludedWorkerProcessMap.count(db.blobManager.get().address()) ||
nonExcludedWorkerProcessMap[db.blobManager.get().address()].processClass.machineClassFitness(
ProcessClass::BlobManager) > fitnessLowerBound)) {

View File

@ -45,6 +45,9 @@ struct WriteDuringReadWorkload : TestWorkload {
bool useSystemKeys;
std::string keyPrefix;
int64_t maximumTotalData;
int64_t maximumDataWritten;
int64_t dataWritten = 0;
bool success;
Database extraDB;
@ -57,6 +60,8 @@ struct WriteDuringReadWorkload : TestWorkload {
numOps = getOption(options, LiteralStringRef("numOps"), 21);
rarelyCommit = getOption(options, LiteralStringRef("rarelyCommit"), false);
maximumTotalData = getOption(options, LiteralStringRef("maximumTotalData"), 3e6);
maximumDataWritten =
getOption(options, LiteralStringRef("maximumDataWritten"), std::numeric_limits<int64_t>::max());
minNode = getOption(options, LiteralStringRef("minNode"), 0);
useSystemKeys = getOption(options, LiteralStringRef("useSystemKeys"), deterministicRandom()->random01() < 0.5);
adjacentKeys = deterministicRandom()->random01() < 0.5;
@ -73,6 +78,7 @@ struct WriteDuringReadWorkload : TestWorkload {
nodes = deterministicRandom()->randomInt(1, 4 << deterministicRandom()->randomInt(0, 20));
}
dataWritten = 0;
int newNodes = std::min<int>(nodes, maximumTotalData / (getKeyForIndex(nodes).size() + valueSizeRange.second));
minNode = std::max(minNode, nodes - newNodes);
nodes = newNodes;
@ -539,11 +545,13 @@ struct WriteDuringReadWorkload : TestWorkload {
}
}
state int64_t txnSize = tr->getApproximateSize();
state std::map<Key, Value> committedDB = self->memoryDatabase;
*doingCommit = true;
wait(tr->commit());
*doingCommit = false;
self->finished.trigger();
self->dataWritten += txnSize;
if (readYourWritesDisabled)
tr->setOption(FDBTransactionOptions::READ_YOUR_WRITES_DISABLE);
@ -616,11 +624,12 @@ struct WriteDuringReadWorkload : TestWorkload {
state Transaction tr(cx);
loop {
try {
if (now() - startTime > self->testDuration)
if (now() - startTime > self->testDuration || self->dataWritten >= self->maximumDataWritten)
return Void();
if (self->useSystemKeys)
tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
state int64_t txnSize = 0;
if (i == 0) {
tr.clear(normalKeys);
}
@ -641,10 +650,13 @@ struct WriteDuringReadWorkload : TestWorkload {
value.substr(0, std::min<int>(value.size(), CLIENT_KNOBS->VALUE_SIZE_LIMIT));
self->memoryDatabase[key] = value;
tr.set(key, value);
int64_t rowSize = key.expectedSize() + value.expectedSize();
txnSize += rowSize;
}
}
}
wait(tr.commit());
self->dataWritten += txnSize;
//TraceEvent("WDRInitBatch").detail("I", i).detail("CommittedVersion", tr.getCommittedVersion());
break;
} catch (Error& e) {
@ -671,7 +683,7 @@ struct WriteDuringReadWorkload : TestWorkload {
throw;
break;
}
if (now() - startTime > self->testDuration)
if (now() - startTime > self->testDuration || self->dataWritten >= self->maximumDataWritten)
return Void();
}
}

View File

@ -84,6 +84,9 @@ ERROR( change_feed_not_registered, 1060, "Change feed not registered" )
ERROR( granule_assignment_conflict, 1061, "Conflicting attempts to assign blob granules" )
ERROR( change_feed_cancelled, 1062, "Change feed was cancelled" )
ERROR( blob_granule_file_load_error, 1063, "Error loading a blob file during granule materialization" )
ERROR( blob_granule_transaction_too_old, 1064, "Read version is older than blob granule history supports" )
ERROR( blob_manager_replaced, 1065, "This blob manager has been replaced." )
ERROR( change_feed_popped, 1066, "Tried to read a version older than what has been popped from the change feed" )
ERROR( broken_promise, 1100, "Broken promise" )
ERROR( operation_cancelled, 1101, "Asynchronous operation cancelled" )
@ -171,7 +174,7 @@ ERROR( quick_get_key_values_has_more, 2033, "One of the mapped range queries is
ERROR( quick_get_value_miss, 2034, "Found a mapped key that is not served in the same SS" )
ERROR( quick_get_key_values_miss, 2035, "Found a mapped range that is not served in the same SS" )
ERROR( blob_granule_no_ryw, 2036, "Blob Granule Read Transactions must be specified as ryw-disabled" )
ERROR( blob_granule_not_materialized, 2037, "Blob Granule Read Transactions must be specified as ryw-disabled" )
ERROR( blob_granule_not_materialized, 2037, "Blob Granule Read was not materialized" )
ERROR( get_mapped_key_values_has_more, 2038, "getMappedRange does not support continuation for now" )
ERROR( get_mapped_range_reads_your_writes, 2039, "getMappedRange tries to read data that were previously written in the transaction" )
ERROR( checkpoint_not_found, 2040, "Checkpoint not found" )

View File

@ -970,8 +970,10 @@ struct NotifiedQueue : private SingleCallback<T>, FastAllocated<NotifiedQueue<T>
std::queue<T, Deque<T>> queue;
Promise<Void> onEmpty;
Error error;
Promise<Void> onError;
NotifiedQueue(int futures, int promises) : promises(promises), futures(futures), onEmpty(nullptr) {
NotifiedQueue(int futures, int promises)
: promises(promises), futures(futures), onEmpty(nullptr), onError(nullptr) {
SingleCallback<T>::next = this;
}
@ -979,6 +981,7 @@ struct NotifiedQueue : private SingleCallback<T>, FastAllocated<NotifiedQueue<T>
bool isReady() const { return !queue.empty() || error.isValid(); }
bool isError() const { return queue.empty() && error.isValid(); } // the *next* thing queued is an error
bool hasError() const { return error.isValid(); } // there is an error queued
uint32_t size() const { return queue.size(); }
virtual T pop() {
@ -1013,7 +1016,17 @@ struct NotifiedQueue : private SingleCallback<T>, FastAllocated<NotifiedQueue<T>
if (error.isValid())
return;
ASSERT(this->error.code() != error_code_success);
this->error = err;
// end_of_stream error is "expected", don't terminate reading stream early for this
// onError must be triggered before callback, otherwise callback could cause delPromiseRef/delFutureRef. This
// could destroy *this* in the callback, causing onError to be referenced after this object is destroyed.
if (err.code() != error_code_end_of_stream && err.code() != error_code_broken_promise && onError.isValid()) {
ASSERT(onError.canBeSet());
onError.sendError(err);
}
if (shouldFireImmediately()) {
SingleCallback<T>::next->error(err);
}
@ -1062,11 +1075,14 @@ protected:
}
auto copy = std::move(queue.front());
queue.pop();
if (onEmpty.isValid() && queue.empty()) {
Promise<Void> hold = onEmpty;
onEmpty = Promise<Void>(nullptr);
hold.send(Void());
}
return copy;
}
bool hasError() { return error.isValid(); }
bool shouldFireImmediately() { return SingleCallback<T>::next != this; }
};

View File

@ -50,8 +50,8 @@ if(WITH_PYTHON)
add_fdb_test(TEST_FILES s3VersionHeaders.txt IGNORE)
add_fdb_test(TEST_FILES BandwidthThrottle.txt IGNORE)
add_fdb_test(TEST_FILES BigInsert.txt IGNORE)
add_fdb_test(TEST_FILES BlobGranuleFileUnit.txt IGNORE)
add_fdb_test(TEST_FILES BlobManagerUnit.txt IGNORE)
add_fdb_test(TEST_FILES BlobGranuleFileUnit.txt)
add_fdb_test(TEST_FILES BlobManagerUnit.txt)
add_fdb_test(TEST_FILES ConsistencyCheck.txt IGNORE)
add_fdb_test(TEST_FILES DDMetricsExclude.txt IGNORE)
add_fdb_test(TEST_FILES DataDistributionMetrics.txt IGNORE)
@ -129,8 +129,10 @@ if(WITH_PYTHON)
add_fdb_test(TEST_FILES fast/BackupCorrectnessClean.toml)
add_fdb_test(TEST_FILES fast/BackupToDBCorrectness.toml)
add_fdb_test(TEST_FILES fast/BackupToDBCorrectnessClean.toml)
add_fdb_test(TEST_FILES fast/BlobGranuleCorrectness.toml IGNORE)
add_fdb_test(TEST_FILES fast/BlobGranuleCorrectnessClean.toml IGNORE)
add_fdb_test(TEST_FILES fast/BlobGranuleVerifySmall.toml)
add_fdb_test(TEST_FILES fast/BlobGranuleVerifySmallClean.toml)
add_fdb_test(TEST_FILES fast/BlobGranuleVerifyAtomicOps.toml)
add_fdb_test(TEST_FILES fast/BlobGranuleVerifyCycle.toml)
add_fdb_test(TEST_FILES fast/CacheTest.toml)
add_fdb_test(TEST_FILES fast/CloggedSideband.toml)
add_fdb_test(TEST_FILES fast/ConfigureLocked.toml)
@ -273,8 +275,12 @@ if(WITH_PYTHON)
add_fdb_test(TEST_FILES slow/ApiCorrectness.toml)
add_fdb_test(TEST_FILES slow/ApiCorrectnessAtomicRestore.toml)
add_fdb_test(TEST_FILES slow/ApiCorrectnessSwitchover.toml)
add_fdb_test(TEST_FILES slow/BlobGranuleCorrectnessLarge.toml IGNORE)
add_fdb_test(TEST_FILES slow/BlobGranuleCorrectnessLargeClean.toml IGNORE)
add_fdb_test(TEST_FILES slow/BlobGranuleVerifyLarge.toml)
add_fdb_test(TEST_FILES slow/BlobGranuleVerifyLargeClean.toml)
add_fdb_test(TEST_FILES slow/BlobGranuleVerifyBalance.toml)
add_fdb_test(TEST_FILES slow/BlobGranuleVerifyBalanceClean.toml)
add_fdb_test(TEST_FILES slow/BlobGranuleCorrectnessClean.toml)
add_fdb_test(TEST_FILES slow/BlobGranuleCorrectness.toml)
add_fdb_test(TEST_FILES slow/ClogWithRollbacks.toml)
add_fdb_test(TEST_FILES slow/CloggedCycleTest.toml)
add_fdb_test(TEST_FILES slow/CloggedStorefront.toml)

View File

@ -1,10 +0,0 @@
[[test]]
testTitle = 'BlobGranuleCorrectnessCleanTest'
[[test.workload]]
testName = 'WriteDuringRead'
testDuration = 120.0
[[test.workload]]
testName = 'BlobGranuleVerifier'
testDuration = 120.0

View File

@ -0,0 +1,37 @@
[configuration]
blobGranulesEnabled = true
[[test]]
testTitle = 'BlobGranuleVerifyAtomicOps'
[[test.workload]]
testName = 'AtomicOps'
transactionsPerSecond = 2500.0
testDuration = 30.0
[[test.workload]]
testName = 'BlobGranuleVerifier'
testDuration = 30.0
[[test.workload]]
testName = 'RandomClogging'
testDuration = 30.0
[[test.workload]]
testName = 'Rollback'
meanDelay = 30.0
testDuration = 30.0
[[test.workload]]
testName = 'Attrition'
machinesToKill = 10
machinesToLeave = 3
reboot = true
testDuration = 30.0
[[test.workload]]
testName = 'Attrition'
machinesToKill = 10
machinesToLeave = 3
reboot = true
testDuration = 30.0

View File

@ -0,0 +1,38 @@
[configuration]
blobGranulesEnabled = true
[[test]]
testTitle = 'BlobGranuleVerifyCycle'
[[test.workload]]
testName = 'Cycle'
transactionsPerSecond = 250.0
testDuration = 60.0
expectedRate = 0
[[test.workload]]
testName = 'BlobGranuleVerifier'
testDuration = 60.0
[[test.workload]]
testName = 'RandomClogging'
testDuration = 60.0
[[test.workload]]
testName = 'Rollback'
meanDelay = 60.0
testDuration = 60.0
[[test.workload]]
testName = 'Attrition'
machinesToKill = 10
machinesToLeave = 3
reboot = true
testDuration = 60.0
[[test.workload]]
testName = 'Attrition'
machinesToKill = 10
machinesToLeave = 3
reboot = true
testDuration = 60.0

View File

@ -0,0 +1,40 @@
[configuration]
blobGranulesEnabled = true
storageEngineExcludeTypes = [3] # FIXME: exclude redwood because WriteDuringRead can write massive KV pairs and we don't chunk change feed data on disk yet
[[test]]
testTitle = 'BlobGranuleVerifySmall'
[[test.workload]]
testName = 'WriteDuringRead'
testDuration = 60.0
useSystemKeys = false
maximumDataWritten=50000000
[[test.workload]]
testName = 'BlobGranuleVerifier'
testDuration = 60.0
[[test.workload]]
testName = 'RandomClogging'
testDuration = 60.0
[[test.workload]]
testName = 'Rollback'
meanDelay = 30.0
testDuration = 60.0
[[test.workload]]
testName = 'Attrition'
machinesToKill = 10
machinesToLeave = 3
reboot = true
testDuration = 60.0
[[test.workload]]
testName = 'Attrition'
machinesToKill = 10
machinesToLeave = 3
reboot = true
testDuration = 60.0

View File

@ -0,0 +1,16 @@
[configuration]
blobGranulesEnabled = true
storageEngineExcludeTypes = [3] # FIXME: exclude redwood because WriteDuringRead can write massive KV pairs and we don't chunk change feed data on disk yet
[[test]]
testTitle = 'BlobGranuleVerifySmallClean'
[[test.workload]]
testName = 'WriteDuringRead'
testDuration = 60.0
useSystemKeys = false
maximumDataWritten=50000000
[[test.workload]]
testName = 'BlobGranuleVerifier'
testDuration = 60.0

View File

@ -1,34 +1,32 @@
[configuration]
blobGranulesEnabled = true
[[test]]
testTitle = 'BlobGranuleCorrectnessTest'
testTitle = 'BlobGranuleCorrectness'
[[test.workload]]
testName = 'WriteDuringRead'
testDuration = 120.0
[[test.workload]]
testName = 'BlobGranuleVerifier'
testName = 'BlobGranuleCorrectnessWorkload'
testDuration = 120.0
[[test.workload]]
testName = 'RandomClogging'
testDuration = 120.0
testDuration = 120.0
[[test.workload]]
testName = 'Rollback'
meanDelay = 30.0
testDuration = 120.0
testDuration = 120.0
[[test.workload]]
testName = 'Attrition'
machinesToKill = 10
machinesToLeave = 3
reboot = true
testDuration = 120.0
testDuration = 120.0
[[test.workload]]
testName = 'Attrition'
machinesToKill = 10
machinesToLeave = 3
reboot = true
testDuration = 120.0
testDuration = 120.0

View File

@ -0,0 +1,9 @@
[configuration]
blobGranulesEnabled = true
[[test]]
testTitle = 'BlobGranuleCorrectness'
[[test.workload]]
testName = 'BlobGranuleCorrectnessWorkload'
testDuration = 120.0

View File

@ -0,0 +1,51 @@
[configuration]
blobGranulesEnabled = true
[[test]]
testTitle = 'BlobGranuleVerifyBalance'
[[test.workload]]
testName = 'DDBalance'
testDuration = 120.0
transactionsPerSecond = 250.0
binCount = 1000
writesPerTransaction = 5
keySpaceDriftFactor = 10
moversPerClient = 10
actorsPerClient = 100
nodes = 100000
[[test.workload]]
testName = 'BlobGranuleVerifier'
testDuration = 120.0
[[test.workload]]
testName = 'RandomClogging'
testDuration = 120.0
swizzle = 1
[[test.workload]]
testName = 'Rollback'
testDuration = 120.0
meanDelay = 10.0
[[test.workload]]
testName = 'Attrition'
machinesToKill = 10
machinesToLeave = 3
reboot = true
testDuration = 120.0
[[test.workload]]
testName = 'Attrition'
machinesToKill = 10
machinesToLeave = 3
reboot = true
testDuration = 120.0
[[test.workload]]
testName = 'RemoveServersSafely'
minDelay = 0
maxDelay = 100
kill1Timeout = 30
kill2Timeout = 6000

View File

@ -0,0 +1,20 @@
[configuration]
blobGranulesEnabled = true
[[test]]
testTitle = 'BlobGranuleVerifyBalanceClean'
[[test.workload]]
testName = 'DDBalance'
testDuration = 120.0
transactionsPerSecond = 250.0
binCount = 1000
writesPerTransaction = 5
keySpaceDriftFactor = 10
moversPerClient = 10
actorsPerClient = 100
nodes = 100000
[[test.workload]]
testName = 'BlobGranuleVerifier'
testDuration = 120.0

View File

@ -1,12 +1,15 @@
[configuration]
blobGranulesEnabled = true
[[test]]
testTitle = 'BlobGranuleCorrectnessLargeTest'
testTitle = 'BlobGranuleVerifyLarge'
[[test.workload]]
testName = 'ReadWrite'
testDuration = 200.0
testDuration = 120.0
transactionsPerSecond = 200
writesPerTransactionA = 0
readsPerTransactionA = 10
writesPerTransactionA = 5
readsPerTransactionA = 1
writesPerTransactionB = 10
readsPerTransactionB = 1
alpha = 0.5
@ -18,28 +21,28 @@ testTitle = 'BlobGranuleCorrectnessLargeTest'
[[test.workload]]
testName = 'BlobGranuleVerifier'
testDuration = 200.0
testDuration = 120.0
[[test.workload]]
testName = 'RandomClogging'
testDuration = 200.0
testDuration = 120.0
[[test.workload]]
testName = 'Rollback'
meanDelay = 30.0
testDuration = 200.0
testDuration = 120.0
[[test.workload]]
testName = 'Attrition'
machinesToKill = 10
machinesToLeave = 3
reboot = true
testDuration = 200.0
testDuration = 120.0
[[test.workload]]
testName = 'Attrition'
machinesToKill = 10
machinesToLeave = 3
reboot = true
testDuration = 200.0
testDuration = 120.0

View File

@ -1,12 +1,15 @@
[configuration]
blobGranulesEnabled = true
[[test]]
testTitle = 'BlobGranuleCorrectnessLargeCleanTest'
testTitle = 'BlobGranuleVerifyLargeClean'
[[test.workload]]
testName = 'ReadWrite'
testDuration = 200.0
testDuration = 120.0
transactionsPerSecond = 200
writesPerTransactionA = 0
readsPerTransactionA = 10
writesPerTransactionA = 5
readsPerTransactionA = 1
writesPerTransactionB = 10
readsPerTransactionB = 1
alpha = 0.5
@ -18,4 +21,4 @@ testTitle = 'BlobGranuleCorrectnessLargeCleanTest'
[[test.workload]]
testName = 'BlobGranuleVerifier'
testDuration = 200.0
testDuration = 120.0