wbtiger
/
foundationdb
mirror of https://github.com/apple/foundationdb.git
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
foundationdb/fdbserver/workloads/RandomMoveKeys.cpp

289 lines
11 KiB
C++
Raw Permalink Blame History

/*
* RandomMoveKeys.cpp
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2026 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/FDBOptions.g.h"
#include "fdbrpc/simulator.h"
#include "fdbclient/StorageServerInterface.h"
#include "fdbclient/ManagementAPI.h"
#include "fdbserver/core/MoveKeys.h"
#include "fdbclient/NativeAPI.actor.h"
#include "fdbserver/tester/workloads.h"
#include "fdbserver/core/Knobs.h"
#include "fdbserver/core/ServerDBInfo.h"
#include "fdbserver/core/QuietDatabase.h"
#include "flow/DeterministicRandom.h"
struct MoveKeysWorkload : FailureInjectionWorkload {
static constexpr auto NAME = "RandomMoveKeys";
bool enabled;
double testDuration = 10.0, meanDelay = 0.05;
double maxKeyspace = 0.1;
DatabaseConfiguration configuration;
MoveKeysWorkload(WorkloadContext const& wcx, NoOptions) : FailureInjectionWorkload(wcx) {
enabled = !clientId && g_network->isSimulated(); // only do this on the "first" client
}
explicit MoveKeysWorkload(WorkloadContext const& wcx) : FailureInjectionWorkload(wcx) {
enabled = !clientId && g_network->isSimulated(); // only do this on the "first" client
meanDelay = getOption(options, "meanDelay"_sr, meanDelay);
testDuration = getOption(options, "testDuration"_sr, testDuration);
maxKeyspace = getOption(options, "maxKeyspace"_sr, maxKeyspace);
}
Future<Void> setup(Database const& cx) override { return Void(); }
bool shouldInject(DeterministicRandom& random,
const WorkloadRequest& work,
const unsigned alreadyAdded) const override {
return alreadyAdded < 1 && work.useDatabase && 0.1 / (1 + alreadyAdded) > random.random01();
}
Future<Void> start(Database const& cx) override {
if (enabled) {
// Get the database configuration so as to use proper team size
Transaction tr(cx);
while (true) {
tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS);
tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE);
Error err;
try {
RangeResult res = co_await tr.getRange(configKeys, 1000);
ASSERT(res.size() < 1000);
for (int i = 0; i < res.size(); i++)
configuration.set(res[i].key, res[i].value);
break;
} catch (Error& e) {
err = e;
}
co_await tr.onError(err);
}
int oldMode = co_await setDDMode(cx, 0);
TraceEvent("RMKStartModeSetting").log();
co_await timeout(reportErrors(worker(cx, this), "MoveKeysWorkloadWorkerError"), testDuration, Void());
// Always set the DD mode back, even if we die with an error
TraceEvent("RMKDoneMoving").log();
co_await setDDMode(cx, oldMode);
TraceEvent("RMKDoneModeSetting").log();
}
}
double getCheckTimeout() const override { return testDuration / 2 + 1; }
Future<bool> check(Database const& cx) override {
return tag(delay(testDuration / 2), true);
} // Give the database time to recover from our damage
void getMetrics(std::vector<PerfMetric>& m) override {}
KeyRange getRandomKeys() const {
double len = deterministicRandom()->random01() * this->maxKeyspace;
double pos = deterministicRandom()->random01() * (1.0 - len);
return KeyRangeRef(doubleToTestKey(pos), doubleToTestKey(pos + len));
}
std::vector<StorageServerInterface> getRandomTeam(std::vector<StorageServerInterface> storageServers,
int teamSize) {
if (storageServers.size() < teamSize) {
TraceEvent(SevWarnAlways, "LessThanThreeStorageServers").log();
throw operation_failed();
}
deterministicRandom()->randomShuffle(storageServers);
std::set<StorageServerInterface> t;
std::set<Optional<Standalone<StringRef>>> machines;
std::set<Optional<Standalone<StringRef>>> dataHalls;
while (t.size() < teamSize && !storageServers.empty()) {
auto s = storageServers.back();
storageServers.pop_back();
if (!machines.contains(s.locality.zoneId()) &&
(!s.locality.dataHallId().present() || !dataHalls.contains(s.locality.dataHallId()))) {
machines.insert(s.locality.zoneId());
dataHalls.insert(s.locality.dataHallId());
t.insert(s);
}
}
if (t.size() < teamSize) {
TraceEvent(SevWarnAlways, "LessThanUniqueMachines")
.detail("TargetTeamSize", teamSize)
.detail("TeamSelected", t.size());
throw operation_failed();
}
return std::vector<StorageServerInterface>(t.begin(), t.end());
}
Future<Void> doMoveKeys(Database cx,
MoveKeysWorkload* self,
KeyRange keys,
std::vector<StorageServerInterface> destinationTeam,
MoveKeysLock lock) {
TraceInterval relocateShardInterval("RelocateShard");
FlowLock fl1(1);
FlowLock fl2(1);
std::string desc;
for (int s = 0; s < destinationTeam.size(); s++)
desc +=
format("%s (%llx),", destinationTeam[s].address().toString().c_str(), destinationTeam[s].id().first());
std::vector<UID> destinationTeamIDs;
destinationTeamIDs.reserve(destinationTeam.size());
for (int s = 0; s < destinationTeam.size(); s++)
destinationTeamIDs.push_back(destinationTeam[s].id());
TraceEvent(relocateShardInterval.begin())
.detail("KeyBegin", printable(keys.begin))
.detail("KeyEnd", printable(keys.end))
.detail("Priority", 0)
.detail("Source", "RandomMoveKeys")
.detail("DestinationTeam", desc);
try {
Promise<Void> signal;
DDEnabledState ddEnabledState;
std::unique_ptr<MoveKeysParams> params;
if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) {
UID dataMoveId =
newDataMoveId(deterministicRandom()->randomUInt64(),
AssignEmptyRange(false),
deterministicRandom()->random01() < SERVER_KNOBS->DD_PHYSICAL_SHARD_MOVE_PROBABILITY
? DataMoveType::PHYSICAL
: DataMoveType::LOGICAL,
DataMovementReason::TEAM_HEALTHY,
UnassignShard(false));
params = std::make_unique<MoveKeysParams>(dataMoveId,
std::vector<KeyRange>{ keys },
destinationTeamIDs,
destinationTeamIDs,
lock,
signal,
&fl1,
&fl2,
false,
relocateShardInterval.pairID,
&ddEnabledState,
CancelConflictingDataMoves::True,
Optional<BulkLoadTaskState>());
} else {
UID dataMoveId = newDataMoveId(deterministicRandom()->randomUInt64(),
AssignEmptyRange(false),
DataMoveType::LOGICAL,
DataMovementReason::TEAM_HEALTHY,
UnassignShard(false));
params = std::make_unique<MoveKeysParams>(dataMoveId,
keys,
destinationTeamIDs,
destinationTeamIDs,
lock,
signal,
&fl1,
&fl2,
false,
relocateShardInterval.pairID,
&ddEnabledState,
CancelConflictingDataMoves::True,
Optional<BulkLoadTaskState>());
}
co_await moveKeys(cx, *params);
TraceEvent(relocateShardInterval.end()).detail("Result", "Success");
co_return;
} catch (Error& e) {
TraceEvent(relocateShardInterval.end(), self->dbInfo->get().master.id()).errorUnsuppressed(e);
throw;
}
}
static void eliminateDuplicates(std::vector<StorageServerInterface>& servers) {
// The real data distribution algorithm doesn't want to deal with multiple servers
// with the same address having keys. So if there are two servers with the same address,
// don't use either one (so we don't have to find out which of them, if any, already has keys).
// Also get rid of tss since we don't want to move a shard to a tss.
std::map<NetworkAddress, int> count;
for (int s = 0; s < servers.size(); s++)
count[servers[s].address()]++;
int o = 0;
for (int s = 0; s < servers.size(); s++)
if (count[servers[s].address()] == 1 && !servers[s].isTss())
servers[o++] = servers[s];
servers.resize(o);
}
Future<Void> forceMasterFailure(Database cx, MoveKeysWorkload* self) {
ASSERT(g_network->isSimulated());
while (true) {
if (g_simulator->killZone(self->dbInfo->get().master.locality.zoneId(), ISimulator::KillType::Reboot, true))
co_return;
co_await delay(1.0);
}
}
Future<Void> worker(Database cx, MoveKeysWorkload* self) {
KeyRangeMap<std::vector<StorageServerInterface>> inFlight;
KeyRangeActorMap inFlightActors;
double lastTime = now();
ASSERT(self->configuration.storageTeamSize > 0);
if (self->configuration.usableRegions > 1) { // FIXME: add support for generating random teams across DCs
co_return;
}
while (true) {
Error err;
try {
MoveKeysLock lock = co_await takeMoveKeysLock(cx, UID());
std::vector<StorageServerInterface> storageServers = co_await getStorageServers(cx);
eliminateDuplicates(storageServers);
while (true) {
co_await poisson(&lastTime, self->meanDelay);
KeyRange keys = self->getRandomKeys();
std::vector<StorageServerInterface> team =
self->getRandomTeam(storageServers, self->configuration.storageTeamSize);
// update both inFlightActors and inFlight key range maps, cancelling deleted RelocateShards
std::vector<KeyRange> ranges;
inFlightActors.getRangesAffectedByInsertion(keys, ranges);
inFlightActors.cancel(KeyRangeRef(ranges.front().begin, ranges.back().end));
inFlight.insert(keys, team);
for (int r = 0; r < ranges.size(); r++) {
auto& rTeam = inFlight.rangeContaining(ranges[r].begin)->value();
inFlightActors.insert(ranges[r], self->doMoveKeys(cx, self, ranges[r], rTeam, lock));
}
}
} catch (Error& e) {
err = e;
}
if (!err.isValid()) {
continue;
}
if (err.code() != error_code_movekeys_conflict && err.code() != error_code_operation_failed)
throw err;
co_await delay(FLOW_KNOBS->PREVENT_FAST_SPIN_DELAY);
// Keep trying to get the moveKeysLock
}
}
};
WorkloadFactory<MoveKeysWorkload> MoveKeysWorkloadFactory;
FailureInjectorFactory<MoveKeysWorkload> MoveKeysFailureInjectionFactory;
Reference in New Issue View Git Blame Copy Permalink