Merge branch 'release-6.2' of github.com:apple/foundationdb
Needed to pull in changes to build docker
This commit is contained in:
Alvin Moore
commit
0f64505d0b
|
|
@ -817,9 +817,9 @@ public class DirectoryLayer implements Directory {
|
|||
|
||||
private static long unpackLittleEndian(byte[] bytes) {
|
||||
assert bytes.length == 8;
|
||||
int value = 0;
|
||||
long value = 0;
|
||||
for(int i = 0; i < 8; ++i) {
|
||||
value += (bytes[i] << (i * 8));
|
||||
value += (Byte.toUnsignedLong(bytes[i]) << (i * 8));
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,4 @@
|
|||
FROM centos:6
|
||||
LABEL version=0.1.9
|
||||
ENV DOCKER_IMAGEVER=0.1.9
|
||||
|
||||
# Install dependencies for developer tools, bindings,\
|
||||
# documentation, actorcompiler, and packaging tools\
|
||||
|
|
@ -8,9 +6,10 @@ RUN yum install -y yum-utils &&\
|
|||
yum-config-manager --enable rhel-server-rhscl-7-rpms &&\
|
||||
yum -y install centos-release-scl epel-release &&\
|
||||
yum -y install devtoolset-8-8.1-1.el6 java-1.8.0-openjdk-devel \
|
||||
devtoolset-8-gcc-8.3.1-3.1.el6 devtoolset-8-gcc-c++-8.3.1-3.1.el6 \
|
||||
rh-python36-python-devel devtoolset-8-valgrind-devel \
|
||||
mono-core rh-ruby24 golang python27 rpm-build debbuild \
|
||||
python-pip npm dos2unix valgrind-devel ccache distcc devtoolset-8-libubsan-devel libubsan-devel &&\
|
||||
python-pip dos2unix valgrind-devel ccache distcc devtoolset-8-libubsan-devel libubsan-devel &&\
|
||||
pip install boto3==1.1.1
|
||||
|
||||
USER root
|
||||
|
|
@ -19,32 +18,42 @@ RUN adduser --comment '' fdb && chown fdb /opt
|
|||
|
||||
# wget of bintray without forcing UTF-8 encoding results in 403 Forbidden
|
||||
RUN cd /opt/ &&\
|
||||
curl -L https://dl.bintray.com/boostorg/release/1.67.0/source/boost_1_67_0.tar.bz2 > boost_1_67_0.tar.bz2 &&\
|
||||
echo "2684c972994ee57fc5632e03bf044746f6eb45d4920c343937a465fd67a5adba boost_1_67_0.tar.bz2" > boost-sha.txt &&\
|
||||
sha256sum -c boost-sha.txt &&\
|
||||
curl -L https://dl.bintray.com/boostorg/release/1.67.0/source/boost_1_67_0.tar.bz2 -o boost_1_67_0.tar.bz2 &&\
|
||||
echo "2684c972994ee57fc5632e03bf044746f6eb45d4920c343937a465fd67a5adba boost_1_67_0.tar.bz2" > boost-sha-67.txt &&\
|
||||
sha256sum -c boost-sha-67.txt &&\
|
||||
tar -xjf boost_1_67_0.tar.bz2 &&\
|
||||
rm -rf boost_1_67_0.tar.bz2 boost-sha.txt boost_1_67_0/libs
|
||||
rm -rf boost_1_67_0.tar.bz2 boost-sha-67.txt boost_1_67_0/libs &&\
|
||||
curl -L https://dl.bintray.com/boostorg/release/1.72.0/source/boost_1_72_0.tar.bz2 -o boost_1_72_0.tar.bz2 &&\
|
||||
echo "59c9b274bc451cf91a9ba1dd2c7fdcaf5d60b1b3aa83f2c9fa143417cc660722 boost_1_72_0.tar.bz2" > boost-sha-72.txt &&\
|
||||
sha256sum -c boost-sha-72.txt &&\
|
||||
tar -xjf boost_1_72_0.tar.bz2 &&\
|
||||
rm -rf boost_1_72_0.tar.bz2 boost-sha-72.txt boost_1_72_0/libs
|
||||
|
||||
# install cmake
|
||||
RUN curl -L https://github.com/Kitware/CMake/releases/download/v3.13.4/cmake-3.13.4-Linux-x86_64.tar.gz > /tmp/cmake.tar.gz &&\
|
||||
RUN curl -L https://github.com/Kitware/CMake/releases/download/v3.13.4/cmake-3.13.4-Linux-x86_64.tar.gz -o /tmp/cmake.tar.gz &&\
|
||||
echo "563a39e0a7c7368f81bfa1c3aff8b590a0617cdfe51177ddc808f66cc0866c76 /tmp/cmake.tar.gz" > /tmp/cmake-sha.txt &&\
|
||||
sha256sum -c /tmp/cmake-sha.txt &&\
|
||||
cd /tmp && tar xf cmake.tar.gz &&\
|
||||
cp -r cmake-3.13.4-Linux-x86_64/* /usr/local/ &&\
|
||||
rm -rf cmake.tar.gz cmake-3.13.4-Linux-x86_64 cmake-sha.txt
|
||||
|
||||
# install LibreSSL
|
||||
RUN cd /tmp && curl -L https://github.com/ninja-build/ninja/archive/v1.9.0.zip > ninja.zip &&\
|
||||
# install Ninja
|
||||
RUN cd /tmp && curl -L https://github.com/ninja-build/ninja/archive/v1.9.0.zip -o ninja.zip &&\
|
||||
unzip ninja.zip && cd ninja-1.9.0 && scl enable devtoolset-8 -- ./configure.py --bootstrap && cp ninja /usr/bin &&\
|
||||
cd .. && rm -rf ninja-1.9.0 ninja.zip &&\
|
||||
curl -L https://ftp.openbsd.org/pub/OpenBSD/LibreSSL/libressl-2.8.2.tar.gz > /tmp/libressl.tar.gz &&\
|
||||
cd /tmp && echo "b8cb31e59f1294557bfc80f2a662969bc064e83006ceef0574e2553a1c254fd5 libressl.tar.gz" > libressl-sha.txt &&\
|
||||
sha256sum -c libressl-sha.txt && tar xf libressl.tar.gz &&\
|
||||
cd libressl-2.8.2 && cd /tmp/libressl-2.8.2 && scl enable devtoolset-8 -- ./configure --prefix=/usr/local/stow/libressl CFLAGS="-fPIC -O3" --prefix=/usr/local &&\
|
||||
cd /tmp/libressl-2.8.2 && scl enable devtoolset-8 -- make -j`nproc` install &&\
|
||||
rm -rf /tmp/libressl-2.8.2 /tmp/libressl.tar.gz
|
||||
cd .. && rm -rf ninja-1.9.0 ninja.zip
|
||||
|
||||
# install openssl
|
||||
RUN cd /tmp && curl -L https://www.openssl.org/source/openssl-1.1.1d.tar.gz -o openssl.tar.gz &&\
|
||||
echo "1e3a91bc1f9dfce01af26026f856e064eab4c8ee0a8f457b5ae30b40b8b711f2 openssl.tar.gz" > openssl-sha.txt &&\
|
||||
sha256sum -c openssl-sha.txt && tar -xzf openssl.tar.gz &&\
|
||||
cd openssl-1.1.1d && scl enable devtoolset-8 -- ./config CFLAGS="-fPIC -O3" --prefix=/usr/local &&\
|
||||
scl enable devtoolset-8 -- make -j`nproc` && scl enable devtoolset-8 -- make -j1 install &&\
|
||||
ln -sv /usr/local/lib64/lib*.so.1.1 /usr/lib64/ &&\
|
||||
cd /tmp/ && rm -rf /tmp/openssl-1.1.1d /tmp/openssl.tar.gz
|
||||
|
||||
LABEL version=0.1.12
|
||||
ENV DOCKER_IMAGEVER=0.1.12
|
||||
ENV JAVA_HOME=/usr/lib/jvm/java-1.8.0
|
||||
ENV CC=/opt/rh/devtoolset-8/root/usr/bin/gcc
|
||||
ENV CXX=/opt/rh/devtoolset-8/root/usr/bin/g++
|
||||
CMD scl enable devtoolset-8 python27 rh-python36 rh-ruby24 -- bash
|
||||
CMD scl enable devtoolset-8 rh-python36 rh-ruby24 -- bash
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ version: "3"
|
|||
|
||||
services:
|
||||
common: &common
|
||||
image: foundationdb/foundationdb-build:0.1.11
|
||||
image: foundationdb/foundationdb-build:0.1.12
|
||||
|
||||
build-setup: &build-setup
|
||||
<<: *common
|
||||
|
|
@ -36,11 +36,11 @@ services:
|
|||
|
||||
release-packages: &release-packages
|
||||
<<: *release-setup
|
||||
command: scl enable devtoolset-8 python27 rh-python36 rh-ruby24 -- bash -c 'make -j "$${MAKEJOBS}" packages'
|
||||
command: scl enable devtoolset-8 rh-python36 rh-ruby24 -- bash -c 'make -j "$${MAKEJOBS}" packages'
|
||||
|
||||
snapshot-packages: &snapshot-packages
|
||||
<<: *build-setup
|
||||
command: scl enable devtoolset-8 python27 rh-python36 rh-ruby24 -- bash -c 'make -j "$${MAKEJOBS}" packages'
|
||||
command: scl enable devtoolset-8 rh-python36 rh-ruby24 -- bash -c 'make -j "$${MAKEJOBS}" packages'
|
||||
|
||||
prb-packages:
|
||||
<<: *snapshot-packages
|
||||
|
|
@ -48,11 +48,11 @@ services:
|
|||
|
||||
release-bindings: &release-bindings
|
||||
<<: *release-setup
|
||||
command: scl enable devtoolset-8 python27 rh-python36 rh-ruby24 -- bash -c 'make -j "$${MAKEJOBS}" bindings'
|
||||
command: scl enable devtoolset-8 rh-python36 rh-ruby24 -- bash -c 'make -j "$${MAKEJOBS}" bindings'
|
||||
|
||||
snapshot-bindings: &snapshot-bindings
|
||||
<<: *build-setup
|
||||
command: scl enable devtoolset-8 python27 rh-python36 rh-ruby24 -- bash -c 'make -j "$${MAKEJOBS}" bindings'
|
||||
command: scl enable devtoolset-8 rh-python36 rh-ruby24 -- bash -c 'make -j "$${MAKEJOBS}" bindings'
|
||||
|
||||
prb-bindings:
|
||||
<<: *snapshot-bindings
|
||||
|
|
@ -60,7 +60,7 @@ services:
|
|||
|
||||
snapshot-cmake: &snapshot-cmake
|
||||
<<: *build-setup
|
||||
command: scl enable devtoolset-8 python27 rh-python36 rh-ruby24 -- bash -c 'mkdir -p "$${BUILD_DIR}" && cd "$${BUILD_DIR}" && cmake -G "Ninja" -DCMAKE_COLOR_MAKEFILE=0 -DUSE_WERROR=1 -DFDB_RELEASE=0 -DVALGRIND=0 /__this_is_some_very_long_name_dir_needed_to_fix_a_bug_with_debug_rpms__/foundationdb && ninja -v -j "$${MAKEJOBS}" "packages" "strip_targets" && cpack'
|
||||
command: scl enable devtoolset-8 rh-python36 rh-ruby24 -- bash -c 'mkdir -p "$${BUILD_DIR}" && cd "$${BUILD_DIR}" && cmake -G "Ninja" -DCMAKE_COLOR_MAKEFILE=0 -DFDB_RELEASE=0 -DVALGRIND=0 /__this_is_some_very_long_name_dir_needed_to_fix_a_bug_with_debug_rpms__/foundationdb && ninja -v -j "$${MAKEJOBS}" "packages" "strip_targets" && cpack'
|
||||
|
||||
prb-cmake:
|
||||
<<: *snapshot-cmake
|
||||
|
|
@ -68,7 +68,7 @@ services:
|
|||
|
||||
snapshot-ctest: &snapshot-ctest
|
||||
<<: *build-setup
|
||||
command: scl enable devtoolset-8 python27 rh-python36 rh-ruby24 -- bash -c 'mkdir -p "$${BUILD_DIR}" && cd "$${BUILD_DIR}" && cmake -G "Ninja" -DCMAKE_COLOR_MAKEFILE=0 -DUSE_WERROR=1 -DFDB_RELEASE=1 /__this_is_some_very_long_name_dir_needed_to_fix_a_bug_with_debug_rpms__/foundationdb && ninja -v -j "$${MAKEJOBS}" && ctest -L fast -j "$${MAKEJOBS}" --output-on-failure'
|
||||
command: scl enable devtoolset-8 rh-python36 rh-ruby24 -- bash -c 'mkdir -p "$${BUILD_DIR}" && cd "$${BUILD_DIR}" && cmake -G "Ninja" -DCMAKE_COLOR_MAKEFILE=0 -DFDB_RELEASE=1 /__this_is_some_very_long_name_dir_needed_to_fix_a_bug_with_debug_rpms__/foundationdb && ninja -v -j "$${MAKEJOBS}" && ctest -L fast -j "$${MAKEJOBS}" --output-on-failure'
|
||||
|
||||
prb-ctest:
|
||||
<<: *snapshot-ctest
|
||||
|
|
@ -76,7 +76,7 @@ services:
|
|||
|
||||
snapshot-correctness: &snapshot-correctness
|
||||
<<: *build-setup
|
||||
command: scl enable devtoolset-8 python27 rh-python36 rh-ruby24 -- bash -c 'mkdir -p "$${BUILD_DIR}" && cd "$${BUILD_DIR}" && cmake -G "Ninja" -DCMAKE_COLOR_MAKEFILE=0 -DUSE_WERROR=1 -DFDB_RELEASE=1 /__this_is_some_very_long_name_dir_needed_to_fix_a_bug_with_debug_rpms__/foundationdb && ninja -v -j "$${MAKEJOBS}" && ctest -j "$${MAKEJOBS}" --output-on-failure'
|
||||
command: scl enable devtoolset-8 rh-python36 rh-ruby24 -- bash -c 'mkdir -p "$${BUILD_DIR}" && cd "$${BUILD_DIR}" && cmake -G "Ninja" -DCMAKE_COLOR_MAKEFILE=0 -DFDB_RELEASE=1 /__this_is_some_very_long_name_dir_needed_to_fix_a_bug_with_debug_rpms__/foundationdb && ninja -v -j "$${MAKEJOBS}" && ctest -j "$${MAKEJOBS}" --output-on-failure'
|
||||
|
||||
prb-correctness:
|
||||
<<: *snapshot-correctness
|
||||
|
|
|
|||
|
|
@ -2546,8 +2546,8 @@ ACTOR void checkWrites( Database cx, Future<Void> committed, Promise<Void> outCo
|
|||
} else {
|
||||
Optional<Value> val = wait( tr.get( it->range().begin ) );
|
||||
if( !val.present() || val.get() != m.setValue ) {
|
||||
TraceEvent evt = TraceEvent(SevError, "CheckWritesFailed")
|
||||
.detail("Class", "Set")
|
||||
TraceEvent evt(SevError, "CheckWritesFailed");
|
||||
evt.detail("Class", "Set")
|
||||
.detail("Key", it->range().begin)
|
||||
.detail("Expected", m.setValue);
|
||||
if( !val.present() )
|
||||
|
|
|
|||
|
|
@ -1136,8 +1136,10 @@ ACTOR Future<Void> dataDistributionRelocator( DDQueueData *self, RelocateData rd
|
|||
}
|
||||
|
||||
// Move a random shard of sourceTeam's to destTeam if sourceTeam has much more data than destTeam
|
||||
ACTOR Future<bool> rebalanceTeams( DDQueueData* self, int priority, Reference<IDataDistributionTeam> sourceTeam, Reference<IDataDistributionTeam> destTeam, bool primary ) {
|
||||
ACTOR Future<bool> rebalanceTeams( DDQueueData* self, int priority, Reference<IDataDistributionTeam> sourceTeam,
|
||||
Reference<IDataDistributionTeam> destTeam, bool primary, TraceEvent *traceEvent ) {
|
||||
if(g_network->isSimulated() && g_simulator.speedUpSimulation) {
|
||||
traceEvent->detail("CancelingDueToSimulationSpeedup", true);
|
||||
return false;
|
||||
}
|
||||
|
||||
|
|
@ -1147,6 +1149,9 @@ ACTOR Future<bool> rebalanceTeams( DDQueueData* self, int priority, Reference<ID
|
|||
state int64_t averageShardBytes = wait(req.getFuture());
|
||||
state std::vector<KeyRange> shards = self->shardsAffectedByTeamFailure->getShardsFor( ShardsAffectedByTeamFailure::Team( sourceTeam->getServerIDs(), primary ) );
|
||||
|
||||
traceEvent->detail("AverageShardBytes", averageShardBytes)
|
||||
.detail("ShardsInSource", shards.size());
|
||||
|
||||
if( !shards.size() )
|
||||
return false;
|
||||
|
||||
|
|
@ -1168,28 +1173,28 @@ ACTOR Future<bool> rebalanceTeams( DDQueueData* self, int priority, Reference<ID
|
|||
|
||||
int64_t sourceBytes = sourceTeam->getLoadBytes(false);
|
||||
int64_t destBytes = destTeam->getLoadBytes();
|
||||
if( sourceBytes - destBytes <= 3 * std::max<int64_t>( SERVER_KNOBS->MIN_SHARD_BYTES, metrics.bytes ) || metrics.bytes == 0 )
|
||||
|
||||
bool sourceAndDestTooSimilar = sourceBytes - destBytes <= 3 * std::max<int64_t>(SERVER_KNOBS->MIN_SHARD_BYTES, metrics.bytes);
|
||||
traceEvent->detail("SourceBytes", sourceBytes)
|
||||
.detail("DestBytes", destBytes)
|
||||
.detail("ShardBytes", metrics.bytes)
|
||||
.detail("SourceAndDestTooSimilar", sourceAndDestTooSimilar);
|
||||
|
||||
if( sourceAndDestTooSimilar || metrics.bytes == 0 ) {
|
||||
return false;
|
||||
}
|
||||
|
||||
{
|
||||
//verify the shard is still in sabtf
|
||||
std::vector<KeyRange> shards = self->shardsAffectedByTeamFailure->getShardsFor( ShardsAffectedByTeamFailure::Team( sourceTeam->getServerIDs(), primary ) );
|
||||
for( int i = 0; i < shards.size(); i++ ) {
|
||||
if( moveShard == shards[i] ) {
|
||||
TraceEvent(priority == SERVER_KNOBS->PRIORITY_REBALANCE_OVERUTILIZED_TEAM ? "BgDDMountainChopper" : "BgDDValleyFiller", self->distributorId)
|
||||
.detail("SourceBytes", sourceBytes)
|
||||
.detail("DestBytes", destBytes)
|
||||
.detail("ShardBytes", metrics.bytes)
|
||||
.detail("AverageShardBytes", averageShardBytes)
|
||||
.detail("SourceTeam", sourceTeam->getDesc())
|
||||
.detail("DestTeam", destTeam->getDesc());
|
||||
|
||||
self->output.send( RelocateShard( moveShard, priority ) );
|
||||
return true;
|
||||
}
|
||||
//verify the shard is still in sabtf
|
||||
shards = self->shardsAffectedByTeamFailure->getShardsFor( ShardsAffectedByTeamFailure::Team( sourceTeam->getServerIDs(), primary ) );
|
||||
for( int i = 0; i < shards.size(); i++ ) {
|
||||
if( moveShard == shards[i] ) {
|
||||
traceEvent->detail("ShardStillPresent", true);
|
||||
self->output.send( RelocateShard( moveShard, priority ) );
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
traceEvent->detail("ShardStillPresent", false);
|
||||
return false;
|
||||
}
|
||||
|
||||
|
|
@ -1200,6 +1205,15 @@ ACTOR Future<Void> BgDDMountainChopper( DDQueueData* self, int teamCollectionInd
|
|||
state double lastRead = 0;
|
||||
state bool skipCurrentLoop = false;
|
||||
loop {
|
||||
state bool moved = false;
|
||||
state TraceEvent traceEvent("BgDDMountainChopper", self->distributorId);
|
||||
traceEvent.suppressFor(5.0)
|
||||
.detail("PollingInterval", rebalancePollingInterval);
|
||||
|
||||
if(*self->lastLimited > 0) {
|
||||
traceEvent.detail("SecondsSinceLastLimited", now() - *self->lastLimited);
|
||||
}
|
||||
|
||||
try {
|
||||
state Future<Void> delayF = delay(rebalancePollingInterval, TaskPriority::DataDistributionLaunch);
|
||||
if ((now() - lastRead) > SERVER_KNOBS->BG_REBALANCE_SWITCH_CHECK_INTERVAL) {
|
||||
|
|
@ -1212,6 +1226,9 @@ ACTOR Future<Void> BgDDMountainChopper( DDQueueData* self, int teamCollectionInd
|
|||
}
|
||||
skipCurrentLoop = val.present();
|
||||
}
|
||||
|
||||
traceEvent.detail("Enabled", !skipCurrentLoop);
|
||||
|
||||
wait(delayF);
|
||||
if (skipCurrentLoop) {
|
||||
// set loop interval to avoid busy wait here.
|
||||
|
|
@ -1219,18 +1236,31 @@ ACTOR Future<Void> BgDDMountainChopper( DDQueueData* self, int teamCollectionInd
|
|||
std::max(rebalancePollingInterval, SERVER_KNOBS->BG_REBALANCE_SWITCH_CHECK_INTERVAL);
|
||||
continue;
|
||||
}
|
||||
|
||||
traceEvent.detail("QueuedRelocations", self->priority_relocations[SERVER_KNOBS->PRIORITY_REBALANCE_OVERUTILIZED_TEAM]);
|
||||
if (self->priority_relocations[SERVER_KNOBS->PRIORITY_REBALANCE_OVERUTILIZED_TEAM] <
|
||||
SERVER_KNOBS->DD_REBALANCE_PARALLELISM) {
|
||||
state Optional<Reference<IDataDistributionTeam>> randomTeam = wait(brokenPromiseToNever(
|
||||
self->teamCollections[teamCollectionIndex].getTeam.getReply(GetTeamRequest(true, false, true, false))));
|
||||
|
||||
traceEvent.detail("DestTeam", printable(randomTeam.map<std::string>([](const Reference<IDataDistributionTeam>& team){
|
||||
return team->getDesc();
|
||||
})));
|
||||
|
||||
if (randomTeam.present()) {
|
||||
state Optional<Reference<IDataDistributionTeam>> loadedTeam =
|
||||
wait(brokenPromiseToNever(self->teamCollections[teamCollectionIndex].getTeam.getReply(
|
||||
GetTeamRequest(true, true, false, true))));
|
||||
|
||||
traceEvent.detail("SourceTeam", printable(loadedTeam.map<std::string>([](const Reference<IDataDistributionTeam>& team){
|
||||
return team->getDesc();
|
||||
})));
|
||||
|
||||
if (loadedTeam.present()) {
|
||||
bool moved =
|
||||
bool _moved =
|
||||
wait(rebalanceTeams(self, SERVER_KNOBS->PRIORITY_REBALANCE_OVERUTILIZED_TEAM, loadedTeam.get(),
|
||||
randomTeam.get(), teamCollectionIndex == 0));
|
||||
randomTeam.get(), teamCollectionIndex == 0, &traceEvent));
|
||||
moved = _moved;
|
||||
if (moved) {
|
||||
resetCount = 0;
|
||||
} else {
|
||||
|
|
@ -1253,10 +1283,16 @@ ACTOR Future<Void> BgDDMountainChopper( DDQueueData* self, int teamCollectionInd
|
|||
rebalancePollingInterval = SERVER_KNOBS->BG_REBALANCE_POLLING_INTERVAL;
|
||||
resetCount = SERVER_KNOBS->DD_REBALANCE_RESET_AMOUNT;
|
||||
}
|
||||
|
||||
traceEvent.detail("ResetCount", resetCount);
|
||||
tr.reset();
|
||||
} catch (Error& e) {
|
||||
traceEvent.error(e, true); // Log actor_cancelled because it's not legal to suppress an event that's initialized
|
||||
wait(tr.onError(e));
|
||||
}
|
||||
|
||||
traceEvent.detail("Moved", moved);
|
||||
traceEvent.log();
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1267,6 +1303,15 @@ ACTOR Future<Void> BgDDValleyFiller( DDQueueData* self, int teamCollectionIndex)
|
|||
state double lastRead = 0;
|
||||
state bool skipCurrentLoop = false;
|
||||
loop {
|
||||
state bool moved = false;
|
||||
state TraceEvent traceEvent("BgDDValleyFiller", self->distributorId);
|
||||
traceEvent.suppressFor(5.0)
|
||||
.detail("PollingInterval", rebalancePollingInterval);
|
||||
|
||||
if(*self->lastLimited > 0) {
|
||||
traceEvent.detail("SecondsSinceLastLimited", now() - *self->lastLimited);
|
||||
}
|
||||
|
||||
try {
|
||||
state Future<Void> delayF = delay(rebalancePollingInterval, TaskPriority::DataDistributionLaunch);
|
||||
if ((now() - lastRead) > SERVER_KNOBS->BG_REBALANCE_SWITCH_CHECK_INTERVAL) {
|
||||
|
|
@ -1279,6 +1324,9 @@ ACTOR Future<Void> BgDDValleyFiller( DDQueueData* self, int teamCollectionIndex)
|
|||
}
|
||||
skipCurrentLoop = val.present();
|
||||
}
|
||||
|
||||
traceEvent.detail("Enabled", !skipCurrentLoop);
|
||||
|
||||
wait(delayF);
|
||||
if (skipCurrentLoop) {
|
||||
// set loop interval to avoid busy wait here.
|
||||
|
|
@ -1286,17 +1334,30 @@ ACTOR Future<Void> BgDDValleyFiller( DDQueueData* self, int teamCollectionIndex)
|
|||
std::max(rebalancePollingInterval, SERVER_KNOBS->BG_REBALANCE_SWITCH_CHECK_INTERVAL);
|
||||
continue;
|
||||
}
|
||||
|
||||
traceEvent.detail("QueuedRelocations", self->priority_relocations[SERVER_KNOBS->PRIORITY_REBALANCE_UNDERUTILIZED_TEAM]);
|
||||
if (self->priority_relocations[SERVER_KNOBS->PRIORITY_REBALANCE_UNDERUTILIZED_TEAM] <
|
||||
SERVER_KNOBS->DD_REBALANCE_PARALLELISM) {
|
||||
state Optional<Reference<IDataDistributionTeam>> randomTeam = wait(brokenPromiseToNever(
|
||||
self->teamCollections[teamCollectionIndex].getTeam.getReply(GetTeamRequest(true, false, false, true))));
|
||||
|
||||
traceEvent.detail("SourceTeam", printable(randomTeam.map<std::string>([](const Reference<IDataDistributionTeam>& team){
|
||||
return team->getDesc();
|
||||
})));
|
||||
|
||||
if (randomTeam.present()) {
|
||||
state Optional<Reference<IDataDistributionTeam>> unloadedTeam = wait(brokenPromiseToNever(
|
||||
self->teamCollections[teamCollectionIndex].getTeam.getReply(GetTeamRequest(true, true, true, false))));
|
||||
|
||||
traceEvent.detail("DestTeam", printable(unloadedTeam.map<std::string>([](const Reference<IDataDistributionTeam>& team){
|
||||
return team->getDesc();
|
||||
})));
|
||||
|
||||
if (unloadedTeam.present()) {
|
||||
bool moved =
|
||||
bool _moved =
|
||||
wait(rebalanceTeams(self, SERVER_KNOBS->PRIORITY_REBALANCE_UNDERUTILIZED_TEAM, randomTeam.get(),
|
||||
unloadedTeam.get(), teamCollectionIndex == 0));
|
||||
unloadedTeam.get(), teamCollectionIndex == 0, &traceEvent));
|
||||
moved = _moved;
|
||||
if (moved) {
|
||||
resetCount = 0;
|
||||
} else {
|
||||
|
|
@ -1319,10 +1380,16 @@ ACTOR Future<Void> BgDDValleyFiller( DDQueueData* self, int teamCollectionIndex)
|
|||
rebalancePollingInterval = SERVER_KNOBS->BG_REBALANCE_POLLING_INTERVAL;
|
||||
resetCount = SERVER_KNOBS->DD_REBALANCE_RESET_AMOUNT;
|
||||
}
|
||||
|
||||
traceEvent.detail("ResetCount", resetCount);
|
||||
tr.reset();
|
||||
} catch (Error& e) {
|
||||
traceEvent.error(e, true); // Log actor_cancelled because it's not legal to suppress an event that's initialized
|
||||
wait(tr.onError(e));
|
||||
}
|
||||
|
||||
traceEvent.detail("Moved", moved);
|
||||
traceEvent.log();
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -684,6 +684,50 @@ void removeTraceRole(std::string role) {
|
|||
g_traceLog.removeRole(role);
|
||||
}
|
||||
|
||||
TraceEvent::TraceEvent() : initialized(true), enabled(false), logged(true) {}
|
||||
|
||||
TraceEvent::TraceEvent(TraceEvent &&ev) {
|
||||
enabled = ev.enabled;
|
||||
err = ev.err;
|
||||
fields = std::move(ev.fields);
|
||||
id = ev.id;
|
||||
initialized = ev.initialized;
|
||||
logged = ev.logged;
|
||||
maxEventLength = ev.maxEventLength;
|
||||
maxFieldLength = ev.maxFieldLength;
|
||||
severity = ev.severity;
|
||||
tmpEventMetric = ev.tmpEventMetric;
|
||||
trackingKey = ev.trackingKey;
|
||||
type = ev.type;
|
||||
|
||||
ev.initialized = true;
|
||||
ev.enabled = false;
|
||||
ev.logged = true;
|
||||
ev.tmpEventMetric = nullptr;
|
||||
}
|
||||
|
||||
TraceEvent& TraceEvent::operator=(TraceEvent &&ev) {
|
||||
enabled = ev.enabled;
|
||||
err = ev.err;
|
||||
fields = std::move(ev.fields);
|
||||
id = ev.id;
|
||||
initialized = ev.initialized;
|
||||
logged = ev.logged;
|
||||
maxEventLength = ev.maxEventLength;
|
||||
maxFieldLength = ev.maxFieldLength;
|
||||
severity = ev.severity;
|
||||
tmpEventMetric = ev.tmpEventMetric;
|
||||
trackingKey = ev.trackingKey;
|
||||
type = ev.type;
|
||||
|
||||
ev.initialized = true;
|
||||
ev.enabled = false;
|
||||
ev.logged = true;
|
||||
ev.tmpEventMetric = nullptr;
|
||||
|
||||
return *this;
|
||||
}
|
||||
|
||||
TraceEvent::TraceEvent( const char* type, UID id ) : id(id), type(type), severity(SevInfo), initialized(false), enabled(true), logged(false) {
|
||||
g_trace_depth++;
|
||||
setMaxFieldLength(0);
|
||||
|
|
@ -760,7 +804,9 @@ bool TraceEvent::init() {
|
|||
}
|
||||
|
||||
detail("Severity", int(severity));
|
||||
detailf("Time", "%.6f", getCurrentTime());
|
||||
detail("Time", "0.000000");
|
||||
timeIndex = fields.size() - 1;
|
||||
|
||||
detail("Type", type);
|
||||
if(g_network && g_network->isSimulated()) {
|
||||
NetworkAddress local = g_network->getLocalAddress();
|
||||
|
|
@ -968,6 +1014,8 @@ void TraceEvent::log() {
|
|||
init();
|
||||
try {
|
||||
if (enabled) {
|
||||
fields.mutate(timeIndex).second = format("%.6f", TraceEvent::getCurrentTime());
|
||||
|
||||
if (this->severity == SevError) {
|
||||
severity = SevInfo;
|
||||
backtrace();
|
||||
|
|
@ -1181,6 +1229,10 @@ std::string TraceEventFields::getValue(std::string key) const {
|
|||
}
|
||||
}
|
||||
|
||||
TraceEventFields::Field& TraceEventFields::mutate(int index) {
|
||||
return fields.at(index);
|
||||
}
|
||||
|
||||
namespace {
|
||||
void parseNumericValue(std::string const& s, double &outValue, bool permissive = false) {
|
||||
double d = 0;
|
||||
|
|
@ -1306,6 +1358,9 @@ void TraceEventFields::validateFormat() const {
|
|||
}
|
||||
|
||||
std::string traceableStringToString(const char* value, size_t S) {
|
||||
ASSERT_WE_THINK(S > 0 && value[S - 1] == '\0');
|
||||
if(g_network) {
|
||||
ASSERT_WE_THINK(S > 0 && value[S - 1] == '\0');
|
||||
}
|
||||
|
||||
return std::string(value, S - 1); // Exclude trailing \0 byte
|
||||
}
|
||||
|
|
|
|||
|
|
@ -81,6 +81,8 @@ public:
|
|||
int64_t getInt64(std::string key, bool permissive=false) const;
|
||||
double getDouble(std::string key, bool permissive=false) const;
|
||||
|
||||
Field &mutate(int index);
|
||||
|
||||
std::string toString() const;
|
||||
void validateFormat() const;
|
||||
template<class Archiver>
|
||||
|
|
@ -374,11 +376,15 @@ struct SpecialTraceMetricType
|
|||
TRACE_METRIC_TYPE(double, double);
|
||||
|
||||
struct TraceEvent {
|
||||
TraceEvent();
|
||||
TraceEvent( const char* type, UID id = UID() ); // Assumes SevInfo severity
|
||||
TraceEvent( Severity, const char* type, UID id = UID() );
|
||||
TraceEvent( struct TraceInterval&, UID id = UID() );
|
||||
TraceEvent( Severity severity, struct TraceInterval& interval, UID id = UID() );
|
||||
|
||||
TraceEvent( TraceEvent &&ev );
|
||||
TraceEvent& operator=( TraceEvent &&ev );
|
||||
|
||||
static void setNetworkThread();
|
||||
static bool isNetworkThread();
|
||||
|
||||
|
|
@ -490,6 +496,7 @@ private:
|
|||
|
||||
int maxFieldLength;
|
||||
int maxEventLength;
|
||||
int timeIndex;
|
||||
|
||||
void setSizeLimits();
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue