diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index b0fa724b955..8183584f3f3 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -92,72 +92,26 @@ jobs: test_deploy: needs: [calc_ver, build] runs-on: ubuntu-latest - env: - RHEL_IMAGE_NAME: "registry.access.redhat.com/ubi8/ubi:latest" - UBUNTU_IMAGE_NAME: "ubuntu:18.04" steps: - - name: Download RPMs and DEBs packages - uses: actions/download-artifact@v3 - id: download_packages - - - name: Echo download path - run: echo ${{steps.download_packages.outputs.download-path}} - - - name: Display structure of downloaded files - run: ls -R - working-directory: ${{github.workspace}} + - name: Checkout + uses: actions/checkout@v3 - - name: Download RHEL and Ubuntu images - run: | - podman pull ${{ env.RHEL_IMAGE_NAME }} - podman pull ${{ env.UBUNTU_IMAGE_NAME }} + - name: Download the foundationdb distro + uses: actions/download-artifact@v3 + with: + path: ${{github.workspace}}/bld/download - - name: Test RPMs deploy - shell: bash - run: | - MY_ARCH=`uname -m` - curr_path="${{ github.workspace }}" - server="foundationdb-server-${{ needs.calc_ver.outputs.full_ver }}.${MY_ARCH}.rpm" - clients="foundationdb-clients-${{ needs.calc_ver.outputs.full_ver }}.${MY_ARCH}.rpm" - podman run --rm -v ${curr_path}/${server}/${server}:/home/${server} ${{ env.RHEL_IMAGE_NAME }} /bin/bash -c "dnf install -y /home/${server}" && { echo "Installation $server have to fail"; exit 1; } - podman run --rm -v ${curr_path}/${server}/${server}:/home/${server} -v ${curr_path}/${clients}/${clients}:/home/${clients} ${{ env.RHEL_IMAGE_NAME }} /bin/bash -c "dnf install -y /home/${server} /home/${clients}" || { echo "Installation $server and $clients have to success"; exit 2; } - podman run --rm -v ${curr_path}/${clients}/${clients}:/home/${clients} ${{ env.RHEL_IMAGE_NAME }} /bin/bash -c "dnf install -y /home/${clients} && grep -q foundationdb /etc/passwd" || { echo "User foundationdb is not created after $clients installation"; exit 3; } - - - name: Test versioned RPMs deploy - shell: bash - run: | - MY_ARCH=`uname -m` - curr_path="${{ github.workspace }}" - server="foundationdb-${{ needs.calc_ver.outputs.full_ver }}-server-versioned-${{ needs.calc_ver.outputs.full_ver }}.${MY_ARCH}.rpm" - clients="foundationdb-${{ needs.calc_ver.outputs.full_ver }}-clients-versioned-${{ needs.calc_ver.outputs.full_ver }}.${MY_ARCH}.rpm" - podman run --rm -v ${curr_path}/${server}/${server}:/home/${server} ${{ env.RHEL_IMAGE_NAME }} /bin/bash -c "dnf install -y /home/${server}" && { echo "Installation $server have to fail"; exit 1; } - podman run --rm -v ${curr_path}/${server}/${server}:/home/${server} -v ${curr_path}/${clients}/${clients}:/home/${clients} ${{ env.RHEL_IMAGE_NAME }} /bin/bash -c "dnf install -y /home/${server} /home/${clients}" || { echo "Installation $server and $clients have to success"; exit 2; } - podman run --rm -v ${curr_path}/${clients}/${clients}:/home/${clients} ${{ env.RHEL_IMAGE_NAME }} /bin/bash -c "dnf install -y /home/${clients} && grep -q foundationdb /etc/passwd" && { echo "User foundationdb was created after $clients installation"; exit 3; } - podman run --rm -v ${curr_path}/${server}/${server}:/home/${server} -v ${curr_path}/${clients}/${clients}:/home/${clients} ${{ env.RHEL_IMAGE_NAME }} /bin/bash -c "dnf install -y /home/${server} /home/${clients} && grep -q foundationdb /etc/passwd" || { echo "User foundationdb is not created after $server installation"; exit 4; } - - - name: Test DEBs deploy + - name: Make a normal distro directory from the downloaded artifacts shell: bash run: | - MY_ARCH=`dpkg-architecture -q DEB_BUILD_ARCH` - curr_path="${{ github.workspace }}" - server="foundationdb-server_${{ needs.calc_ver.outputs.full_ver }}_${MY_ARCH}.deb" - clients="foundationdb-clients_${{ needs.calc_ver.outputs.full_ver }}_${MY_ARCH}.deb" - podman run --rm -v ${curr_path}/${server}/${server}:/home/${server} ${{ env.UBUNTU_IMAGE_NAME }} /bin/bash -c "apt-get update && apt-get install -y /home/${server}" && { echo "Installation $server have to fail"; exit 1; } - podman run --rm -v ${curr_path}/${server}/${server}:/home/${server} -v ${curr_path}/${clients}/${clients}:/home/${clients} ${{ env.UBUNTU_IMAGE_NAME }} /bin/bash -c "apt-get update && apt-get install -y /home/${server} /home/${clients}" || { echo "Installation $server and $clients have to success"; exit 2; } - podman run --rm -v ${curr_path}/${clients}/${clients}:/home/${clients} ${{ env.UBUNTU_IMAGE_NAME }} /bin/bash -c "apt-get update && apt-get install -y /home/${clients} && grep -q foundationdb /etc/passwd" || { echo "User foundationdb is not created after $clients installation"; exit 3; } - - - name: Test versioned DEBs deploy + ls -Rl ${{github.workspace}}/bld/download + mkdir -p ${{github.workspace}}/bld/linux/packages + mv -v ${{github.workspace}}/bld/download/*/* ${{github.workspace}}/bld/linux/packages/ + + - name: Run tests shell: bash - run: | - MY_ARCH=`dpkg-architecture -q DEB_BUILD_ARCH` - curr_path="${{ github.workspace }}" - server="foundationdb-${{ needs.calc_ver.outputs.full_ver }}-server-versioned_${{ needs.calc_ver.outputs.full_ver }}_${MY_ARCH}.deb" - clients="foundationdb-${{ needs.calc_ver.outputs.full_ver }}-clients-versioned_${{ needs.calc_ver.outputs.full_ver }}_${MY_ARCH}.deb" - podman run --rm -v ${curr_path}/${server}/${server}:/home/${server} ${{ env.UBUNTU_IMAGE_NAME }} /bin/bash -c "apt-get update && apt-get install -y /home/${server}" && { echo "Installation $server have to fail"; exit 1; } - podman run --rm -v ${curr_path}/${server}/${server}:/home/${server} -v ${curr_path}/${clients}/${clients}:/home/${clients} ${{ env.UBUNTU_IMAGE_NAME }} /bin/bash -c "apt-get update && apt-get install -y /home/${server} /home/${clients}" || { echo "Installation $server and $clients have to success"; exit 2; } - podman run --rm -v ${curr_path}/${clients}/${clients}:/home/${clients} ${{ env.UBUNTU_IMAGE_NAME }} /bin/bash -c "apt-get update && apt-get install -y /home/${clients} && grep -q foundationdb /etc/passwd" && { echo "User foundationdb was created after $clients installation"; exit 3; } - podman run --rm -v ${curr_path}/${server}/${server}:/home/${server} -v ${curr_path}/${clients}/${clients}:/home/${clients} ${{ env.UBUNTU_IMAGE_NAME }} /bin/bash -c "apt-get update && apt-get install -y /home/${server} /home/${clients} && grep -q foundationdb /etc/passwd" || { echo "User foundationdb is not created after $server installation"; exit 4; } + run: ${{github.workspace}}/build-scripts/for-linux/test-deploy.bash ${{ needs.calc_ver.outputs.full_ver }} ${{github.workspace}}/bld/linux/packages release: needs: [calc_ver, build] diff --git a/CMakeLists.txt b/CMakeLists.txt index bdedb09eb34..4f4857db0ab 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -23,7 +23,7 @@ else() endif() project(foundationdb - VERSION 7.1.25 + VERSION 7.1.29 DESCRIPTION "FoundationDB is a scalable, fault-tolerant, ordered key-value store with full ACID transactions." HOMEPAGE_URL "http://www.foundationdb.org/" LANGUAGES C CXX ASM) diff --git a/bindings/c/CMakeLists.txt b/bindings/c/CMakeLists.txt index 676fe824e30..b1a187b994f 100644 --- a/bindings/c/CMakeLists.txt +++ b/bindings/c/CMakeLists.txt @@ -184,7 +184,11 @@ endif() # Make sure that fdb_c.h is compatible with c90 add_executable(fdb_c90_test test/fdb_c90_test.c) set_property(TARGET fdb_c90_test PROPERTY C_STANDARD 90) - target_compile_options(fdb_c90_test PRIVATE -Wall -Wextra -Wpedantic -Werror) + if (CMAKE_CXX_COMPILER_ID STREQUAL "Clang") + target_compile_options(fdb_c90_test PRIVATE -Wall -Wextra -Wpedantic -Wno-gnu-line-marker -Werror) + else () + target_compile_options(fdb_c90_test PRIVATE -Wall -Wextra -Wpedantic -Werror) + endif () target_link_libraries(fdb_c90_test PRIVATE fdb_c) endif() diff --git a/bindings/c/test/apitester/TesterOptions.h b/bindings/c/test/apitester/TesterOptions.h index 0f60ae436fa..8c73f5d3e1a 100644 --- a/bindings/c/test/apitester/TesterOptions.h +++ b/bindings/c/test/apitester/TesterOptions.h @@ -42,6 +42,7 @@ class TesterOptions { int numClients; std::vector> knobs; TestSpec testSpec; + bool retainClientLibCopies = false; }; } // namespace FdbApiTester diff --git a/bindings/c/test/apitester/fdb_c_api_tester.cpp b/bindings/c/test/apitester/fdb_c_api_tester.cpp index 062ffb95f71..4ee484db9aa 100644 --- a/bindings/c/test/apitester/fdb_c_api_tester.cpp +++ b/bindings/c/test/apitester/fdb_c_api_tester.cpp @@ -45,7 +45,8 @@ enum TesterOptionId { OPT_TRACE_FORMAT, OPT_KNOB, OPT_EXTERNAL_CLIENT_LIBRARY, - OPT_TEST_FILE + OPT_TEST_FILE, + OPT_RETAIN_CLIENT_LIB_COPIES }; CSimpleOpt::SOption TesterOptionDefs[] = // @@ -61,6 +62,7 @@ CSimpleOpt::SOption TesterOptionDefs[] = // { OPT_EXTERNAL_CLIENT_LIBRARY, "--external-client-library", SO_REQ_SEP }, { OPT_TEST_FILE, "-f", SO_REQ_SEP }, { OPT_TEST_FILE, "--test-file", SO_REQ_SEP }, + { OPT_RETAIN_CLIENT_LIB_COPIES, "--retain-client-lib-copies", SO_NONE }, SO_END_OF_OPTIONS }; void printProgramUsage(const char* execName) { @@ -144,6 +146,10 @@ bool processArg(TesterOptions& options, const CSimpleOpt& args) { options.testFile = args.OptionArg(); options.testSpec = readTomlTestSpec(options.testFile); break; + + case OPT_RETAIN_CLIENT_LIB_COPIES: + options.retainClientLibCopies = true; + break; } return true; } @@ -205,6 +211,10 @@ void applyNetworkOptions(TesterOptions& options) { fdb_check(FdbApi::setOption(FDBNetworkOption::FDB_NET_OPTION_TRACE_LOG_GROUP, options.logGroup)); } + if (options.retainClientLibCopies) { + fdb_check(FdbApi::setOption(FDBNetworkOption::FDB_NET_OPTION_RETAIN_CLIENT_LIBRARY_COPIES)); + } + for (auto knob : options.knobs) { fdb_check(FdbApi::setOption(FDBNetworkOption::FDB_NET_OPTION_KNOB, fmt::format("{}={}", knob.first.c_str(), knob.second.c_str()))); diff --git a/bindings/c/test/apitester/run_c_api_tests.py b/bindings/c/test/apitester/run_c_api_tests.py index 8f79e6d8b11..39067ae4a86 100755 --- a/bindings/c/test/apitester/run_c_api_tests.py +++ b/bindings/c/test/apitester/run_c_api_tests.py @@ -54,6 +54,9 @@ def run_tester(args, test_file): if args.external_client_library is not None: cmd += ["--external-client-library", args.external_client_library] + if args.retain_client_lib_copies: + cmd += ["--retain-client-lib-copies"] + get_logger().info('\nRunning tester \'%s\'...' % ' '.join(cmd)) proc = Popen(cmd, stdout=sys.stdout, stderr=sys.stderr) timed_out = False @@ -111,7 +114,12 @@ def parse_args(argv): help='The timeout in seconds for running each individual test. (default 300)') parser.add_argument('--logging-level', type=str, default='INFO', choices=['ERROR', 'WARNING', 'INFO', 'DEBUG'], help='Specifies the level of detail in the tester output (default=\'INFO\').') - + parser.add_argument( + "--retain-client-lib-copies", + action="store_true", + default=False, + help="Retain temporary external client library copies.", + ) return parser.parse_args(argv) diff --git a/bindings/go/src/fdb/generated.go b/bindings/go/src/fdb/generated.go index b636484408b..61a5b2517d0 100644 --- a/bindings/go/src/fdb/generated.go +++ b/bindings/go/src/fdb/generated.go @@ -239,6 +239,11 @@ func (o NetworkOptions) SetClientThreadsPerVersion(param int64) error { return o.setOpt(65, int64ToBytes(param)) } +// Retain temporary external client library copies that are created for enabling multi-threading. +func (o NetworkOptions) SetRetainClientLibraryCopies() error { + return o.setOpt(67, nil) +} + // Disables logging of client statistics, such as sampled transaction activity. func (o NetworkOptions) SetDisableClientStatisticsLogging() error { return o.setOpt(70, nil) @@ -617,12 +622,12 @@ const ( StreamingModeWantAll StreamingMode = -1 // The default. The client doesn't know how much of the range it is likely - // to used and wants different performance concerns to be balanced. Only a - // small portion of data is transferred to the client initially (in order to - // minimize costs if the client doesn't read the entire range), and as the - // caller iterates over more items in the range larger batches will be - // transferred in order to minimize latency. After enough iterations, the - // iterator mode will eventually reach the same byte limit as “WANT_ALL“ + // to used and wants different performance concerns to be balanced. + // Only a small portion of data is transferred to the client initially (in + // order to minimize costs if the client doesn't read the entire range), and + // as the caller iterates over more items in the range larger batches will + // be transferred in order to minimize latency. After enough iterations, + // the iterator mode will eventually reach the same byte limit as “WANT_ALL“ StreamingModeIterator StreamingMode = 0 // Infrequently used. The client has passed a specific row limit and wants @@ -632,8 +637,8 @@ const ( // mode is used. StreamingModeExact StreamingMode = 1 - // Infrequently used. Transfer data in batches small enough to not be much - // more expensive than reading individual rows, to minimize cost if + // Infrequently used. Transfer data in batches small enough to not be + // much more expensive than reading individual rows, to minimize cost if // iteration stops early. StreamingModeSmall StreamingMode = 2 @@ -641,16 +646,16 @@ const ( // large. StreamingModeMedium StreamingMode = 3 - // Infrequently used. Transfer data in batches large enough to be, in a - // high-concurrency environment, nearly as efficient as possible. If the - // client stops iteration early, some disk and network bandwidth may be - // wasted. The batch size may still be too small to allow a single client to - // get high throughput from the database, so if that is what you need + // Infrequently used. Transfer data in batches large enough to be, + // in a high-concurrency environment, nearly as efficient as possible. + // If the client stops iteration early, some disk and network bandwidth may + // be wasted. The batch size may still be too small to allow a single client + // to get high throughput from the database, so if that is what you need // consider the SERIAL StreamingMode. StreamingModeLarge StreamingMode = 4 - // Transfer data in batches large enough that an individual client can get - // reasonable read bandwidth from the database. If the client stops + // Transfer data in batches large enough that an individual client can + // get reasonable read bandwidth from the database. If the client stops // iteration early, considerable disk and network bandwidth may be wasted. StreamingModeSerial StreamingMode = 5 ) diff --git a/build-scripts/for-linux/build-find-libatomic.bash b/build-scripts/for-linux/build-find-libatomic.bash deleted file mode 100755 index 189bf0e7055..00000000000 --- a/build-scripts/for-linux/build-find-libatomic.bash +++ /dev/null @@ -1,30 +0,0 @@ -#!/bin/bash - -# should be called with "source" command -# try to find libatomic library and set -DATOMIC_LIBRARY_FILE to the APP_PRMS - -set -e - -# try to find the static library -LIBRARY_CANDIDATE=$(ls -t1 /usr/lib/gcc/x86_64*/*/libatomic.a | head -n 1) || true - -if [[ -n "$LIBRARY_CANDIDATE" ]]; then - # test if the library is linked relocatable - PROBE_O=glfree.o - TMP_DIR=`mktemp -d` - pushd $TMP_DIR - ar x $LIBRARY_CANDIDATE $PROBE_O - REL_TYPE=`objdump -r glfree.o | awk '/.rodata/{ print $2;exit;}'` - popd - rm -rf $TMP_DIR - # R_X86_64_32S is prohibited for linking shared libraries - [[ "$REL_TYPE" == "R_X86_64_32S" ]] && LIBRARY_CANDIDATE="" -fi - -# try to find a dynamic library -[[ -z "$LIBRARY_CANDIDATE" ]] \ - && LIBRARY_CANDIDATE=$(ls -1 /usr/lib64/libatomic.so* /usr/lib/gcc/x86_64-*/*/libatomic.so* | sort | head -n 1) \ - || true - -# if found then put in to APP_PRMS -[[ -n "$LIBRARY_CANDIDATE" ]] && APP_PRMS="$APP_PRMS -DATOMIC_LIBRARY_FILE=$LIBRARY_CANDIDATE" diff --git a/build-scripts/for-linux/build-on-linux.bash b/build-scripts/for-linux/build-on-linux.bash index dc8b5e3d198..e74228d26ec 100755 --- a/build-scripts/for-linux/build-on-linux.bash +++ b/build-scripts/for-linux/build-on-linux.bash @@ -34,9 +34,6 @@ APP_PRMS="\ [ ! -e /usr/lib64/libcrypto.a -a -e /opt/openssl/lib/libcrypto.a ] && \ APP_PRMS="$APP_PRMS -DOPENSSL_ROOT_DIR=/opt/openssl" -# find libatomic -source $BASE_DIR/build-find-libatomic.bash - echo "env CC=clang CXX=clang++ cmake -G Ninja $APP_PRMS . $SRC_DIR" env CC=clang CXX=clang++ cmake -G Ninja $APP_PRMS . $SRC_DIR diff --git a/build-scripts/for-linux/test-deploy.bash b/build-scripts/for-linux/test-deploy.bash new file mode 100755 index 00000000000..728b71dd6ce --- /dev/null +++ b/build-scripts/for-linux/test-deploy.bash @@ -0,0 +1,131 @@ +#!/bin/bash + +# Test deploying foundationdb on rpm-and deb-based linux +# $1 - full Foundationdb version, ex. 7.1.29-0.ow.1 +# $2 - distr dir. Default is bld/linux/packages relative to the current dir +# $3 - a rpm-based linux docker image. Default is oraclelinux:8 +# $4 - a deb-based linux docker image. Default is debian:10 + +set -e + +BASE_DIR="$(readlink -f $(dirname $0))" +FULL_VERSION="$1" +DISTR_DIR="$(readlink -f ${2:-bld/linux/packages})" +RPM_IMAGE=${3:-oraclelinux:8} +DEB_IMAGE=${4:-debian:10} + +print_usage() { + echo >&2 "Usage: $0 FullFdbVersion [FdbDistrDir] [RpmImage] [DebImage]" +} + +# testing parameters +if [[ -z "$FULL_VERSION" ]]; then + echo >&2 "FullFdbVersion is not specified." + print_usage + return 1 2>/dev/null || exit 1 +fi + +if ! ls "$DISTR_DIR"/foundationdb-*$FULL_VERSION*.{rpm,deb}; then + echo >&2 "No $DISTR_DIR/foundationdb-*$FULL_VERSION*.{rpm,deb} files have been found." + echo >&2 "Possible FdbDistrDir is not correct." + print_usage + return 1 2>/dev/null || exit 1 +fi + +# check images +podman pull $RPM_IMAGE +podman pull $DEB_IMAGE + +MY_ARCH_RPM=`uname -m` +MY_ARCH_DEB=`dpkg-architecture -q DEB_HOST_ARCH` + +test_deploy_pkgs() { + IMAGE=$1 + INSTALL_CMD=$2 + SERVER_FILE=$3 + CLIENT_FILE=$4 + USER_AFTER_CLIENT=${5:-Y} + + echo SERVER_FILE=$SERVER_FILE + echo CLIENT_FILE=$CLIENT_FILE + + if [[ $USER_AFTER_CLIENT == Y ]]; then + CLIENT_CHECK_WITH="" + ERRMSG_CLIENT="not created" + else + CLIENT_CHECK_WITH="!" + ERRMSG_CLIENT="created unexpectedly" + fi + + + echo "Trying to install the client package only..." + set -x + if ! podman run --rm -v "$DISTR_DIR:/mnt/distr:Z,ro" $IMAGE \ + /bin/bash -c "$INSTALL_CMD /mnt/distr/$CLIENT_FILE && $CLIENT_CHECK_WITH getent passwd foundationdb"; then + echo >&2 "Installation of $CLIENT_FILE failed or the foundationdb user was $ERRMSG_CLIENT." + return 3 + fi + set +x + echo "Installation test of the client package only is successful" + echo "" + + echo "Trying to install the both client and server package packages..." + set -x + if ! podman run --rm -v "$DISTR_DIR:/mnt/distr:Z,ro" $IMAGE \ + /bin/bash -c "$INSTALL_CMD /mnt/distr/$SERVER_FILE /mnt/distr/$CLIENT_FILE && getent passwd foundationdb" + then + echo >&2 "Installation $SERVER_FILE and $CLIENT_FILE failed or the foundationdb user was not created." + return 2 + fi + set +x + echo "Installation test of the both client and server packages is successful" + echo "" + + echo "Trying to install the server package only..." + set -x + if podman run --rm -v "$DISTR_DIR:/mnt/distr:Z,ro" $IMAGE \ + /bin/bash -c "$INSTALL_CMD /mnt/distr/$SERVER_FILE"; then + echo >&2 "Installation $SERVER_FILE without a client must fail." + return 1 + fi + set +x + echo "Installation test of the server package only is successful" + echo "" + +} + +RPM_INSTALL_CMD="dnf install -y" +DEB_INSTALL_CMD="apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y" + +echo "Testing DEBs deploy..." +test_deploy_pkgs \ + "$DEB_IMAGE" \ + "$DEB_INSTALL_CMD" \ + "foundationdb-server_${FULL_VERSION}_$MY_ARCH_DEB.deb" \ + "foundationdb-clients_${FULL_VERSION}_$MY_ARCH_DEB.deb" \ + Y + +echo "Testing versioned DEBs deploy..." +test_deploy_pkgs \ + "$DEB_IMAGE" \ + "$DEB_INSTALL_CMD" \ + "foundationdb-$FULL_VERSION-server-versioned_${FULL_VERSION}_$MY_ARCH_DEB.deb" \ + "foundationdb-$FULL_VERSION-clients-versioned_${FULL_VERSION}_$MY_ARCH_DEB.deb" \ + N + +echo "Testing RPMs deploy..." +test_deploy_pkgs \ + "$RPM_IMAGE" \ + "$RPM_INSTALL_CMD" \ + "foundationdb-server-$FULL_VERSION.$MY_ARCH_RPM.rpm" \ + "foundationdb-clients-$FULL_VERSION.$MY_ARCH_RPM.rpm" \ + Y + +echo "Testing versioned RPMs deploy..." +test_deploy_pkgs \ + "$RPM_IMAGE" \ + "$RPM_INSTALL_CMD" \ + "foundationdb-$FULL_VERSION-server-versioned-$FULL_VERSION.$MY_ARCH_RPM.rpm" \ + "foundationdb-$FULL_VERSION-clients-versioned-$FULL_VERSION.$MY_ARCH_RPM.rpm" \ + N + diff --git a/cmake/CompileRocksDB.cmake b/cmake/CompileRocksDB.cmake index 3a75c5b7f94..62405d04df9 100644 --- a/cmake/CompileRocksDB.cmake +++ b/cmake/CompileRocksDB.cmake @@ -1,75 +1,74 @@ # FindRocksDB -find_package(RocksDB 6.27.3) +find_package(RocksDB 7.7.3) include(ExternalProject) -if (RocksDB_FOUND) +set(RocksDB_CMAKE_ARGS + -DUSE_RTTI=1 + -DPORTABLE=${PORTABLE_ROCKSDB} + -DFORCE_SSE42=${ROCKSDB_SSE42} + -DFORCE_AVX=${ROCKSDB_AVX} + -DFORCE_AVX2=${ROCKSDB_AVX2} + -DCMAKE_C_COMPILER=${CMAKE_C_COMPILER} + -DCMAKE_C_FLAGS=${CMAKE_C_FLAGS} + -DCMAKE_CXX_STANDARD=${CMAKE_CXX_STANDARD} + -DCMAKE_CXX_COMPILER=${CMAKE_CXX_COMPILER} + -DCMAKE_CXX_FLAGS=${CMAKE_CXX_FLAGS} + -DCMAKE_SHARED_LINKER_FLAGS=${CMAKE_SHARED_LINKER_FLAGS} + -DCMAKE_STATIC_LINKER_FLAGS=${CMAKE_STATIC_LINKER_FLAGS} + -DCMAKE_EXE_LINKER_FLAGS=${CMAKE_EXE_LINKER_FLAGS} + -DCMAKE_BUILD_TYPE=${CMAKE_BUILD_TYPE} + -DFAIL_ON_WARNINGS=OFF + -DWITH_GFLAGS=OFF + -DWITH_TESTS=OFF + -DWITH_TOOLS=OFF + -DWITH_CORE_TOOLS=OFF + -DWITH_BENCHMARK_TOOLS=OFF + -DWITH_BZ2=OFF + -DWITH_LZ4=ON + -DWITH_SNAPPY=OFF + -DWITH_ZLIB=OFF + -DWITH_ZSTD=OFF + -DWITH_LIBURING=${WITH_LIBURING} + -DWITH_TSAN=${USE_TSAN} + -DWITH_ASAN=${USE_ASAN} + -DWITH_UBSAN=${USE_UBSAN} + -DROCKSDB_BUILD_SHARED=OFF + -DCMAKE_POSITION_INDEPENDENT_CODE=True +) + +if(ROCKSDB_FOUND) ExternalProject_Add(rocksdb SOURCE_DIR "${RocksDB_ROOT}" DOWNLOAD_COMMAND "" - CMAKE_ARGS -DUSE_RTTI=1 -DPORTABLE=${PORTABLE_ROCKSDB} - -DCMAKE_CXX_STANDARD=${CMAKE_CXX_STANDARD} - -DCMAKE_CXX_COMPILER=${CMAKE_CXX_COMPILER} - -DCMAKE_BUILD_TYPE=${CMAKE_BUILD_TYPE} - -DWITH_GFLAGS=OFF - -DWITH_TESTS=OFF - -DWITH_TOOLS=OFF - -DWITH_CORE_TOOLS=OFF - -DWITH_BENCHMARK_TOOLS=OFF - -DWITH_BZ2=OFF - -DWITH_LZ4=ON - -DWITH_SNAPPY=OFF - -DWITH_ZLIB=OFF - -DWITH_ZSTD=OFF - -DWITH_LIBURING=${WITH_LIBURING} - -DWITH_TSAN=${USE_TSAN} - -DWITH_ASAN=${USE_ASAN} - -DWITH_UBSAN=${USE_UBSAN} - -DROCKSDB_BUILD_SHARED=OFF - -DCMAKE_POSITION_INDEPENDENT_CODE=True + CMAKE_ARGS ${RocksDB_CMAKE_ARGS} BUILD_BYPRODUCTS /librocksdb.a INSTALL_COMMAND "" + LOG_CONFIGURE TRUE + LOG_BUILD TRUE ) ExternalProject_Get_Property(rocksdb BINARY_DIR) set(ROCKSDB_LIBRARIES - ${BINARY_DIR}/librocksdb.a) + ${BINARY_DIR}/librocksdb.a) else() ExternalProject_Add(rocksdb - URL https://github.com/facebook/rocksdb/archive/refs/tags/v6.27.3.tar.gz - URL_HASH SHA256=ee29901749b9132692b26f0a6c1d693f47d1a9ed8e3771e60556afe80282bf58 - CMAKE_ARGS -DUSE_RTTI=1 -DPORTABLE=${PORTABLE_ROCKSDB} - -DCMAKE_CXX_STANDARD=${CMAKE_CXX_STANDARD} - -DCMAKE_CXX_COMPILER=${CMAKE_CXX_COMPILER} - -DCMAKE_BUILD_TYPE=${CMAKE_BUILD_TYPE} - -DWITH_GFLAGS=OFF - -DWITH_TESTS=OFF - -DWITH_TOOLS=OFF - -DWITH_CORE_TOOLS=OFF - -DWITH_BENCHMARK_TOOLS=OFF - -DWITH_BZ2=OFF - -DWITH_LZ4=ON - -DWITH_SNAPPY=OFF - -DWITH_ZLIB=OFF - -DWITH_ZSTD=OFF - -DWITH_LIBURING=${WITH_LIBURING} - -DWITH_TSAN=${USE_TSAN} - -DWITH_ASAN=${USE_ASAN} - -DWITH_UBSAN=${USE_UBSAN} - -DROCKSDB_BUILD_SHARED=OFF - -DCMAKE_POSITION_INDEPENDENT_CODE=True - -DFAIL_ON_WARNINGS=${USE_WERROR} + URL https://github.com/facebook/rocksdb/archive/refs/tags/v7.7.3.tar.gz + URL_HASH SHA256=b8ac9784a342b2e314c821f6d701148912215666ac5e9bdbccd93cf3767cb611 + CMAKE_ARGS ${RocksDB_CMAKE_ARGS} BUILD_BYPRODUCTS /librocksdb.a INSTALL_COMMAND "" + LOG_CONFIGURE TRUE + LOG_BUILD TRUE ) ExternalProject_Get_Property(rocksdb BINARY_DIR) set(ROCKSDB_LIBRARIES - ${BINARY_DIR}/librocksdb.a) + ${BINARY_DIR}/librocksdb.a) ExternalProject_Get_Property(rocksdb SOURCE_DIR) - set (ROCKSDB_INCLUDE_DIR "${SOURCE_DIR}/include") + set(ROCKSDB_INCLUDE_DIR "${SOURCE_DIR}/include") set(ROCKSDB_FOUND TRUE) endif() @@ -78,6 +77,6 @@ message(STATUS "Found RocksDB library: ${ROCKSDB_LIBRARIES}") message(STATUS "Found RocksDB includes: ${ROCKSDB_INCLUDE_DIR}") mark_as_advanced( - ROCKSDB_LIBRARIES - ROCKSDB_INCLUDE_DIR + ROCKSDB_LIBRARIES + ROCKSDB_INCLUDE_DIR ) diff --git a/cmake/ConfigureCompiler.cmake b/cmake/ConfigureCompiler.cmake index d6b8436a989..a230d305396 100644 --- a/cmake/ConfigureCompiler.cmake +++ b/cmake/ConfigureCompiler.cmake @@ -278,34 +278,20 @@ else() if (CLANG) add_compile_options() if (APPLE OR USE_LIBCXX) - add_compile_options($<$:-stdlib=libc++>) + set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -stdlib=libc++") if (NOT APPLE) if (STATIC_LINK_LIBCXX) - add_link_options(-static-libgcc -nostdlib++ -Wl,-Bstatic -lc++ -lc++abi -Wl,-Bdynamic) + set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -static-libgcc -nostdlib++ -Wl,-Bstatic -lc++ -lc++abi -Wl,-Bdynamic") + set(CMAKE_SHARED_LINKER_FLAGS "${CMAKE_SHARED_LINKER_FLAGS} -static-libgcc -nostdlib++ -Wl,-Bstatic -lc++ -lc++abi -Wl,-Bdynamic") endif() - add_link_options(-stdlib=libc++ -Wl,-build-id=sha1) + set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -stdlib=libc++ -Wl,-build-id=sha1") + set(CMAKE_SHARED_LINKER_FLAGS "${CMAKE_SHARED_LINKER_FLAGS} -stdlib=libc++ -Wl,-build-id=sha1") endif() - else() - # clang compiler does not fully support builtin atomic operations - # See https://reviews.llvm.org/D85044?id=287068 - # So it needs linking to an atomic library (static or dynamic) - if(ATOMIC_LIBRARY_FILE) - set(Atomic_LIBRARY "${ATOMIC_LIBRARY_FILE}") - else() - # Try to find an atomic static library - execute_process( - COMMAND ${CMAKE_C_COMPILER} -print-search-dirs - OUTPUT_VARIABLE MY_CLANG_SEARCH_DIRS - RESULT_VARIABLE MY_CLANG_SEARCH_DIRS_FOUND - OUTPUT_STRIP_TRAILING_WHITESPACE) - string(REGEX MATCH "libraries: =([^\n].*)" _dummy "${MY_CLANG_SEARCH_DIRS}\n") - string(REPLACE ":" ";" MY_CLANG_LIB_DIRS "${CMAKE_MATCH_1}") - find_library(Atomic_LIBRARY NAMES libatomic.a PATHS ${MY_CLANG_LIB_DIRS}) - if(NOT Atomic_LIBRARY) - message(FATAL_ERROR "atomic static library not found") - endif() - endif() - message(STATUS "Atomic_LIBRARY=${Atomic_LIBRARY}") + endif() + if (NOT APPLE AND NOT USE_LIBCXX) + message(STATUS "Linking libatomic") + set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -latomic") + set(CMAKE_SHARED_LINKER_FLAGS "${CMAKE_SHARED_LINKER_FLAGS} -latomic") endif() if (OPEN_FOR_IDE) add_compile_options( diff --git a/cmake/FDBComponents.cmake b/cmake/FDBComponents.cmake index 865e2f79dc9..e12b54f5bb8 100644 --- a/cmake/FDBComponents.cmake +++ b/cmake/FDBComponents.cmake @@ -163,6 +163,9 @@ endif() set(SSD_ROCKSDB_EXPERIMENTAL ON CACHE BOOL "Build with experimental RocksDB support") set(PORTABLE_ROCKSDB ON CACHE BOOL "Compile RocksDB in portable mode") # Set this to OFF to compile RocksDB with `-march=native` +set(ROCKSDB_SSE42 OFF CACHE BOOL "Compile RocksDB with SSE42 enabled") +set(ROCKSDB_AVX ${USE_AVX} CACHE BOOL "Compile RocksDB with AVX enabled") +set(ROCKSDB_AVX2 OFF CACHE BOOL "Compile RocksDB with AVX2 enabled") set(WITH_LIBURING OFF CACHE BOOL "Build with liburing enabled") # Set this to ON to include liburing # RocksDB is currently enabled by default for all compilers. Clang 14 could build it too. #if (SSD_ROCKSDB_EXPERIMENTAL AND GCC) diff --git a/cmake/InstallLayout.cmake b/cmake/InstallLayout.cmake index 073d01e334e..920c11826e9 100644 --- a/cmake/InstallLayout.cmake +++ b/cmake/InstallLayout.cmake @@ -341,19 +341,22 @@ set(CPACK_DEBIAN_ENABLE_COMPONENT_DEPENDS ON) set(CPACK_DEBIAN_DEBUGINFO_PACKAGE ${GENERATE_DEBUG_PACKAGES}) set(CPACK_DEBIAN_PACKAGE_SECTION "database") +set(CPACK_DEBIAN_PACKAGE_SHLIBDEPS ON) +set(CPACK_DEBIAN_PACKAGE_GENERATE_SHLIBS ON) + set(CPACK_DEBIAN_SERVER-DEB_PACKAGE_NAME "${server_package_basename}") set(CPACK_DEBIAN_CLIENTS-DEB_PACKAGE_NAME "${clients_package_basename}") set(CPACK_DEBIAN_SERVER-VERSIONED_PACKAGE_NAME "${server_versioned_package_basename}") set(CPACK_DEBIAN_CLIENTS-VERSIONED_PACKAGE_NAME "${clients_versioned_package_basename}") -set(CPACK_DEBIAN_SERVER-DEB_PACKAGE_DEPENDS "adduser, libc6 (>= 2.12)") -set(CPACK_DEBIAN_SERVER-DEB_PACKAGE_RECOMMENDS "python (>= 2.6)") -set(CPACK_DEBIAN_CLIENTS-DEB_PACKAGE_DEPENDS "adduser, libc6 (>= 2.12)") +set(CPACK_DEBIAN_SERVER-DEB_PACKAGE_DEPENDS "adduser") +# set(CPACK_DEBIAN_SERVER-DEB_PACKAGE_RECOMMENDS "python (>= 2.6)") +set(CPACK_DEBIAN_CLIENTS-DEB_PACKAGE_DEPENDS "adduser") -set(CPACK_DEBIAN_SERVER-VERSIONED_PACKAGE_DEPENDS "adduser, libc6 (>= 2.12)") -set(CPACK_DEBIAN_SERVER-VERSIONED_PACKAGE_RECOMMENDS "python (>= 2.6)") -set(CPACK_DEBIAN_CLIENTS-VERSIONED_PACKAGE_DEPENDS "adduser, libc6 (>= 2.12)") +set(CPACK_DEBIAN_SERVER-VERSIONED_PACKAGE_DEPENDS "adduser") +# set(CPACK_DEBIAN_SERVER-VERSIONED_PACKAGE_RECOMMENDS "python (>= 2.6)") +set(CPACK_DEBIAN_CLIENTS-VERSIONED_PACKAGE_DEPENDS "adduser") set(CPACK_DEBIAN_PACKAGE_HOMEPAGE "https://www.foundationdb.org") set(CPACK_DEBIAN_CLIENTS-DEB_PACKAGE_CONTROL_EXTRA diff --git a/documentation/sphinx/source/api-python.rst b/documentation/sphinx/source/api-python.rst index a92f5257a1e..7cd7a8e295f 100644 --- a/documentation/sphinx/source/api-python.rst +++ b/documentation/sphinx/source/api-python.rst @@ -206,7 +206,6 @@ After importing the ``fdb`` module and selecting an API version, you probably wa |option-tls-password| - .. method :: fdb.options.set_disable_local_client() |option-set-disable-local-client| diff --git a/documentation/sphinx/source/index.rst b/documentation/sphinx/source/index.rst index 40c7a762797..167bebda43f 100644 --- a/documentation/sphinx/source/index.rst +++ b/documentation/sphinx/source/index.rst @@ -50,6 +50,7 @@ The latest changes are detailed in :ref:`release-notes`. The documentation has t :hidden: local-dev + internal-dev-tools why-foundationdb technical-overview client-design diff --git a/documentation/sphinx/source/internal-dev-tools.rst b/documentation/sphinx/source/internal-dev-tools.rst new file mode 100644 index 00000000000..ea809473126 --- /dev/null +++ b/documentation/sphinx/source/internal-dev-tools.rst @@ -0,0 +1,58 @@ +################## +Internal Dev Tools +################## + +Code Probes +=========== + +Code probes are a mechanism in FDB to prove that certain code-paths are being tested under the right conditions. They differ from code coverage in multiple ways (explained below). + +The general format of a code probe is: + +.. code-block:: C++ + + CODE_PROBE(, "Comment", [annotations...]); + +A simple example of a code probe could look as follows: + +.. code-block:: C++ + + CODE_PROBE(self->forceRecovery, "Resolver detects forced recovery", probe::context::sim2); + +On a very high level, the above code will indicate that whenever this line is executed and ``self->forceRecovery`` is ``true``, we ran into some interesting case. In addition this probe is also annotated with ``probe::context::sim2``. This indicates that we expect this code to be eventually hit in simulation. + +By default, FDB simply will write a trace-line when this code is hit and the condition is ``true``. If the code is never hit, the simulator will, at the end of the run, print the code probe but set the ``covered`` field to ``false``. This all happens in the context of a single simulation run (``fdbserver`` doesn't have a concept of ensembles). This information is written into the log file. ``TestHarness`` (see below) will then use this information to write code probe statistics to the ensemble in the Joshua cluster (if the test is run in Joshua). + +We expect that ALL code probes will be hit in a nightly run. In the future we can potentially use this feature for other things (like instructing the simulator to do an extensive search starting when one of these probes is being hit). + +In addition to ``context`` annotations, users can also define and pass assertions. For example: + +.. code-block:: C++ + + CODE_PROBE(condition, "Some comment", assert::simOnly); + +These will add an assertion to the code. In addition to that, the simulator will not print missed code probes that asserted that the probe won't be hit in simulation. + +Test Harness +============ + +TestHarness is our primary testing tool. It has multiple jobs: + +* *Running*: It can run a test in Joshua. +* *Statistics*: It will choose a test to run based on previous runs (within the same ensemble) spent CPU time for each test. It does that by writing statistics about the test at the end of each run. +* *Reporting*: After an ensemble has finished (or while it is running), ``TestHarness`` can be used to generate a report in ``xml`` or ``json``. + +Test Harness can be found in the FDB source repository under ``contrib/TestHarness2``. It has a weak dependency to `joshua `_ (if Test Harness can find joshua it will report back about failed tests, otherwise it will just print out general statistics about the ensemble). Joshua will call Test Harness as follows: + +.. code-block:: shell + + python3 -m test_harness.app -s ${JOSHUA_SEED} --old-binaries-path ${OLDBINDIR} + +Here the seed is a random number generated by joshua and ``OLDBINDIR`` is a directory path where the old fdb binaries can be found (this is needed for restart tests). If one wants to retry a test they can pass the previous joshua seed, a directory path that has *exactly* the same content as ``OLDBINARYDIR``, plus the reported statistics to the test harness app. This should then re-run the same code as before. + +In order to figure out what command line arguments ``test_harness.app`` (and ``test_harness.results``) accepts, one can check the contents of ``contrib/TestHarness2/test_harness/config.py``. + +Reporting +--------- + +After a joshua ensemble completed, ``test_harness.results`` can be used in order to get a report on the ensemble. This will include, by default, a list of all failed tests (similar to ``joshua tail --errors``, though in a more human readable file). For completed ensemble it will also print code probes that weren't hit often enough. An ensemble is considered to be successful if no simulation runs completed with an error AND all code probes have been hit sufficiently often. diff --git a/documentation/sphinx/source/release-notes/release-notes-710.rst b/documentation/sphinx/source/release-notes/release-notes-710.rst index 2df927160c5..b2922f80818 100644 --- a/documentation/sphinx/source/release-notes/release-notes-710.rst +++ b/documentation/sphinx/source/release-notes/release-notes-710.rst @@ -4,6 +4,45 @@ Release Notes ############# +7.1.29 +====== +* Same as 7.1.28 release with AVX enabled. + +7.1.28 +====== +* Released with AVX disabled. +* Changed log router to detect slow peeks and to automatically switch DC for peeking. `(PR #9640) `_ +* Added multiple prefix filter support for fdbdecode. `(PR #9483) `_, `(PR #9489) `_, `(PR #9511) `_, and `(PR #9560) `_ +* Enhanced fdbbackup query command to estimate data processing from a specific snapshot to a target version. `(PR #9506) `_ +* Improved PTree insertion and erase performance for storage servers. `(PR #9508) `_ +* Added exclude to fdbcli's configure command to prevent faulty TLogs from affecting recovery. `(PR #9404) `_ +* Fixed getMappedRange metrics. `(PR #9331) `_ + +7.1.27 +====== +* Same as 7.1.26 release with AVX enabled. + +7.1.26 +====== +* Released with AVX disabled. +* Fixed status json to not report fdbserver processes as clients. `(PR #9252) `_ +* Added detection of disconnection to satellite TLog in gray failure detection. `(PR #9107) `_ +* Fixed (non)empty peeks stats in TLogMetrics. `(PR #9074) `_ +* Fixed a data distribution bug where exclusions can become stuck because DD cannot build new teams. `(PR #9035) `_ +* Added FoundationDB version to ProcessMetrics. `(PR #9037) `_ +* Removed RocksDB read iterator destruction from the commit path. `(PR #8971) `_ +* Added determinstic degraded server selection in gray failure detection. `(PR #9001) `_ +* Fixed an interger overflow bug that causes fetching backup files to fail. `(PR #8996) `_ +* Fixed a log router race condition that blocks remote tlogs forever. `(PR #8966) `_ +* Fixed a backup worker assertion failure. `(PR #8887) `_ +* Upgraded RocksDB to 7.7.3 version. `(PR #8880) `_ +* Added byte limit for index prefetch. `(PR #8802) `_ +* Added storage server read range bytes metrics. `(PR #8724) `_ +* Added counters for single key clear requests. `(PR #8792) `_ +* Added more RocksDB knobs. `(PR #8713) `_, `(PR #8862) `_, and `(PR #9165) `_ +* Added a new network option "retain_client_library_copies" to keep the client library copies. `(PR #8740) `_ +* Fixed a transaction_too_old error on storage servers when version vector is enabled. `(PR #8710) `_ + 7.1.25 ====== * Same as 7.1.24 release with AVX enabled. diff --git a/documentation/tutorial/tutorial.actor.cpp b/documentation/tutorial/tutorial.actor.cpp index 67c542b6326..32f6c10eca6 100644 --- a/documentation/tutorial/tutorial.actor.cpp +++ b/documentation/tutorial/tutorial.actor.cpp @@ -63,7 +63,9 @@ ACTOR Future simpleTimer() { ACTOR Future someFuture(Future ready) { // loop choose {} works as well here - the braces are optional loop choose { - when(wait(delay(0.5))) { std::cout << "Still waiting...\n"; } + when(wait(delay(0.5))) { + std::cout << "Still waiting...\n"; + } when(int r = wait(ready)) { std::cout << format("Ready %d\n", r); wait(delay(double(r))); @@ -84,8 +86,12 @@ ACTOR Future promiseDemo() { ACTOR Future eventLoop(AsyncTrigger* trigger) { loop choose { - when(wait(delay(0.5))) { std::cout << "Still waiting...\n"; } - when(wait(trigger->onTrigger())) { std::cout << "Triggered!\n"; } + when(wait(delay(0.5))) { + std::cout << "Still waiting...\n"; + } + when(wait(trigger->onTrigger())) { + std::cout << "Triggered!\n"; + } } } @@ -185,7 +191,9 @@ ACTOR Future echoServer() { when(GetInterfaceRequest req = waitNext(echoServer.getInterface.getFuture())) { req.reply.send(echoServer); } - when(EchoRequest req = waitNext(echoServer.echo.getFuture())) { req.reply.send(req.message); } + when(EchoRequest req = waitNext(echoServer.echo.getFuture())) { + req.reply.send(req.message); + } when(ReverseRequest req = waitNext(echoServer.reverse.getFuture())) { req.reply.send(std::string(req.message.rbegin(), req.message.rend())); } diff --git a/fdbbackup/FileConverter.h b/fdbbackup/FileConverter.h index 0aa1d105a6c..dfb23d03fa1 100644 --- a/fdbbackup/FileConverter.h +++ b/fdbbackup/FileConverter.h @@ -31,6 +31,7 @@ namespace file_converter { // File format convertion constants enum { OPT_CONTAINER, + OPT_FILE_TYPE, OPT_BEGIN_VERSION, OPT_BLOB_CREDENTIALS, OPT_CRASHONERROR, @@ -42,7 +43,9 @@ enum { OPT_INPUT_FILE, OPT_BUILD_FLAGS, OPT_LIST_ONLY, + OPT_VALIDATE_FILTERS, OPT_KEY_PREFIX, + OPT_FILTERS, OPT_HEX_KEY_PREFIX, OPT_BEGIN_VERSION_FILTER, OPT_END_VERSION_FILTER, @@ -53,6 +56,8 @@ enum { CSimpleOpt::SOption gConverterOptions[] = { { OPT_CONTAINER, "-r", SO_REQ_SEP }, { OPT_CONTAINER, "--container", SO_REQ_SEP }, + { OPT_FILE_TYPE, "-t", SO_REQ_SEP }, + { OPT_FILE_TYPE, "--file-type", SO_REQ_SEP }, { OPT_BEGIN_VERSION, "-b", SO_REQ_SEP }, { OPT_BEGIN_VERSION, "--begin", SO_REQ_SEP }, { OPT_CRASHONERROR, "--crash", SO_NONE }, @@ -70,7 +75,9 @@ CSimpleOpt::SOption gConverterOptions[] = { { OPT_CONTAINER, "-r", SO_REQ_SEP }, #endif { OPT_BUILD_FLAGS, "--build-flags", SO_NONE }, { OPT_LIST_ONLY, "--list-only", SO_NONE }, + { OPT_VALIDATE_FILTERS, "--validate-filters", SO_NONE }, { OPT_KEY_PREFIX, "-k", SO_REQ_SEP }, + { OPT_FILTERS, "--filters", SO_REQ_SEP }, { OPT_HEX_KEY_PREFIX, "--hex-prefix", SO_REQ_SEP }, { OPT_BEGIN_VERSION_FILTER, "--begin-version-filter", SO_REQ_SEP }, { OPT_END_VERSION_FILTER, "--end-version-filter", SO_REQ_SEP }, diff --git a/fdbbackup/FileDecoder.actor.cpp b/fdbbackup/FileDecoder.actor.cpp index c13e6f68771..1d079a61f63 100644 --- a/fdbbackup/FileDecoder.actor.cpp +++ b/fdbbackup/FileDecoder.actor.cpp @@ -20,12 +20,15 @@ #include #include +#include #include #include #include +#include #include #include #include + #ifdef _WIN32 #include #endif @@ -35,12 +38,16 @@ #include "fdbbackup/FileConverter.h" #include "fdbclient/BackupAgent.actor.h" #include "fdbclient/BackupContainer.h" +#include "fdbclient/BackupContainerFileSystem.h" #include "fdbclient/CommitTransaction.h" #include "fdbclient/FDBTypes.h" #include "fdbclient/IKnobCollection.h" +#include "fdbclient/KeyRangeMap.h" #include "fdbclient/Knobs.h" #include "fdbclient/MutationList.h" +#include "fdbclient/SystemData.h" #include "flow/ArgParseUtil.h" +#include "flow/FastRef.h" #include "flow/IRandom.h" #include "flow/Platform.h" #include "flow/Trace.h" @@ -76,14 +83,20 @@ void printDecodeUsage() { " --blob-credentials FILE\n" " File containing blob credentials in JSON format.\n" " The same credential format/file fdbbackup uses.\n" + " -t, --file-type [log|range|both]\n" + " Specifies the backup file type to decode.\n" #ifndef TLS_DISABLED TLS_HELP #endif " --build-flags Print build information and exit.\n" " --list-only Print file list and exit.\n" - " -k KEY_PREFIX Use the prefix for filtering mutations\n" + " --validate-filters Validate the default RangeMap filtering logic with a slower one.\n" + " -k KEY_PREFIX Use a single prefix for filtering mutations.\n" + " --filters PREFIX_FILTER_FILE\n" + " A file containing a list of prefix filters in HEX format separated by \";\",\n" + " e.g., \"\\x05\\x01;\\x15\\x2b\"\n" " --hex-prefix HEX_PREFIX\n" - " The prefix specified in HEX format, e.g., \"\\\\x05\\\\x01\".\n" + " The prefix specified in HEX format, e.g., --hex-prefix \"\\\\x05\\\\x01\".\n" " --begin-version-filter BEGIN_VERSION\n" " The version range's begin version (inclusive) for filtering.\n" " --end-version-filter END_VERSION\n" @@ -99,7 +112,7 @@ void printBuildInformation() { std::cout << jsonBuildInformation() << "\n"; } -struct DecodeParams { +struct DecodeParams : public ReferenceCounted { std::string container_url; Optional proxy; std::string fileFilter; // only files match the filter will be decoded @@ -107,8 +120,13 @@ struct DecodeParams { std::string log_dir, trace_format, trace_log_group; BackupTLSConfig tlsConfig; bool list_only = false; + bool decode_logs = true; + bool decode_range = true; bool save_file_locally = false; - std::string prefix; // Key prefix for filtering + bool validate_filters = false; + std::vector prefixes; // Key prefixes for filtering + // more efficient data structure for intersection queries than "prefixes" + fileBackup::RangeMapFilters filters; Version beginVersionFilter = 0; Version endVersionFilter = std::numeric_limits::max(); @@ -120,6 +138,70 @@ struct DecodeParams { return !(begin >= endVersionFilter || end <= beginVersionFilter); } + bool overlap(Version version) const { return version >= beginVersionFilter && version < endVersionFilter; } + + void updateRangeMap() { filters.updateFilters(prefixes); } + + bool matchFilters(const MutationRef& m) const { + bool match = filters.match(m); + if (!validate_filters) { + return match; + } + + // If we choose to validate the filters, go through filters one by one + for (const auto& prefix : prefixes) { + if (isSingleKeyMutation((MutationRef::Type)m.type)) { + if (m.param1.startsWith(StringRef(prefix))) { + ASSERT(match); + return true; + } + } else if (m.type == MutationRef::ClearRange) { + KeyRange range(KeyRangeRef(m.param1, m.param2)); + KeyRange range2 = prefixRange(StringRef(prefix)); + if (range.intersects(range2)) { + ASSERT(match); + return true; + } + } else { + ASSERT(false); + } + } + ASSERT(!match); + return false; + } + + bool matchFilters(const KeyRange& range) const { + bool match = filters.match(range); + if (!validate_filters) { + return match; + } + + for (const auto& prefix : prefixes) { + if (range.intersects(prefixRange(StringRef(prefix)))) { + ASSERT(match); + return true; + } + } + return false; + } + + bool matchFilters(KeyValueRef kv) const { + bool match = filters.match(kv); + + if (!validate_filters) { + return match; + } + + for (const auto& prefix : prefixes) { + if (kv.key.startsWith(StringRef(prefix))) { + ASSERT(match); + return true; + } + } + + return match; + } + std::string toString() { std::string s; s.append("ContainerURL: "); @@ -142,14 +224,15 @@ struct DecodeParams { } } s.append(", list_only: ").append(list_only ? "true" : "false"); + s.append(", validate_filters: ").append(validate_filters ? "true" : "false"); if (beginVersionFilter != 0) { s.append(", beginVersionFilter: ").append(std::to_string(beginVersionFilter)); } if (endVersionFilter < std::numeric_limits::max()) { s.append(", endVersionFilter: ").append(std::to_string(endVersionFilter)); } - if (!prefix.empty()) { - s.append(", KeyPrefix: ").append(printable(KeyRef(prefix))); + if (!prefixes.empty()) { + s.append(", KeyPrefixes: ").append(printable(describe(prefixes))); } for (const auto& [knob, value] : knobs) { s.append(", KNOB-").append(knob).append(" = ").append(value); @@ -167,8 +250,10 @@ struct DecodeParams { }; // Decode an ASCII string, e.g., "\x15\x1b\x19\x04\xaf\x0c\x28\x0a", -// into the binary string. -std::string decode_hex_string(std::string line) { +// into the binary string. Set "err" to true if the format is invalid. +// Note ',' '\' '," ';' are escaped by '\'. Normal characters can be +// unencoded into HEX, but not recommended. +std::string decode_hex_string(std::string line, bool& err) { size_t i = 0; std::string ret; @@ -177,6 +262,7 @@ std::string decode_hex_string(std::string line) { case '\\': if (i + 2 > line.length()) { std::cerr << "Invalid hex string at: " << i << "\n"; + err = true; return ret; } switch (line[i + 1]) { @@ -190,6 +276,7 @@ std::string decode_hex_string(std::string line) { case 'x': if (i + 4 > line.length()) { std::cerr << "Invalid hex string at: " << i << "\n"; + err = true; return ret; } char* pEnd; @@ -205,6 +292,7 @@ std::string decode_hex_string(std::string line) { break; default: std::cerr << "Invalid hex string at: " << i << "\n"; + err = true; return ret; } default: @@ -215,7 +303,37 @@ std::string decode_hex_string(std::string line) { return line.substr(0, i); } -int parseDecodeCommandLine(DecodeParams* param, CSimpleOpt* args) { +// Parses and returns a ";" separated HEX encoded strings. So the ";" in +// the string should be escaped as "\;". +// Sets "err" to true if there is any parsing error. +std::vector parsePrefixesLine(const std::string& line, bool& err) { + std::vector results; + err = false; + + int p = 0; + while (p < line.size()) { + int end = line.find_first_of(';', p); + if (end == line.npos) { + end = line.size(); + } + auto prefix = decode_hex_string(line.substr(p, end - p), err); + if (err) { + return results; + } + results.push_back(prefix); + p = end + 1; + } + return results; +} + +std::vector parsePrefixFile(const std::string& filename, bool& err) { + std::string line = readFileBytes(filename, 64 * 1024 * 1024); + return parsePrefixesLine(line, err); +} + +int parseDecodeCommandLine(Reference param, CSimpleOpt* args) { + bool err = false; + while (args->Next()) { auto lastError = args->LastError(); switch (lastError) { @@ -236,16 +354,41 @@ int parseDecodeCommandLine(DecodeParams* param, CSimpleOpt* args) { param->container_url = args->OptionArg(); break; + case OPT_FILE_TYPE: { + auto ftype = std::string(args->OptionArg()); + if (ftype == "log") { + param->decode_range = false; + } else if (ftype == "range") { + param->decode_logs = false; + } else if (ftype != "both" && ftype != "") { + err = true; + std::cerr << "ERROR: Unrecognized backup file type option: " << args->OptionArg() << "\n"; + return FDB_EXIT_ERROR; + } + break; + } + case OPT_LIST_ONLY: param->list_only = true; break; + case OPT_VALIDATE_FILTERS: + param->validate_filters = true; + break; + case OPT_KEY_PREFIX: - param->prefix = args->OptionArg(); + param->prefixes.push_back(args->OptionArg()); + break; + + case OPT_FILTERS: + param->prefixes = parsePrefixFile(args->OptionArg(), err); + if (err) { + throw std::runtime_error("ERROR:" + std::string(args->OptionArg()) + "contains invalid prefix(es)"); + } break; case OPT_HEX_KEY_PREFIX: - param->prefix = decode_hex_string(args->OptionArg()); + param->prefixes.push_back(decode_hex_string(args->OptionArg(), err)); break; case OPT_BEGIN_VERSION_FILTER: @@ -337,19 +480,31 @@ int parseDecodeCommandLine(DecodeParams* param, CSimpleOpt* args) { return FDB_EXIT_SUCCESS; } -void printLogFiles(std::string msg, const std::vector& files) { - std::cout << msg << " " << files.size() << " log files\n"; +template +void printLogFiles(std::string msg, const std::vector& files) { + std::cout << msg << " " << files.size() << " total\n"; for (const auto& file : files) { std::cout << file.toString() << "\n"; } std::cout << std::endl; } -std::vector getRelevantLogFiles(const std::vector& files, const DecodeParams& params) { +std::vector getRelevantLogFiles(const std::vector& files, const Reference params) { std::vector filtered; for (const auto& file : files) { - if (file.fileName.find(params.fileFilter) != std::string::npos && - params.overlap(file.beginVersion, file.endVersion + 1)) { + if (file.fileName.find(params->fileFilter) != std::string::npos && + params->overlap(file.beginVersion, file.endVersion + 1)) { + filtered.push_back(file); + } + } + return filtered; +} + +std::vector getRelevantRangeFiles(const std::vector& files, + const Reference params) { + std::vector filtered; + for (const auto& file : files) { + if (file.fileName.find(params->fileFilter) != std::string::npos && params->overlap(file.version)) { filtered.push_back(file); } } @@ -424,6 +579,12 @@ class DecodeProgress { ACTOR static Future openFileImpl(DecodeProgress* self, Reference container) { Reference fd = wait(container->readFile(self->file.fileName)); self->fd = fd; + state Standalone buf = makeString(self->file.fileSize); + int rLen = wait(self->fd->read(mutateString(buf), self->file.fileSize, 0)); + if (rLen != self->file.fileSize) { + throw restore_bad_read(); + } + if (self->save) { std::string dir = self->file.fileName; std::size_t found = self->file.fileName.find_last_of('/'); @@ -438,10 +599,17 @@ class DecodeProgress { TraceEvent(SevError, "OpenLocalFileFailed").detail("File", self->file.fileName); throw platform_error(); } + int wlen = write(self->lfd, buf.begin(), self->file.fileSize); + if (wlen != self->file.fileSize) { + TraceEvent(SevError, "WriteLocalFileFailed") + .detail("File", self->file.fileName) + .detail("Len", self->file.fileSize); + throw platform_error(); + } + TraceEvent("WriteLocalFile").detail("Name", self->file.fileName).detail("Len", self->file.fileSize); } - while (!self->eof) { - wait(readAndDecodeFile(self)); - } + + self->decodeFile(buf); return Void(); } @@ -453,77 +621,182 @@ class DecodeProgress { } } - // Reads a file block, decodes it into key/value pairs, and stores these pairs. - ACTOR static Future readAndDecodeFile(DecodeProgress* self) { + // Reads a file a file content in the buffer, decodes it into key/value pairs, and stores these pairs. + void decodeFile(const Standalone& buf) { try { - state int64_t len = std::min(self->file.blockSize, self->file.fileSize - self->offset); - if (len == 0) { - self->eof = true; - return Void(); + loop { + int64_t len = std::min(file.blockSize, file.fileSize - offset); + if (len == 0) { + return; + } + + // Decode a file block into log_key and log_value chunks + Standalone> chunks = + fileBackup::decodeMutationLogFileBlock(buf.substr(offset, len)); + blocks.push_back(chunks); + addBlockKVPairs(chunks); + offset += len; } + } catch (Error& e) { + TraceEvent(SevWarn, "CorruptLogFileBlock") + .error(e) + .detail("Filename", file.fileName) + .detail("BlockOffset", offset) + .detail("BlockLen", file.blockSize); + throw; + } + } - // Decode a file block into log_key and log_value chunks - state Standalone> chunks = - wait(fileBackup::decodeMutationLogFileBlock(self->fd, self->offset, len)); - self->blocks.push_back(chunks); - - if (self->save) { - ASSERT(self->lfd != -1); - - // Read the chunck one more time - state Standalone buf = makeString(len); - int rLen = wait(self->fd->read(mutateString(buf), len, self->offset)); - if (rLen != len) - throw restore_bad_read(); - - int wlen = write(self->lfd, buf.begin(), len); - if (wlen != len) { - TraceEvent(SevError, "WriteLocalFileFailed") - .detail("File", self->file.fileName) - .detail("Offset", self->offset) - .detail("Len", len) - .detail("Wrote", wlen); - throw platform_error(); + LogFile file; + Reference fd; + int64_t offset = 0; + bool eof = false; + bool save = false; + int lfd = -1; // local file descriptor +}; + +class DecodeRangeProgress { +public: + std::vector>> blocks; + + DecodeRangeProgress() = default; + DecodeRangeProgress(const RangeFile& file, bool save) : file(file), save(save) {} + ~DecodeRangeProgress() { + if (lfd != -1) { + close(lfd); + } + } + + // Open and loads file into memory + Future openFile(Reference container) { return openFileImpl(this, container); } + + ACTOR static Future openFileImpl(DecodeRangeProgress* self, Reference container) { + TraceEvent("ReadFile").detail("Name", self->file.fileName).detail("Len", self->file.fileSize); + + Reference fd = wait(container->readFile(self->file.fileName)); + self->fd = fd; + state Standalone buf = makeString(self->file.fileSize); + int rLen = wait(self->fd->read(mutateString(buf), self->file.fileSize, 0)); + if (rLen != self->file.fileSize) { + throw restore_bad_read(); + } + + if (self->save) { + std::string dir = self->file.fileName; + std::size_t found = self->file.fileName.find_last_of('/'); + if (found != std::string::npos) { + std::string path = self->file.fileName.substr(0, found); + if (!directoryExists(path)) { + platform::createDirectory(path); } - TraceEvent("WriteLocalFile") - .detail("Name", self->file.fileName) - .detail("Len", len) - .detail("Offset", self->offset); } - TraceEvent("ReadFile") - .detail("Name", self->file.fileName) - .detail("Len", len) - .detail("Offset", self->offset); - self->addBlockKVPairs(chunks); - self->offset += len; + self->lfd = open(self->file.fileName.c_str(), O_WRONLY | O_CREAT | O_TRUNC); + if (self->lfd == -1) { + TraceEvent(SevError, "OpenLocalFileFailed").detail("File", self->file.fileName); + throw platform_error(); + } + int wlen = write(self->lfd, buf.begin(), self->file.fileSize); + if (wlen != self->file.fileSize) { + TraceEvent(SevError, "WriteLocalFileFailed") + .detail("File", self->file.fileName) + .detail("Len", self->file.fileSize); + throw platform_error(); + } + TraceEvent("WriteLocalFile").detail("Name", self->file.fileName).detail("Len", self->file.fileSize); + } + + self->decodeFile(buf); + return Void(); + } + + // Reads a file content in the buffer, decodes it into key/value pairs, and stores these pairs. + void decodeFile(const Standalone& buf) { + try { + loop { + // process one block at a time + int64_t len = std::min(file.blockSize, file.fileSize - offset); + if (len == 0) { + return; + } - return Void(); + Standalone> chunks = fileBackup::decodeRangeFileBlock(buf.substr(offset, len)); + blocks.push_back(chunks); + offset += len; + } } catch (Error& e) { - TraceEvent(SevWarn, "CorruptLogFileBlock") + TraceEvent(SevWarn, "CorruptRangeFileBlock") .error(e) - .detail("Filename", self->file.fileName) - .detail("BlockOffset", self->offset) - .detail("BlockLen", self->file.blockSize); + .detail("Filename", file.fileName) + .detail("BlockOffset", offset) + .detail("BlockLen", file.blockSize); throw; } } - LogFile file; + RangeFile file; Reference fd; int64_t offset = 0; - bool eof = false; bool save = false; int lfd = -1; // local file descriptor }; -ACTOR Future process_file(Reference container, LogFile file, UID uid, DecodeParams params) { +// convert a StringRef to Hex string +std::string hexStringRef(const StringRef& s) { + std::string result; + result.reserve(s.size() * 2); + for (int i = 0; i < s.size(); i++) { + result.append(format("%02x", s[i])); + } + return result; +} + +ACTOR Future process_range_file(Reference container, + RangeFile file, + UID uid, + Reference params) { + + if (file.fileSize == 0) { + TraceEvent("SkipEmptyFile", uid).detail("Name", file.fileName); + return Void(); + } + + state DecodeRangeProgress progress(file, params->save_file_locally); + wait(progress.openFile(container)); + + for (auto& block : progress.blocks) { + for (const auto& kv : block) { + bool print = params->prefixes.empty(); // no filtering + + if (!print) { + print = params->matchFilters(kv); + } + + if (print) { + TraceEvent(format("KVPair_%llu", file.version).c_str(), uid) + .detail("Version", file.version) + .setMaxFieldLength(1000) + .detail("KV", kv); + std::cout << file.version << " key: " << hexStringRef(kv.key) << " value: " << hexStringRef(kv.value) + << std::endl; + } + } + } + + TraceEvent("ProcessRangeFileDone", uid).detail("File", file.fileName); + return Void(); +} + +ACTOR Future process_file(Reference container, + LogFile file, + UID uid, + Reference params) { if (file.fileSize == 0) { TraceEvent("SkipEmptyFile", uid).detail("Name", file.fileName); return Void(); } - state DecodeProgress progress(file, params.save_file_locally); + state DecodeProgress progress(file, params->save_file_locally); wait(progress.openFile(container)); while (true) { auto batch = progress.getNextBatch(); @@ -531,7 +804,7 @@ ACTOR Future process_file(Reference container, LogFile f break; const VersionedMutations& vms = batch.get(); - if (vms.version < params.beginVersionFilter || vms.version >= params.endVersionFilter) { + if (vms.version < params->beginVersionFilter || vms.version >= params->endVersionFilter) { TraceEvent("SkipVersion").detail("Version", vms.version); continue; } @@ -539,25 +812,18 @@ ACTOR Future process_file(Reference container, LogFile f int sub = 0; for (const auto& m : vms.mutations) { sub++; // sub sequence number starts at 1 - bool print = params.prefix.empty(); // no filtering + bool print = params->prefixes.empty(); // no filtering if (!print) { - if (isSingleKeyMutation((MutationRef::Type)m.type)) { - print = m.param1.startsWith(StringRef(params.prefix)); - } else if (m.type == MutationRef::ClearRange) { - KeyRange range(KeyRangeRef(m.param1, m.param2)); - KeyRange range2 = prefixRange(StringRef(params.prefix)); - print = range.intersects(range2); - } else { - ASSERT(false); - } + print = params->matchFilters(m); } if (print) { TraceEvent(format("Mutation_%llu_%d", vms.version, sub).c_str(), uid) .detail("Version", vms.version) - .setMaxFieldLength(10000) + .setMaxFieldLength(1000) .detail("M", m.toString()); - std::cout << vms.version << " " << m.toString() << "\n"; + std::cout << vms.version << "." << sub << " " << typeString[(int)m.type] + << " param1: " << hexStringRef(m.param1) << " param2: " << hexStringRef(m.param2) << "\n"; } } } @@ -565,9 +831,37 @@ ACTOR Future process_file(Reference container, LogFile f return Void(); } -ACTOR Future decode_logs(DecodeParams params) { +// Use the snapshot metadata to quickly identify relevant range files and +// then filter by versions. +ACTOR Future> getRangeFiles(Reference bc, Reference params) { + state std::vector snapshots = + wait((dynamic_cast(bc.getPtr()))->listKeyspaceSnapshots()); + state std::vector files; + + state int i = 0; + for (; i < snapshots.size(); i++) { + try { + std::pair, std::map> results = + wait((dynamic_cast(bc.getPtr()))->readKeyspaceSnapshot(snapshots[i])); + for (const auto& rangeFile : results.first) { + const auto& keyRange = results.second.at(rangeFile.fileName); + if (params->matchFilters(keyRange)) { + files.push_back(rangeFile); + } + } + } catch (Error& e) { + TraceEvent("ReadKeyspaceSnapshotError").error(e).detail("I", i); + if (e.code() != error_code_restore_missing_data) { + throw; + } + } + } + return getRelevantRangeFiles(files, params); +} + +ACTOR Future decode_logs(Reference params) { state Reference container = - IBackupContainer::openContainer(params.container_url, params.proxy, {}); + IBackupContainer::openContainer(params->container_url, params->proxy, {}); state UID uid = deterministicRandom()->randomUniqueID(); state BackupFileList listing = wait(container->dumpFileList()); // remove partitioned logs @@ -579,25 +873,54 @@ ACTOR Future decode_logs(DecodeParams params) { }), listing.logs.end()); std::sort(listing.logs.begin(), listing.logs.end()); - TraceEvent("Container", uid).detail("URL", params.container_url).detail("Logs", listing.logs.size()); - TraceEvent("DecodeParam", uid).setMaxFieldLength(100000).detail("Value", params.toString()); + TraceEvent("Container", uid).detail("URL", params->container_url).detail("Logs", listing.logs.size()); + TraceEvent("DecodeParam", uid).setMaxFieldLength(100000).detail("Value", params->toString()); BackupDescription desc = wait(container->describeBackup()); std::cout << "\n" << desc.toString() << "\n"; - state std::vector logs = getRelevantLogFiles(listing.logs, params); - printLogFiles("Relevant files are: ", logs); + state std::vector logFiles; + state std::vector rangeFiles; - if (params.list_only) + if (params->decode_logs) { + logFiles = getRelevantLogFiles(listing.logs, params); + printLogFiles("Relevant log files are: ", logFiles); + } + + if (params->decode_range) { + // rangeFiles = getRelevantRangeFiles(filteredRangeFiles, params); + std::vector files = wait(getRangeFiles(container, params)); + rangeFiles = files; + printLogFiles("Releavant range files are: ", rangeFiles); + } + + TraceEvent("TotalFiles", uid).detail("LogFiles", logFiles.size()).detail("RangeFiles", rangeFiles.size()); + + if (params->list_only) return Void(); + // Decode log files. state int idx = 0; - while (idx < logs.size()) { - TraceEvent("ProcessFile").detail("Name", logs[idx].fileName).detail("I", idx); - wait(process_file(container, logs[idx], uid, params)); - idx++; + if (params->decode_logs) { + while (idx < logFiles.size()) { + TraceEvent("ProcessFile").detail("Name", logFiles[idx].fileName).detail("I", idx); + wait(process_file(container, logFiles[idx], uid, params)); + idx++; + } + TraceEvent("DecodeLogsDone", uid).log(); } - TraceEvent("DecodeDone", uid).log(); + + // Decode range files. + if (params->decode_range) { + idx = 0; + while (idx < rangeFiles.size()) { + TraceEvent("ProcessFile").detail("Name", rangeFiles[idx].fileName).detail("I", idx); + wait(process_range_file(container, rangeFiles[idx], uid, params)); + idx++; + } + TraceEvent("DecodeRangeFileDone", uid).log(); + } + return Void(); } @@ -607,31 +930,32 @@ int main(int argc, char** argv) { try { std::unique_ptr args( new CSimpleOpt(argc, argv, file_converter::gConverterOptions, SO_O_EXACT | SO_O_HYPHEN_TO_UNDERSCORE)); - file_converter::DecodeParams param; - int status = file_converter::parseDecodeCommandLine(¶m, args.get()); - std::cout << "Params: " << param.toString() << "\n"; + auto param = makeReference(); + int status = file_converter::parseDecodeCommandLine(param, args.get()); + std::cout << "Params: " << param->toString() << "\n"; + param->updateRangeMap(); if (status != FDB_EXIT_SUCCESS) { file_converter::printDecodeUsage(); return status; } - if (param.log_enabled) { - if (param.log_dir.empty()) { + if (param->log_enabled) { + if (param->log_dir.empty()) { setNetworkOption(FDBNetworkOptions::TRACE_ENABLE); } else { - setNetworkOption(FDBNetworkOptions::TRACE_ENABLE, StringRef(param.log_dir)); + setNetworkOption(FDBNetworkOptions::TRACE_ENABLE, StringRef(param->log_dir)); } - if (!param.trace_format.empty()) { - setNetworkOption(FDBNetworkOptions::TRACE_FORMAT, StringRef(param.trace_format)); + if (!param->trace_format.empty()) { + setNetworkOption(FDBNetworkOptions::TRACE_FORMAT, StringRef(param->trace_format)); } else { setNetworkOption(FDBNetworkOptions::TRACE_FORMAT, "json"_sr); } - if (!param.trace_log_group.empty()) { - setNetworkOption(FDBNetworkOptions::TRACE_LOG_GROUP, StringRef(param.trace_log_group)); + if (!param->trace_log_group.empty()) { + setNetworkOption(FDBNetworkOptions::TRACE_LOG_GROUP, StringRef(param->trace_log_group)); } } - if (!param.tlsConfig.setupTLS()) { + if (!param->tlsConfig.setupTLS()) { TraceEvent(SevError, "TLSError").log(); throw tls_error(); } @@ -639,15 +963,15 @@ int main(int argc, char** argv) { platformInit(); Error::init(); - StringRef url(param.container_url); + StringRef url(param->container_url); setupNetwork(0, UseMetrics::True); // Must be called after setupNetwork() to be effective - param.updateKnobs(); + param->updateKnobs(); TraceEvent::setNetworkThread(); - openTraceFile(NetworkAddress(), 10 << 20, 500 << 20, param.log_dir, "decode", param.trace_log_group); - param.tlsConfig.setupBlobCredentials(); + openTraceFile(NetworkAddress(), 10 << 20, 500 << 20, param->log_dir, "decode", param->trace_log_group); + param->tlsConfig.setupBlobCredentials(); auto f = stopAfter(decode_logs(param)); diff --git a/fdbbackup/backup.actor.cpp b/fdbbackup/backup.actor.cpp index 81db05f01d6..c3bd7e869bf 100644 --- a/fdbbackup/backup.actor.cpp +++ b/fdbbackup/backup.actor.cpp @@ -134,6 +134,7 @@ enum { OPT_PROXY, OPT_TAGNAME, OPT_BACKUPKEYS, + OPT_BACKUPKEYS_FILE, OPT_WAITFORDONE, OPT_BACKUPKEYS_FILTER, OPT_INCREMENTALONLY, @@ -146,6 +147,7 @@ enum { // Restore constants OPT_RESTORECONTAINER, OPT_RESTORE_VERSION, + OPT_RESTORE_SNAPSHOT_VERSION, OPT_RESTORE_TIMESTAMP, OPT_PREFIX_ADD, OPT_PREFIX_REMOVE, @@ -248,6 +250,7 @@ CSimpleOpt::SOption g_rgBackupStartOptions[] = { { OPT_TAGNAME, "-t", SO_REQ_SEP }, { OPT_TAGNAME, "--tagname", SO_REQ_SEP }, { OPT_BACKUPKEYS, "-k", SO_REQ_SEP }, + { OPT_BACKUPKEYS_FILE, "--keys-file", SO_REQ_SEP }, { OPT_BACKUPKEYS, "--keys", SO_REQ_SEP }, { OPT_DRYRUN, "-n", SO_NONE }, { OPT_DRYRUN, "--dryrun", SO_NONE }, @@ -679,6 +682,7 @@ CSimpleOpt::SOption g_rgBackupQueryOptions[] = { { OPT_PROXY, "--proxy", SO_REQ_SEP }, { OPT_RESTORE_VERSION, "-qrv", SO_REQ_SEP }, { OPT_RESTORE_VERSION, "--query-restore-version", SO_REQ_SEP }, + { OPT_RESTORE_SNAPSHOT_VERSION, "--query-restore-snapshot-version", SO_REQ_SEP }, { OPT_BACKUPKEYS_FILTER, "-k", SO_REQ_SEP }, { OPT_BACKUPKEYS_FILTER, "--keys", SO_REQ_SEP }, { OPT_TRACE, "--log", SO_NONE }, @@ -721,6 +725,7 @@ CSimpleOpt::SOption g_rgRestoreOptions[] = { { OPT_TAGNAME, "-t", SO_REQ_SEP }, { OPT_TAGNAME, "--tagname", SO_REQ_SEP }, { OPT_BACKUPKEYS, "-k", SO_REQ_SEP }, + { OPT_BACKUPKEYS_FILE, "--keys-file", SO_REQ_SEP }, { OPT_BACKUPKEYS, "--keys", SO_REQ_SEP }, { OPT_WAITFORDONE, "-w", SO_NONE }, { OPT_WAITFORDONE, "--waitfordone", SO_NONE }, @@ -798,6 +803,7 @@ CSimpleOpt::SOption g_rgDBStartOptions[] = { { OPT_TAGNAME, "-t", SO_REQ_SEP }, { OPT_TAGNAME, "--tagname", SO_REQ_SEP }, { OPT_BACKUPKEYS, "-k", SO_REQ_SEP }, + { OPT_BACKUPKEYS_FILE, "--keys-file", SO_REQ_SEP }, { OPT_BACKUPKEYS, "--keys", SO_REQ_SEP }, { OPT_TRACE, "--log", SO_NONE }, { OPT_TRACE_DIR, "--logdir", SO_REQ_SEP }, @@ -1088,9 +1094,13 @@ static void printBackupUsage(bool devhelp) { "containing no data at or after a\n" " version approximately NUM_DAYS days worth of versions prior to the latest log version in " "the backup.\n"); + printf(" --query-restore-snapshot-version VERSION\n" + " For query operations, set the snapshot version, inclusive, used to restore a backup.\n" + " Set -1 to use the latest valid snapshot,\n" + " Set -3 to use the oldest valid snapshot.\n"); printf(" -qrv --query-restore-version VERSION\n" " For query operations, set target version for restoring a backup. Set -1 for maximum\n" - " restorable version (default) and -2 for minimum restorable version.\n"); + " restorable version (default) and -3 for minimum restorable version.\n"); printf( " --query-restore-timestamp DATETIME\n" " For query operations, instead of a numeric version, use this to specify a timestamp in %s\n", @@ -1125,6 +1135,8 @@ static void printBackupUsage(bool devhelp) { printf(" -e ERRORLIMIT The maximum number of errors printed by status (default is 10).\n"); printf(" -k KEYS List of key ranges to backup or to filter the backup in query operations.\n" " If not specified, the entire database will be backed up or no filter will be applied.\n"); + printf(" --keys-file FILE\n" + " Same as -k option, except keys are specified in the input file.\n"); printf(" --partitioned-log-experimental Starts with new type of backup system using partitioned logs.\n"); printf(" -n, --dryrun For backup start or restore start, performs a trial run with no actual changes made.\n"); printf(" --log Enables trace file logging for the CLI session.\n" @@ -1198,6 +1210,8 @@ static void printRestoreUsage(bool devhelp) { printf(" -w, --waitfordone\n"); printf(" Wait for the restore to complete before exiting. Prints progress updates.\n"); printf(" -k KEYS List of key ranges from the backup to restore.\n"); + printf(" --keys-file FILE\n" + " Same as -k option, except keys are specified in the input file.\n"); printf(" --remove-prefix PREFIX\n"); printf(" Prefix to remove from the restored keys.\n"); printf(" --add-prefix PREFIX\n"); @@ -1321,6 +1335,8 @@ static void printDBBackupUsage(bool devhelp) { printf(" -e ERRORLIMIT The maximum number of errors printed by status (default is 10).\n"); printf(" -k KEYS List of key ranges to backup.\n" " If not specified, the entire database will be backed up.\n"); + printf(" --keys-file FILE\n" + " Same as -k option, except keys are specified in the input file.\n"); printf(" --cleanup Abort will attempt to stop mutation logging on the source cluster.\n"); printf(" --dstonly Abort will not make any changes on the source cluster.\n"); #ifndef TLS_DISABLED @@ -2636,7 +2652,9 @@ ACTOR Future expireBackupData(const char* name, lastProgress = p; } } - when(wait(expire)) { break; } + when(wait(expire)) { + break; + } } } @@ -2679,7 +2697,9 @@ ACTOR Future deleteBackupContainer(const char* name, loop { choose { - when(wait(done)) { break; } + when(wait(done)) { + break; + } when(wait(delay(5))) { if (numDeleted != lastUpdate) { printf("\r%d...", numDeleted); @@ -2729,6 +2749,24 @@ static void reportBackupQueryError(UID operationId, JsonBuilderObject& result, s TraceEvent("BackupQueryFailure").detail("OperationId", operationId).detail("Reason", errorMessage); } +std::pair getMaxMinRestorableVersions(const BackupDescription& desc, bool mayOnlyApplyMutationLog) { + Version maxRestorableVersion = invalidVersion; + Version minRestorableVersion = invalidVersion; + if (desc.maxRestorableVersion.present()) { + maxRestorableVersion = desc.maxRestorableVersion.get(); + } else if (mayOnlyApplyMutationLog && desc.contiguousLogEnd.present()) { + maxRestorableVersion = desc.contiguousLogEnd.get(); + } + + if (desc.minRestorableVersion.present()) { + minRestorableVersion = desc.minRestorableVersion.get(); + } else if (mayOnlyApplyMutationLog && desc.minLogBegin.present()) { + minRestorableVersion = desc.minLogBegin.get(); + } + + return std::make_pair(maxRestorableVersion, minRestorableVersion); +} + // If restoreVersion is invalidVersion or latestVersion, use the maximum or minimum restorable version respectively for // selected key ranges. If restoreTimestamp is specified, any specified restoreVersion will be overriden to the version // resolved to that timestamp. @@ -2737,6 +2775,7 @@ ACTOR Future queryBackup(const char* name, Optional proxy, Standalone> keyRangesFilter, Version restoreVersion, + Version snapshotVersion, std::string originalClusterFile, std::string restoreTimestamp, Verbose verbose) { @@ -2751,6 +2790,7 @@ ACTOR Future queryBackup(const char* name, .detail("DestinationContainer", destinationContainer) .detail("KeyRangesFilter", printable(keyRangesFilter)) .detail("SpecifiedRestoreVersion", restoreVersion) + .detail("SpecifiedSnapshotVersion", snapshotVersion) .detail("RestoreTimestamp", restoreTimestamp) .detail("BackupClusterFile", originalClusterFile); @@ -2780,36 +2820,93 @@ ACTOR Future queryBackup(const char* name, restoreVersion = v; } + state int64_t totalRangeFilesSize = 0; + state int64_t totalLogFilesSize = 0; + state JsonBuilderArray rangeFilesJson; + state JsonBuilderArray logFilesJson; try { state Reference bc = openBackupContainer(name, destinationContainer, proxy, {}); + BackupDescription desc = wait(bc->describeBackup()); + // Use continuous log end version for the maximum restorable version for the key ranges when a restorable + // version doesn't exist. + auto [maxRestorableVersion, minRestorableVersion] = getMaxMinRestorableVersions(desc, !keyRangesFilter.empty()); if (restoreVersion == invalidVersion) { - BackupDescription desc = wait(bc->describeBackup()); - if (desc.maxRestorableVersion.present()) { - restoreVersion = desc.maxRestorableVersion.get(); - // Use continuous log end version for the maximum restorable version for the key ranges. - } else if (keyRangesFilter.size() && desc.contiguousLogEnd.present()) { - restoreVersion = desc.contiguousLogEnd.get(); - } else { - reportBackupQueryError( - operationId, - result, - errorMessage = format("the backup for the specified key ranges is not restorable to any version")); - } + restoreVersion = maxRestorableVersion; } - - if (restoreVersion < 0 && restoreVersion != latestVersion) { + if (restoreVersion == earliestVersion) { + restoreVersion = minRestorableVersion; + } + if (snapshotVersion == earliestVersion) { + snapshotVersion = minRestorableVersion; + } + TraceEvent("BackupQueryResolveRestoreVersion") + .detail("RestoreVersion", restoreVersion) + .detail("SnapshotVersion", snapshotVersion); + if (restoreVersion < 0) { reportBackupQueryError(operationId, result, errorMessage = format("the specified restorable version %lld is not valid", restoreVersion)); return Void(); } - Optional fileSet = wait(bc->getRestoreSet(restoreVersion, keyRangesFilter)); + + state Optional fileSet; + if (snapshotVersion != invalidVersion) { + // When a snapshot version is specified, we will first get a restore set using the latest snapshot file to + // restore to the snapshot version. After snapshot version, we will only use mutation logs to restore. + wait(store(fileSet, bc->getRestoreSet(snapshotVersion, keyRangesFilter))); + if (fileSet.present()) { + result["snapshot_version"] = fileSet.get().targetVersion; + for (const auto& rangeFile : fileSet.get().ranges) { + JsonBuilderObject object; + object["file_name"] = rangeFile.fileName; + object["file_size"] = rangeFile.fileSize; + object["version"] = rangeFile.version; + object["key_range"] = fileSet.get().keyRanges.count(rangeFile.fileName) == 0 + ? "none" + : fileSet.get().keyRanges.at(rangeFile.fileName).toString(); + rangeFilesJson.push_back(object); + totalRangeFilesSize += rangeFile.fileSize; + } + for (const auto& log : fileSet.get().logs) { + JsonBuilderObject object; + object["file_name"] = log.fileName; + object["file_size"] = log.fileSize; + object["begin_version"] = log.beginVersion; + object["end_version"] = log.endVersion; + logFilesJson.push_back(object); + totalLogFilesSize += log.fileSize; + } + + snapshotVersion = fileSet.get().targetVersion; + + TraceEvent("BackupQueryReceivedRestorableFilesSetFromSnapshot") + .detail("SnapshotVersion", snapshotVersion) + .detail("DestinationContainer", destinationContainer) + .detail("KeyRangesFilter", printable(keyRangesFilter)) + .detail("ActualRestoreVersion", fileSet.get().targetVersion) + .detail("NumRangeFiles", fileSet.get().ranges.size()) + .detail("NumLogFiles", fileSet.get().logs.size()) + .detail("RangeFilesBytes", totalRangeFilesSize) + .detail("LogFilesBytes", totalLogFilesSize); + } else { + reportBackupQueryError( + operationId, + result, + format("no restorable files set found for specified key ranges from snapshotVersion %lld", + snapshotVersion)); + return Void(); + } + + // We only need to know all the mutation logs from `snapshotVersion` to `restoreVersion`. + wait(store(fileSet, bc->getRestoreSet(restoreVersion, keyRangesFilter, /*logOnly=*/true, snapshotVersion))); + } else { + // When a snapshot version is not specified, we use the latest snapshot to restore to the `restoreVersion`. + wait(store(fileSet, bc->getRestoreSet(restoreVersion, keyRangesFilter))); + } + if (fileSet.present()) { - int64_t totalRangeFilesSize = 0, totalLogFilesSize = 0; result["restore_version"] = fileSet.get().targetVersion; - JsonBuilderArray rangeFilesJson; - JsonBuilderArray logFilesJson; for (const auto& rangeFile : fileSet.get().ranges) { JsonBuilderObject object; object["file_name"] = rangeFile.fileName; @@ -2831,14 +2928,6 @@ ACTOR Future queryBackup(const char* name, totalLogFilesSize += log.fileSize; } - result["total_range_files_size"] = totalRangeFilesSize; - result["total_log_files_size"] = totalLogFilesSize; - - if (verbose) { - result["ranges"] = rangeFilesJson; - result["logs"] = logFilesJson; - } - TraceEvent("BackupQueryReceivedRestorableFilesSet") .detail("DestinationContainer", destinationContainer) .detail("KeyRangesFilter", printable(keyRangesFilter)) @@ -2847,7 +2936,7 @@ ACTOR Future queryBackup(const char* name, .detail("NumLogFiles", fileSet.get().logs.size()) .detail("RangeFilesBytes", totalRangeFilesSize) .detail("LogFilesBytes", totalLogFilesSize); - } else { + } else if (snapshotVersion == invalidVersion) { reportBackupQueryError(operationId, result, "no restorable files set found for specified key ranges"); return Void(); } @@ -2857,6 +2946,14 @@ ACTOR Future queryBackup(const char* name, return Void(); } + result["total_range_files_size"] = totalRangeFilesSize; + result["total_log_files_size"] = totalLogFilesSize; + + if (verbose) { + result["ranges"] = rangeFilesJson; + result["logs"] = logFilesJson; + } + printf("%s\n", result.getJson().c_str()); return Void(); } @@ -3084,7 +3181,7 @@ static std::vector> parseLine(std::string& line, bool& er static void addKeyRange(std::string optionValue, Standalone>& keyRanges) { bool err = false, partial = false; - int tokenArray = 0; + [[maybe_unused]] int tokenArray = 0; auto parsed = parseLine(optionValue, err, partial); @@ -3419,6 +3516,7 @@ int main(int argc, char* argv[]) { int maxErrors = 20; Version beginVersion = invalidVersion; Version restoreVersion = invalidVersion; + Version snapshotVersion = invalidVersion; std::string restoreTimestamp; WaitForComplete waitForDone{ false }; StopWhenDone stopWhenDone{ true }; @@ -3641,6 +3739,15 @@ int main(int argc, char* argv[]) { return FDB_EXIT_ERROR; } break; + case OPT_BACKUPKEYS_FILE: + try { + std::string line = readFileBytes(args->OptionArg(), 64 * 1024 * 1024); + addKeyRange(line, backupKeys); + } catch (Error&) { + printHelpTeaser(argv[0]); + return FDB_EXIT_ERROR; + } + break; case OPT_BACKUPKEYS_FILTER: try { addKeyRange(args->OptionArg(), backupKeysFilter); @@ -3752,6 +3859,17 @@ int main(int argc, char* argv[]) { restoreVersion = ver; break; } + case OPT_RESTORE_SNAPSHOT_VERSION: { + const char* a = args->OptionArg(); + long long ver = 0; + if (!sscanf(a, "%lld", &ver)) { + fprintf(stderr, "ERROR: Could not parse database version `%s'\n", a); + printHelpTeaser(argv[0]); + return FDB_EXIT_ERROR; + } + snapshotVersion = ver; + break; + } case OPT_RESTORE_INCONSISTENT_SNAPSHOT_ONLY: { inconsistentSnapshotOnly.set(true); break; @@ -4136,6 +4254,7 @@ int main(int argc, char* argv[]) { proxy, backupKeysFilter, restoreVersion, + snapshotVersion, restoreClusterFileOrig, restoreTimestamp, Verbose{ !quietDisplay })); diff --git a/fdbcli/ConfigureCommand.actor.cpp b/fdbcli/ConfigureCommand.actor.cpp index f6209f2f410..16e98db58f3 100644 --- a/fdbcli/ConfigureCommand.actor.cpp +++ b/fdbcli/ConfigureCommand.actor.cpp @@ -275,7 +275,8 @@ CommandFactory configureFactory( "commit_proxies=|grv_proxies=|logs=|resolvers=>*|" "count=|perpetual_storage_wiggle=|perpetual_storage_wiggle_locality=" "<:|0>|storage_migration_type={disabled|gradual|aggressive}" - "|tenant_mode={disabled|optional_experimental|required_experimental}|blob_granules_enabled={0|1}", + "|tenant_mode={disabled|optional_experimental|required_experimental}|blob_granules_enabled={0|1}" + "|exclude=", "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 " @@ -309,6 +310,10 @@ CommandFactory configureFactory( "tenant_mode=: Sets the tenant mode for the cluster. If " "optional, then transactions can be run with or without specifying tenants. If required, all data must be " "accessed using tenants.\n\n" + "exclude=: Sets the addresses in the format of IP1:port1,IP2:port2 pairs to be excluded during " + "recruitment. Note this should be only used when the database is unavailable because of the faulty processes " + "that are blocking the recovery from completion. The number of addresses should be less than the replication " + "factor to avoid data loss.\n\n" "See the FoundationDB Administration Guide for more information.")); diff --git a/fdbcli/fdbcli.actor.cpp b/fdbcli/fdbcli.actor.cpp index c132695f326..f918307cc7f 100644 --- a/fdbcli/fdbcli.actor.cpp +++ b/fdbcli/fdbcli.actor.cpp @@ -648,7 +648,9 @@ ACTOR template Future makeInterruptable(Future f) { Future interrupt = LineNoise::onKeyboardInterrupt(); choose { - when(T t = wait(f)) { return t; } + when(T t = wait(f)) { + return t; + } when(wait(interrupt)) { f.cancel(); throw operation_cancelled(); diff --git a/fdbclient/BackupAgent.actor.h b/fdbclient/BackupAgent.actor.h index 44fa7921d3b..43cd8ee811d 100644 --- a/fdbclient/BackupAgent.actor.h +++ b/fdbclient/BackupAgent.actor.h @@ -1002,10 +1002,14 @@ struct StringRefReader { }; namespace fileBackup { +Standalone> decodeRangeFileBlock(const Standalone& buf); + ACTOR Future>> decodeRangeFileBlock(Reference file, int64_t offset, int len); +Standalone> decodeMutationLogFileBlock(const Standalone& buf); + // Reads a mutation log block from file and parses into batch mutation blocks for further parsing. ACTOR Future>> decodeMutationLogFileBlock(Reference file, int64_t offset, diff --git a/fdbclient/BackupContainer.h b/fdbclient/BackupContainer.h index 36c9ff7cfab..f6523bd141c 100644 --- a/fdbclient/BackupContainer.h +++ b/fdbclient/BackupContainer.h @@ -315,6 +315,39 @@ class IBackupContainer { }; namespace fileBackup { +// Use RangeMap to store a list of ranges for efficient query if a mutation +// matches to any of the range. +class RangeMapFilters { +public: + RangeMapFilters() = default; + + explicit RangeMapFilters(const std::vector& ranges) { + for (const auto& range : ranges) { + rangeMap.insert(range, 1); + } + rangeMap.coalesce(allKeys); + } + + void updateFilters(std::vector& prefixes) { + for (const auto& prefix : prefixes) { + rangeMap.insert(prefixRange(StringRef(prefix)), 1); + } + rangeMap.coalesce(allKeys); + } + + // Returns if the mutation matches any filter ranges. + bool match(const MutationRef& m) const; + + // Returns if the key-value pair matches any filter ranges. + bool match(const KeyValueRef& kv) const; + + // Returns if the range intersects with any filter ranges. + bool match(const KeyRangeRef& range) const; + +private: + KeyRangeMap rangeMap; +}; + // Accumulates mutation log value chunks, as both a vector of chunks and as a combined chunk, // in chunk order, and can check the chunk set for completion or intersection with a set // of ranges. @@ -335,7 +368,7 @@ struct AccumulatedMutations { // Returns true if a complete chunk contains any MutationRefs which intersect with any // range in ranges. // It is undefined behavior to run this if isComplete() does not return true. - bool matchesAnyRange(const std::vector& ranges) const; + bool matchesAnyRange(const RangeMapFilters& rangeMap) const; std::vector kvs; std::string serializedMutations; diff --git a/fdbclient/BackupContainerFileSystem.actor.cpp b/fdbclient/BackupContainerFileSystem.actor.cpp index a4778ecc104..634cba1c15e 100644 --- a/fdbclient/BackupContainerFileSystem.actor.cpp +++ b/fdbclient/BackupContainerFileSystem.actor.cpp @@ -912,6 +912,7 @@ class BackupContainerFileSystemImpl { if (logsOnly) { state RestorableFileSet restorableSet; + restorableSet.targetVersion = targetVersion; state std::vector logFiles; Version begin = beginVersion == invalidVersion ? 0 : beginVersion; wait(store(logFiles, bc->listLogFiles(begin, targetVersion, false))); diff --git a/fdbclient/ClientKnobs.cpp b/fdbclient/ClientKnobs.cpp index 7fa8d890294..16c8191cee4 100644 --- a/fdbclient/ClientKnobs.cpp +++ b/fdbclient/ClientKnobs.cpp @@ -203,7 +203,6 @@ void ClientKnobs::initialize(Randomize randomize) { init( DEFAULT_AUTO_LOGS, 3 ); init( DEFAULT_COMMIT_GRV_PROXIES_RATIO, 3 ); init( DEFAULT_MAX_GRV_PROXIES, 4 ); - init( UNLINKONLOAD_FDBCLIB, true ); // if false, don't delete libfdb_c in tmp directory on client connect. init( IS_ACCEPTABLE_DELAY, 1.5 ); diff --git a/fdbclient/CoordinationInterface.h b/fdbclient/CoordinationInterface.h index eb0f99895b5..7897ded33e0 100644 --- a/fdbclient/CoordinationInterface.h +++ b/fdbclient/CoordinationInterface.h @@ -241,6 +241,7 @@ struct OpenDatabaseCoordRequest { std::vector hostnames; std::vector coordinators; ReplyPromise> reply; + bool internal{ true }; template void serialize(Ar& ar) { @@ -252,7 +253,8 @@ struct OpenDatabaseCoordRequest { clusterKey, coordinators, reply, - hostnames); + hostnames, + internal); } }; diff --git a/fdbclient/DatabaseConfiguration.cpp b/fdbclient/DatabaseConfiguration.cpp index 657129f86e2..bd9554018fe 100644 --- a/fdbclient/DatabaseConfiguration.cpp +++ b/fdbclient/DatabaseConfiguration.cpp @@ -639,6 +639,8 @@ bool DatabaseConfiguration::setInternal(KeyRef key, ValueRef value) { } else if (ck == LiteralStringRef("blob_granules_enabled")) { parse((&type), value); blobGranulesEnabled = (type != 0); + } else if (ck.startsWith("excluded/"_sr)) { + // excluded servers: don't keep the state internally } else { return false; } diff --git a/fdbclient/FDBTypes.h b/fdbclient/FDBTypes.h index 447124e3e1f..c494762743e 100644 --- a/fdbclient/FDBTypes.h +++ b/fdbclient/FDBTypes.h @@ -514,7 +514,12 @@ using KeySelector = Standalone; using RangeResult = Standalone; using MappedRangeResult = Standalone; -enum { invalidVersion = -1, latestVersion = -2, MAX_VERSION = std::numeric_limits::max() }; +enum { + invalidVersion = -1, + latestVersion = -2, + earliestVersion = -3, + MAX_VERSION = std::numeric_limits::max() +}; inline Key keyAfter(const KeyRef& key) { if (key == LiteralStringRef("\xff\xff")) diff --git a/fdbclient/FileBackupAgent.actor.cpp b/fdbclient/FileBackupAgent.actor.cpp index e53042bdc7d..482baaac604 100644 --- a/fdbclient/FileBackupAgent.actor.cpp +++ b/fdbclient/FileBackupAgent.actor.cpp @@ -47,7 +47,7 @@ FDB_DEFINE_BOOLEAN_PARAM(IncrementalBackupOnly); FDB_DEFINE_BOOLEAN_PARAM(OnlyApplyMutationLogs); #define SevFRTestInfo SevVerbose -//#define SevFRTestInfo SevInfo +// #define SevFRTestInfo SevInfo static std::string boolToYesOrNo(bool val) { return val ? std::string("Yes") : std::string("No"); @@ -584,66 +584,67 @@ struct RangeFileWriter { Key lastValue; }; -ACTOR Future>> decodeRangeFileBlock(Reference file, - int64_t offset, - int len) { - state Standalone buf = makeString(len); - int rLen = wait(uncancellable(holdWhile(buf, file->read(mutateString(buf), len, offset)))); - if (rLen != len) - throw restore_bad_read(); +Standalone> decodeRangeFileBlock(const Standalone& buf) { + Standalone> results({}, buf.arena()); + StringRefReader reader(buf, restore_corrupted_data()); - simulateBlobFailure(); + // Read header, currently only decoding BACKUP_AGENT_SNAPSHOT_FILE_VERSION + if (reader.consume() != BACKUP_AGENT_SNAPSHOT_FILE_VERSION) + throw restore_unsupported_file_version(); - Standalone> results({}, buf.arena()); - state StringRefReader reader(buf, restore_corrupted_data()); + // Read begin key, if this fails then block was invalid. + uint32_t kLen = reader.consumeNetworkUInt32(); + const uint8_t* k = reader.consume(kLen); + results.push_back(results.arena(), KeyValueRef(KeyRef(k, kLen), ValueRef())); - try { - // Read header, currently only decoding BACKUP_AGENT_SNAPSHOT_FILE_VERSION - if (reader.consume() != BACKUP_AGENT_SNAPSHOT_FILE_VERSION) - throw restore_unsupported_file_version(); + // Read kv pairs and end key + while (1) { + // Read a key. + kLen = reader.consumeNetworkUInt32(); + k = reader.consume(kLen); - // Read begin key, if this fails then block was invalid. - uint32_t kLen = reader.consumeNetworkUInt32(); - const uint8_t* k = reader.consume(kLen); - results.push_back(results.arena(), KeyValueRef(KeyRef(k, kLen), ValueRef())); + // If eof reached or first value len byte is 0xFF then a valid block end was reached. + if (reader.eof() || *reader.rptr == 0xFF) { + results.push_back(results.arena(), KeyValueRef(KeyRef(k, kLen), ValueRef())); + break; + } - // Read kv pairs and end key - while (1) { - // Read a key. - kLen = reader.consumeNetworkUInt32(); - k = reader.consume(kLen); + // Read a value, which must exist or the block is invalid + uint32_t vLen = reader.consumeNetworkUInt32(); + const uint8_t* v = reader.consume(vLen); + results.push_back(results.arena(), KeyValueRef(KeyRef(k, kLen), ValueRef(v, vLen))); - // If eof reached or first value len byte is 0xFF then a valid block end was reached. - if (reader.eof() || *reader.rptr == 0xFF) { - results.push_back(results.arena(), KeyValueRef(KeyRef(k, kLen), ValueRef())); - break; - } + // If eof reached or first byte of next key len is 0xFF then a valid block end was reached. + if (reader.eof() || *reader.rptr == 0xFF) + break; + } - // Read a value, which must exist or the block is invalid - uint32_t vLen = reader.consumeNetworkUInt32(); - const uint8_t* v = reader.consume(vLen); - results.push_back(results.arena(), KeyValueRef(KeyRef(k, kLen), ValueRef(v, vLen))); + // Make sure any remaining bytes in the block are 0xFF + for (auto b : reader.remainder()) + if (b != 0xFF) + throw restore_corrupted_data_padding(); - // If eof reached or first byte of next key len is 0xFF then a valid block end was reached. - if (reader.eof() || *reader.rptr == 0xFF) - break; - } + return results; +} - // Make sure any remaining bytes in the block are 0xFF - for (auto b : reader.remainder()) - if (b != 0xFF) - throw restore_corrupted_data_padding(); +ACTOR Future>> decodeRangeFileBlock(Reference file, + int64_t offset, + int len) { + state Standalone buf = makeString(len); + int rLen = wait(uncancellable(holdWhile(buf, file->read(mutateString(buf), len, offset)))); + if (rLen != len) + throw restore_bad_read(); - return results; + simulateBlobFailure(); + try { + return decodeRangeFileBlock(buf); } catch (Error& e) { TraceEvent(SevWarn, "FileRestoreDecodeRangeFileBlockFailed") .error(e) .detail("Filename", file->getFilename()) .detail("BlockOffset", offset) - .detail("BlockLen", len) - .detail("ErrorRelativeOffset", reader.rptr - buf.begin()) - .detail("ErrorAbsoluteOffset", reader.rptr - buf.begin() + offset); + .detail("BlockLen", len); throw; } } @@ -692,6 +693,37 @@ struct LogFileWriter { int64_t blockEnd; }; +Standalone> decodeMutationLogFileBlock(const Standalone& buf) { + Standalone> results({}, buf.arena()); + StringRefReader reader(buf, restore_corrupted_data()); + + // Read header, currently only decoding version BACKUP_AGENT_MLOG_VERSION + if (reader.consume() != BACKUP_AGENT_MLOG_VERSION) + throw restore_unsupported_file_version(); + + // Read k/v pairs. Block ends either at end of last value exactly or with 0xFF as first key len byte. + while (1) { + // If eof reached or first key len bytes is 0xFF then end of block was reached. + if (reader.eof() || *reader.rptr == 0xFF) + break; + + // Read key and value. If anything throws then there is a problem. + uint32_t kLen = reader.consumeNetworkUInt32(); + const uint8_t* k = reader.consume(kLen); + uint32_t vLen = reader.consumeNetworkUInt32(); + const uint8_t* v = reader.consume(vLen); + + results.push_back(results.arena(), KeyValueRef(KeyRef(k, kLen), ValueRef(v, vLen))); + } + + // Make sure any remaining bytes in the block are 0xFF + for (auto b : reader.remainder()) + if (b != 0xFF) + throw restore_corrupted_data_padding(); + + return results; +} + ACTOR Future>> decodeMutationLogFileBlock(Reference file, int64_t offset, int len) { @@ -700,44 +732,14 @@ ACTOR Future>> decodeMutationLogFileBlock(Refe if (rLen != len) throw restore_bad_read(); - Standalone> results({}, buf.arena()); - state StringRefReader reader(buf, restore_corrupted_data()); - try { - // Read header, currently only decoding version BACKUP_AGENT_MLOG_VERSION - if (reader.consume() != BACKUP_AGENT_MLOG_VERSION) - throw restore_unsupported_file_version(); - - // Read k/v pairs. Block ends either at end of last value exactly or with 0xFF as first key len byte. - while (1) { - // If eof reached or first key len bytes is 0xFF then end of block was reached. - if (reader.eof() || *reader.rptr == 0xFF) - break; - - // Read key and value. If anything throws then there is a problem. - uint32_t kLen = reader.consumeNetworkUInt32(); - const uint8_t* k = reader.consume(kLen); - uint32_t vLen = reader.consumeNetworkUInt32(); - const uint8_t* v = reader.consume(vLen); - - results.push_back(results.arena(), KeyValueRef(KeyRef(k, kLen), ValueRef(v, vLen))); - } - - // Make sure any remaining bytes in the block are 0xFF - for (auto b : reader.remainder()) - if (b != 0xFF) - throw restore_corrupted_data_padding(); - - return results; - + return decodeMutationLogFileBlock(buf); } catch (Error& e) { TraceEvent(SevWarn, "FileRestoreCorruptLogFileBlock") .error(e) .detail("Filename", file->getFilename()) .detail("BlockOffset", offset) - .detail("BlockLen", len) - .detail("ErrorRelativeOffset", reader.rptr - buf.begin()) - .detail("ErrorAbsoluteOffset", reader.rptr - buf.begin() + offset); + .detail("BlockLen", len); throw; } } @@ -3331,28 +3333,49 @@ bool AccumulatedMutations::isComplete() const { // Returns true if a complete chunk contains any MutationRefs which intersect with any // range in ranges. // It is undefined behavior to run this if isComplete() does not return true. -bool AccumulatedMutations::matchesAnyRange(const std::vector& ranges) const { +bool AccumulatedMutations::matchesAnyRange(const RangeMapFilters& filters) const { std::vector mutations = decodeMutationLogValue(serializedMutations); for (auto& m : mutations) { - for (auto& r : ranges) { - if (m.type == MutationRef::ClearRange) { - if (r.intersects(KeyRangeRef(m.param1, m.param2))) { - return true; - } - } else { - if (r.contains(m.param1)) { - return true; - } - } + if (filters.match(m)) { + return true; } } return false; } +bool RangeMapFilters::match(const MutationRef& m) const { + if (isSingleKeyMutation((MutationRef::Type)m.type)) { + if (match(singleKeyRange(m.param1))) { + return true; + } + } else if (m.type == MutationRef::ClearRange) { + if (match(KeyRangeRef(m.param1, m.param2))) { + return true; + } + } else { + ASSERT(false); + } + return false; +} + +bool RangeMapFilters::match(const KeyValueRef& kv) const { + return match(singleKeyRange(kv.key)); +} + +bool RangeMapFilters::match(const KeyRangeRef& range) const { + auto ranges = rangeMap.intersectingRanges(range); + for (const auto& r : ranges) { + if (r.cvalue() == 1) { + return true; + } + } + return false; +} + // Returns a vector of filtered KV refs from data which are either part of incomplete mutation groups OR complete // and have data relevant to one of the KV ranges in ranges -std::vector filterLogMutationKVPairs(VectorRef data, const std::vector& ranges) { +std::vector filterLogMutationKVPairs(VectorRef data, const RangeMapFilters& filters) { std::unordered_map mutationBlocksByVersion; for (auto& kv : data) { @@ -3366,7 +3389,7 @@ std::vector filterLogMutationKVPairs(VectorRef data, c AccumulatedMutations& m = vb.second; // If the mutations are incomplete or match one of the ranges, include in results. - if (!m.isComplete() || m.matchesAnyRange(ranges)) { + if (!m.isComplete() || m.matchesAnyRange(filters)) { output.insert(output.end(), m.kvs.begin(), m.kvs.end()); } } @@ -3432,7 +3455,8 @@ struct RestoreLogDataTaskFunc : RestoreFileTaskFuncBase { // Filter the KV pairs extracted from the log file block to remove any records known to not be needed for this // restore based on the restore range set. - state std::vector dataFiltered = filterLogMutationKVPairs(dataOriginal, ranges); + RangeMapFilters filters(ranges); + state std::vector dataFiltered = filterLogMutationKVPairs(dataOriginal, filters); state int start = 0; state int end = dataFiltered.size(); diff --git a/fdbclient/GenericManagementAPI.actor.h b/fdbclient/GenericManagementAPI.actor.h index 8ea27b33cf4..87e11b79a99 100644 --- a/fdbclient/GenericManagementAPI.actor.h +++ b/fdbclient/GenericManagementAPI.actor.h @@ -291,7 +291,6 @@ Future changeConfig(Reference db, std::maprandomUniqueID(), Unversioned()); state bool oldReplicationUsesDcId = false; state bool warnPPWGradual = false; - state bool warnChangeStorageNoMigrate = false; state bool warnRocksDBIsExperimental = false; loop { try { diff --git a/fdbclient/ManagementAPI.actor.cpp b/fdbclient/ManagementAPI.actor.cpp index f627168e8bd..ff06b3e18ac 100644 --- a/fdbclient/ManagementAPI.actor.cpp +++ b/fdbclient/ManagementAPI.actor.cpp @@ -200,6 +200,23 @@ std::map configForToken(std::string const& mode) { } out[p + key] = format("%d", tenantMode); } + if (key == "exclude") { + int p = 0; + while (p < value.size()) { + int end = value.find_first_of(',', p); + if (end == value.npos) { + end = value.size(); + } + auto addrRef = StringRef(value).substr(p, end - p); + AddressExclusion addr = AddressExclusion::parse(addrRef); + if (addr.isValid()) { + out[encodeExcludedServersKey(addr)] = ""; + } else { + printf("Error: invalid address format: %s\n", addrRef.toString().c_str()); + } + p = end + 1; + } + } return out; } @@ -879,7 +896,9 @@ ACTOR Future> changeQuorumChecker(Transaction* tr, choose { when(wait(waitForAll(leaderServers))) {} - when(wait(delay(5.0))) { return CoordinatorsResult::COORDINATOR_UNREACHABLE; } + when(wait(delay(5.0))) { + return CoordinatorsResult::COORDINATOR_UNREACHABLE; + } } tr->set(coordinatorsKey, conn->toString()); return Optional(); @@ -976,7 +995,9 @@ ACTOR Future changeQuorum(Database cx, Reference asyncDeserializeClusterInterface(Reference> s state Future deserializer = asyncDeserialize(serializedInfo, knownLeader); loop { choose { - when(wait(deserializer)) { UNSTOPPABLE_ASSERT(false); } + when(wait(deserializer)) { + UNSTOPPABLE_ASSERT(false); + } when(wait(knownLeader->onChange())) { if (knownLeader->get().present()) { outKnownLeader->set(knownLeader->get().get().clientInterface); @@ -860,7 +862,8 @@ ACTOR Future monitorProxiesOneGeneration( Reference>> coordinator, MonitorLeaderInfo info, Reference>>> supportedVersions, - Key traceLogGroup) { + Key traceLogGroup, + IsInternal internal) { state ClusterConnectionString cs = info.intermediateConnRecord->getConnectionString(); state int coordinatorsSize = cs.hostnames.size() + cs.coords.size(); state int index = 0; @@ -892,6 +895,7 @@ ACTOR Future monitorProxiesOneGeneration( req.knownClientInfoID = clientInfo->get().id; req.supportedVersions = supportedVersions->get(); req.traceLogGroup = traceLogGroup; + req.internal = internal; state ClusterConnectionString storedConnectionString; if (connRecord) { @@ -980,12 +984,13 @@ ACTOR Future monitorProxies( Reference> clientInfo, Reference>> coordinator, Reference>>> supportedVersions, - Key traceLogGroup) { + Key traceLogGroup, + IsInternal internal) { state MonitorLeaderInfo info(connRecord->get()); loop { choose { when(MonitorLeaderInfo _info = wait(monitorProxiesOneGeneration( - connRecord->get(), clientInfo, coordinator, info, supportedVersions, traceLogGroup))) { + connRecord->get(), clientInfo, coordinator, info, supportedVersions, traceLogGroup, internal))) { info = _info; } when(wait(connRecord->onChange())) { diff --git a/fdbclient/MonitorLeader.h b/fdbclient/MonitorLeader.h index 8819b5702b4..86f2545fe9c 100644 --- a/fdbclient/MonitorLeader.h +++ b/fdbclient/MonitorLeader.h @@ -26,6 +26,7 @@ #include "fdbclient/CoordinationInterface.h" #include "fdbclient/ClusterInterface.h" #include "fdbclient/CommitProxyInterface.h" +#include "fdbclient/ClientBooleanParams.h" #define CLUSTER_FILE_ENV_VAR_NAME "FDB_CLUSTER_FILE" @@ -85,7 +86,8 @@ Future monitorProxies( Reference> const& clientInfo, Reference>> const& coordinator, Reference>>> const& supportedVersions, - Key const& traceLogGroup); + Key const& traceLogGroup, + IsInternal const& internal); void shrinkProxyList(ClientDBInfo& ni, std::vector& lastCommitProxyUIDs, diff --git a/fdbclient/MultiVersionTransaction.actor.cpp b/fdbclient/MultiVersionTransaction.actor.cpp index c21273975ca..d5cbae914f8 100644 --- a/fdbclient/MultiVersionTransaction.actor.cpp +++ b/fdbclient/MultiVersionTransaction.actor.cpp @@ -2134,6 +2134,9 @@ void MultiVersionApi::setNetworkOptionInternal(FDBNetworkOptions::Option option, // multiple client threads are not supported on windows. threadCount = extractIntOption(value, 1, 1); #endif + } else if (option == FDBNetworkOptions::RETAIN_CLIENT_LIBRARY_COPIES) { + validateOption(value, false, true); + retainClientLibCopies = true; } else { forwardOption = true; } @@ -2177,10 +2180,7 @@ void MultiVersionApi::setupNetwork() { if (externalClients.count(filename) == 0) { externalClients[filename] = {}; for (const auto& tmp : copyExternalLibraryPerThread(path)) { - bool unlinkOnLoad = tmp.second; - if (!CLIENT_KNOBS->UNLINKONLOAD_FDBCLIB) { - unlinkOnLoad = false; - } + bool unlinkOnLoad = tmp.second && !retainClientLibCopies; externalClients[filename].push_back(Reference( new ClientInfo(new DLApi(tmp.first, unlinkOnLoad /*unlink on load*/), path))); } @@ -2513,7 +2513,8 @@ void MultiVersionApi::loadEnvironmentVariableNetworkOptions() { MultiVersionApi::MultiVersionApi() : callbackOnMainThread(true), localClientDisabled(false), networkStartSetup(false), networkSetup(false), - bypassMultiClientApi(false), externalClient(false), apiVersion(0), threadCount(0), envOptionsLoaded(false) {} + bypassMultiClientApi(false), externalClient(false), retainClientLibCopies(false), apiVersion(0), threadCount(0), + envOptionsLoaded(false) {} MultiVersionApi* MultiVersionApi::api = new MultiVersionApi(); diff --git a/fdbclient/MultiVersionTransaction.h b/fdbclient/MultiVersionTransaction.h index e2b0c7e027c..6e172d33513 100644 --- a/fdbclient/MultiVersionTransaction.h +++ b/fdbclient/MultiVersionTransaction.h @@ -894,6 +894,8 @@ class MultiVersionApi : public IClientApi { volatile bool networkSetup; volatile bool bypassMultiClientApi; volatile bool externalClient; + bool retainClientLibCopies; + int apiVersion; int nextThread = 0; diff --git a/fdbclient/NativeAPI.actor.cpp b/fdbclient/NativeAPI.actor.cpp index 978c708c643..cd6865028bd 100644 --- a/fdbclient/NativeAPI.actor.cpp +++ b/fdbclient/NativeAPI.actor.cpp @@ -971,7 +971,9 @@ ACTOR static Future monitorClientDBInfoChange(DatabaseContext* cx, proxiesChangeTrigger->trigger(); } } - when(wait(actors.getResult())) { UNSTOPPABLE_ASSERT(false); } + when(wait(actors.getResult())) { + UNSTOPPABLE_ASSERT(false); + } } } } @@ -2252,7 +2254,8 @@ Database Database::createDatabase(Reference connRecord clientInfo, coordinator, networkOptions.supportedVersions, - StringRef(networkOptions.traceLogGroup)); + StringRef(networkOptions.traceLogGroup), + internal); DatabaseContext* db; if (preallocatedDb) { @@ -3330,7 +3333,9 @@ ACTOR Future> getValue(Reference trState, std::vector{ transaction_too_old(), future_version() }); } choose { - when(wait(trState->cx->connectionFileChanged())) { throw transaction_too_old(); } + when(wait(trState->cx->connectionFileChanged())) { + throw transaction_too_old(); + } when(GetValueReply _reply = wait(loadBalance( trState->cx.getPtr(), locationInfo.locations, @@ -3472,7 +3477,9 @@ ACTOR Future getKey(Reference trState, state GetKeyReply reply; try { choose { - when(wait(trState->cx->connectionFileChanged())) { throw transaction_too_old(); } + when(wait(trState->cx->connectionFileChanged())) { + throw transaction_too_old(); + } when(GetKeyReply _reply = wait(loadBalance( trState->cx.getPtr(), locationInfo.locations, @@ -3634,7 +3641,9 @@ ACTOR Future watchValue(Database cx, Reference p TaskPriority::DefaultPromiseEndpoint))) { resp = r; } - when(wait(cx->connectionRecord ? cx->connectionRecord->onChange() : Never())) { wait(Never()); } + when(wait(cx->connectionRecord ? cx->connectionRecord->onChange() : Never())) { + wait(Never()); + } } if (watchValueID.present()) { g_traceBatch.addEvent("WatchValueDebug", watchValueID.get().first(), "NativeAPI.watchValue.After"); @@ -3934,7 +3943,9 @@ Future getExactRange(Reference trState, state GetKeyValuesFamilyReply rep; try { choose { - when(wait(trState->cx->connectionFileChanged())) { throw transaction_too_old(); } + when(wait(trState->cx->connectionFileChanged())) { + throw transaction_too_old(); + } when(GetKeyValuesFamilyReply _rep = wait(loadBalance( trState->cx.getPtr(), locations[shard].locations, @@ -4399,9 +4410,11 @@ Future getRange(Reference trState, output.readToBegin = readToBegin; output.readThroughEnd = readThroughEnd; - if (BUGGIFY && limits.hasByteLimit() && output.size() > std::max(1, originalLimits.minRows)) { + if (BUGGIFY && limits.hasByteLimit() && output.size() > std::max(1, originalLimits.minRows) && + (!std::is_same::value)) { // Copy instead of resizing because TSS maybe be using output's arena for comparison. This only // happens in simulation so it's fine + // disable it on prefetch, because boundary entries serve as continuations RangeResultFamily copy; int newSize = deterministicRandom()->randomInt(std::max(1, originalLimits.minRows), output.size()); @@ -4823,7 +4836,9 @@ ACTOR Future getRangeStreamFragment(Reference trState, return Void(); } - when(GetKeyValuesStreamReply _rep = waitNext(replyStream.getFuture())) { rep = _rep; } + when(GetKeyValuesStreamReply _rep = waitNext(replyStream.getFuture())) { + rep = _rep; + } } ++trState->cx->transactionPhysicalReadsCompleted; } catch (Error& e) { @@ -5277,7 +5292,9 @@ ACTOR Future watch(Reference watch, loop { choose { // NativeAPI watchValue future finishes or errors - when(wait(watch->watchFuture)) { break; } + when(wait(watch->watchFuture)) { + break; + } when(wait(cx->connectionFileChanged())) { TEST(true); // Recreated a watch after switch @@ -6657,7 +6674,9 @@ ACTOR Future getConsistentReadVersion(SpanID parentSpan, state Future onProxiesChanged = cx->onProxiesChanged(); choose { - when(wait(onProxiesChanged)) { onProxiesChanged = cx->onProxiesChanged(); } + when(wait(onProxiesChanged)) { + onProxiesChanged = cx->onProxiesChanged(); + } when(GetReadVersionReply v = wait(basicLoadBalance(cx->getGrvProxies(UseProvisionalProxies( flags & GetReadVersionRequest::FLAG_USE_PROVISIONAL_PROXIES)), @@ -7081,7 +7100,9 @@ ACTOR Future getClusterProtocolImpl( needToConnect = false; } choose { - when(wait(coordinator->onChange())) { needToConnect = true; } + when(wait(coordinator->onChange())) { + needToConnect = true; + } when(ProtocolVersion pv = wait(protocolVersion)) { if (!expectedVersion.present() || expectedVersion.get() != pv) { @@ -8209,8 +8230,12 @@ ACTOR Future> getCheckpointMetaData(Database cx, } choose { - when(wait(cx->connectionFileChanged())) { cx->invalidateCache(KeyRef(), keys); } - when(wait(waitForAll(futures))) { break; } + when(wait(cx->connectionFileChanged())) { + cx->invalidateCache(KeyRef(), keys); + } + when(wait(waitForAll(futures))) { + break; + } when(wait(delay(timeout))) { TraceEvent("GetCheckpointTimeout").detail("Range", keys).detail("Version", version); } @@ -8493,15 +8518,11 @@ Version ChangeFeedData::getVersion() { // native api has consumed and processed, them, and then the fdb client has consumed all of the mutations. ACTOR Future changeFeedWaitLatest(Reference self, Version version) { // wait on SS to have sent up through version - int desired = 0; - int waiting = 0; std::vector> allAtLeast; for (auto& it : self->storageData) { if (it->version.get() < version) { - waiting++; if (version > it->desired.get()) { it->desired.set(version); - desired++; } allAtLeast.push_back(it->version.whenAtLeast(version)); } @@ -8560,8 +8581,12 @@ ACTOR Future changeFeedWhenAtLatest(Reference self, Versio // only allowed to use empty versions if you're caught up Future waitEmptyVersion = (self->notAtLatest.get() == 0) ? changeFeedWaitLatest(self, version) : Never(); choose { - when(wait(waitEmptyVersion)) { break; } - when(wait(lastReturned)) { break; } + when(wait(waitEmptyVersion)) { + break; + } + when(wait(lastReturned)) { + break; + } when(wait(self->refresh.getFuture())) {} when(wait(self->notAtLatest.onChange())) {} } diff --git a/fdbclient/PaxosConfigTransaction.actor.cpp b/fdbclient/PaxosConfigTransaction.actor.cpp index 4b7c19c05af..d20ced3ace3 100644 --- a/fdbclient/PaxosConfigTransaction.actor.cpp +++ b/fdbclient/PaxosConfigTransaction.actor.cpp @@ -194,8 +194,12 @@ class GetGenerationQuorum { } try { choose { - when(ConfigGeneration generation = wait(self->result.getFuture())) { return generation; } - when(wait(self->actors.getResult())) { ASSERT(false); } + when(ConfigGeneration generation = wait(self->result.getFuture())) { + return generation; + } + when(wait(self->actors.getResult())) { + ASSERT(false); + } } } catch (Error& e) { if (e.code() == error_code_failed_to_reach_quorum) { diff --git a/fdbclient/ReadYourWrites.actor.cpp b/fdbclient/ReadYourWrites.actor.cpp index c651adad32a..67fb6939534 100644 --- a/fdbclient/ReadYourWrites.actor.cpp +++ b/fdbclient/ReadYourWrites.actor.cpp @@ -351,16 +351,24 @@ class RYWImpl { Req req, Snapshot snapshot) { choose { - when(typename Req::Result result = wait(readThrough(ryw, req, snapshot))) { return result; } - when(wait(ryw->resetPromise.getFuture())) { throw internal_error(); } + when(typename Req::Result result = wait(readThrough(ryw, req, snapshot))) { + return result; + } + when(wait(ryw->resetPromise.getFuture())) { + throw internal_error(); + } } } ACTOR template static Future readWithConflictRangeSnapshot(ReadYourWritesTransaction* ryw, Req req) { state SnapshotCache::iterator it(&ryw->cache, &ryw->writes); choose { - when(typename Req::Result result = wait(read(ryw, req, &it))) { return result; } - when(wait(ryw->resetPromise.getFuture())) { throw internal_error(); } + when(typename Req::Result result = wait(read(ryw, req, &it))) { + return result; + } + when(wait(ryw->resetPromise.getFuture())) { + throw internal_error(); + } } } ACTOR template @@ -376,7 +384,9 @@ class RYWImpl { addConflictRange(ryw, req, it.extractWriteMapIterator(), result); return result; } - when(wait(ryw->resetPromise.getFuture())) { throw internal_error(); } + when(wait(ryw->resetPromise.getFuture())) { + throw internal_error(); + } } } template @@ -1191,7 +1201,9 @@ class RYWImpl { addConflictRangeAndMustUnmodified(ryw, req, writes, result); return result; } - when(wait(ryw->resetPromise.getFuture())) { throw internal_error(); } + when(wait(ryw->resetPromise.getFuture())) { + throw internal_error(); + } } } @@ -1437,9 +1449,13 @@ class RYWImpl { ACTOR static Future getReadVersion(ReadYourWritesTransaction* ryw) { choose { - when(Version v = wait(ryw->tr.getReadVersion())) { return v; } + when(Version v = wait(ryw->tr.getReadVersion())) { + return v; + } - when(wait(ryw->resetPromise.getFuture())) { throw internal_error(); } + when(wait(ryw->resetPromise.getFuture())) { + throw internal_error(); + } } } }; diff --git a/fdbclient/S3BlobStore.actor.cpp b/fdbclient/S3BlobStore.actor.cpp index 80e3611690e..3dc1f551020 100644 --- a/fdbclient/S3BlobStore.actor.cpp +++ b/fdbclient/S3BlobStore.actor.cpp @@ -422,7 +422,9 @@ ACTOR Future deleteRecursively_impl(Reference b, loop { choose { // Throw if done throws, otherwise don't stop until end_of_stream - when(wait(done)) { done = Never(); } + when(wait(done)) { + done = Never(); + } when(S3BlobStoreEndpoint::ListResult list = waitNext(resultStream.getFuture())) { for (auto& object : list.objects) { @@ -1078,7 +1080,9 @@ ACTOR Future listObjects_impl(Referencerandom01() < 0.1 ? true : false; // false by default. disable the consistency check when it's true @@ -361,11 +364,11 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi // KeyValueStoreRocksDB init( ROCKSDB_LEVEL_COMPACTION_DYNAMIC_LEVEL_BYTES, true ); if( randomize && BUGGIFY ) ROCKSDB_LEVEL_COMPACTION_DYNAMIC_LEVEL_BYTES = false; - init( ROCKSDB_SUGGEST_COMPACT_CLEAR_RANGE, true ); if( randomize && BUGGIFY ) ROCKSDB_SUGGEST_COMPACT_CLEAR_RANGE = false; + init( ROCKSDB_SUGGEST_COMPACT_CLEAR_RANGE, false ); init( ROCKSDB_THREAD_PROMISE_PRIORITY, 7500 ); init( ROCKSDB_READER_THREAD_PRIORITY, 0 ); init( ROCKSDB_WRITER_THREAD_PRIORITY, 0 ); - init( ROCKSDB_BACKGROUND_PARALLELISM, 4 ); + init( ROCKSDB_BACKGROUND_PARALLELISM, 2 ); init( ROCKSDB_READ_PARALLELISM, 4 ); // Use a smaller memtable in simulation to avoid OOMs. int64_t memtableBytes = isSimulated ? 32 * 1024 : 512 * 1024 * 1024; @@ -375,12 +378,15 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi init( ROCKSDB_PERIODIC_COMPACTION_SECONDS, 0 ); init( ROCKSDB_PREFIX_LEN, 0 ); // If rocksdb block cache size is 0, the default 8MB is used. - int64_t blockCacheSize = isSimulated ? 0 : 1024 * 1024 * 1024 /* 1GB */; + int64_t blockCacheSize = isSimulated ? 16 * 1024 * 1024 : 2147483648 /* 2GB */; init( ROCKSDB_BLOCK_CACHE_SIZE, blockCacheSize ); init( ROCKSDB_METRICS_DELAY, 60.0 ); - init( ROCKSDB_READ_VALUE_TIMEOUT, 5.0 ); - init( ROCKSDB_READ_VALUE_PREFIX_TIMEOUT, 5.0 ); - init( ROCKSDB_READ_RANGE_TIMEOUT, 5.0 ); + // ROCKSDB_READ_VALUE_TIMEOUT, ROCKSDB_READ_VALUE_PREFIX_TIMEOUT, ROCKSDB_READ_RANGE_TIMEOUT knobs: + // In simulation, increasing the read operation timeouts to 5 minutes, as some of the tests have + // very high load and single read thread cannot process all the load within the timeouts. + init( ROCKSDB_READ_VALUE_TIMEOUT, isSimulated ? 300.0 : 5.0 ); + init( ROCKSDB_READ_VALUE_PREFIX_TIMEOUT, isSimulated ? 300.0 : 5.0 ); + init( ROCKSDB_READ_RANGE_TIMEOUT, isSimulated ? 300.0 : 5.0 ); init( ROCKSDB_READ_QUEUE_WAIT, 1.0 ); init( ROCKSDB_READ_QUEUE_HARD_MAX, 1000 ); init( ROCKSDB_READ_QUEUE_SOFT_MAX, 500 ); @@ -388,8 +394,8 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi init( ROCKSDB_FETCH_QUEUE_SOFT_MAX, 50 ); init( ROCKSDB_HISTOGRAMS_SAMPLE_RATE, 0.001 ); if( randomize && BUGGIFY ) ROCKSDB_HISTOGRAMS_SAMPLE_RATE = 0; init( ROCKSDB_READ_RANGE_ITERATOR_REFRESH_TIME, 30.0 ); if( randomize && BUGGIFY ) ROCKSDB_READ_RANGE_ITERATOR_REFRESH_TIME = 0.1; - init( ROCKSDB_READ_RANGE_REUSE_ITERATORS, true ); if( randomize && BUGGIFY ) ROCKSDB_READ_RANGE_REUSE_ITERATORS = deterministicRandom()->coinflip() ? true : false; - init( ROCKSDB_READ_RANGE_REUSE_BOUNDED_ITERATORS, false ); if( randomize && BUGGIFY ) ROCKSDB_READ_RANGE_REUSE_BOUNDED_ITERATORS = deterministicRandom()->coinflip() ? true : false; + init( ROCKSDB_READ_RANGE_REUSE_ITERATORS, true ); if( randomize && BUGGIFY ) ROCKSDB_READ_RANGE_REUSE_ITERATORS = deterministicRandom()->coinflip(); + init( ROCKSDB_READ_RANGE_REUSE_BOUNDED_ITERATORS, false ); if( randomize && BUGGIFY ) ROCKSDB_READ_RANGE_REUSE_BOUNDED_ITERATORS = deterministicRandom()->coinflip(); init( ROCKSDB_READ_RANGE_BOUNDED_ITERATORS_MAX_LIMIT, 200 ); // Set to 0 to disable rocksdb write rate limiting. Rate limiter unit: bytes per second. init( ROCKSDB_WRITE_RATE_LIMITER_BYTES_PER_SEC, 0 ); @@ -398,9 +404,9 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi init( DEFAULT_FDB_ROCKSDB_COLUMN_FAMILY, "fdb"); init( ROCKSDB_DISABLE_AUTO_COMPACTIONS, false ); // RocksDB default - init( ROCKSDB_PERFCONTEXT_ENABLE, false ); if( randomize && BUGGIFY ) ROCKSDB_PERFCONTEXT_ENABLE = deterministicRandom()->coinflip() ? false : true; - init( ROCKSDB_PERFCONTEXT_SAMPLE_RATE, 0.0001 ); - init( ROCKSDB_MAX_SUBCOMPACTIONS, 2 ); + init( ROCKSDB_PERFCONTEXT_ENABLE, false ); if( randomize && BUGGIFY ) ROCKSDB_PERFCONTEXT_ENABLE = deterministicRandom()->coinflip(); + init( ROCKSDB_PERFCONTEXT_SAMPLE_RATE, 0.0001 ); + init( ROCKSDB_MAX_SUBCOMPACTIONS, 0 ); init( ROCKSDB_SOFT_PENDING_COMPACT_BYTES_LIMIT, 64000000000 ); // 64GB, Rocksdb option, Writes will slow down. init( ROCKSDB_HARD_PENDING_COMPACT_BYTES_LIMIT, 100000000000 ); // 100GB, Rocksdb option, Writes will stall. init( ROCKSDB_CAN_COMMIT_COMPACT_BYTES_LIMIT, 50000000000 ); // 50GB, Commit waits. @@ -409,9 +415,18 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi init( ROCKSDB_DISABLE_WAL_EXPERIMENTAL, false ); // If ROCKSDB_SINGLEKEY_DELETES_ON_CLEARRANGE is enabled, disable ROCKSDB_ENABLE_CLEAR_RANGE_EAGER_READS knob. // These knobs have contrary functionality. - init( ROCKSDB_SINGLEKEY_DELETES_ON_CLEARRANGE, false ); if( randomize && BUGGIFY ) ROCKSDB_SINGLEKEY_DELETES_ON_CLEARRANGE = deterministicRandom()->coinflip() ? false : true; - init( ROCKSDB_SINGLEKEY_DELETES_BYTES_LIMIT, 200000 ); // 200KB - init( ROCKSDB_ENABLE_CLEAR_RANGE_EAGER_READS, true ); if( randomize && BUGGIFY ) ROCKSDB_ENABLE_CLEAR_RANGE_EAGER_READS = deterministicRandom()->coinflip() ? false : true; + init( ROCKSDB_SINGLEKEY_DELETES_ON_CLEARRANGE, true ); + init( ROCKSDB_SINGLEKEY_DELETES_BYTES_LIMIT, 10000 ); // 10KB + init( ROCKSDB_ENABLE_CLEAR_RANGE_EAGER_READS, false ); + // ROCKSDB_STATS_LEVEL=1 indicates rocksdb::StatsLevel::kExceptHistogramOrTimers + // Refer StatsLevel: https://github.com/facebook/rocksdb/blob/main/include/rocksdb/statistics.h#L594 + init( ROCKSDB_STATS_LEVEL, 1 ); if( randomize && BUGGIFY ) ROCKSDB_STATS_LEVEL = deterministicRandom()->randomInt(0, 6); + init( ROCKSDB_ENABLE_COMPACT_ON_DELETION, false ); + // CDCF: CompactOnDeletionCollectorFactory. The below 3 are parameters of the CompactOnDeletionCollectorFactory + // which controls the compaction on deleted data. + init( ROCKSDB_CDCF_SLIDING_WINDOW_SIZE, 128 ); + init( ROCKSDB_CDCF_DELETION_TRIGGER, 1 ); + init( ROCKSDB_CDCF_DELETION_RATIO, 0 ); // Can commit will delay ROCKSDB_CAN_COMMIT_DELAY_ON_OVERLOAD seconds for // ROCKSDB_CAN_COMMIT_DELAY_TIMES_ON_OVERLOAD times, if rocksdb overloaded. // Set ROCKSDB_CAN_COMMIT_DELAY_TIMES_ON_OVERLOAD to 0, to disable @@ -521,7 +536,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi init( BACKUP_TIMEOUT, 0.4 ); init( BACKUP_NOOP_POP_DELAY, 5.0 ); init( BACKUP_FILE_BLOCK_BYTES, 1024 * 1024 ); - init( BACKUP_LOCK_BYTES, 3e9 ); if(randomize && BUGGIFY) BACKUP_LOCK_BYTES = deterministicRandom()->randomInt(1024, 4096) * 15 * 1024; + init( BACKUP_LOCK_BYTES, 3e9 ); if(randomize && BUGGIFY) BACKUP_LOCK_BYTES = deterministicRandom()->randomInt(1024, 4096) * 256 * 1024; init( BACKUP_UPLOAD_DELAY, 10.0 ); if(randomize && BUGGIFY) BACKUP_UPLOAD_DELAY = deterministicRandom()->random01() * 60; //Cluster Controller @@ -736,11 +751,13 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi 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 ); if( randomize && BUGGIFY ) ENABLE_CLEAR_RANGE_EAGER_READS = deterministicRandom()->coinflip() ? false : true; + init( ENABLE_CLEAR_RANGE_EAGER_READS, true ); if( randomize && BUGGIFY ) ENABLE_CLEAR_RANGE_EAGER_READS = deterministicRandom()->coinflip(); init( CHECKPOINT_TRANSFER_BLOCK_BYTES, 40e6 ); init( QUICK_GET_VALUE_FALLBACK, true ); init( QUICK_GET_KEY_VALUES_FALLBACK, true ); - init( MAX_PARALLEL_QUICK_GET_VALUE, 50 ); if ( randomize && BUGGIFY ) MAX_PARALLEL_QUICK_GET_VALUE = deterministicRandom()->randomInt(1, 100); + init( STRICTLY_ENFORCE_BYTE_LIMIT, false); if( randomize && BUGGIFY ) STRICTLY_ENFORCE_BYTE_LIMIT = deterministicRandom()->coinflip(); + init( FRACTION_INDEX_BYTELIMIT_PREFETCH, 0.2); if( randomize && BUGGIFY ) FRACTION_INDEX_BYTELIMIT_PREFETCH = 0.01 + deterministicRandom()->random01(); + init( MAX_PARALLEL_QUICK_GET_VALUE, 10 ); if ( randomize && BUGGIFY ) MAX_PARALLEL_QUICK_GET_VALUE = deterministicRandom()->randomInt(1, 100); init( QUICK_GET_KEY_VALUES_LIMIT, 2000 ); init( QUICK_GET_KEY_VALUES_LIMIT_BYTES, 1e7 ); diff --git a/fdbclient/ServerKnobs.h b/fdbclient/ServerKnobs.h index 2034b61f4e7..33fdcaf4f37 100644 --- a/fdbclient/ServerKnobs.h +++ b/fdbclient/ServerKnobs.h @@ -86,6 +86,7 @@ class ServerKnobs : public KnobsImpl { double DESIRED_GET_MORE_DELAY; int CONCURRENT_LOG_ROUTER_READS; int LOG_ROUTER_PEEK_FROM_SATELLITES_PREFERRED; // 0==peek from primary, non-zero==peek from satellites + double LOG_ROUTER_PEEK_SWITCH_DC_TIME; double DISK_QUEUE_ADAPTER_MIN_SWITCH_TIME; double DISK_QUEUE_ADAPTER_MAX_SWITCH_TIME; int64_t TLOG_SPILL_REFERENCE_MAX_PEEK_MEMORY_BYTES; @@ -116,6 +117,7 @@ class ServerKnobs : public KnobsImpl { bool PEEK_BATCHING_EMPTY_MSG; double PEEK_BATCHING_EMPTY_MSG_INTERVAL; double POP_FROM_LOG_DELAY; + double TLOG_PULL_ASYNC_DATA_WARNING_TIMEOUT_SECS; // Data distribution queue double HEALTH_POLL_TIME; @@ -240,6 +242,7 @@ class ServerKnobs : public KnobsImpl { double DD_FAILURE_TIME; double DD_ZERO_HEALTHY_TEAM_DELAY; + int DD_BUILD_EXTRA_TEAMS_OVERRIDE; // build extra teams to allow data movement to progress. must be larger than 0 // Run storage enginee on a child process on the same machine with storage process bool REMOTE_KV_STORE; @@ -292,7 +295,7 @@ class ServerKnobs : public KnobsImpl { // KeyValueStoreRocksDB bool ROCKSDB_LEVEL_COMPACTION_DYNAMIC_LEVEL_BYTES; - int ROCKSDB_SUGGEST_COMPACT_CLEAR_RANGE; + bool ROCKSDB_SUGGEST_COMPACT_CLEAR_RANGE; int ROCKSDB_THREAD_PROMISE_PRIORITY; int ROCKSDB_READER_THREAD_PRIORITY; int ROCKSDB_WRITER_THREAD_PRIORITY; @@ -336,6 +339,11 @@ class ServerKnobs : public KnobsImpl { bool ROCKSDB_SINGLEKEY_DELETES_ON_CLEARRANGE; int64_t ROCKSDB_SINGLEKEY_DELETES_BYTES_LIMIT; bool ROCKSDB_ENABLE_CLEAR_RANGE_EAGER_READS; + bool ROCKSDB_ENABLE_COMPACT_ON_DELETION; + int64_t ROCKSDB_CDCF_SLIDING_WINDOW_SIZE; // CDCF: CompactOnDeletionCollectorFactory + int64_t ROCKSDB_CDCF_DELETION_TRIGGER; // CDCF: CompactOnDeletionCollectorFactory + double ROCKSDB_CDCF_DELETION_RATIO; // CDCF: CompactOnDeletionCollectorFactory + int ROCKSDB_STATS_LEVEL; int64_t ROCKSDB_COMPACTION_READAHEAD_SIZE; int64_t ROCKSDB_BLOCK_SIZE; @@ -675,6 +683,8 @@ class ServerKnobs : public KnobsImpl { bool ENABLE_CLEAR_RANGE_EAGER_READS; bool QUICK_GET_VALUE_FALLBACK; bool QUICK_GET_KEY_VALUES_FALLBACK; + bool STRICTLY_ENFORCE_BYTE_LIMIT; + double FRACTION_INDEX_BYTELIMIT_PREFETCH; int MAX_PARALLEL_QUICK_GET_VALUE; int CHECKPOINT_TRANSFER_BLOCK_BYTES; int QUICK_GET_KEY_VALUES_LIMIT; diff --git a/fdbclient/SpecialKeySpace.actor.cpp b/fdbclient/SpecialKeySpace.actor.cpp index 6d2d717e49b..451febfbeeb 100644 --- a/fdbclient/SpecialKeySpace.actor.cpp +++ b/fdbclient/SpecialKeySpace.actor.cpp @@ -320,7 +320,9 @@ ACTOR Future SpecialKeySpace::checkRYWValid(SpecialKeySpace* sks, wait(SpecialKeySpace::getRangeAggregationActor(sks, ryw, begin, end, limits, reverse))) { return result; } - when(wait(ryw->resetFuture())) { throw internal_error(); } + when(wait(ryw->resetFuture())) { + throw internal_error(); + } } } @@ -973,7 +975,7 @@ ACTOR Future checkExclusion(Database db, state int ssTotalCount = 0; state int ssExcludedCount = 0; - state std::unordered_set diskLocalities(); + state std::unordered_set diskLocalities; state int64_t totalKvStoreFreeBytes = 0; state int64_t totalKvStoreUsedBytes = 0; state int64_t totalKvStoreUsedBytesNonExcluded = 0; diff --git a/fdbclient/VersionedMap.h b/fdbclient/VersionedMap.h index ad5d9f32f0f..2b47c96fc8d 100644 --- a/fdbclient/VersionedMap.h +++ b/fdbclient/VersionedMap.h @@ -284,12 +284,17 @@ void insert(Reference>& p, Version at, const T& x) { if (!p) { p = makeReference>(x, at); } else { - bool direction = !(x < p->data); - Reference> child = p->child(direction, at); - insert(child, at, x); - p = update(p, direction, child, at); - if (p->child(direction, at)->priority > p->priority) - rotate(p, at, !direction); + int c = ::compare(x, p->data); + if (c == 0) { + p = makeReference>(p->priority, x, p->left(at), p->right(at), at); + } else { + const bool direction = !(c < 0); + Reference> child = p->child(direction, at); + insert(child, at, x); + p = update(p, direction, child, at); + if (p->child(direction, at)->priority > p->priority) + rotate(p, at, !direction); + } } } @@ -345,6 +350,41 @@ void removeRoot(Reference>& p, Version at) { } } +// changes p to point to a PTree with finger removed. p must be the root of the +// tree associated with finger. +// +// Invalidates finger. +template +void removeFinger(Reference>& p, Version at, PTreeFinger finger) { + ASSERT_GT(finger.size(), 0); + // Start at the end of the finger, remove, and propagate copies up along the + // search path (finger) as needed. + auto node = Reference>::addRef(const_cast*>(finger.back())); + auto* before = node.getPtr(); + removeRoot(node, at); + for (;;) { + if (before == node.getPtr()) { + // Done propagating copies + return; + } + if (finger.size() == 1) { + // Check we passed the correct root for this finger + ASSERT(p.getPtr() == before); + // Propagate copy to root + p = node; + return; + } + finger.pop_back(); + auto parent = Reference>::addRef(const_cast*>(finger.back())); + bool isLeftChild = parent->left(at).getPtr() == before; + bool isRightChild = parent->right(at).getPtr() == before; + ASSERT(isLeftChild || isRightChild); // Corrupt finger? + // Prepare for next iteration + before = parent.getPtr(); + node = update(parent, isRightChild, node, at); + } +} + // changes p to point to a PTree with x removed template void remove(Reference>& p, Version at, const X& x) { @@ -684,7 +724,7 @@ class VersionedMap : NonCopyable { } Future forgetVersionsBeforeAsync(Version newOldestVersion, TaskPriority taskID = TaskPriority::DefaultYield) { - ASSERT(newOldestVersion <= latestVersion); + ASSERT_LE(newOldestVersion, latestVersion); auto r = upper_bound(roots.begin(), roots.end(), newOldestVersion, rootsComparator()); auto upper = r; --r; @@ -731,10 +771,6 @@ class VersionedMap : NonCopyable { // insert() and erase() invalidate atLatest() and all iterators into it void insert(const K& k, const T& t) { insert(k, t, latestVersion); } void insert(const K& k, const T& t, Version insertAt) { - if (PTreeImpl::contains(roots.back().second, latestVersion, k)) - PTreeImpl::remove(roots.back().second, - latestVersion, - k); // FIXME: Make PTreeImpl::insert do this automatically (see also WriteMap.h FIXME) PTreeImpl::insert( roots.back().second, latestVersion, MapPair>(k, std::make_pair(t, insertAt))); } @@ -743,9 +779,8 @@ class VersionedMap : NonCopyable { PTreeImpl::remove(roots.back().second, latestVersion, key); } void erase(iterator const& item) { // iterator must be in latest version! - // SOMEDAY: Optimize to use item.finger and avoid repeated search - K key = item.key(); - erase(key); + ASSERT_EQ(item.at, latestVersion); + PTreeImpl::removeFinger(roots.back().second, latestVersion, item.finger); } void printDetail() { PTreeImpl::printTreeDetails(roots.back().second, 0); } diff --git a/fdbclient/WriteMap.h b/fdbclient/WriteMap.h index 33fb6aee373..b0529b82c77 100644 --- a/fdbclient/WriteMap.h +++ b/fdbclient/WriteMap.h @@ -134,6 +134,8 @@ struct WriteMapEntry { int compare(ExtStringRef const& r) const { return -r.compare(key); } + int compare(WriteMapEntry const& r) const { return key.compare(r.key); } + std::string toString() const { return printable(key); } }; diff --git a/fdbclient/json_spirit/json_spirit_reader_template.h b/fdbclient/json_spirit/json_spirit_reader_template.h index 207e2e8e74e..9005a09764b 100644 --- a/fdbclient/json_spirit/json_spirit_reader_template.h +++ b/fdbclient/json_spirit/json_spirit_reader_template.h @@ -13,7 +13,7 @@ #include "json_spirit_value.h" #include "json_spirit_error_position.h" -//#define BOOST_SPIRIT_THREADSAFE // uncomment for multithreaded use, requires linking to boost.thread +// #define BOOST_SPIRIT_THREADSAFE // uncomment for multithreaded use, requires linking to boost.thread #include #include diff --git a/fdbclient/vexillographer/fdb.options b/fdbclient/vexillographer/fdb.options index 9f2f9e52f93..98abe24e48e 100644 --- a/fdbclient/vexillographer/fdb.options +++ b/fdbclient/vexillographer/fdb.options @@ -115,6 +115,8 @@ description is not currently required but encouraged.