diff --git a/bindings/c/CMakeLists.txt b/bindings/c/CMakeLists.txt index d08bb48344..0c1bdbc554 100644 --- a/bindings/c/CMakeLists.txt +++ b/bindings/c/CMakeLists.txt @@ -1,7 +1,6 @@ set(FDB_C_SRCS fdb_c.cpp - foundationdb/fdb_c.h - ThreadCleanup.cpp) + foundationdb/fdb_c.h) file(MAKE_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/foundationdb) @@ -18,6 +17,8 @@ endif() if(CMAKE_SYSTEM_PROCESSOR MATCHES "aarch64" OR CMAKE_SYSTEM_PROCESSOR MATCHES "arm64") set(cpu "aarch64") +elseif(CMAKE_SYSTEM_PROCESSOR MATCHES "^(ppc64le|powerpc64le)") + set(cpu "ppc64le") endif() set(IS_ARM_MAC NO) @@ -49,7 +50,7 @@ endif() add_dependencies(fdb_c fdb_c_generated fdb_c_options) add_dependencies(fdbclient fdb_c_options) add_dependencies(fdbclient_sampling fdb_c_options) -target_link_libraries(fdb_c PUBLIC $) +target_link_libraries(fdb_c PRIVATE $) if(APPLE) set(symbols ${CMAKE_CURRENT_BINARY_DIR}/fdb_c.symbols) add_custom_command(OUTPUT ${symbols} @@ -74,10 +75,8 @@ if(WIN32) set_property(SOURCE ${asm_file} PROPERTY LANGUAGE ASM_MASM) endif() -# The tests don't build on windows and ARM macs -# doctest doesn't seem to compile on ARM macs, we should -# check later whether this works -if(NOT WIN32 AND NOT IS_ARM_MAC) +# The tests don't build on windows +if(NOT WIN32) set(MAKO_SRCS test/mako/mako.c test/mako/mako.h @@ -121,9 +120,9 @@ if(NOT WIN32 AND NOT IS_ARM_MAC) strip_debug_symbols(fdb_c_ryw_benchmark) strip_debug_symbols(fdb_c_txn_size_test) endif() - target_link_libraries(fdb_c_performance_test PRIVATE fdb_c) - target_link_libraries(fdb_c_ryw_benchmark PRIVATE fdb_c) - target_link_libraries(fdb_c_txn_size_test PRIVATE fdb_c) + target_link_libraries(fdb_c_performance_test PRIVATE fdb_c Threads::Threads) + target_link_libraries(fdb_c_ryw_benchmark PRIVATE fdb_c Threads::Threads) + target_link_libraries(fdb_c_txn_size_test PRIVATE fdb_c Threads::Threads) add_dependencies(fdb_c_setup_tests doctest) add_dependencies(fdb_c_unit_tests doctest) @@ -134,14 +133,14 @@ if(NOT WIN32 AND NOT IS_ARM_MAC) target_include_directories(fdb_c_unit_tests_version_510 PUBLIC ${DOCTEST_INCLUDE_DIR}) target_include_directories(disconnected_timeout_unit_tests PUBLIC ${DOCTEST_INCLUDE_DIR}) target_link_libraries(fdb_c_setup_tests PRIVATE fdb_c Threads::Threads) - target_link_libraries(fdb_c_unit_tests PRIVATE fdb_c Threads::Threads) + target_link_libraries(fdb_c_unit_tests PRIVATE fdb_c Threads::Threads fdbclient) target_link_libraries(fdb_c_unit_tests_version_510 PRIVATE fdb_c Threads::Threads) - target_link_libraries(trace_partial_file_suffix_test PRIVATE fdb_c Threads::Threads) + target_link_libraries(trace_partial_file_suffix_test PRIVATE fdb_c Threads::Threads flow) target_link_libraries(disconnected_timeout_unit_tests PRIVATE fdb_c Threads::Threads) # do not set RPATH for mako set_property(TARGET mako PROPERTY SKIP_BUILD_RPATH TRUE) - target_link_libraries(mako PRIVATE fdb_c) + target_link_libraries(mako PRIVATE fdb_c fdbclient) if(NOT OPEN_FOR_IDE) # Make sure that fdb_c.h is compatible with c90 diff --git a/bindings/c/ThreadCleanup.cpp b/bindings/c/ThreadCleanup.cpp deleted file mode 100644 index 86f61ba608..0000000000 --- a/bindings/c/ThreadCleanup.cpp +++ /dev/null @@ -1,64 +0,0 @@ -/* - * ThreadCleanup.cpp - * - * This source file is part of the FoundationDB open source project - * - * Copyright 2013-2018 Apple Inc. and the FoundationDB project authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "flow/Platform.h" -#include "flow/FastAlloc.h" - -#if defined(WIN32) - -#include - -BOOL WINAPI DllMain(HINSTANCE dll, DWORD reason, LPVOID reserved) { - - if (reason == DLL_THREAD_DETACH) - releaseAllThreadMagazines(); - return TRUE; -} - -#elif defined(__unixish__) - -#ifdef __INTEL_COMPILER -#pragma warning(disable : 2415) -#endif - -static pthread_key_t threadDestructorKey; - -static void threadDestructor(void*) { - releaseAllThreadMagazines(); -} - -void registerThread() { - pthread_setspecific(threadDestructorKey, (const void*)1); -} - -static int initThreadDestructorKey() { - if (!pthread_key_create(&threadDestructorKey, &threadDestructor)) { - registerThread(); - setFastAllocatorThreadInitFunction(®isterThread); - } - - return 0; -} - -static int threadDestructorKeyInit = initThreadDestructorKey(); - -#else -#error Port me! -#endif diff --git a/bindings/c/generate_asm.py b/bindings/c/generate_asm.py index b0e9a0e7ad..3acd14abed 100755 --- a/bindings/c/generate_asm.py +++ b/bindings/c/generate_asm.py @@ -59,9 +59,10 @@ def write_windows_asm(asmfile, functions): def write_unix_asm(asmfile, functions, prefix): - if cpu != "aarch64": + if cpu != "aarch64" and cpu!= "ppc64le": asmfile.write(".intel_syntax noprefix\n") + i = 0 if os == 'linux' or os == 'freebsd': asmfile.write("\n.data\n") for f in functions: @@ -70,8 +71,13 @@ def write_unix_asm(asmfile, functions, prefix): if os == 'linux' or os == 'freebsd': asmfile.write("\n.text\n") for f in functions: + if cpu == "ppc64le": + asmfile.write("\n.LC%d:\n" % (i)) + asmfile.write("\t.quad \tfdb_api_ptr_%s\n" % (f)) + asmfile.write("\t.align 2\n") + i = i + 1 asmfile.write("\t.global %s\n\t.type %s, @function\n" % (f, f)) - + i = 0 for f in functions: asmfile.write("\n.globl %s%s\n" % (prefix, f)) if cpu == 'aarch64' and os == 'osx': @@ -118,6 +124,46 @@ def write_unix_asm(asmfile, functions, prefix): assert False, '{} not supported for Arm yet'.format(os) asmfile.write("\tldr x8, [x8]\n") asmfile.write("\tbr x8\n") + elif cpu == "ppc64le": + asmfile.write("\n.LCF%d:\n" % (i)) + asmfile.write("\taddis 2,12,.TOC.-.LCF%d@ha\n" % (i)) + asmfile.write("\taddi 2,2,.TOC.-.LCF%d@l\n" % (i)) + asmfile.write("\tmflr 0\n") + asmfile.write("\tstd 31, -8(1)\n") + asmfile.write("\tstd 0,16(1)\n") + asmfile.write("\tstdu 1,-192(1)\n") + #asmfile.write("\tstd 2,24(1)\n") + asmfile.write("\taddis 11,2,.LC%d@toc@ha\n" % (i)) + asmfile.write("\tld 11,.LC%d@toc@l(11)\n" % (i)) + asmfile.write("\tld 12,0(11)\n") + asmfile.write("\tstd 2,24(1)\n") + asmfile.write("\tlwa 11,344(1)\n") + asmfile.write("\tmtctr 12\n") + asmfile.write("\tstd 11,152(1)\n") + asmfile.write("\tlwa 11,352(1)\n") + asmfile.write("\tstd 11,160(1)\n") + asmfile.write("\tlwa 11,336(1)\n") + asmfile.write("\tstd 11,144(1)\n") + asmfile.write("\tlwa 11,328(1)\n") + asmfile.write("\tstd 11,136(1)\n") + asmfile.write("\tlwa 11,320(1)\n") + asmfile.write("\tstd 11,128(1)\n") + asmfile.write("\tlwa 11,312(1)\n") + asmfile.write("\tstd 11,120(1)\n") + asmfile.write("\tlwa 11,304(1)\n") + asmfile.write("\tstd 11,112(1)\n") + asmfile.write("\tld 11,296(1)\n") + asmfile.write("\tstd 11,104(1)\n") + asmfile.write("\tlwa 11,288(1)\n") + asmfile.write("\tstd 11,96(1)\n") + asmfile.write("\tbctrl\n") + asmfile.write("\tld 2,24(1)\n") + asmfile.write("\taddi 1,1,192\n") + asmfile.write("\tld 0,16(1)\n") + asmfile.write("\tld 31, -8(1)\n") + asmfile.write("\tmtlr 0\n") + asmfile.write("\tblr\n") + i = i + 1 else: asmfile.write( "\tmov r11, qword ptr [%sfdb_api_ptr_%s@GOTPCREL+rip]\n" % (prefix, f)) diff --git a/bindings/c/test/mako/mako.c b/bindings/c/test/mako/mako.c index 7c887dd48a..fc69ae06b5 100644 --- a/bindings/c/test/mako/mako.c +++ b/bindings/c/test/mako/mako.c @@ -1,17 +1,18 @@ #include +#include #include #include +#include #include -#include #include #include #include #include +#include +#include #include #include #include -#include -#include #if defined(__linux__) #include @@ -615,7 +616,7 @@ int64_t granule_start_load(const char* filename, // don't seek if offset == 0 if (offset && fseek(fp, offset, SEEK_SET)) { // if fseek was non-zero, it failed - fprintf(stderr, "ERROR: BG could not seek to %ld in file %s\n", offset, full_fname); + fprintf(stderr, "ERROR: BG could not seek to %" PRId64 " in file %s\n", offset, full_fname); fclose(fp); return -1; } @@ -625,7 +626,7 @@ int64_t granule_start_load(const char* filename, fclose(fp); if (readSize != length) { - fprintf(stderr, "ERROR: BG could not read %ld bytes from file: %s\n", length, full_fname); + fprintf(stderr, "ERROR: BG could not read %" PRId64 " bytes from file: %s\n", length, full_fname); return -1; } @@ -636,7 +637,7 @@ int64_t granule_start_load(const char* filename, uint8_t* granule_get_load(int64_t loadId, void* userContext) { BGLocalFileContext* context = (BGLocalFileContext*)userContext; if (context->data_by_id[loadId] == 0) { - fprintf(stderr, "ERROR: BG loadId invalid for get_load: %ld\n", loadId); + fprintf(stderr, "ERROR: BG loadId invalid for get_load: %" PRId64 "\n", loadId); return 0; } return context->data_by_id[loadId]; @@ -645,7 +646,7 @@ uint8_t* granule_get_load(int64_t loadId, void* userContext) { void granule_free_load(int64_t loadId, void* userContext) { BGLocalFileContext* context = (BGLocalFileContext*)userContext; if (context->data_by_id[loadId] == 0) { - fprintf(stderr, "ERROR: BG loadId invalid for free_load: %ld\n", loadId); + fprintf(stderr, "ERROR: BG loadId invalid for free_load: %" PRId64 "\n", loadId); } free(context->data_by_id[loadId]); context->data_by_id[loadId] = 0; @@ -1119,7 +1120,7 @@ int run_workload(FDBTransaction* transaction, if (tracetimer == dotrace) { fdb_error_t err; tracetimer = 0; - snprintf(traceid, 32, "makotrace%019ld", total_xacts); + snprintf(traceid, 32, "makotrace%019" PRId64, total_xacts); fprintf(debugme, "DEBUG: txn tracing %s\n", traceid); err = fdb_transaction_set_option(transaction, FDB_TR_OPTION_DEBUG_TRANSACTION_IDENTIFIER, @@ -1283,7 +1284,7 @@ void* worker_thread(void* thread_args) { } fprintf(debugme, - "DEBUG: worker_id:%d (%d) thread_id:%d (%d) database_index:%lu (tid:%lu)\n", + "DEBUG: worker_id:%d (%d) thread_id:%d (%d) database_index:%lu (tid:%" PRIu64 ")\n", worker_id, args->num_processes, thread_id, @@ -1350,6 +1351,11 @@ void* worker_thread(void* thread_args) { char str2[1000]; sprintf(str2, "%s%d", TEMP_DATA_STORE, *parent_id); rc = mkdir(str2, S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH); + if (rc < 0 && errno != EEXIST) { + int ec = errno; + fprintf(stderr, "Failed to make directory: %s because %s\n", str2, strerror(ec)); + goto failExit; + } for (op = 0; op < MAX_OP; op++) { if (args->txnspec.ops[op][OP_COUNT] > 0 || op == OP_COMMIT || op == OP_TRANSACTION) { FILE* fp; @@ -1357,6 +1363,11 @@ void* worker_thread(void* thread_args) { strcat(file_name, str2); get_stats_file_name(file_name, worker_id, thread_id, op); fp = fopen(file_name, "w"); + if (!fp) { + int ec = errno; + fprintf(stderr, "Failed to open file: %s because %s\n", file_name, strerror(ec)); + goto failExit; + } lat_block_t* temp_block = ((thread_args_t*)thread_args)->block[op]; if (is_memory_allocated[op]) { size = stats->latency_samples[op] / LAT_BLOCK_SIZE; @@ -1376,11 +1387,11 @@ void* worker_thread(void* thread_args) { fclose(fp); } } - __sync_fetch_and_add(stopcount, 1); } /* fall through */ failExit: + __sync_fetch_and_add(stopcount, 1); for (op = 0; op < MAX_OP; op++) { lat_block_t* curr = ((thread_args_t*)thread_args)->block[op]; lat_block_t* prev = NULL; @@ -2240,9 +2251,9 @@ void print_stats(mako_args_t* args, mako_stats_t* stats, struct timespec* now, s for (op = 0; op < MAX_OP; op++) { if (args->txnspec.ops[op][OP_COUNT] > 0) { uint64_t ops_total_diff = ops_total[op] - ops_total_prev[op]; - printf("%" STR(STATS_FIELD_WIDTH) "lu ", ops_total_diff); + printf("%" STR(STATS_FIELD_WIDTH) PRIu64 " ", ops_total_diff); if (fp) { - fprintf(fp, "\"%s\": %lu,", get_ops_name(op), ops_total_diff); + fprintf(fp, "\"%s\": %" PRIu64 ",", get_ops_name(op), ops_total_diff); } errors_diff[op] = errors_total[op] - errors_total_prev[op]; print_err = (errors_diff[op] > 0); @@ -2270,7 +2281,7 @@ void print_stats(mako_args_t* args, mako_stats_t* stats, struct timespec* now, s printf("%" STR(STATS_TITLE_WIDTH) "s ", "Errors"); for (op = 0; op < MAX_OP; op++) { if (args->txnspec.ops[op][OP_COUNT] > 0) { - printf("%" STR(STATS_FIELD_WIDTH) "lu ", errors_diff[op]); + printf("%" STR(STATS_FIELD_WIDTH) PRIu64 " ", errors_diff[op]); if (fp) { fprintf(fp, ",\"errors\": %.2f", conflicts_diff); } @@ -2419,10 +2430,10 @@ void print_report(mako_args_t* args, break; } } - printf("Total Xacts: %8lu\n", totalxacts); - printf("Total Conflicts: %8lu\n", conflicts); - printf("Total Errors: %8lu\n", totalerrors); - printf("Overall TPS: %8lu\n\n", totalxacts * 1000000000 / duration_nsec); + printf("Total Xacts: %8" PRIu64 "\n", totalxacts); + printf("Total Conflicts: %8" PRIu64 "\n", conflicts); + printf("Total Errors: %8" PRIu64 "\n", totalerrors); + printf("Overall TPS: %8" PRIu64 "\n\n", totalxacts * 1000000000 / duration_nsec); if (fp) { fprintf(fp, "\"results\": {"); @@ -2430,10 +2441,10 @@ void print_report(mako_args_t* args, fprintf(fp, "\"totalProcesses\": %d,", args->num_processes); fprintf(fp, "\"totalThreads\": %d,", args->num_threads); fprintf(fp, "\"targetTPS\": %d,", args->tpsmax); - fprintf(fp, "\"totalXacts\": %lu,", totalxacts); - fprintf(fp, "\"totalConflicts\": %lu,", conflicts); - fprintf(fp, "\"totalErrors\": %lu,", totalerrors); - fprintf(fp, "\"overallTPS\": %lu,", totalxacts * 1000000000 / duration_nsec); + fprintf(fp, "\"totalXacts\": %" PRIu64 ",", totalxacts); + fprintf(fp, "\"totalConflicts\": %" PRIu64 ",", conflicts); + fprintf(fp, "\"totalErrors\": %" PRIu64 ",", totalerrors); + fprintf(fp, "\"overallTPS\": %" PRIu64 ",", totalxacts * 1000000000 / duration_nsec); } /* per-op stats */ @@ -2446,14 +2457,14 @@ void print_report(mako_args_t* args, } for (op = 0; op < MAX_OP; op++) { if ((args->txnspec.ops[op][OP_COUNT] > 0 && op != OP_TRANSACTION) || op == OP_COMMIT) { - printf("%" STR(STATS_FIELD_WIDTH) "lu ", ops_total[op]); + printf("%" STR(STATS_FIELD_WIDTH) PRIu64 " ", ops_total[op]); if (fp) { if (first_op) { first_op = 0; } else { fprintf(fp, ","); } - fprintf(fp, "\"%s\": %lu", get_ops_name(op), ops_total[op]); + fprintf(fp, "\"%s\": %" PRIu64, get_ops_name(op), ops_total[op]); } } } @@ -2475,14 +2486,14 @@ void print_report(mako_args_t* args, first_op = 1; for (op = 0; op < MAX_OP; op++) { if (args->txnspec.ops[op][OP_COUNT] > 0 && op != OP_TRANSACTION) { - printf("%" STR(STATS_FIELD_WIDTH) "lu ", errors_total[op]); + printf("%" STR(STATS_FIELD_WIDTH) PRIu64 " ", errors_total[op]); if (fp) { if (first_op) { first_op = 0; } else { fprintf(fp, ","); } - fprintf(fp, "\"%s\": %lu", get_ops_name(op), errors_total[op]); + fprintf(fp, "\"%s\": %" PRIu64, get_ops_name(op), errors_total[op]); } } } @@ -2500,7 +2511,7 @@ void print_report(mako_args_t* args, for (op = 0; op < MAX_OP; op++) { if (args->txnspec.ops[op][OP_COUNT] > 0 || op == OP_TRANSACTION || op == OP_COMMIT) { if (lat_total[op]) { - printf("%" STR(STATS_FIELD_WIDTH) "lu ", lat_samples[op]); + printf("%" STR(STATS_FIELD_WIDTH) PRIu64 " ", lat_samples[op]); } else { printf("%" STR(STATS_FIELD_WIDTH) "s ", "N/A"); } @@ -2510,7 +2521,7 @@ void print_report(mako_args_t* args, } else { fprintf(fp, ","); } - fprintf(fp, "\"%s\": %lu", get_ops_name(op), lat_samples[op]); + fprintf(fp, "\"%s\": %" PRIu64, get_ops_name(op), lat_samples[op]); } } } @@ -2527,14 +2538,14 @@ void print_report(mako_args_t* args, if (lat_min[op] == -1) { printf("%" STR(STATS_FIELD_WIDTH) "s ", "N/A"); } else { - printf("%" STR(STATS_FIELD_WIDTH) "lu ", lat_min[op]); + printf("%" STR(STATS_FIELD_WIDTH) PRIu64 " ", lat_min[op]); if (fp) { if (first_op) { first_op = 0; } else { fprintf(fp, ","); } - fprintf(fp, "\"%s\": %lu", get_ops_name(op), lat_min[op]); + fprintf(fp, "\"%s\": %" PRIu64, get_ops_name(op), lat_min[op]); } } } @@ -2550,14 +2561,14 @@ void print_report(mako_args_t* args, for (op = 0; op < MAX_OP; op++) { if (args->txnspec.ops[op][OP_COUNT] > 0 || op == OP_TRANSACTION || op == OP_COMMIT) { if (lat_total[op]) { - printf("%" STR(STATS_FIELD_WIDTH) "lu ", lat_total[op] / lat_samples[op]); + printf("%" STR(STATS_FIELD_WIDTH) PRIu64 " ", lat_total[op] / lat_samples[op]); if (fp) { if (first_op) { first_op = 0; } else { fprintf(fp, ","); } - fprintf(fp, "\"%s\": %lu", get_ops_name(op), lat_total[op] / lat_samples[op]); + fprintf(fp, "\"%s\": %" PRIu64, get_ops_name(op), lat_total[op] / lat_samples[op]); } } else { printf("%" STR(STATS_FIELD_WIDTH) "s ", "N/A"); @@ -2577,14 +2588,14 @@ void print_report(mako_args_t* args, if (lat_max[op] == 0) { printf("%" STR(STATS_FIELD_WIDTH) "s ", "N/A"); } else { - printf("%" STR(STATS_FIELD_WIDTH) "lu ", lat_max[op]); + printf("%" STR(STATS_FIELD_WIDTH) PRIu64 " ", lat_max[op]); if (fp) { if (first_op) { first_op = 0; } else { fprintf(fp, ","); } - fprintf(fp, "\"%s\": %lu", get_ops_name(op), lat_max[op]); + fprintf(fp, "\"%s\": %" PRIu64, get_ops_name(op), lat_max[op]); } } } @@ -2635,14 +2646,14 @@ void print_report(mako_args_t* args, } else { median = (dataPoints[op][num_points[op] / 2] + dataPoints[op][num_points[op] / 2 - 1]) >> 1; } - printf("%" STR(STATS_FIELD_WIDTH) "lu ", median); + printf("%" STR(STATS_FIELD_WIDTH) PRIu64 " ", median); if (fp) { if (first_op) { first_op = 0; } else { fprintf(fp, ","); } - fprintf(fp, "\"%s\": %lu", get_ops_name(op), median); + fprintf(fp, "\"%s\": %" PRIu64, get_ops_name(op), median); } } else { printf("%" STR(STATS_FIELD_WIDTH) "s ", "N/A"); @@ -2665,14 +2676,14 @@ void print_report(mako_args_t* args, } if (lat_total[op]) { point_95pct = ((float)(num_points[op]) * 0.95) - 1; - printf("%" STR(STATS_FIELD_WIDTH) "lu ", dataPoints[op][point_95pct]); + printf("%" STR(STATS_FIELD_WIDTH) PRIu64 " ", dataPoints[op][point_95pct]); if (fp) { if (first_op) { first_op = 0; } else { fprintf(fp, ","); } - fprintf(fp, "\"%s\": %lu", get_ops_name(op), dataPoints[op][point_95pct]); + fprintf(fp, "\"%s\": %" PRIu64, get_ops_name(op), dataPoints[op][point_95pct]); } } else { printf("%" STR(STATS_FIELD_WIDTH) "s ", "N/A"); @@ -2695,14 +2706,14 @@ void print_report(mako_args_t* args, } if (lat_total[op]) { point_99pct = ((float)(num_points[op]) * 0.99) - 1; - printf("%" STR(STATS_FIELD_WIDTH) "lu ", dataPoints[op][point_99pct]); + printf("%" STR(STATS_FIELD_WIDTH) PRIu64 " ", dataPoints[op][point_99pct]); if (fp) { if (first_op) { first_op = 0; } else { fprintf(fp, ","); } - fprintf(fp, "\"%s\": %lu", get_ops_name(op), dataPoints[op][point_99pct]); + fprintf(fp, "\"%s\": %" PRIu64, get_ops_name(op), dataPoints[op][point_99pct]); } } else { printf("%" STR(STATS_FIELD_WIDTH) "s ", "N/A"); @@ -2725,14 +2736,14 @@ void print_report(mako_args_t* args, } if (lat_total[op]) { point_99_9pct = ((float)(num_points[op]) * 0.999) - 1; - printf("%" STR(STATS_FIELD_WIDTH) "lu ", dataPoints[op][point_99_9pct]); + printf("%" STR(STATS_FIELD_WIDTH) PRIu64 " ", dataPoints[op][point_99_9pct]); if (fp) { if (first_op) { first_op = 0; } else { fprintf(fp, ","); } - fprintf(fp, "\"%s\": %lu", get_ops_name(op), dataPoints[op][point_99_9pct]); + fprintf(fp, "\"%s\": %" PRIu64, get_ops_name(op), dataPoints[op][point_99_9pct]); } } else { printf("%" STR(STATS_FIELD_WIDTH) "s ", "N/A"); diff --git a/bindings/c/test/txn_size_test.c b/bindings/c/test/txn_size_test.c index b8be90ceb1..1fb08eb3f5 100644 --- a/bindings/c/test/txn_size_test.c +++ b/bindings/c/test/txn_size_test.c @@ -67,25 +67,25 @@ void runTests(struct ResultSet* rs) { fdb_transaction_set(tr, keys[i], KEY_SIZE, valueStr, VALUE_SIZE); e = getSize(rs, tr, sizes + i); checkError(e, "transaction get size", rs); - printf("size %d: %ld\n", i, sizes[i]); + printf("size %d: %" PRId64 "\n", i, sizes[i]); i++; fdb_transaction_set(tr, keys[i], KEY_SIZE, valueStr, VALUE_SIZE); e = getSize(rs, tr, sizes + i); checkError(e, "transaction get size", rs); - printf("size %d: %ld\n", i, sizes[i]); + printf("size %d: %" PRId64 "\n", i, sizes[i]); i++; fdb_transaction_clear(tr, keys[i], KEY_SIZE); e = getSize(rs, tr, sizes + i); checkError(e, "transaction get size", rs); - printf("size %d: %ld\n", i, sizes[i]); + printf("size %d: %" PRId64 "\n", i, sizes[i]); i++; fdb_transaction_clear_range(tr, keys[i], KEY_SIZE, keys[i + 1], KEY_SIZE); e = getSize(rs, tr, sizes + i); checkError(e, "transaction get size", rs); - printf("size %d: %ld\n", i, sizes[i]); + printf("size %d: %" PRId64 "\n", i, sizes[i]); i++; for (j = 0; j + 1 < i; j++) { diff --git a/bindings/c/test/unit/third_party/CMakeLists.txt b/bindings/c/test/unit/third_party/CMakeLists.txt index 6229abb0c9..014157553c 100644 --- a/bindings/c/test/unit/third_party/CMakeLists.txt +++ b/bindings/c/test/unit/third_party/CMakeLists.txt @@ -6,7 +6,7 @@ ExternalProject_Add( doctest PREFIX ${CMAKE_BINARY_DIR}/doctest GIT_REPOSITORY https://github.com/onqtam/doctest.git - GIT_TAG 1c8da00c978c19e00a434b2b1f854fcffc9fba35 # v2.4.0 + GIT_TAG 8424be522357e68d8c6178375546bb0cf9d5f6b3 # v2.4.1 TIMEOUT 10 CONFIGURE_COMMAND "" BUILD_COMMAND "" diff --git a/bindings/c/test/unit/unit_tests.cpp b/bindings/c/test/unit/unit_tests.cpp index 58a44de098..420a16dc3c 100644 --- a/bindings/c/test/unit/unit_tests.cpp +++ b/bindings/c/test/unit/unit_tests.cpp @@ -2358,6 +2358,19 @@ TEST_CASE("commit_does_not_reset") { } } +TEST_CASE("Fast alloc thread cleanup") { + // Try to cause an OOM if thread cleanup doesn't work + for (int i = 0; i < 50000; ++i) { + auto thread = std::thread([]() { + fdb::Transaction tr(db); + for (int s = 0; s < 11; ++s) { + tr.set(key("foo"), std::string(8 << s, '\x00')); + } + }); + thread.join(); + } +} + int main(int argc, char** argv) { if (argc < 3) { std::cout << "Unit tests for the FoundationDB C API.\n" diff --git a/bindings/flow/CMakeLists.txt b/bindings/flow/CMakeLists.txt index 29d5d8fcb9..34dd13e051 100644 --- a/bindings/flow/CMakeLists.txt +++ b/bindings/flow/CMakeLists.txt @@ -18,6 +18,7 @@ set(SRCS add_flow_target(STATIC_LIBRARY NAME fdb_flow SRCS ${SRCS}) target_link_libraries(fdb_flow PUBLIC fdb_c) +target_link_libraries(fdb_flow PUBLIC fdbclient) target_include_directories(fdb_flow PUBLIC "${CMAKE_CURRENT_BINARY_DIR}" "${CMAKE_CURRENT_SOURCE_DIR}" diff --git a/bindings/go/src/fdb/generated.go b/bindings/go/src/fdb/generated.go index 2acd3c4a21..8be49f5c79 100644 --- a/bindings/go/src/fdb/generated.go +++ b/bindings/go/src/fdb/generated.go @@ -598,6 +598,11 @@ func (o TransactionOptions) SetBypassUnreadable() error { return o.setOpt(1100, nil) } +// Allows this transaction to use cached GRV from the database context. Defaults to off. Upon first usage, starts a background updater to periodically update the cache to avoid stale read versions. +func (o TransactionOptions) SetUseGrvCache() error { + return o.setOpt(1101, nil) +} + type StreamingMode int const ( diff --git a/bindings/java/CMakeLists.txt b/bindings/java/CMakeLists.txt index 40727063f6..9dfb349040 100644 --- a/bindings/java/CMakeLists.txt +++ b/bindings/java/CMakeLists.txt @@ -154,6 +154,7 @@ endif() set_target_properties(java_workloads PROPERTIES LIBRARY_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}/share/foundationdb") target_link_libraries(java_workloads PUBLIC fdb_c ${JNI_LIBRARIES}) +target_link_libraries(java_workloads PRIVATE flow) # mostly for boost target_include_directories(java_workloads PUBLIC ${JNI_INCLUDE_DIRS}) set(CMAKE_JAVA_COMPILE_FLAGS "-source" "1.8" "-target" "1.8" "-XDignore.symbol.file") @@ -228,6 +229,8 @@ if(NOT OPEN_FOR_IDE) else() if(CMAKE_SYSTEM_PROCESSOR MATCHES "aarch64") set(lib_destination "linux/aarch64") + elseif(CMAKE_SYSTEM_PROCESSOR MATCHES "ppc64le") + set(lib_destination "linux/ppc64le") else() set(lib_destination "linux/amd64") endif() diff --git a/bindings/java/src/main/com/apple/foundationdb/JNIUtil.java b/bindings/java/src/main/com/apple/foundationdb/JNIUtil.java index a3811bd029..cf4e0ade06 100644 --- a/bindings/java/src/main/com/apple/foundationdb/JNIUtil.java +++ b/bindings/java/src/main/com/apple/foundationdb/JNIUtil.java @@ -182,7 +182,7 @@ public class JNIUtil { private static OS getRunningOS() { String osname = System.getProperty("os.name").toLowerCase(); String arch = System.getProperty("os.arch"); - if (!arch.equals("amd64") && !arch.equals("x86_64") && !arch.equals("aarch64")) { + if (!arch.equals("amd64") && !arch.equals("x86_64") && !arch.equals("aarch64") && !arch.equals("ppc64le")) { throw new IllegalStateException("Unknown or unsupported arch: " + arch); } if (osname.startsWith("windows")) { diff --git a/cmake/ConfigureCompiler.cmake b/cmake/ConfigureCompiler.cmake index e2856ad0fd..f05ca279be 100644 --- a/cmake/ConfigureCompiler.cmake +++ b/cmake/ConfigureCompiler.cmake @@ -219,7 +219,7 @@ else() endif() if(STATIC_LINK_LIBCXX) if (NOT USE_LIBCXX AND NOT APPLE) - add_link_options(-static-libstdc++ -static-libgcc) + add_link_options(-static-libstdc++ -static-libgcc) endif() endif() # # Instruction sets we require to be supported by the CPU @@ -309,7 +309,7 @@ else() if (PROFILE_INSTR_GENERATE) message(FATAL_ERROR "Can't set both PROFILE_INSTR_GENERATE and PROFILE_INSTR_USE") endif() - add_compile_options(-Wno-error=profile-instr-out-of-date) + add_compile_options(-Wno-error=profile-instr-out-of-date -Wno-error=profile-instr-unprofiled) add_compile_options(-fprofile-instr-use=${PROFILE_INSTR_USE}) add_link_options(-fprofile-instr-use=${PROFILE_INSTR_USE}) endif() @@ -349,6 +349,9 @@ else() add_compile_options(-march=armv8.2-a+crc+simd) endif() + if (CMAKE_SYSTEM_PROCESSOR MATCHES "ppc64le") + add_compile_options(-m64 -mcpu=power9 -mtune=power9 -DNO_WARN_X86_INTRINSICS) + endif() # Check whether we can use dtrace probes include(CheckSymbolExists) check_symbol_exists(DTRACE_PROBE sys/sdt.h SUPPORT_DTRACE) diff --git a/cmake/InstallLayout.cmake b/cmake/InstallLayout.cmake index 61af9df658..8da0fce34f 100644 --- a/cmake/InstallLayout.cmake +++ b/cmake/InstallLayout.cmake @@ -303,7 +303,9 @@ set(CPACK_RPM_EXCLUDE_FROM_AUTO_FILELIST_ADDITION "/etc/rc.d/init.d" "/usr/lib/pkgconfig" "/usr/lib/foundationdb" - "/usr/lib/cmake") + "/usr/lib/cmake" + "/usr/lib/foundationdb-${FDB_VERSION}/etc/foundationdb" + ) set(CPACK_RPM_DEBUGINFO_PACKAGE ${GENERATE_DEBUG_PACKAGES}) #set(CPACK_RPM_BUILD_SOURCE_FDB_INSTALL_DIRS_PREFIX /usr/src) set(CPACK_RPM_COMPONENT_INSTALL ON) diff --git a/contrib/CMakeLists.txt b/contrib/CMakeLists.txt index 63ed617212..47705943be 100644 --- a/contrib/CMakeLists.txt +++ b/contrib/CMakeLists.txt @@ -1,5 +1,6 @@ add_subdirectory(fmt-8.0.1) if(NOT WIN32) + add_subdirectory(debug_determinism) add_subdirectory(monitoring) add_subdirectory(TraceLogHelper) add_subdirectory(TestHarness) diff --git a/contrib/debug_determinism/CMakeLists.txt b/contrib/debug_determinism/CMakeLists.txt new file mode 100644 index 0000000000..0aefcfc59f --- /dev/null +++ b/contrib/debug_determinism/CMakeLists.txt @@ -0,0 +1,5 @@ +add_library(debug_determinism STATIC debug_determinism.cpp) + +# So that we can link to libfdb_c.so. Not strictly necessary but convenient for use with our +# TRACE_PC_GUARD_INSTRUMENTATION_LIB cmake option +target_compile_options(debug_determinism PRIVATE -fPIC) \ No newline at end of file diff --git a/contrib/debug_determinism/README.md b/contrib/debug_determinism/README.md new file mode 100644 index 0000000000..3b327394dd --- /dev/null +++ b/contrib/debug_determinism/README.md @@ -0,0 +1,45 @@ +Utilities for debugging unseed mismatches for foundationdb simulation tests. + +99/100 times the source of the nondeterminism is use of uninitialized memory and +what you want to do is build with `-DUSE_VALGRIND=ON` and run simulations under +valgrind. + +Common sources of nondeterminism and specialized tools to find them. +1. Use of uninitialized memory (use valgrind!) +1. Memory errors (use valgrind and/or asan) +1. Undefined behavior (use ubsan. You can also try _GLIBCXX_DEBUG) + +If it's not any of these then now it's time to try this technique. Look for + +1. Call to some kind of "get current time" function that's not in `INetwork` +1. Depending on the relative ordering of allocated memory. E.g. Using heap-allocated pointers as keys in a `std::map`. +1. Inspecting something about the current state of the system (e.g. free disk space) +1. Depending on iteration order of an unordered map + +# Quickstart + +Set these cmake flags + +``` +-DTRACE_PC_GUARD_INSTRUMENTATION_LIB=$BUILDDIR/lib/libdebug_determinism.a +``` + +and change `#define DEBUG_DETERMINISM 0` to `#define DEBUG_DETERMINISM 1` in +flow/Platform.h. This disables several known sources of nondeterminism that +don't affect unseeds. + +For reasons I don't fully understand, it appears that sqlite exhibits some +nondeterminism if you don't add `#define SQLITE_OMIT_LOOKASIDE` to the top of +fdbserver/sqlite/sqlite3.amalgamation.c, so you probably want to do that too. + +Now when you run an fdbserver simulation, it will write a file `out.bin` in the +current directory which contains the sequence of edges in the control flow graph +that were encountered during the simulation. If you rename `out.bin` to `in.bin` +and then re-run, the simulation will validate that the sequence of edges is the +same as the last run. If it's not, then the simulation will enter an infinite +loop at the first difference and print a message. Then you probably want to +attach gdb to the process and investigate from there. + +You'll need to make sure you delete the `simfdb` folder before each run, because +otherwise you'll take a different codepath for deleting the `simfdb` folder at +the beginning of simulation. diff --git a/contrib/debug_determinism/debug_determinism.cpp b/contrib/debug_determinism/debug_determinism.cpp new file mode 100644 index 0000000000..02982a8cbb --- /dev/null +++ b/contrib/debug_determinism/debug_determinism.cpp @@ -0,0 +1,52 @@ +#include +#include + +namespace { +FILE* out = nullptr; +FILE* in = nullptr; +void loop_forever() { + // Try to convince the optimizer not to optimize away this loop + static volatile uint64_t x = 0; + for (;;) { + ++x; + } +} +} // namespace + +// This callback is inserted by the compiler as a module constructor +// into every DSO. 'start' and 'stop' correspond to the +// beginning and end of the section with the guards for the entire +// binary (executable or DSO). The callback will be called at least +// once per DSO and may be called multiple times with the same parameters. +extern "C" void __sanitizer_cov_trace_pc_guard_init(uint32_t* start, uint32_t* stop) { + in = fopen("in.bin", "r"); + out = fopen("out.bin", "w"); + static uint64_t N; // Counter for the guards. + if (start == stop || *start) + return; // Initialize only once. + for (uint32_t* x = start; x < stop; x++) { + *x = ++N; // Guards should start from 1. + } +} + +// This callback is inserted by the compiler on every edge in the +// control flow (some optimizations apply). +// Typically, the compiler will emit the code like this: +// if(*guard) +// __sanitizer_cov_trace_pc_guard(guard); +// But for large functions it will emit a simple call: +// __sanitizer_cov_trace_pc_guard(guard); +extern "C" void __sanitizer_cov_trace_pc_guard(uint32_t* guard) { + if (!guard) { + return; + } + fwrite(guard, 1, sizeof(*guard), out); + if (in) { + uint32_t theirs; + fread(&theirs, 1, sizeof(theirs), in); + if (*guard != theirs) { + printf("Non-determinism detected\n"); + loop_forever(); + } + } +} \ No newline at end of file diff --git a/contrib/generate_profile.sh b/contrib/generate_profile.sh new file mode 100755 index 0000000000..9b6a446bd2 --- /dev/null +++ b/contrib/generate_profile.sh @@ -0,0 +1,37 @@ +#!/bin/bash +if [ $# -eq 0 ] || [ $# -gt 2 ] + then + echo "Usage: generate_profile.sh Path_Of_Foundation_Build_Directory Storage_Engine" + exit 1 +fi +fdbdir=$1 +storage_engine='ssd' +if [ $# -eq 2 ] + then + storage_engine=$2 +fi + +export LD_LIBRARY_PATH=$fdbdir/lib:$LD_LIBRARY_PATH +export FDB_CLUSTER_FILE=$fdbdir/fdb.cluster +export LLVM_PROFILE_FILE=$fdbdir/sandbox/fdb-%p.profraw +$fdbdir/bin/fdbmonitor --conffile $fdbdir/sandbox/foundationdb.conf --lockfile $fdbdir/sandbox/fdbmonitor.pid & +# This profile will be ignored +export LLVM_PROFILE_FILE=$fdbdir/sandbox/cli-%m.profraw +$fdbdir/bin/fdbcli -C $fdbdir/fdb.cluster --exec "configure new $storage_engine single" +export LLVM_PROFILE_FILE=$fdbdir/sandbox/mako-build-%m.profraw +$fdbdir/bin/mako -p 64 -t 1 --keylen 32 --vallen 16 --mode build --rows 10000 --trace --trace_format json +export LLVM_PROFILE_FILE=$fdbdir/sandbox/mako-run-%m.profraw +$fdbdir/bin/mako -p 1 -t 2 --keylen 32 --vallen 16 --mode run --rows 10000 --transaction grvg7i2gr1:48cr1:48 --seconds 60 --trace $fdbdir/sandbox/logs --trace_format json + +# Shutdown fdbserver to trigger profile dumping +fdbmonitor_pid=$(cat $fdbdir/sandbox/fdbmonitor.pid) +fdbserver_pid=$(cat /proc/$fdbmonitor_pid/task/$fdbmonitor_pid/children) +gdb --batch --eval-command 'call (void)exit(0)' --pid $fdbserver_pid + +# Clean up +kill -9 $fdbmonitor_pid + +# Profile for server +llvm-profdata merge -output=$fdbdir/fdb.profdata $fdbdir/sandbox/fdb-*.profraw +# Profile for client +llvm-profdata merge -output=$fdbdir/mako.profdata $fdbdir/sandbox/mako-*.profraw diff --git a/contrib/pkg_tester/test_fdb_pkgs.py b/contrib/pkg_tester/test_fdb_pkgs.py index 3b6c4855e1..db8a1ae27b 100644 --- a/contrib/pkg_tester/test_fdb_pkgs.py +++ b/contrib/pkg_tester/test_fdb_pkgs.py @@ -147,7 +147,7 @@ def centos_image_with_fdb_helper(versioned: bool) -> Iterator[Optional[Image]]: container = None image = None try: - container = Container("centos", initd=True) + container = Container("centos:7", initd=True) for rpm in rpms: container.copy_to(rpm, "/opt") container.run(["bash", "-c", "yum update -y"]) @@ -237,10 +237,6 @@ def test_write(linux_container: Container, snapshot): assert snapshot == linux_container.run(["fdbcli", "--exec", "get x"]) -def test_fdbcli_help_text(linux_container: Container, snapshot): - assert snapshot == linux_container.run(["fdbcli", "--help"]) - - def test_execstack_permissions_libfdb_c(linux_container: Container, snapshot): linux_container.run(["ldconfig"]) assert snapshot == linux_container.run( diff --git a/documentation/sphinx/source/release-notes/release-notes-630.rst b/documentation/sphinx/source/release-notes/release-notes-630.rst index 9deaf9ec27..0876faa1fc 100644 --- a/documentation/sphinx/source/release-notes/release-notes-630.rst +++ b/documentation/sphinx/source/release-notes/release-notes-630.rst @@ -2,8 +2,16 @@ Release Notes ############# +6.3.24 +====== +* Fixed a bug where get key location can overload proxies. `(PR #6453) `_ +* Added a mechanism that can reduce the number of empty peek reply by not always returning empty peek reply immediately. `(PR #6413) `_ +* Enable TLS support for Windows. `(PR #6193) `_ +* Fixed a bug where a shard gets merged too soon. `(PR #6115) `_ + 6.3.23 ====== +* Add AWS v4 header support for backup. `(PR #6025) `_ * Fixed a bug that remoteDCIsHealthy logic is not guarded by CC_ENABLE_WORKER_HEALTH_MONITOR, which may prevent HA failback. `(PR #6106) `_ * Fixed a race condition with updating the coordinated state and updating the master registration. `(PR #6088) `_ * Changed dbinfo broadcast to be explicitly requested by the worker registration message. `(PR #6073) `_ diff --git a/documentation/tutorial/tutorial.actor.cpp b/documentation/tutorial/tutorial.actor.cpp index 326f0246f2..8fdb32b0c2 100644 --- a/documentation/tutorial/tutorial.actor.cpp +++ b/documentation/tutorial/tutorial.actor.cpp @@ -19,6 +19,7 @@ * limitations under the License. */ +#include "contrib/fmt-8.0.1/include/fmt/format.h" #include "flow/flow.h" #include "flow/Platform.h" #include "flow/DeterministicRandom.h" @@ -413,7 +414,7 @@ ACTOR Future logThroughput(int64_t* v, Key* next) { loop { state int64_t last = *v; wait(delay(1)); - printf("throughput: %ld bytes/s, next: %s\n", *v - last, printable(*next).c_str()); + fmt::print("throughput: {} bytes/s, next: {}\n", *v - last, printable(*next).c_str()); } } diff --git a/fdbbackup/backup.actor.cpp b/fdbbackup/backup.actor.cpp index 38a43eef08..e3cdc01ce7 100644 --- a/fdbbackup/backup.actor.cpp +++ b/fdbbackup/backup.actor.cpp @@ -1690,7 +1690,7 @@ ACTOR Future cleanupStatus(Reference tr, readMore = true; } catch (Error& e) { // If doc can't be parsed or isn't alive, delete it. - TraceEvent(SevWarn, "RemovedDeadBackupLayerStatus").detail("Key", docs[i].key).error(e, true); + TraceEvent(SevWarn, "RemovedDeadBackupLayerStatus").errorUnsuppressed(e).detail("Key", docs[i].key); tr->clear(docs[i].key); // If limit is 1 then read more. if (limit == 1) @@ -2754,7 +2754,7 @@ ACTOR Future queryBackup(const char* name, reportBackupQueryError(operationId, result, errorMessage = - format("the specified restorable version %ld is not valid", restoreVersion)); + format("the specified restorable version %lld is not valid", restoreVersion)); return Void(); } Optional fileSet = wait(bc->getRestoreSet(restoreVersion, keyRangesFilter)); @@ -3081,7 +3081,7 @@ static void addKeyRange(std::string optionValue, Standalone advanceVersionCommandActor(Reference db, std::vect } else { state Version v; int n = 0; - if (sscanf(tokens[1].toString().c_str(), "%ld%n", &v, &n) != 1 || n != tokens[1].size()) { + if (sscanf(tokens[1].toString().c_str(), "%" PRId64 "%n", &v, &n) != 1 || n != tokens[1].size()) { printUsage(tokens[0]); return false; } else { @@ -53,7 +53,7 @@ ACTOR Future advanceVersionCommandActor(Reference db, std::vect tr->set(advanceVersionSpecialKey, boost::lexical_cast(v)); wait(safeThreadFutureToFuture(tr->commit())); } else { - printf("Current read version is %ld\n", rv); + fmt::print("Current read version is {}\n", rv); return true; } } catch (Error& e) { diff --git a/fdbcli/ChangeFeedCommand.actor.cpp b/fdbcli/ChangeFeedCommand.actor.cpp index e08eadb37e..94415be921 100644 --- a/fdbcli/ChangeFeedCommand.actor.cpp +++ b/fdbcli/ChangeFeedCommand.actor.cpp @@ -115,7 +115,7 @@ ACTOR Future changeFeedCommandActor(Database localDb, std::vector::max(); if (tokens.size() > 3) { int n = 0; - if (sscanf(tokens[3].toString().c_str(), "%ld%n", &begin, &n) != 1 || n != tokens[3].size()) { + if (sscanf(tokens[3].toString().c_str(), "%" PRId64 "%n", &begin, &n) != 1 || n != tokens[3].size()) { printUsage(tokens[0]); return false; } @@ -168,7 +168,7 @@ ACTOR Future changeFeedCommandActor(Database localDb, std::vector changeCoordinators(Reference db, std::vectorstartsWith(nameTokenBegin)) { new_cluster_description = tok->substr(nameTokenBegin.size()); std::copy(tok + 1, tokens.end(), tok); tokens.resize(tokens.size() - 1); break; } + } state bool automatic = tokens.size() == 2 && tokens[1] == LiteralStringRef("auto"); state Reference tr = db->createTransaction(); @@ -96,18 +97,32 @@ ACTOR Future changeCoordinators(Reference db, std::vectorset(fdb_cli::coordinatorsProcessSpecialKey, auto_coordinators_str); } else if (tokens.size() > 1) { state std::set new_coordinators_addresses; - state std::vector newAddresslist; + state std::set new_coordinators_hostnames; + state std::vector newCoordinatorslist; state std::vector::iterator t; for (t = tokens.begin() + 1; t != tokens.end(); ++t) { try { - // TODO(renxuan): add hostname parsing here. - auto const& addr = NetworkAddress::parse(t->toString()); - if (new_coordinators_addresses.count(addr)) { - fprintf(stderr, "ERROR: passed redundant coordinators: `%s'\n", addr.toString().c_str()); - return true; + if (Hostname::isHostname(t->toString())) { + // We do not resolve hostnames here. We commit them as is. + const auto& hostname = Hostname::parse(t->toString()); + if (new_coordinators_hostnames.count(hostname)) { + fprintf(stderr, + "ERROR: passed redundant coordinators: `%s'\n", + hostname.toString().c_str()); + return true; + } + new_coordinators_hostnames.insert(hostname); + newCoordinatorslist.push_back(hostname.toString()); + } else { + const auto& addr = NetworkAddress::parse(t->toString()); + if (new_coordinators_addresses.count(addr)) { + fprintf( + stderr, "ERROR: passed redundant coordinators: `%s'\n", addr.toString().c_str()); + return true; + } + new_coordinators_addresses.insert(addr); + newCoordinatorslist.push_back(addr.toString()); } - new_coordinators_addresses.insert(addr); - newAddresslist.push_back(addr.toString()); } catch (Error& e) { if (e.code() == error_code_connection_string_invalid) { fprintf( @@ -117,12 +132,12 @@ ACTOR Future changeCoordinators(Reference db, std::vectorset(fdb_cli::coordinatorsProcessSpecialKey, new_addresses_str); + std::string new_coordinators_str = boost::algorithm::join(newCoordinatorslist, ", "); + tr->set(fdb_cli::coordinatorsProcessSpecialKey, new_coordinators_str); } wait(safeThreadFutureToFuture(tr->commit())); // commit should always fail here - // if coordinators are changed, we should get commit_unknown() error + // If the commit succeeds, the coordinators change and the commit will fail with commit_unknown_result(). ASSERT(false); } catch (Error& e) { state Error err(e); diff --git a/fdbcli/IncludeCommand.actor.cpp b/fdbcli/IncludeCommand.actor.cpp index fda54f6fcb..2a72a214e2 100644 --- a/fdbcli/IncludeCommand.actor.cpp +++ b/fdbcli/IncludeCommand.actor.cpp @@ -59,7 +59,7 @@ ACTOR Future includeLocalities(Reference db, wait(safeThreadFutureToFuture(tr->commit())); return Void(); } catch (Error& e) { - TraceEvent("IncludeLocalitiesError").error(e, true); + TraceEvent("IncludeLocalitiesError").errorUnsuppressed(e); wait(safeThreadFutureToFuture(tr->onError(e))); } } @@ -99,7 +99,7 @@ ACTOR Future includeServers(Reference db, std::vectorcommit())); return Void(); } catch (Error& e) { - TraceEvent("IncludeServersError").error(e, true); + TraceEvent("IncludeServersError").errorUnsuppressed(e); wait(safeThreadFutureToFuture(tr->onError(e))); } } diff --git a/fdbcli/StatusCommand.actor.cpp b/fdbcli/StatusCommand.actor.cpp index be30c1d47e..d1a772e0a0 100644 --- a/fdbcli/StatusCommand.actor.cpp +++ b/fdbcli/StatusCommand.actor.cpp @@ -705,12 +705,12 @@ void printStatus(StatusObjectReader statusObj, } } outputString += format( - " %s log epoch: %ld begin: %ld end: %s, missing " + " %s log epoch: %lld begin: %lld end: %s, missing " "log interfaces(id,address): %s\n", current ? "Current" : "Old", epoch, beginVersion, - endVersion == invalidVersion ? "(unknown)" : format("%ld", endVersion).c_str(), + endVersion == invalidVersion ? "(unknown)" : format("%lld", endVersion).c_str(), missing_log_interfaces.c_str()); } } @@ -810,6 +810,28 @@ void printStatus(StatusObjectReader statusObj, outputString = outputStringCache; outputString += "\n Unable to retrieve data status"; } + // Storage Wiggle section + StatusObjectReader storageWigglerObj; + std::string storageWigglerString; + try { + if (statusObjCluster.get("storage_wiggler", storageWigglerObj)) { + int size = 0; + if (storageWigglerObj.has("wiggle_server_addresses")) { + storageWigglerString += "\n Wiggle server addresses-"; + for (auto& v : storageWigglerObj.obj().at("wiggle_server_addresses").get_array()) { + storageWigglerString += " " + v.get_str(); + size += 1; + } + } + storageWigglerString += "\n Wiggle server count - " + std::to_string(size); + } + } catch (std::runtime_error&) { + storageWigglerString += "\n Unable to retrieve storage wiggler status"; + } + if (storageWigglerString.size()) { + outputString += "\n\nStorage wiggle:"; + outputString += storageWigglerString; + } // Operating space section outputString += "\n\nOperating space:"; diff --git a/fdbcli/fdbcli.actor.cpp b/fdbcli/fdbcli.actor.cpp index c94f912845..f7720f9efb 100644 --- a/fdbcli/fdbcli.actor.cpp +++ b/fdbcli/fdbcli.actor.cpp @@ -1014,9 +1014,9 @@ struct CLIOptions { } else { fprintf(stderr, "ERROR: Failed to set knob option '%s': %s\n", knobName.c_str(), e.what()); TraceEvent(SevError, "FailedToSetKnob") + .error(e) .detail("Knob", printable(knobName)) - .detail("Value", printable(knobValueString)) - .error(e); + .detail("Value", printable(knobValueString)); exit_code = FDB_EXIT_ERROR; } } @@ -1615,7 +1615,7 @@ ACTOR Future cli(CLIOptions opt, LineNoise* plinenoise) { } else { Version v = wait(makeInterruptable( safeThreadFutureToFuture(getTransaction(db, tr, options, intrans)->getReadVersion()))); - printf("%ld\n", v); + fmt::print("{}\n", v); } continue; } diff --git a/fdbclient/BackupContainer.actor.cpp b/fdbclient/BackupContainer.actor.cpp index a5219ab524..bfef037be5 100644 --- a/fdbclient/BackupContainer.actor.cpp +++ b/fdbclient/BackupContainer.actor.cpp @@ -305,9 +305,9 @@ Reference IBackupContainer::openContainer(const std::string& u throw; TraceEvent m(SevWarn, "BackupContainer"); + m.error(e); m.detail("Description", "Invalid container specification. See help."); m.detail("URL", url); - m.error(e); if (e.code() == error_code_backup_invalid_url) m.detail("LastOpenError", lastOpenError); @@ -360,10 +360,9 @@ ACTOR Future> listContainers_impl(std::string baseURL) throw; TraceEvent m(SevWarn, "BackupContainer"); - + m.error(e); m.detail("Description", "Invalid backup container URL prefix. See help."); m.detail("URL", baseURL); - m.error(e); if (e.code() == error_code_backup_invalid_url) m.detail("LastOpenError", IBackupContainer::lastOpenError); diff --git a/fdbclient/BackupContainerFileSystem.actor.cpp b/fdbclient/BackupContainerFileSystem.actor.cpp index 676ae8165e..22c5087bc0 100644 --- a/fdbclient/BackupContainerFileSystem.actor.cpp +++ b/fdbclient/BackupContainerFileSystem.actor.cpp @@ -1149,8 +1149,8 @@ public: keyFile = _keyFile; } catch (Error& e) { TraceEvent(SevWarnAlways, "FailedToOpenEncryptionKeyFile") - .detail("FileName", encryptionKeyFileName) - .error(e); + .error(e) + .detail("FileName", encryptionKeyFileName); throw e; } int bytesRead = wait(keyFile->read(cipherKey->data(), cipherKey->size(), 0)); @@ -1377,8 +1377,8 @@ ACTOR static Future getSnapshotFileKeyRange_impl(Reference BackupContainerFileSystem::openContainerFS( throw; TraceEvent m(SevWarn, "BackupContainer"); + m.error(e); m.detail("Description", "Invalid container specification. See help."); m.detail("URL", url); - m.error(e); if (e.code() == error_code_backup_invalid_url) m.detail("LastOpenError", lastOpenError); diff --git a/fdbclient/ClientKnobs.cpp b/fdbclient/ClientKnobs.cpp index 9bd3e8568f..f98f0faa02 100644 --- a/fdbclient/ClientKnobs.cpp +++ b/fdbclient/ClientKnobs.cpp @@ -87,6 +87,8 @@ void ClientKnobs::initialize(Randomize randomize) { init( LOCATION_CACHE_EVICTION_SIZE, 600000 ); init( LOCATION_CACHE_EVICTION_SIZE_SIM, 10 ); if( randomize && BUGGIFY ) LOCATION_CACHE_EVICTION_SIZE_SIM = 3; + init( LOCATION_CACHE_ENDPOINT_FAILURE_GRACE_PERIOD, 60 ); + init( LOCATION_CACHE_FAILED_ENDPOINT_RETRY_INTERVAL, 60 ); init( GET_RANGE_SHARD_LIMIT, 2 ); init( WARM_RANGE_SHARD_LIMIT, 100 ); @@ -118,6 +120,12 @@ void ClientKnobs::initialize(Randomize randomize) { init( CORE_VERSIONSPERSECOND, 1e6 ); init( LOG_RANGE_BLOCK_SIZE, CORE_VERSIONSPERSECOND ); init( MUTATION_BLOCK_SIZE, 10000); + init( MAX_VERSION_CACHE_LAG, 0.1 ); + init( MAX_PROXY_CONTACT_LAG, 0.2 ); + init( DEBUG_USE_GRV_CACHE_CHANCE, -1.0 ); // For 100% chance at 1.0, this means 0.0 is not 0%. We don't want the default to be 0. + init( FORCE_GRV_CACHE_OFF, false ); + init( GRV_CACHE_RK_COOLDOWN, 60.0 ); + init( GRV_SUSTAINED_THROTTLING_THRESHOLD, 0.1 ); // TaskBucket init( TASKBUCKET_LOGGING_DELAY, 5.0 ); diff --git a/fdbclient/ClientKnobs.h b/fdbclient/ClientKnobs.h index 36bff8bfe1..61c2dcd9d1 100644 --- a/fdbclient/ClientKnobs.h +++ b/fdbclient/ClientKnobs.h @@ -87,6 +87,8 @@ public: // When locationCache in DatabaseContext gets to be this size, items will be evicted int LOCATION_CACHE_EVICTION_SIZE; int LOCATION_CACHE_EVICTION_SIZE_SIM; + double LOCATION_CACHE_ENDPOINT_FAILURE_GRACE_PERIOD; + double LOCATION_CACHE_FAILED_ENDPOINT_RETRY_INTERVAL; int GET_RANGE_SHARD_LIMIT; int WARM_RANGE_SHARD_LIMIT; @@ -121,6 +123,13 @@ public: int64_t CORE_VERSIONSPERSECOND; // This is defined within the server but used for knobs based on server value int LOG_RANGE_BLOCK_SIZE; int MUTATION_BLOCK_SIZE; + double MAX_VERSION_CACHE_LAG; // The upper bound in seconds for OK amount of staleness when using a cached RV + double MAX_PROXY_CONTACT_LAG; // The upper bound in seconds for how often we want a response from the GRV proxies + double DEBUG_USE_GRV_CACHE_CHANCE; // Debug setting to change the chance for a regular GRV request to use the cache + bool FORCE_GRV_CACHE_OFF; // Panic button to turn off cache. Holds priority over other options. + double GRV_CACHE_RK_COOLDOWN; // Required number of seconds to pass after throttling to re-allow cache use + double GRV_SUSTAINED_THROTTLING_THRESHOLD; // If ALL GRV requests have been throttled in the last number of seconds + // specified here, ratekeeper is throttling and not a false positive // Taskbucket double TASKBUCKET_LOGGING_DELAY; diff --git a/fdbclient/CommitProxyInterface.h b/fdbclient/CommitProxyInterface.h index ad71dc14d7..80a7871532 100644 --- a/fdbclient/CommitProxyInterface.h +++ b/fdbclient/CommitProxyInterface.h @@ -196,6 +196,8 @@ struct GetReadVersionReply : public BasicLoadBalancedReply { bool locked; Optional metadataVersion; int64_t midShardSize = 0; + bool rkDefaultThrottled = false; + bool rkBatchThrottled = false; TransactionTagMap tagThrottleInfo; @@ -213,6 +215,8 @@ struct GetReadVersionReply : public BasicLoadBalancedReply { metadataVersion, tagThrottleInfo, midShardSize, + rkDefaultThrottled, + rkBatchThrottled, ssVersionVectorDelta, proxyId); } diff --git a/fdbclient/ConfigKnobs.cpp b/fdbclient/ConfigKnobs.cpp index 208fb8ff62..9318c949a3 100644 --- a/fdbclient/ConfigKnobs.cpp +++ b/fdbclient/ConfigKnobs.cpp @@ -27,7 +27,7 @@ ConfigKey ConfigKeyRef::decodeKey(KeyRef const& key) { try { tuple = Tuple::unpack(key); } catch (Error& e) { - TraceEvent(SevWarnAlways, "FailedToUnpackConfigKey").detail("Key", printable(key)).error(e); + TraceEvent(SevWarnAlways, "FailedToUnpackConfigKey").error(e).detail("Key", printable(key)); throw invalid_config_db_key(); } if (tuple.size() != 2) { @@ -96,7 +96,7 @@ public: struct ToStringFunc { std::string operator()(int v) const { return format("int:%d", v); } - std::string operator()(int64_t v) const { return format("int64_t:%ld", v); } + std::string operator()(int64_t v) const { return format("int64_t:%lld", v); } std::string operator()(bool v) const { return format("bool:%d", v); } std::string operator()(ValueRef v) const { return "string:" + v.toString(); } std::string operator()(double v) const { return format("double:%lf", v); } diff --git a/fdbclient/CoordinationInterface.h b/fdbclient/CoordinationInterface.h index 9873817a88..bd37fb3f55 100644 --- a/fdbclient/CoordinationInterface.h +++ b/fdbclient/CoordinationInterface.h @@ -93,7 +93,11 @@ public: // This one should only be used when resolving asynchronously is impossible. For all other cases, resolveHostnames() // should be preferred. void resolveHostnamesBlocking(); + // This function derives the member connectionString from the current key, coordinators and hostnames. + void resetConnectionString(); + void resetToUnresolved(); + void parseKey(const std::string& key); ConnectionStringStatus status = RESOLVED; AsyncTrigger resolveFinish; @@ -103,7 +107,6 @@ public: private: void parseConnString(); - void parseKey(const std::string& key); Key key, keyDesc; std::string connectionString; }; diff --git a/fdbclient/DatabaseBackupAgent.actor.cpp b/fdbclient/DatabaseBackupAgent.actor.cpp index fe072e29fa..7e41b58eb6 100644 --- a/fdbclient/DatabaseBackupAgent.actor.cpp +++ b/fdbclient/DatabaseBackupAgent.actor.cpp @@ -2142,7 +2142,7 @@ struct StartFullBackupTaskFunc : TaskFuncBase { wait(tr->commit()); break; } catch (Error& e) { - TraceEvent("SetDestUidOrBeginVersionError").error(e, true); + TraceEvent("SetDestUidOrBeginVersionError").errorUnsuppressed(e); wait(tr->onError(e)); } } @@ -2907,7 +2907,7 @@ public: TraceEvent("DBA_Abort").detail("CommitVersion", tr->getCommittedVersion()); break; } catch (Error& e) { - TraceEvent("DBA_AbortError").error(e, true); + TraceEvent("DBA_AbortError").errorUnsuppressed(e); wait(tr->onError(e)); } } diff --git a/fdbclient/DatabaseContext.h b/fdbclient/DatabaseContext.h index 8a54ee23f3..e7581ea4c2 100644 --- a/fdbclient/DatabaseContext.h +++ b/fdbclient/DatabaseContext.h @@ -199,6 +199,11 @@ struct ChangeFeedData : ReferenceCounted { ChangeFeedData() : notAtLatest(1) {} }; +struct EndpointFailureInfo { + double startTime = 0; + double lastRefreshTime = 0; +}; + class DatabaseContext : public ReferenceCounted, public FastAllocated, NonCopyable { public: static DatabaseContext* allocateOnForeignThread() { @@ -242,6 +247,14 @@ public: void invalidateCache(const KeyRef&, Reverse isBackward = Reverse::False); void invalidateCache(const KeyRangeRef&); + // Records that `endpoint` is failed on a healthy server. + void setFailedEndpointOnHealthyServer(const Endpoint& endpoint); + + // Updates `endpoint` refresh time if the `endpoint` is a failed endpoint. If not, this does nothing. + void updateFailedEndpointRefreshTime(const Endpoint& endpoint); + Optional getEndpointFailureInfo(const Endpoint& endpoint); + void clearFailedEndpointOnHealthyServer(const Endpoint& endpoint); + bool sampleReadTags() const; bool sampleOnCost(uint64_t cost) const; @@ -352,6 +365,7 @@ public: Future monitorTssInfoChange; Future tssMismatchHandler; PromiseStream>> tssMismatchStream; + Future grvUpdateHandler; Reference commitProxies; Reference grvProxies; bool proxyProvisional; // Provisional commit proxy and grv proxy are used at the same time. @@ -396,6 +410,7 @@ public: // Cache of location information int locationCacheSize; CoalescedKeyRangeMap> locationCache; + std::unordered_map failedEndpointsOnHealthyServersInfo; std::map server_interf; std::map blobWorker_interf; // blob workers don't change endpoints for the same ID @@ -474,6 +489,20 @@ public: int outstandingWatches; int maxOutstandingWatches; + // GRV Cache + // Database-level read version cache storing the most recent successful GRV as well as the time it was requested. + double lastGrvTime; + Version cachedReadVersion; + void updateCachedReadVersion(double t, Version v); + Version getCachedReadVersion(); + double getLastGrvTime(); + double lastRkBatchThrottleTime; + double lastRkDefaultThrottleTime; + // Cached RVs can be updated through commits, and using cached RVs avoids the proxies altogether + // Because our checks for ratekeeper throttling requires communication with the proxies, + // we want to track the last time in order to periodically contact the proxy to check for throttling + double lastProxyRequestTime; + int snapshotRywEnabled; bool transactionTracingSample; diff --git a/fdbclient/FDBTypes.h b/fdbclient/FDBTypes.h index 8c65fa19ae..4875fea79d 100644 --- a/fdbclient/FDBTypes.h +++ b/fdbclient/FDBTypes.h @@ -1219,10 +1219,12 @@ struct ReadBlobGranuleContext { struct StorageMetadataType { constexpr static FileIdentifier file_identifier = 732123; // when the SS is initialized - uint64_t createdTime; // comes from Platform::timer_int() + uint64_t createdTime; // comes from currentTime() StorageMetadataType() : createdTime(0) {} StorageMetadataType(uint64_t t) : createdTime(t) {} + static uint64_t currentTime() { return g_network->timer() * 1e9; } + // To change this serialization, ProtocolVersion::StorageMetadata must be updated, and downgrades need // to be considered template diff --git a/fdbclient/FileBackupAgent.actor.cpp b/fdbclient/FileBackupAgent.actor.cpp index 7546226673..251e0ab9a4 100644 --- a/fdbclient/FileBackupAgent.actor.cpp +++ b/fdbclient/FileBackupAgent.actor.cpp @@ -87,7 +87,7 @@ std::string secondsToTimeFormat(int64_t seconds) { else if (seconds >= 60) return format("%.2f minute(s)", seconds / 60.0); else - return format("%ld second(s)", seconds); + return format("%lld second(s)", seconds); } const Key FileBackupAgent::keyLastRestorable = LiteralStringRef("last_restorable"); @@ -4407,9 +4407,9 @@ public: break; } catch (Error& e) { TraceEvent(numTries > 50 ? SevError : SevInfo, "FastRestoreToolSubmitRestoreRequestsMayFail") + .error(e) .detail("Reason", "DB is not properly locked") - .detail("ExpectedLockID", randomUID) - .error(e); + .detail("ExpectedLockID", randomUID); numTries++; wait(tr->onError(e)); } @@ -4443,8 +4443,8 @@ public: break; } catch (Error& e) { TraceEvent(numTries > 50 ? SevError : SevInfo, "FastRestoreToolSubmitRestoreRequestsRetry") - .detail("RestoreIndex", restoreIndex) - .error(e); + .error(e) + .detail("RestoreIndex", restoreIndex); numTries++; wait(tr->onError(e)); } @@ -5183,7 +5183,7 @@ public: else statusText += "The initial snapshot is still running.\n"; - statusText += format("\nDetails:\n LogBytes written - %ld\n RangeBytes written - %ld\n " + statusText += format("\nDetails:\n LogBytes written - %lld\n RangeBytes written - %lld\n " "Last complete log version and timestamp - %s, %s\n " "Last complete snapshot version and timestamp - %s, %s\n " "Current Snapshot start version and timestamp - %s, %s\n " @@ -5800,9 +5800,9 @@ ACTOR static Future transformDatabaseContents(Database cx, break; } catch (Error& e) { TraceEvent("FastRestoreWorkloadTransformDatabaseContentsGetAllKeys") + .error(e) .detail("Index", i) - .detail("RestoreRange", restoreRanges[i]) - .error(e); + .detail("RestoreRange", restoreRanges[i]); oldData = Standalone>(); // clear the vector wait(tr.onError(e)); } diff --git a/fdbclient/HTTP.actor.cpp b/fdbclient/HTTP.actor.cpp index b34504b641..d10445d38d 100644 --- a/fdbclient/HTTP.actor.cpp +++ b/fdbclient/HTTP.actor.cpp @@ -448,6 +448,7 @@ ACTOR Future> doRequest(Reference conn, err = http_bad_request_id(); TraceEvent(SevError, "HTTPRequestFailedIDMismatch") + .error(err.get()) .detail("DebugID", conn->getDebugID()) .detail("RemoteAddress", conn->getPeerAddress()) .detail("Verb", verb) @@ -456,8 +457,7 @@ ACTOR Future> doRequest(Reference conn, .detail("ResponseCode", r->code) .detail("ResponseContentLen", r->contentLen) .detail("RequestIDSent", requestID) - .detail("RequestIDReceived", responseID) - .error(err.get()); + .detail("RequestIDReceived", responseID); } } @@ -501,7 +501,7 @@ ACTOR Future> doRequest(Reference conn, contentLen, total_sent); } - event.error(e); + event.errorUnsuppressed(e); throw; } } diff --git a/fdbclient/ManagementAPI.actor.cpp b/fdbclient/ManagementAPI.actor.cpp index cf5763b8b5..bd016538a7 100644 --- a/fdbclient/ManagementAPI.actor.cpp +++ b/fdbclient/ManagementAPI.actor.cpp @@ -772,7 +772,7 @@ ACTOR Future> getCoordinators(Database cx) { ACTOR Future> changeQuorumChecker(Transaction* tr, Reference change, - std::vector* desiredCoordinators) { + ClusterConnectionString* conn) { tr->setOption(FDBTransactionOptions::LOCK_AWARE); tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); tr->setOption(FDBTransactionOptions::USE_PROVISIONAL_PROXIES); @@ -783,44 +783,47 @@ ACTOR Future> changeQuorumChecker(Transaction* tr, return CoordinatorsResult::BAD_DATABASE_STATE; // Someone deleted this key entirely? state ClusterConnectionString old(currentKey.get().toString()); + wait(old.resolveHostnames()); if (tr->getDatabase()->getConnectionRecord() && old.clusterKeyName().toString() != tr->getDatabase()->getConnectionRecord()->getConnectionString().clusterKeyName()) return CoordinatorsResult::BAD_DATABASE_STATE; // Someone changed the "name" of the database?? state CoordinatorsResult result = CoordinatorsResult::SUCCESS; - if (!desiredCoordinators->size()) { - std::vector _desiredCoordinators = wait(change->getDesiredCoordinators( + if (!conn->coords.size()) { + std::vector desiredCoordinatorAddresses = wait(change->getDesiredCoordinators( tr, old.coordinators(), Reference(new ClusterConnectionMemoryRecord(old)), result)); - *desiredCoordinators = _desiredCoordinators; + conn->coords = desiredCoordinatorAddresses; } if (result != CoordinatorsResult::SUCCESS) return result; - if (!desiredCoordinators->size()) + if (!conn->coordinators().size()) return CoordinatorsResult::INVALID_NETWORK_ADDRESSES; - std::sort(desiredCoordinators->begin(), desiredCoordinators->end()); + std::sort(conn->coords.begin(), conn->coords.end()); + std::sort(conn->hostnames.begin(), conn->hostnames.end()); std::string newName = change->getDesiredClusterKeyName(); if (newName.empty()) newName = old.clusterKeyName().toString(); - if (old.coordinators() == *desiredCoordinators && old.clusterKeyName() == newName) + if (old.coordinators() == conn->coordinators() && old.clusterKeyName() == newName) return CoordinatorsResult::SAME_NETWORK_ADDRESSES; - state ClusterConnectionString conn(*desiredCoordinators, - StringRef(newName + ':' + deterministicRandom()->randomAlphaNumeric(32))); + std::string key(newName + ':' + deterministicRandom()->randomAlphaNumeric(32)); + conn->parseKey(key); + conn->resetConnectionString(); if (g_network->isSimulated()) { int i = 0; int protectedCount = 0; - while ((protectedCount < ((desiredCoordinators->size() / 2) + 1)) && (i < desiredCoordinators->size())) { - auto process = g_simulator.getProcessByAddress((*desiredCoordinators)[i]); + while ((protectedCount < ((conn->coordinators().size() / 2) + 1)) && (i < conn->coordinators().size())) { + auto process = g_simulator.getProcessByAddress(conn->coordinators()[i]); auto addresses = process->addresses; if (!process->isReliable()) { @@ -832,14 +835,14 @@ ACTOR Future> changeQuorumChecker(Transaction* tr, if (addresses.secondaryAddress.present()) { g_simulator.protectedAddresses.insert(process->addresses.secondaryAddress.get()); } - TraceEvent("ProtectCoordinator").detail("Address", (*desiredCoordinators)[i]).backtrace(); + TraceEvent("ProtectCoordinator").detail("Address", conn->coordinators()[i]).backtrace(); protectedCount++; i++; } } std::vector>> leaderServers; - ClientCoordinators coord(Reference(new ClusterConnectionMemoryRecord(conn))); + ClientCoordinators coord(Reference(new ClusterConnectionMemoryRecord(*conn))); leaderServers.reserve(coord.clientLeaderServers.size()); for (int i = 0; i < coord.clientLeaderServers.size(); i++) @@ -851,7 +854,7 @@ ACTOR Future> changeQuorumChecker(Transaction* tr, when(wait(waitForAll(leaderServers))) {} when(wait(delay(5.0))) { return CoordinatorsResult::COORDINATOR_UNREACHABLE; } } - tr->set(coordinatorsKey, conn.toString()); + tr->set(coordinatorsKey, conn->toString()); return Optional(); } @@ -1273,7 +1276,7 @@ ACTOR Future excludeServers(Database cx, std::vector ser wait(ryw.commit()); return Void(); } catch (Error& e) { - TraceEvent("ExcludeServersError").error(e, true); + TraceEvent("ExcludeServersError").errorUnsuppressed(e); wait(ryw.onError(e)); } } @@ -1285,7 +1288,7 @@ ACTOR Future excludeServers(Database cx, std::vector ser wait(tr.commit()); return Void(); } catch (Error& e) { - TraceEvent("ExcludeServersError").error(e, true); + TraceEvent("ExcludeServersError").errorUnsuppressed(e); wait(tr.onError(e)); } } @@ -1336,7 +1339,7 @@ ACTOR Future excludeLocalities(Database cx, std::unordered_set excludeLocalities(Database cx, std::unordered_set includeServers(Database cx, std::vector ser wait(ryw.commit()); return Void(); } catch (Error& e) { - TraceEvent("IncludeServersError").error(e, true); + TraceEvent("IncludeServersError").errorUnsuppressed(e); wait(ryw.onError(e)); } } @@ -1449,7 +1452,7 @@ ACTOR Future includeServers(Database cx, std::vector ser wait(tr.commit()); return Void(); } catch (Error& e) { - TraceEvent("IncludeServersError").error(e, true); + TraceEvent("IncludeServersError").errorUnsuppressed(e); wait(tr.onError(e)); } } @@ -1487,7 +1490,7 @@ ACTOR Future includeLocalities(Database cx, std::vector local wait(ryw.commit()); return Void(); } catch (Error& e) { - TraceEvent("IncludeLocalitiesError").error(e, true); + TraceEvent("IncludeLocalitiesError").errorUnsuppressed(e); wait(ryw.onError(e)); } } @@ -1535,7 +1538,7 @@ ACTOR Future includeLocalities(Database cx, std::vector local wait(tr.commit()); return Void(); } catch (Error& e) { - TraceEvent("IncludeLocalitiesError").error(e, true); + TraceEvent("IncludeLocalitiesError").errorUnsuppressed(e); wait(tr.onError(e)); } } @@ -1907,7 +1910,7 @@ ACTOR Future mgmtSnapCreate(Database cx, Standalone snapCmd, UI TraceEvent("SnapCreateSucceeded").detail("snapUID", snapUID); return Void(); } catch (Error& e) { - TraceEvent(SevWarn, "SnapCreateFailed").detail("snapUID", snapUID).error(e); + TraceEvent(SevWarn, "SnapCreateFailed").error(e).detail("snapUID", snapUID); throw; } } @@ -2198,7 +2201,7 @@ ACTOR Future advanceVersion(Database cx, Version v) { tr.set(minRequiredCommitVersionKey, BinaryWriter::toValue(v + 1, Unversioned())); wait(tr.commit()); } else { - printf("Current read version is %ld\n", rv); + fmt::print("Current read version is {}\n", rv); return Void(); } } catch (Error& e) { diff --git a/fdbclient/ManagementAPI.actor.h b/fdbclient/ManagementAPI.actor.h index d07a743a5e..a2adbb7025 100644 --- a/fdbclient/ManagementAPI.actor.h +++ b/fdbclient/ManagementAPI.actor.h @@ -56,7 +56,7 @@ struct IQuorumChange : ReferenceCounted { // Change to use the given set of coordination servers ACTOR Future> changeQuorumChecker(Transaction* tr, Reference change, - std::vector* desiredCoordinators); + ClusterConnectionString* conn); ACTOR Future changeQuorum(Database cx, Reference change); Reference autoQuorumChange(int desired = -1); Reference noQuorumChange(); diff --git a/fdbclient/MonitorLeader.actor.cpp b/fdbclient/MonitorLeader.actor.cpp index abfda0aefd..e63824466f 100644 --- a/fdbclient/MonitorLeader.actor.cpp +++ b/fdbclient/MonitorLeader.actor.cpp @@ -178,6 +178,10 @@ void ClusterConnectionString::resetToUnresolved() { } } +void ClusterConnectionString::resetConnectionString() { + connectionString = toString(); +} + void ClusterConnectionString::parseConnString() { // Split on '@' into key@addrs int pAt = connectionString.find_first_of('@'); @@ -444,26 +448,14 @@ ClusterConnectionString::ClusterConnectionString(const std::vector& hosts, Key key) : status(UNRESOLVED), hostnames(hosts) { std::string keyString = key.toString(); parseKey(keyString); - connectionString = keyString + "@"; - for (int i = 0; i < hostnames.size(); i++) { - if (i) { - connectionString += ','; - } - connectionString += hostnames[i].toString(); - } + resetConnectionString(); } void ClusterConnectionString::parseKey(const std::string& key) { @@ -562,9 +554,9 @@ ACTOR Future monitorNominee(Key key, if (rep.isError()) { // Connecting to nominee failed, most likely due to connection failed. TraceEvent("MonitorNomineeError") + .error(rep.getError()) .detail("Hostname", hostname.present() ? hostname.get().toString() : "UnknownHostname") - .detail("OldAddr", coord.getLeader.getEndpoint().getPrimaryAddress().toString()) - .error(rep.getError()); + .detail("OldAddr", coord.getLeader.getEndpoint().getPrimaryAddress().toString()); if (rep.getError().code() == error_code_request_maybe_delivered) { // 50 milliseconds delay to prevent tight resolving loop due to outdated DNS cache wait(delay(0.05)); diff --git a/fdbclient/MultiVersionTransaction.actor.cpp b/fdbclient/MultiVersionTransaction.actor.cpp index 8d02995b04..8af1df1fa2 100644 --- a/fdbclient/MultiVersionTransaction.actor.cpp +++ b/fdbclient/MultiVersionTransaction.actor.cpp @@ -1226,9 +1226,9 @@ MultiVersionDatabase::MultiVersionDatabase(MultiVersionApi* api, // but we may not see trace logs from this client until a successful connection // is established. TraceEvent(SevWarnAlways, "FailedToInitializeExternalClient") + .error(e) .detail("LibraryPath", client->libPath) - .detail("ClusterFilePath", clusterFilePath) - .error(e); + .detail("ClusterFilePath", clusterFilePath); } } }); @@ -1242,9 +1242,9 @@ MultiVersionDatabase::MultiVersionDatabase(MultiVersionApi* api, } catch (Error& e) { // This connection is discarded TraceEvent(SevWarnAlways, "FailedToCreateLegacyDatabaseConnection") + .error(e) .detail("LibraryPath", client->libPath) - .detail("ClusterFilePath", clusterFilePath) - .error(e); + .detail("ClusterFilePath", clusterFilePath); } } }); @@ -1384,8 +1384,8 @@ ThreadFuture MultiVersionDatabase::DatabaseState::monitorProtocolVersion() } TraceEvent("ErrorGettingClusterProtocolVersion") - .detail("ExpectedProtocolVersion", expected) - .error(cv.getError()); + .error(cv.getError()) + .detail("ExpectedProtocolVersion", expected); } ProtocolVersion clusterVersion = @@ -1433,10 +1433,10 @@ void MultiVersionDatabase::DatabaseState::protocolVersionChanged(ProtocolVersion newDb = client->api->createDatabase(clusterFilePath.c_str()); } catch (Error& e) { TraceEvent(SevWarnAlways, "MultiVersionClientFailedToCreateDatabase") + .error(e) .detail("LibraryPath", client->libPath) .detail("External", client->external) - .detail("ClusterFilePath", clusterFilePath) - .error(e); + .detail("ClusterFilePath", clusterFilePath); // Put the client in a disconnected state until the version changes again updateDatabase(Reference(), Reference()); @@ -1510,8 +1510,8 @@ void MultiVersionDatabase::DatabaseState::updateDatabase(Reference ne // We can't create a new database to monitor the cluster version. This means we will continue using the // previous one, which should hopefully continue to work. TraceEvent(SevWarnAlways, "FailedToCreateDatabaseForVersionMonitoring") - .detail("ClusterFilePath", clusterFilePath) - .error(e); + .error(e) + .detail("ClusterFilePath", clusterFilePath); } } } else { @@ -1523,8 +1523,8 @@ void MultiVersionDatabase::DatabaseState::updateDatabase(Reference ne // We can't create a new database to monitor the cluster version. This means we will continue using the // previous one, which should hopefully continue to work. TraceEvent(SevWarnAlways, "FailedToCreateDatabaseForVersionMonitoring") - .detail("ClusterFilePath", clusterFilePath) - .error(e); + .error(e) + .detail("ClusterFilePath", clusterFilePath); } } diff --git a/fdbclient/NativeAPI.actor.cpp b/fdbclient/NativeAPI.actor.cpp index 13d99c9a48..45e5fddca6 100644 --- a/fdbclient/NativeAPI.actor.cpp +++ b/fdbclient/NativeAPI.actor.cpp @@ -63,6 +63,7 @@ #include "fdbrpc/LoadBalance.h" #include "fdbrpc/Net2FileSystem.h" #include "fdbrpc/simulator.h" +#include "fdbrpc/sim_validation.h" #include "flow/Arena.h" #include "flow/ActorCollection.h" #include "flow/DeterministicRandom.h" @@ -251,6 +252,32 @@ void DatabaseContext::getLatestCommitVersions(const Reference& loc } } +void DatabaseContext::updateCachedReadVersion(double t, Version v) { + if (v >= cachedReadVersion) { + TraceEvent(SevDebug, "CachedReadVersionUpdate") + .detail("Version", v) + .detail("GrvStartTime", t) + .detail("LastVersion", cachedReadVersion) + .detail("LastTime", lastGrvTime); + cachedReadVersion = v; + // Since the time is based on the start of the request, it's possible that we + // get a newer version with an older time. + // (Request started earlier, but was latest to reach the proxy) + // Only update time when strictly increasing (?) + if (t > lastGrvTime) { + lastGrvTime = t; + } + } +} + +Version DatabaseContext::getCachedReadVersion() { + return cachedReadVersion; +} + +double DatabaseContext::getLastGrvTime() { + return lastGrvTime; +} + Reference StorageServerInfo::getInterface(DatabaseContext* cx, StorageServerInterface const& ssi, LocalityData const& locality) { @@ -1046,6 +1073,53 @@ ACTOR static Future handleTssMismatches(DatabaseContext* cx) { } } +ACTOR static Future backgroundGrvUpdater(DatabaseContext* cx) { + state Transaction tr; + state double grvDelay = 0.001; + try { + loop { + if (CLIENT_KNOBS->FORCE_GRV_CACHE_OFF) + return Void(); + wait(refreshTransaction(cx, &tr)); + state double curTime = now(); + state double lastTime = cx->getLastGrvTime(); + state double lastProxyTime = cx->lastProxyRequestTime; + TraceEvent(SevDebug, "BackgroundGrvUpdaterBefore") + .detail("CurTime", curTime) + .detail("LastTime", lastTime) + .detail("GrvDelay", grvDelay) + .detail("CachedReadVersion", cx->getCachedReadVersion()) + .detail("CachedTime", cx->getLastGrvTime()) + .detail("Gap", curTime - lastTime) + .detail("Bound", CLIENT_KNOBS->MAX_VERSION_CACHE_LAG - grvDelay); + if (curTime - lastTime >= (CLIENT_KNOBS->MAX_VERSION_CACHE_LAG - grvDelay) || + curTime - lastProxyTime > CLIENT_KNOBS->MAX_PROXY_CONTACT_LAG) { + try { + tr.setOption(FDBTransactionOptions::SKIP_GRV_CACHE); + wait(success(tr.getReadVersion())); + cx->lastProxyRequestTime = curTime; + grvDelay = (grvDelay + (now() - curTime)) / 2.0; + TraceEvent(SevDebug, "BackgroundGrvUpdaterSuccess") + .detail("GrvDelay", grvDelay) + .detail("CachedReadVersion", cx->getCachedReadVersion()) + .detail("CachedTime", cx->getLastGrvTime()); + } catch (Error& e) { + TraceEvent(SevInfo, "BackgroundGrvUpdaterTxnError").errorUnsuppressed(e); + wait(tr.onError(e)); + } + } else { + wait( + delay(std::max(0.001, + std::min(CLIENT_KNOBS->MAX_PROXY_CONTACT_LAG - (curTime - lastProxyTime), + (CLIENT_KNOBS->MAX_VERSION_CACHE_LAG - grvDelay) - (curTime - lastTime))))); + } + } + } catch (Error& e) { + TraceEvent(SevInfo, "BackgroundGrvUpdaterFailed").errorUnsuppressed(e); + throw; + } +} + ACTOR static Future getHealthMetricsActor(DatabaseContext* cx, bool detailed) { if (now() - cx->healthMetricsLastUpdated < CLIENT_KNOBS->AGGREGATE_HEALTH_METRICS_MAX_STALENESS) { if (detailed) { @@ -1275,9 +1349,11 @@ DatabaseContext::DatabaseContext(ReferenceSHARD_STAT_SMOOTH_AMOUNT), + lastGrvTime(0.0), cachedReadVersion(0), lastRkBatchThrottleTime(0.0), lastRkDefaultThrottleTime(0.0), + lastProxyRequestTime(0.0), transactionTracingSample(false), taskID(taskID), clientInfo(clientInfo), + clientInfoMonitor(clientInfoMonitor), coordinator(coordinator), apiVersion(apiVersion), mvCacheInsertLocation(0), + healthMetricsLastUpdated(0), detailedHealthMetricsLastUpdated(0), + smoothMidShardSize(CLIENT_KNOBS->SHARD_STAT_SMOOTH_AMOUNT), specialKeySpace(std::make_unique(specialKeys.begin, specialKeys.end, /* test */ false)), connectToDatabaseEventCacheHolder(format("ConnectToDatabase/%s", dbId.toString().c_str())) { dbId = deterministicRandom()->randomUniqueID(); @@ -1563,6 +1639,9 @@ DatabaseContext::~DatabaseContext() { clientDBInfoMonitor.cancel(); monitorTssInfoChange.cancel(); tssMismatchHandler.cancel(); + if (grvUpdateHandler.isValid()) { + grvUpdateHandler.cancel(); + } for (auto it = server_interf.begin(); it != server_interf.end(); it = server_interf.erase(it)) it->second->notifyContextDestroyed(); ASSERT_ABORT(server_interf.empty()); @@ -1645,6 +1724,32 @@ void DatabaseContext::invalidateCache(const KeyRangeRef& keys) { locationCache.insert(KeyRangeRef(begin, end), Reference()); } +void DatabaseContext::setFailedEndpointOnHealthyServer(const Endpoint& endpoint) { + if (failedEndpointsOnHealthyServersInfo.find(endpoint) == failedEndpointsOnHealthyServersInfo.end()) { + failedEndpointsOnHealthyServersInfo[endpoint] = + EndpointFailureInfo{ .startTime = now(), .lastRefreshTime = now() }; + } +} + +void DatabaseContext::updateFailedEndpointRefreshTime(const Endpoint& endpoint) { + if (failedEndpointsOnHealthyServersInfo.find(endpoint) == failedEndpointsOnHealthyServersInfo.end()) { + // The endpoint is not failed. Nothing to update. + return; + } + failedEndpointsOnHealthyServersInfo[endpoint].lastRefreshTime = now(); +} + +Optional DatabaseContext::getEndpointFailureInfo(const Endpoint& endpoint) { + if (failedEndpointsOnHealthyServersInfo.find(endpoint) == failedEndpointsOnHealthyServersInfo.end()) { + return Optional(); + } + return failedEndpointsOnHealthyServersInfo[endpoint]; +} + +void DatabaseContext::clearFailedEndpointOnHealthyServer(const Endpoint& endpoint) { + failedEndpointsOnHealthyServersInfo.erase(endpoint); +} + Future DatabaseContext::onProxiesChanged() const { return this->proxiesChangeTrigger.onTrigger(); } @@ -2517,6 +2622,35 @@ ACTOR Future>> getKeyLocation_intern } } +// Checks if `endpoint` is failed on a healthy server or not. Returns true if we need to refresh the location cache for +// the endpoint. +bool checkOnlyEndpointFailed(const Database& cx, const Endpoint& endpoint) { + if (IFailureMonitor::failureMonitor().onlyEndpointFailed(endpoint)) { + // This endpoint is failed, but the server is still healthy. There are two cases this can happen: + // - There is a recent bounce in the cluster where the endpoints in SSes get updated. + // - The SS is failed and terminated on a server, but the server is kept running. + // To account for the first case, we invalidate the cache and issue GetKeyLocation requests to the proxy to + // update the cache with the new SS points. However, if the failure is caused by the second case, the + // requested key location will continue to be the failed endpoint until the data movement is finished. But + // every read will generate a GetKeyLocation request to the proxies (and still getting the failed endpoint + // back), which may overload the proxy and affect data movement speed. Therefore, we only refresh the + // location cache for short period of time, and after the initial grace period that we keep retrying + // resolving key location, we will slow it down to resolve it only once every + // `LOCATION_CACHE_FAILED_ENDPOINT_RETRY_INTERVAL`. + cx->setFailedEndpointOnHealthyServer(endpoint); + const auto& failureInfo = cx->getEndpointFailureInfo(endpoint); + ASSERT(failureInfo.present()); + if (now() - failureInfo.get().startTime < CLIENT_KNOBS->LOCATION_CACHE_ENDPOINT_FAILURE_GRACE_PERIOD || + now() - failureInfo.get().lastRefreshTime > CLIENT_KNOBS->LOCATION_CACHE_FAILED_ENDPOINT_RETRY_INTERVAL) { + cx->updateFailedEndpointRefreshTime(endpoint); + return true; + } + } else { + cx->clearFailedEndpointOnHealthyServer(endpoint); + } + return false; +} + template Future>> getKeyLocation(Database const& cx, Key const& key, @@ -2531,14 +2665,19 @@ Future>> getKeyLocation(Database con return getKeyLocation_internal(cx, key, spanID, debugID, useProvisionalProxies, isBackward); } + bool onlyEndpointFailedAndNeedRefresh = false; for (int i = 0; i < ssi.second->size(); i++) { - if (IFailureMonitor::failureMonitor().onlyEndpointFailed(ssi.second->get(i, member).getEndpoint())) { - cx->invalidateCache(key); - ssi.second.clear(); - return getKeyLocation_internal(cx, key, spanID, debugID, useProvisionalProxies, isBackward); + if (checkOnlyEndpointFailed(cx, ssi.second->get(i, member).getEndpoint())) { + onlyEndpointFailedAndNeedRefresh = true; } } + if (onlyEndpointFailedAndNeedRefresh) { + cx->invalidateCache(key); + // Refresh the cache with a new getKeyLocations made to proxies. + return getKeyLocation_internal(cx, key, spanID, debugID, useProvisionalProxies, isBackward); + } + return ssi; } @@ -2622,21 +2761,21 @@ Future>>> getKeyRangeLoc bool foundFailed = false; for (const auto& [range, locInfo] : locations) { - bool onlyEndpointFailed = false; + bool onlyEndpointFailedAndNeedRefresh = false; for (int i = 0; i < locInfo->size(); i++) { - if (IFailureMonitor::failureMonitor().onlyEndpointFailed(locInfo->get(i, member).getEndpoint())) { - onlyEndpointFailed = true; - break; + if (checkOnlyEndpointFailed(cx, locInfo->get(i, member).getEndpoint())) { + onlyEndpointFailedAndNeedRefresh = true; } } - if (onlyEndpointFailed) { + if (onlyEndpointFailedAndNeedRefresh) { cx->invalidateCache(range.begin); foundFailed = true; } } if (foundFailed) { + // Refresh the cache with a new getKeyRangeLocations made to proxies. return getKeyRangeLocations_internal(cx, keys, limit, reverse, spanID, debugID, useProvisionalProxies); } @@ -5030,6 +5169,8 @@ void TransactionOptions::clear() { readTags = TagSet{}; priority = TransactionPriority::DEFAULT; expensiveClearCostEstimation = false; + useGrvCache = false; + skipGrvCache = false; } TransactionOptions::TransactionOptions() { @@ -5200,7 +5341,7 @@ ACTOR static Future commitDummyTransaction(Reference trS return Void(); } catch (Error& e) { TraceEvent("CommitDummyTransactionError") - .error(e, true) + .errorUnsuppressed(e) .detail("Key", range.begin) .detail("Retries", retries); wait(tr.onError(e)); @@ -5360,7 +5501,7 @@ ACTOR static Future tryCommit(Reference trState, TaskPriority::DefaultPromiseEndpoint, AtMostOnce::True); } - + state double grvTime = now(); choose { when(wait(trState->cx->onProxiesChanged())) { reply.cancel(); @@ -5372,6 +5513,7 @@ ACTOR static Future tryCommit(Reference trState, if (CLIENT_BUGGIFY) { throw commit_unknown_result(); } + trState->cx->updateCachedReadVersion(grvTime, v); if (debugID.present()) TraceEvent(interval.end()).detail("CommittedVersion", v); trState->committedVersion = v; @@ -5798,6 +5940,18 @@ void Transaction::setOption(FDBTransactionOptions::Option option, Optionaloptions.expensiveClearCostEstimation = true; break; + case FDBTransactionOptions::USE_GRV_CACHE: + validateOptionValueNotPresent(value); + if (trState->numErrors == 0) { + trState->options.useGrvCache = true; + } + break; + + case FDBTransactionOptions::SKIP_GRV_CACHE: + validateOptionValueNotPresent(value); + trState->options.skipGrvCache = true; + break; + default: break; } @@ -5979,7 +6133,16 @@ ACTOR Future extractReadVersion(Reference trState, Promise> metadataVersion) { state Span span(spanContext, location, { trState->spanID }); GetReadVersionReply rep = wait(f); - double latency = now() - trState->startTime; + double replyTime = now(); + double latency = replyTime - trState->startTime; + trState->cx->lastProxyRequestTime = trState->startTime; + trState->cx->updateCachedReadVersion(trState->startTime, rep.version); + if (rep.rkBatchThrottled) { + trState->cx->lastRkBatchThrottleTime = replyTime; + } + if (rep.rkDefaultThrottled) { + trState->cx->lastRkDefaultThrottleTime = replyTime; + } trState->cx->GRVLatencies.addSample(latency); if (trState->trLogInfo) trState->trLogInfo->addLog(FdbClientLogEvents::EventGetVersion_V3( @@ -6041,8 +6204,42 @@ ACTOR Future extractReadVersion(Reference trState, return rep.version; } +bool rkThrottlingCooledDown(DatabaseContext* cx, TransactionPriority priority) { + if (priority == TransactionPriority::IMMEDIATE) { + return true; + } else if (priority == TransactionPriority::BATCH) { + if (cx->lastRkBatchThrottleTime == 0.0) { + return true; + } + return (now() - cx->lastRkBatchThrottleTime > CLIENT_KNOBS->GRV_CACHE_RK_COOLDOWN); + } else if (priority == TransactionPriority::DEFAULT) { + if (cx->lastRkDefaultThrottleTime == 0.0) { + return true; + } + return (now() - cx->lastRkDefaultThrottleTime > CLIENT_KNOBS->GRV_CACHE_RK_COOLDOWN); + } + return false; +} + Future Transaction::getReadVersion(uint32_t flags) { if (!readVersion.isValid()) { + if (!CLIENT_KNOBS->FORCE_GRV_CACHE_OFF && !trState->options.skipGrvCache && + (deterministicRandom()->random01() <= CLIENT_KNOBS->DEBUG_USE_GRV_CACHE_CHANCE || + trState->options.useGrvCache) && + rkThrottlingCooledDown(getDatabase().getPtr(), trState->options.priority)) { + // Upon our first request to use cached RVs, start the background updater + if (!trState->cx->grvUpdateHandler.isValid()) { + trState->cx->grvUpdateHandler = backgroundGrvUpdater(getDatabase().getPtr()); + } + Version rv = trState->cx->getCachedReadVersion(); + double lastTime = trState->cx->getLastGrvTime(); + double requestTime = now(); + if (requestTime - lastTime <= CLIENT_KNOBS->MAX_VERSION_CACHE_LAG && rv != Version(0)) { + ASSERT(!debug_checkVersionTime(rv, requestTime, "CheckStaleness")); + readVersion = rv; + return readVersion; + } // else go through regular GRV path + } ++trState->cx->transactionReadVersions; flags |= trState->options.getReadVersionFlags; switch (trState->options.priority) { @@ -6224,6 +6421,9 @@ uint32_t Transaction::getSize() { } Future Transaction::onError(Error const& e) { + if (g_network->isSimulated() && ++trState->numErrors % 10 == 0) { + TraceEvent(SevWarnAlways, "TransactionTooManyRetries").detail("NumRetries", trState->numErrors); + } if (e.code() == error_code_success) { return client_invalid_operation(); } @@ -6261,9 +6461,6 @@ Future Transaction::onError(Error const& e) { return delay(std::min(CLIENT_KNOBS->FUTURE_VERSION_RETRY_DELAY, maxBackoff), trState->taskID); } - if (g_network->isSimulated() && ++trState->numErrors % 10 == 0) - TraceEvent(SevWarnAlways, "TransactionTooManyRetries").detail("NumRetries", trState->numErrors); - return e; } ACTOR Future getStorageMetricsLargeKeyRange(Database cx, KeyRange keys); @@ -6845,6 +7042,32 @@ ACTOR Future setPerpetualStorageWiggle(Database cx, bool enable, LockAware return Void(); } +ACTOR Future>> readStorageWiggleValues(Database cx, + bool primary, + bool use_system_priority) { + state const Key readKey = perpetualStorageWiggleIDPrefix.withSuffix(primary ? "primary/"_sr : "remote/"_sr); + state KeyBackedObjectMap metadataMap(readKey, + IncludeVersion()); + state Reference tr(new ReadYourWritesTransaction(cx)); + state std::vector> res; + // read the wiggling pairs + loop { + try { + tr->setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); + tr->setOption(FDBTransactionOptions::READ_LOCK_AWARE); + if (use_system_priority) { + tr->setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); + } + wait(store(res, metadataMap.getRange(tr, UID(0, 0), Optional(), CLIENT_KNOBS->TOO_MANY))); + wait(tr->commit()); + break; + } catch (Error& e) { + wait(tr->onError(e)); + } + } + return res; +} + ACTOR Future>> splitStorageMetrics(Database cx, KeyRange keys, StorageMetrics limit, @@ -6971,7 +7194,7 @@ ACTOR Future snapCreate(Database cx, Standalone snapCmd, UID sn } } } catch (Error& e) { - TraceEvent("SnapCreateError").detail("SnapCmd", snapCmd.toString()).detail("UID", snapUID).error(e); + TraceEvent("SnapCreateError").error(e).detail("SnapCmd", snapCmd.toString()).detail("UID", snapUID); throw; } } @@ -6999,9 +7222,9 @@ ACTOR Future checkSafeExclusions(Database cx, std::vector setPerpetualStorageWiggle(Database cx, bool enable, LockAware lockAware = LockAware::False); +ACTOR Future>> readStorageWiggleValues(Database cx, + bool primary, + bool use_system_priority); + #include "flow/unactorcompiler.h" #endif diff --git a/fdbclient/ReadYourWrites.actor.cpp b/fdbclient/ReadYourWrites.actor.cpp index fdc1d5c771..ee021b6a05 100644 --- a/fdbclient/ReadYourWrites.actor.cpp +++ b/fdbclient/ReadYourWrites.actor.cpp @@ -2585,7 +2585,7 @@ void ReadYourWritesTransaction::debugLogRetries(Optional error) { { TraceEvent trace = TraceEvent("LongTransaction"); if (error.present()) - trace.error(error.get(), true); + trace.errorUnsuppressed(error.get()); if (!transactionDebugInfo->transactionName.empty()) trace.detail("TransactionName", transactionDebugInfo->transactionName); trace.detail("Elapsed", elapsed).detail("Retries", retries).detail("Committed", committed); diff --git a/fdbclient/S3BlobStore.actor.cpp b/fdbclient/S3BlobStore.actor.cpp index aba8d1667e..0abcbe441a 100644 --- a/fdbclient/S3BlobStore.actor.cpp +++ b/fdbclient/S3BlobStore.actor.cpp @@ -500,7 +500,7 @@ ACTOR Future> tryReadJSONFile(std::string path) { } catch (Error& e) { if (e.code() != error_code_actor_cancelled) - TraceEvent(SevWarn, errorEventType).error(e).suppressFor(60).detail("File", path); + TraceEvent(SevWarn, errorEventType).errorUnsuppressed(e).suppressFor(60).detail("File", path); } return Optional(); @@ -744,7 +744,7 @@ ACTOR Future> doRequest_impl(Reference listObjectsStream_impl(Reference bstore, } catch (Error& e) { if (e.code() != error_code_actor_cancelled) TraceEvent(SevWarn, "S3BlobStoreEndpointListResultParseError") - .error(e) + .errorUnsuppressed(e) .suppressFor(60) .detail("Resource", fullResource); throw http_bad_response(); @@ -1080,7 +1080,7 @@ ACTOR Future> listBuckets_impl(Referencecoinflip() ? false : true; + init( ROCKSDB_PERFCONTEXT_SAMPLE_RATE, 0.0001 ); + init( ROCKSDB_MAX_SUBCOMPACTIONS, 2 ); // Leader election bool longLeaderElection = randomize && BUGGIFY; @@ -587,6 +589,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi init( MIN_AVAILABLE_SPACE, 1e8 ); init( MIN_AVAILABLE_SPACE_RATIO, 0.05 ); + init( MIN_AVAILABLE_SPACE_RATIO_SAFETY_BUFFER, 0.01 ); init( TARGET_AVAILABLE_SPACE_RATIO, 0.30 ); init( AVAILABLE_SPACE_UPDATE_DELAY, 5.0 ); diff --git a/fdbclient/ServerKnobs.h b/fdbclient/ServerKnobs.h index d0aa6c4837..1ba07c40ae 100644 --- a/fdbclient/ServerKnobs.h +++ b/fdbclient/ServerKnobs.h @@ -298,6 +298,9 @@ public: bool ROCKSDB_READ_RANGE_REUSE_ITERATORS; int64_t ROCKSDB_WRITE_RATE_LIMITER_BYTES_PER_SEC; bool ROCKSDB_WRITE_RATE_LIMITER_AUTO_TUNE; + bool ROCKSDB_PERFCONTEXT_ENABLE; // Enable rocks perf context metrics. May cause performance overhead + double ROCKSDB_PERFCONTEXT_SAMPLE_RATE; + int ROCKSDB_MAX_SUBCOMPACTIONS; // Leader election int MAX_NOTIFICATIONS; @@ -531,6 +534,7 @@ public: int64_t MIN_AVAILABLE_SPACE; double MIN_AVAILABLE_SPACE_RATIO; + double MIN_AVAILABLE_SPACE_RATIO_SAFETY_BUFFER; double TARGET_AVAILABLE_SPACE_RATIO; double AVAILABLE_SPACE_UPDATE_DELAY; diff --git a/fdbclient/SpecialKeySpace.actor.cpp b/fdbclient/SpecialKeySpace.actor.cpp index 6c1db7639f..af4fe3e470 100644 --- a/fdbclient/SpecialKeySpace.actor.cpp +++ b/fdbclient/SpecialKeySpace.actor.cpp @@ -1628,8 +1628,9 @@ Future CoordinatorsImpl::getRange(ReadYourWritesTransaction* ryw, K ACTOR static Future> coordinatorsCommitActor(ReadYourWritesTransaction* ryw, KeyRangeRef kr) { state Reference change; - state std::vector addressesVec; - state std::vector process_address_strs; + state ClusterConnectionString + conn; // We don't care about the Key here, it will be overrode in changeQuorumChecker(). + state std::vector process_address_or_hostname_strs; state Optional msg; state int index; state bool parse_error = false; @@ -1640,38 +1641,45 @@ ACTOR static Future> coordinatorsCommitActor(ReadYourWrite if (processes_entry.first) { ASSERT(processes_entry.second.present()); // no clear should be seen here auto processesStr = processes_entry.second.get().toString(); - boost::split(process_address_strs, processesStr, [](char c) { return c == ','; }); - if (!process_address_strs.size()) { + boost::split(process_address_or_hostname_strs, processesStr, [](char c) { return c == ','; }); + if (!process_address_or_hostname_strs.size()) { return ManagementAPIError::toJsonString( false, "coordinators", "New coordinators\' processes are empty, please specify new processes\' network addresses with format " - "\"IP:PORT,IP:PORT,...,IP:PORT\""); + "\"IP:PORT,IP:PORT,...,IP:PORT\" or \"HOSTNAME:PORT,HOSTNAME:PORT,...,HOSTNAME:PORT\""); } - for (index = 0; index < process_address_strs.size(); index++) { + for (index = 0; index < process_address_or_hostname_strs.size(); index++) { try { - auto a = NetworkAddress::parse(process_address_strs[index]); - if (!a.isValid()) - parse_error = true; - else - addressesVec.push_back(a); + if (Hostname::isHostname(process_address_or_hostname_strs[index])) { + conn.hostnames.push_back(Hostname::parse(process_address_or_hostname_strs[index])); + conn.status = ClusterConnectionString::ConnectionStringStatus::UNRESOLVED; + } else { + NetworkAddress a = NetworkAddress::parse(process_address_or_hostname_strs[index]); + if (!a.isValid()) { + parse_error = true; + } else { + conn.coords.push_back(a); + } + } } catch (Error& e) { TraceEvent(SevDebug, "SpecialKeysNetworkParseError").error(e); parse_error = true; } if (parse_error) { - std::string error = - "ERROR: \'" + process_address_strs[index] + "\' is not a valid network endpoint address\n"; - if (process_address_strs[index].find(":tls") != std::string::npos) + std::string error = "ERROR: \'" + process_address_or_hostname_strs[index] + + "\' is not a valid network endpoint address\n"; + if (process_address_or_hostname_strs[index].find(":tls") != std::string::npos) error += " Do not include the `:tls' suffix when naming a process\n"; return ManagementAPIError::toJsonString(false, "coordinators", error); } } } - if (addressesVec.size()) - change = specifiedQuorumChange(addressesVec); + wait(conn.resolveHostnames()); + if (conn.coordinators().size()) + change = specifiedQuorumChange(conn.coordinators()); else change = noQuorumChange(); @@ -1693,10 +1701,11 @@ ACTOR static Future> coordinatorsCommitActor(ReadYourWrite ASSERT(change.isValid()); TraceEvent(SevDebug, "SKSChangeCoordinatorsStart") - .detail("NewAddresses", describe(addressesVec)) + .detail("NewHostnames", conn.hostnames.size() ? describe(conn.hostnames) : "N/A") + .detail("NewAddresses", describe(conn.coordinators())) .detail("Description", entry.first ? entry.second.get().toString() : ""); - Optional r = wait(changeQuorumChecker(&ryw->getTransaction(), change, &addressesVec)); + Optional r = wait(changeQuorumChecker(&ryw->getTransaction(), change, &conn)); TraceEvent(SevDebug, "SKSChangeCoordinatorsFinish") .detail("Result", r.present() ? static_cast(r.get()) : -1); // -1 means success diff --git a/fdbclient/StatusClient.actor.cpp b/fdbclient/StatusClient.actor.cpp index 2829771ba8..fc7a77b17a 100644 --- a/fdbclient/StatusClient.actor.cpp +++ b/fdbclient/StatusClient.actor.cpp @@ -306,6 +306,7 @@ ACTOR Future> clientCoordinatorsStatusFetcher(ReferenceresolveHostnames()); state ClientCoordinators coord(connRecord); state StatusObject statusObj; diff --git a/fdbclient/vexillographer/fdb.options b/fdbclient/vexillographer/fdb.options index 0ea8f6bb87..c4845ae8bd 100644 --- a/fdbclient/vexillographer/fdb.options +++ b/fdbclient/vexillographer/fdb.options @@ -292,6 +292,11 @@ description is not currently required but encouraged. description="Asks storage servers for how many bytes a clear key range contains. Otherwise uses the location cache to roughly estimate this." />