apple/foundationdb
1
0
Fork 0
mirror of https://github.com/apple/foundationdb.git synced 2025-06-02 19:25:52 +08:00
Code Issues Packages Projects Releases Wiki Activity

Address merge conflicts

Browse Source
This commit is contained in:
Zhe Wu 2021-03-26 11:36:02 -07:00
commit c3aff4340f
47 changed files with 1192 additions and 613 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@ -70,6 +70,8 @@ trace.*.xml
*.user
.idea/
.project
.projectile
.dir-locals.el
.pydevproject
.vscode
.vs/
@ -85,6 +87,7 @@ flow/coveragetool/obj
/compile_commands.json
/.ccls-cache
/.clangd
/.cache
# Temporary and user configuration files
*~

2
README.md
View File

@ -1,5 +1,7 @@
<img alt="FoundationDB logo" src="documentation/FDB_logo.png?raw=true" width="400">
![Build Status](https://codebuild.us-west-2.amazonaws.com/badges?uuid=eyJlbmNyeXB0ZWREYXRhIjoidTh3TTlZa2FQdkdPL1drZzJUQnA1NWg0MzQ3c1VnMXlaVWQ0MVUwcUJpRlltUExBYmRCc3h2c0p1TXZLdWhDQ3BoS0Jmc2ZZdG5yVmxGUHNJM0JtV0MwPSIsIml2UGFyYW1ldGVyU3BlYyI6InBrclM3R0J2d3hmRUFDTjgiLCJtYXRlcmlhbFNldFNlcmlhbCI6MX0%3D&branch=master)
FoundationDB is a distributed database designed to handle large volumes of structured data across clusters of commodity servers. It organizes data as an ordered key-value store and employs ACID transactions for all operations. It is especially well-suited for read/write workloads but also has excellent performance for write-intensive workloads. Users interact with the database using API language binding.
To learn more about FoundationDB, visit [foundationdb.org](https://www.foundationdb.org/)

7
bindings/c/fdb_c.cpp
View File

@ -357,6 +357,13 @@ extern "C" DLLEXPORT FDBFuture* fdb_database_create_snapshot(FDBDatabase* db,
.extractPtr());
}
// Get network thread busyness (updated every 1s)
// A value of 0 indicates that the client is more or less idle
// A value of 1 (or more) indicates that the client is saturated
extern "C" DLLEXPORT double fdb_database_get_main_thread_busyness(FDBDatabase* d) {
return DB(d)->getMainThreadBusyness();
}
extern "C" DLLEXPORT void fdb_transaction_destroy(FDBTransaction* tr) {
try {
TXN(tr)->delref();

464
bindings/c/foundationdb/fdb_c.h
View File

@ -22,7 +22,6 @@
#define FDB_C_H
#pragma once
#ifndef DLLEXPORT
#define DLLEXPORT
#endif
@ -52,10 +51,9 @@
* ensure a compile error in such cases, and attempt to make the compile error
* slightly informative.
*/
#define This_FoundationDB_API_function_is_removed_at_this_FDB_API_VERSION() \
[== == = ]
#define FDB_REMOVED_FUNCTION \
This_FoundationDB_API_function_is_removed_at_this_FDB_API_VERSION(0)
#define This_FoundationDB_API_function_is_removed_at_this_FDB_API_VERSION() \
{ == == = }
#define FDB_REMOVED_FUNCTION This_FoundationDB_API_function_is_removed_at_this_FDB_API_VERSION(0)
#include <stdint.h>
@ -65,333 +63,351 @@
extern "C" {
#endif
/* Pointers to these opaque types represent objects in the FDB API */
typedef struct FDB_future FDBFuture;
typedef struct FDB_database FDBDatabase;
typedef struct FDB_transaction FDBTransaction;
/* Pointers to these opaque types represent objects in the FDB API */
typedef struct FDB_future FDBFuture;
typedef struct FDB_database FDBDatabase;
typedef struct FDB_transaction FDBTransaction;
typedef int fdb_error_t;
typedef int fdb_bool_t;
typedef int fdb_error_t;
typedef int fdb_bool_t;
DLLEXPORT const char*
fdb_get_error( fdb_error_t code );
DLLEXPORT const char* fdb_get_error(fdb_error_t code);
DLLEXPORT fdb_bool_t
fdb_error_predicate( int predicate_test, fdb_error_t code );
DLLEXPORT fdb_bool_t fdb_error_predicate(int predicate_test, fdb_error_t code);
#define /* fdb_error_t */ fdb_select_api_version(v) fdb_select_api_version_impl(v, FDB_API_VERSION)
#define /* fdb_error_t */ fdb_select_api_version(v) fdb_select_api_version_impl(v, FDB_API_VERSION)
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_network_set_option( FDBNetworkOption option, uint8_t const* value,
int value_length );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_network_set_option(FDBNetworkOption option,
uint8_t const* value,
int value_length);
#if FDB_API_VERSION >= 14
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_setup_network();
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_setup_network();
#endif
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_run_network();
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_run_network();
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_stop_network();
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_stop_network();
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_add_network_thread_completion_hook(void (*hook)(void*), void *hook_parameter);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_add_network_thread_completion_hook(void (*hook)(void*),
void* hook_parameter);
#pragma pack(push, 4)
typedef struct key {
const uint8_t* key;
int key_length;
} FDBKey;
typedef struct key {
const uint8_t* key;
int key_length;
} FDBKey;
#if FDB_API_VERSION >= 700
typedef struct keyvalue {
const uint8_t* key;
int key_length;
const uint8_t* value;
int value_length;
} FDBKeyValue;
typedef struct keyvalue {
const uint8_t* key;
int key_length;
const uint8_t* value;
int value_length;
} FDBKeyValue;
#else
typedef struct keyvalue {
const void* key;
int key_length;
const void* value;
int value_length;
} FDBKeyValue;
typedef struct keyvalue {
const void* key;
int key_length;
const void* value;
int value_length;
} FDBKeyValue;
#endif
#pragma pack(pop)
DLLEXPORT void fdb_future_cancel( FDBFuture* f );
DLLEXPORT void fdb_future_cancel(FDBFuture* f);
DLLEXPORT void fdb_future_release_memory( FDBFuture* f );
DLLEXPORT void fdb_future_release_memory(FDBFuture* f);
DLLEXPORT void fdb_future_destroy( FDBFuture* f );
DLLEXPORT void fdb_future_destroy(FDBFuture* f);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_block_until_ready( FDBFuture* f );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_block_until_ready(FDBFuture* f);
DLLEXPORT fdb_bool_t fdb_future_is_ready( FDBFuture* f );
DLLEXPORT fdb_bool_t fdb_future_is_ready(FDBFuture* f);
typedef void (*FDBCallback)(FDBFuture* future, void* callback_parameter);
typedef void (*FDBCallback)(FDBFuture* future, void* callback_parameter);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_future_set_callback( FDBFuture* f, FDBCallback callback,
void* callback_parameter );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_set_callback(FDBFuture* f,
FDBCallback callback,
void* callback_parameter);
#if FDB_API_VERSION >= 23
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_future_get_error( FDBFuture* f );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_get_error(FDBFuture* f);
#endif
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_future_get_int64( FDBFuture* f, int64_t* out );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_get_int64(FDBFuture* f, int64_t* out);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_future_get_uint64( FDBFuture* f, uint64_t* out );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_get_uint64(FDBFuture* f, uint64_t* out);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_future_get_key( FDBFuture* f, uint8_t const** out_key,
int* out_key_length );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_get_key(FDBFuture* f, uint8_t const** out_key, int* out_key_length);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_future_get_value( FDBFuture* f, fdb_bool_t *out_present,
uint8_t const** out_value,
int* out_value_length );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_get_value(FDBFuture* f,
fdb_bool_t* out_present,
uint8_t const** out_value,
int* out_value_length);
#if FDB_API_VERSION >= 14
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_future_get_keyvalue_array( FDBFuture* f, FDBKeyValue const** out_kv,
int* out_count, fdb_bool_t* out_more );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_get_keyvalue_array(FDBFuture* f,
FDBKeyValue const** out_kv,
int* out_count,
fdb_bool_t* out_more);
#endif
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_future_get_key_array( FDBFuture* f, FDBKey const** out_key_array,
int* out_count);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_get_key_array(FDBFuture* f,
FDBKey const** out_key_array,
int* out_count);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_get_string_array(FDBFuture* f,
const char*** out_strings, int* out_count);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_get_string_array(FDBFuture* f,
const char*** out_strings,
int* out_count);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_create_database( const char* cluster_file_path, FDBDatabase** out_database );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_create_database(const char* cluster_file_path, FDBDatabase** out_database);
DLLEXPORT void fdb_database_destroy( FDBDatabase* d );
DLLEXPORT void fdb_database_destroy(FDBDatabase* d);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_database_set_option( FDBDatabase* d, FDBDatabaseOption option,
uint8_t const* value, int value_length );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_database_set_option(FDBDatabase* d,
FDBDatabaseOption option,
uint8_t const* value,
int value_length);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_database_create_transaction( FDBDatabase* d,
FDBTransaction** out_transaction );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_database_create_transaction(FDBDatabase* d,
FDBTransaction** out_transaction);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture*
fdb_database_reboot_worker( FDBDatabase* db, uint8_t const* address,
int address_length, fdb_bool_t check, int duration);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture*
fdb_database_force_recovery_with_data_loss( FDBDatabase* db, uint8_t const* dcid, int dcid_length);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_database_reboot_worker(FDBDatabase* db,
uint8_t const* address,
int address_length,
fdb_bool_t check,
int duration);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture*
fdb_database_create_snapshot(FDBDatabase *db, uint8_t const *uid,
int uid_length, uint8_t const *snap_command,
int snap_command_length);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_database_force_recovery_with_data_loss(FDBDatabase* db,
uint8_t const* dcid,
int dcid_length);
DLLEXPORT void fdb_transaction_destroy( FDBTransaction* tr);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_database_create_snapshot(FDBDatabase* db,
uint8_t const* uid,
int uid_length,
uint8_t const* snap_command,
int snap_command_length);
DLLEXPORT void fdb_transaction_cancel( FDBTransaction* tr);
DLLEXPORT WARN_UNUSED_RESULT double fdb_database_get_main_thread_busyness(FDBDatabase* db);
DLLEXPORT void fdb_transaction_destroy(FDBTransaction* tr);
DLLEXPORT void fdb_transaction_cancel(FDBTransaction* tr);
#if FDB_API_VERSION >= 14
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_transaction_set_option( FDBTransaction* tr, FDBTransactionOption option,
uint8_t const* value, int value_length );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_transaction_set_option(FDBTransaction* tr,
FDBTransactionOption option,
uint8_t const* value,
int value_length);
#endif
DLLEXPORT void
fdb_transaction_set_read_version( FDBTransaction* tr, int64_t version );
DLLEXPORT void fdb_transaction_set_read_version(FDBTransaction* tr, int64_t version);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get_read_version( FDBTransaction* tr );
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get_read_version(FDBTransaction* tr);
#if FDB_API_VERSION >= 14
DLLEXPORT WARN_UNUSED_RESULT FDBFuture*
fdb_transaction_get( FDBTransaction* tr, uint8_t const* key_name,
int key_name_length, fdb_bool_t snapshot );
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get(FDBTransaction* tr,
uint8_t const* key_name,
int key_name_length,
fdb_bool_t snapshot);
#endif
#if FDB_API_VERSION >= 14
DLLEXPORT WARN_UNUSED_RESULT FDBFuture*
fdb_transaction_get_key( FDBTransaction* tr, uint8_t const* key_name,
int key_name_length, fdb_bool_t or_equal,
int offset, fdb_bool_t snapshot );
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get_key(FDBTransaction* tr,
uint8_t const* key_name,
int key_name_length,
fdb_bool_t or_equal,
int offset,
fdb_bool_t snapshot);
#endif
DLLEXPORT WARN_UNUSED_RESULT FDBFuture*
fdb_transaction_get_addresses_for_key(FDBTransaction* tr, uint8_t const* key_name,
int key_name_length);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get_addresses_for_key(FDBTransaction* tr,
uint8_t const* key_name,
int key_name_length);
#if FDB_API_VERSION >= 14
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get_range(
FDBTransaction* tr, uint8_t const* begin_key_name,
int begin_key_name_length, fdb_bool_t begin_or_equal, int begin_offset,
uint8_t const* end_key_name, int end_key_name_length,
fdb_bool_t end_or_equal, int end_offset, int limit, int target_bytes,
FDBStreamingMode mode, int iteration, fdb_bool_t snapshot,
fdb_bool_t reverse );
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get_range(FDBTransaction* tr,
uint8_t const* begin_key_name,
int begin_key_name_length,
fdb_bool_t begin_or_equal,
int begin_offset,
uint8_t const* end_key_name,
int end_key_name_length,
fdb_bool_t end_or_equal,
int end_offset,
int limit,
int target_bytes,
FDBStreamingMode mode,
int iteration,
fdb_bool_t snapshot,
fdb_bool_t reverse);
#endif
DLLEXPORT void
fdb_transaction_set( FDBTransaction* tr, uint8_t const* key_name,
int key_name_length, uint8_t const* value,
int value_length );
DLLEXPORT void fdb_transaction_set(FDBTransaction* tr,
uint8_t const* key_name,
int key_name_length,
uint8_t const* value,
int value_length);
DLLEXPORT void
fdb_transaction_atomic_op( FDBTransaction* tr, uint8_t const* key_name,
int key_name_length, uint8_t const* param,
int param_length, FDBMutationType operation_type );
DLLEXPORT void fdb_transaction_atomic_op(FDBTransaction* tr,
uint8_t const* key_name,
int key_name_length,
uint8_t const* param,
int param_length,
FDBMutationType operation_type);
DLLEXPORT void
fdb_transaction_clear( FDBTransaction* tr, uint8_t const* key_name,
int key_name_length );
DLLEXPORT void fdb_transaction_clear(FDBTransaction* tr, uint8_t const* key_name, int key_name_length);
DLLEXPORT void fdb_transaction_clear_range(
FDBTransaction* tr, uint8_t const* begin_key_name,
int begin_key_name_length, uint8_t const* end_key_name,
int end_key_name_length );
DLLEXPORT void fdb_transaction_clear_range(FDBTransaction* tr,
uint8_t const* begin_key_name,
int begin_key_name_length,
uint8_t const* end_key_name,
int end_key_name_length);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_watch( FDBTransaction *tr,
uint8_t const* key_name,
int key_name_length);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_watch(FDBTransaction* tr,
uint8_t const* key_name,
int key_name_length);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_commit( FDBTransaction* tr );
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_commit(FDBTransaction* tr);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_transaction_get_committed_version( FDBTransaction* tr,
int64_t* out_version );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_transaction_get_committed_version(FDBTransaction* tr,
int64_t* out_version);
/*
* This function intentionally returns an FDBFuture instead of an integer
* directly, so that calling this API can see the effect of previous
* mutations on the transaction. Specifically, mutations are applied
* asynchronously by the main thread. In order to see them, this call has to
* be serviced by the main thread too.
*/
DLLEXPORT WARN_UNUSED_RESULT FDBFuture *
fdb_transaction_get_approximate_size(FDBTransaction *tr);
/*
* This function intentionally returns an FDBFuture instead of an integer
* directly, so that calling this API can see the effect of previous
* mutations on the transaction. Specifically, mutations are applied
* asynchronously by the main thread. In order to see them, this call has to
* be serviced by the main thread too.
*/
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get_approximate_size(FDBTransaction* tr);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture*
fdb_get_server_protocol(const char* clusterFilePath);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_get_server_protocol(const char* clusterFilePath);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get_versionstamp( FDBTransaction* tr );
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get_versionstamp(FDBTransaction* tr);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture*
fdb_transaction_on_error( FDBTransaction* tr, fdb_error_t error );
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_on_error(FDBTransaction* tr, fdb_error_t error);
DLLEXPORT void fdb_transaction_reset( FDBTransaction* tr );
DLLEXPORT void fdb_transaction_reset(FDBTransaction* tr);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_transaction_add_conflict_range(FDBTransaction *tr,
uint8_t const* begin_key_name,
int begin_key_name_length,
uint8_t const* end_key_name,
int end_key_name_length,
FDBConflictRangeType type);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_transaction_add_conflict_range(FDBTransaction* tr,
uint8_t const* begin_key_name,
int begin_key_name_length,
uint8_t const* end_key_name,
int end_key_name_length,
FDBConflictRangeType type);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture*
fdb_transaction_get_estimated_range_size_bytes( FDBTransaction* tr, uint8_t const* begin_key_name,
int begin_key_name_length, uint8_t const* end_key_name, int end_key_name_length);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get_estimated_range_size_bytes(FDBTransaction* tr,
uint8_t const* begin_key_name,
int begin_key_name_length,
uint8_t const* end_key_name,
int end_key_name_length);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture*
fdb_transaction_get_range_split_points( FDBTransaction* tr, uint8_t const* begin_key_name,
int begin_key_name_length, uint8_t const* end_key_name, int end_key_name_length, int64_t chunk_size);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get_range_split_points(FDBTransaction* tr,
uint8_t const* begin_key_name,
int begin_key_name_length,
uint8_t const* end_key_name,
int end_key_name_length,
int64_t chunk_size);
#define FDB_KEYSEL_LAST_LESS_THAN(k, l) k, l, 0, 0
#define FDB_KEYSEL_LAST_LESS_OR_EQUAL(k, l) k, l, 1, 0
#define FDB_KEYSEL_FIRST_GREATER_THAN(k, l) k, l, 1, 1
#define FDB_KEYSEL_FIRST_GREATER_OR_EQUAL(k, l) k, l, 0, 1
#define FDB_KEYSEL_LAST_LESS_THAN(k, l) k, l, 0, 0
#define FDB_KEYSEL_LAST_LESS_OR_EQUAL(k, l) k, l, 1, 0
#define FDB_KEYSEL_FIRST_GREATER_THAN(k, l) k, l, 1, 1
#define FDB_KEYSEL_FIRST_GREATER_OR_EQUAL(k, l) k, l, 0, 1
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_select_api_version_impl( int runtime_version, int header_version );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_select_api_version_impl(int runtime_version, int header_version);
DLLEXPORT int fdb_get_max_api_version();
DLLEXPORT const char* fdb_get_client_version();
DLLEXPORT int fdb_get_max_api_version();
DLLEXPORT const char* fdb_get_client_version();
/* LEGACY API VERSIONS */
/* LEGACY API VERSIONS */
#if FDB_API_VERSION < 620
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_future_get_version( FDBFuture* f, int64_t* out_version );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_get_version(FDBFuture* f, int64_t* out_version);
#else
#define fdb_future_get_version(f, ov) FDB_REMOVED_FUNCTION
#define fdb_future_get_version(f, ov) FDB_REMOVED_FUNCTION
#endif
#if FDB_API_VERSION < 610 || defined FDB_INCLUDE_LEGACY_TYPES
typedef struct FDB_cluster FDBCluster;
typedef struct FDB_cluster FDBCluster;
typedef enum {
/* This option is only a placeholder for C compatibility and should not be used */
FDB_CLUSTER_OPTION_DUMMY_DO_NOT_USE=-1
} FDBClusterOption;
typedef enum {
/* This option is only a placeholder for C compatibility and should not be used */
FDB_CLUSTER_OPTION_DUMMY_DO_NOT_USE = -1
} FDBClusterOption;
#endif
#if FDB_API_VERSION < 610
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_future_get_cluster( FDBFuture* f, FDBCluster** out_cluster );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_get_cluster(FDBFuture* f, FDBCluster** out_cluster);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_future_get_database( FDBFuture* f, FDBDatabase** out_database );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_get_database(FDBFuture* f, FDBDatabase** out_database);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_create_cluster( const char* cluster_file_path );
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_create_cluster(const char* cluster_file_path);
DLLEXPORT void fdb_cluster_destroy( FDBCluster* c );
DLLEXPORT void fdb_cluster_destroy(FDBCluster* c);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_cluster_set_option( FDBCluster* c, FDBClusterOption option,
uint8_t const* value, int value_length );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_cluster_set_option(FDBCluster* c,
FDBClusterOption option,
uint8_t const* value,
int value_length);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture*
fdb_cluster_create_database( FDBCluster* c, uint8_t const* db_name,
int db_name_length );
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_cluster_create_database(FDBCluster* c,
uint8_t const* db_name,
int db_name_length);
#else
#define fdb_future_get_cluster(f, oc) FDB_REMOVED_FUNCTION
#define fdb_future_get_database(f, od) FDB_REMOVED_FUNCTION
#define fdb_create_cluster(cfp) FDB_REMOVED_FUNCTION
#define fdb_cluster_destroy(c) FDB_REMOVED_FUNCTION
#define fdb_cluster_set_option(c, o, v, vl) FDB_REMOVED_FUNCTION
#define fdb_cluster_create_database(c, dn, dnl) FDB_REMOVED_FUNCTION
#define fdb_future_get_cluster(f, oc) FDB_REMOVED_FUNCTION
#define fdb_future_get_database(f, od) FDB_REMOVED_FUNCTION
#define fdb_create_cluster(cfp) FDB_REMOVED_FUNCTION
#define fdb_cluster_destroy(c) FDB_REMOVED_FUNCTION
#define fdb_cluster_set_option(c, o, v, vl) FDB_REMOVED_FUNCTION
#define fdb_cluster_create_database(c, dn, dnl) FDB_REMOVED_FUNCTION
#endif
#if FDB_API_VERSION < 23
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t
fdb_future_get_error( FDBFuture* f,
const char** out_description /* = NULL */ );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_get_error(FDBFuture* f, const char** out_description /* = NULL */);
DLLEXPORT fdb_bool_t fdb_future_is_error( FDBFuture* f );
DLLEXPORT fdb_bool_t fdb_future_is_error(FDBFuture* f);
#else
#define fdb_future_is_error(x) FDB_REMOVED_FUNCTION
#define fdb_future_is_error(x) FDB_REMOVED_FUNCTION
#endif
#if FDB_API_VERSION < 14
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_get_keyvalue_array(
FDBFuture* f, FDBKeyValue const** out_kv, int* out_count );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_future_get_keyvalue_array(FDBFuture* f,
FDBKeyValue const** out_kv,
int* out_count);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get(
FDBTransaction* tr, uint8_t const* key_name, int key_name_length );
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get(FDBTransaction* tr,
uint8_t const* key_name,
int key_name_length);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get_key(
FDBTransaction* tr, uint8_t const* key_name, int key_name_length,
fdb_bool_t or_equal, int offset );
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get_key(FDBTransaction* tr,
uint8_t const* key_name,
int key_name_length,
fdb_bool_t or_equal,
int offset);
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_setup_network( const char* local_address );
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_setup_network(const char* local_address);
DLLEXPORT void fdb_transaction_set_option(
FDBTransaction* tr, FDBTransactionOption option );
DLLEXPORT void fdb_transaction_set_option(FDBTransaction* tr, FDBTransactionOption option);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get_range(
FDBTransaction* tr, uint8_t const* begin_key_name,
int begin_key_name_length, uint8_t const* end_key_name,
int end_key_name_length, int limit );
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get_range(FDBTransaction* tr,
uint8_t const* begin_key_name,
int begin_key_name_length,
uint8_t const* end_key_name,
int end_key_name_length,
int limit);
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get_range_selector(
FDBTransaction* tr, uint8_t const* begin_key_name,
int begin_key_name_length, fdb_bool_t begin_or_equal,
int begin_offset, uint8_t const* end_key_name,
int end_key_name_length, fdb_bool_t end_or_equal, int end_offset,
int limit );
DLLEXPORT WARN_UNUSED_RESULT FDBFuture* fdb_transaction_get_range_selector(FDBTransaction* tr,
uint8_t const* begin_key_name,
int begin_key_name_length,
fdb_bool_t begin_or_equal,
int begin_offset,
uint8_t const* end_key_name,
int end_key_name_length,
fdb_bool_t end_or_equal,
int end_offset,
int limit);
#else
#define fdb_transaction_get_range_selector(tr,bkn,bknl,boe,bo,ekn,eknl,eoe,eo,lim) FDB_REMOVED_FUNCTION
#define fdb_transaction_get_range_selector(tr, bkn, bknl, boe, bo, ekn, eknl, eoe, eo, lim) FDB_REMOVED_FUNCTION
#endif
#ifdef __cplusplus

33
bindings/c/test/unit/unit_tests.cpp
View File

@ -35,11 +35,14 @@
#include <tuple>
#include <vector>
#include <random>
#include <chrono>
#define DOCTEST_CONFIG_IMPLEMENT
#include "doctest.h"
#include "fdbclient/rapidjson/document.h"
#include "flow/config.h"
#include "fdb_api.hpp"
void fdb_check(fdb_error_t e) {
@ -1511,12 +1514,19 @@ TEST_CASE("fdb_transaction_get_approximate_size") {
}
TEST_CASE("fdb_get_server_protocol") {
// We don't really have any expectations other than "don't crash" here
FDBFuture* protocolFuture = fdb_get_server_protocol(clusterFilePath.c_str());
uint64_t out;
fdb_check(fdb_future_block_until_ready(protocolFuture));
fdb_check(fdb_future_get_uint64(protocolFuture, &out));
fdb_future_destroy(protocolFuture);
// "Default" cluster file version
protocolFuture = fdb_get_server_protocol(nullptr);
fdb_check(fdb_future_block_until_ready(protocolFuture));
fdb_check(fdb_future_get_uint64(protocolFuture, &out));
fdb_future_destroy(protocolFuture);
}
TEST_CASE("fdb_transaction_watch read_your_writes_disable") {
@ -1958,6 +1968,11 @@ std::string get_valid_status_json() {
}
TEST_CASE("fdb_database_reboot_worker") {
#ifdef USE_TSAN
MESSAGE(
"fdb_database_reboot_worker disabled for tsan, since fdbmonitor doesn't seem to restart the killed process");
return;
#endif
std::string status_json = get_valid_status_json();
rapidjson::Document statusJson;
statusJson.Parse(status_json.c_str());
@ -2112,6 +2127,24 @@ TEST_CASE("block_from_callback") {
context.event.wait();
}
// monitors network busyness for 2 sec (40 readings)
TEST_CASE("monitor_network_busyness") {
bool containsGreaterZero = false;
for (int i = 0; i < 40; i++) {
double busyness = fdb_database_get_main_thread_busyness(db);
// make sure the busyness is between 0 and 1
CHECK(busyness >= 0);
CHECK(busyness <= 1);
if (busyness > 0) {
containsGreaterZero = true;
}
std::this_thread::sleep_for(std::chrono::milliseconds(50));
}
// assert that at least one of the busyness readings was greater than 0
CHECK(containsGreaterZero);
}
int main(int argc, char** argv) {
if (argc != 3 && argc != 4) {
std::cout << "Unit tests for the FoundationDB C API.\n"

158
build/docker/buildspec.yml
View File

@ -1,158 +0,0 @@
version: 0.2
env:
secrets-manager:
DOCKERHUB_AUTH: dockerhub_foundationdb:foundationdb
phases:
install:
commands:
- echo "install phase"
- 'ACCOUNT_ID=$(echo $CODEBUILD_BUILD_ARN | cut -d : -f 5)'
- REGISTRY=${ACCOUNT_ID}.dkr.ecr.${AWS_DEFAULT_REGION}.amazonaws.com
- aws ecr get-login-password | docker login --username AWS --password-stdin ${REGISTRY}
- docker pull ${REGISTRY}/centos:6
- docker tag ${REGISTRY}/centos:6 centos:6
- docker pull ${REGISTRY}/centos:7
- docker tag ${REGISTRY}/centos:7 centos:7
pre_build:
commands:
- echo "pre_build phase"
- COMMIT_HASH=$(echo $CODEBUILD_RESOLVED_SOURCE_VERSION | cut -c 1-7)
- DATE_STR=$(date +"%Y%m%d%H%M%S")
build:
commands:
- echo "build phase"
- ################################################################################
- # CENTOS 7 foundationdb/build
- ################################################################################
- cd ${CODEBUILD_SRC_DIR}/build/docker/centos7/build
- docker pull ${REGISTRY}/foundationdb/build:centos7-latest || true
- docker build --cache-from ${REGISTRY}/foundationdb/build:centos7-latest
--tag ${REGISTRY}/foundationdb/build:centos7-${DATE_STR}-${COMMIT_HASH}
--tag ${REGISTRY}/foundationdb/build:centos7-latest
--tag ${REGISTRY}/foundationdb/build:latest
--tag foundationdb/build:centos7-${DATE_STR}-${COMMIT_HASH}
--tag foundationdb/build:centos7-latest
--tag foundationdb/build:latest
.
- ################################################################################
- # CENTOS 7 foundationdb/devel
- ################################################################################
- cd ${CODEBUILD_SRC_DIR}/build/docker/centos7/devel
- docker pull ${REGISTRY}/foundationdb/devel:centos7-latest || true
- docker build --cache-from ${REGISTRY}/foundationdb/devel:centos7-latest
--build-arg REPOSITORY=${REGISTRY}/foundationdb/build
--tag ${REGISTRY}/foundationdb/devel:centos7-${DATE_STR}-${COMMIT_HASH}
--tag ${REGISTRY}/foundationdb/devel:centos7-latest
--tag ${REGISTRY}/foundationdb/devel:latest
--tag foundationdb/devel:centos7-${DATE_STR}-${COMMIT_HASH}
--tag foundationdb/devel:centos7-latest
--tag foundationdb/devel:latest
.
- ################################################################################
- # CENTOS 7 foundationdb/distcc
- ################################################################################
- cd ${CODEBUILD_SRC_DIR}/build/docker/centos7/distcc
- docker pull ${REGISTRY}/foundationdb/distcc:centos7-latest || true
- docker build --cache-from ${REGISTRY}/foundationdb/distcc:centos7-latest
--build-arg REPOSITORY=${REGISTRY}/foundationdb/build
--tag ${REGISTRY}/foundationdb/distcc:centos7-${DATE_STR}-${COMMIT_HASH}
--tag ${REGISTRY}/foundationdb/distcc:centos7-latest
--tag ${REGISTRY}/foundationdb/distcc:latest
--tag foundationdb/distcc:centos7-${DATE_STR}-${COMMIT_HASH}
--tag foundationdb/distcc:centos7-latest
--tag foundationdb/distcc:latest
.
- ################################################################################
- # CENTOS 6 foundationdb/build
- ################################################################################
- cd ${CODEBUILD_SRC_DIR}/build/docker/centos6/build
- docker pull ${REGISTRY}/foundationdb/build:centos6-latest || true
- docker build --cache-from ${REGISTRY}/foundationdb/build:centos6-latest
--tag ${REGISTRY}/foundationdb/build:centos6-${DATE_STR}-${COMMIT_HASH}
--tag ${REGISTRY}/foundationdb/build:centos6-latest
--tag foundationdb/build:centos6-${DATE_STR}-${COMMIT_HASH}
--tag foundationdb/build:centos6-latest
.
- ################################################################################
- # CENTOS 6 foundationdb/devel
- ################################################################################
- cd ${CODEBUILD_SRC_DIR}/build/docker/centos6/devel
- docker pull ${REGISTRY}/foundationdb/devel:centos6-latest || true
- docker build --cache-from ${REGISTRY}/foundationdb/devel:centos6-latest
--build-arg REPOSITORY=${REGISTRY}/foundationdb/build
--tag ${REGISTRY}/foundationdb/devel:centos6-${DATE_STR}-${COMMIT_HASH}
--tag ${REGISTRY}/foundationdb/devel:centos6-latest
--tag foundationdb/devel:centos6-${DATE_STR}-${COMMIT_HASH}
--tag foundationdb/devel:centos6-latest
.
- ################################################################################
- # CENTOS 6 foundationdb/distcc
- ################################################################################
- cd ${CODEBUILD_SRC_DIR}/build/docker/centos6/distcc
- docker pull ${REGISTRY}/foundationdb/distcc:centos6-latest || true
- docker build --cache-from ${REGISTRY}/foundationdb/distcc:centos6-latest
--build-arg REPOSITORY=${REGISTRY}/foundationdb/build
--tag ${REGISTRY}/foundationdb/distcc:centos6-${DATE_STR}-${COMMIT_HASH}
--tag ${REGISTRY}/foundationdb/distcc:centos6-latest
--tag foundationdb/distcc:centos6-${DATE_STR}-${COMMIT_HASH}
--tag foundationdb/distcc:centos6-latest
.
post_build:
commands:
- echo "post_build phase"
- echo ${DOCKERHUB_AUTH} | docker login --username foundationdb --password-stdin
- ################################################################################
- # CENTOS 7 PUSH TO ECR
- ################################################################################
- # PUSH TO build ECR
- docker push ${REGISTRY}/foundationdb/build:centos7-${DATE_STR}-${COMMIT_HASH}
- docker push ${REGISTRY}/foundationdb/build:centos7-latest
- docker push ${REGISTRY}/foundationdb/build:latest
- # PUSH TO devel ECR
- docker push ${REGISTRY}/foundationdb/devel:centos7-${DATE_STR}-${COMMIT_HASH}
- docker push ${REGISTRY}/foundationdb/devel:centos7-latest
- docker push ${REGISTRY}/foundationdb/devel:latest
- # PUSH TO distcc ECR
- docker push ${REGISTRY}/foundationdb/distcc:centos7-${DATE_STR}-${COMMIT_HASH}
- docker push ${REGISTRY}/foundationdb/distcc:centos7-latest
- docker push ${REGISTRY}/foundationdb/distcc:latest
- ################################################################################
- # CENTOS 7 PUSH TO DOCKERHUB
- ################################################################################
- # PUSH TO build DOCKERHUB
- docker push foundationdb/build:centos7-${DATE_STR}-${COMMIT_HASH}
- docker push foundationdb/build:centos7-latest
- docker push foundationdb/build:latest
- # PUSH TO devel DOCKERHUB
- docker push foundationdb/devel:centos7-${DATE_STR}-${COMMIT_HASH}
- docker push foundationdb/devel:centos7-latest
- docker push foundationdb/devel:latest
- # PUSH TO distcc DOCKERHUB
- docker push foundationdb/distcc:centos7-${DATE_STR}-${COMMIT_HASH}
- docker push foundationdb/distcc:centos7-latest
- docker push foundationdb/distcc:latest
- ################################################################################
- # CENTOS 6 PUSH TO ECR
- ################################################################################
- # PUSH TO build ECR
- docker push ${REGISTRY}/foundationdb/build:centos6-${DATE_STR}-${COMMIT_HASH}
- docker push ${REGISTRY}/foundationdb/build:centos6-latest
- # PUSH TO devel ECR
- docker push ${REGISTRY}/foundationdb/devel:centos6-${DATE_STR}-${COMMIT_HASH}
- docker push ${REGISTRY}/foundationdb/devel:centos6-latest
- # PUSH TO distcc ECR
- docker push ${REGISTRY}/foundationdb/distcc:centos6-${DATE_STR}-${COMMIT_HASH}
- docker push ${REGISTRY}/foundationdb/distcc:centos6-latest
- ################################################################################
- # CENTOS 6 PUSH TO DOCKERHUB
- ################################################################################
- # PUSH TO build DOCKERHUB
- docker push foundationdb/build:centos6-${DATE_STR}-${COMMIT_HASH}
- docker push foundationdb/build:centos6-latest
- # PUSH TO devel DOCKERHUB
- docker push foundationdb/devel:centos6-${DATE_STR}-${COMMIT_HASH}
- docker push foundationdb/devel:centos6-latest
- # PUSH TO distcc DOCKERHUB
- docker push foundationdb/distcc:centos6-${DATE_STR}-${COMMIT_HASH}
- docker push foundationdb/distcc:centos6-latest

8
build/docker/centos6/build/Dockerfile
View File

@ -37,13 +37,13 @@ RUN sed -i -e '/enabled/d' /etc/yum.repos.d/CentOS-Base.repo && \
lz4-devel \
lz4-static \
mono-devel \
rh-python36 \
rh-python36-python-devel \
rh-ruby24 \
rpm-build \
tcl-devel \
unzip \
wget && \
wget \
rh-python36 \
rh-python36-python-devel \
rh-ruby24 && \
yum clean all && \
rm -rf /var/cache/yum

60
build/docker/centos6/devel/Dockerfile
View File

@ -5,13 +5,16 @@ FROM ${REPOSITORY}:${VERSION}
# add vscode server
RUN yum repolist && \
yum -y install \
bash-completion \
byobu \
cgdb \
emacs-nox \
jq \
the_silver_searcher \
tmux \
tree \
emacs-nox \
vim \
bash-completion \
jq \
cgdb && \
zsh && \
yum clean all && \
rm -rf /var/cache/yum
@ -19,14 +22,25 @@ WORKDIR /tmp
RUN source /opt/rh/devtoolset-8/enable && \
source /opt/rh/rh-python36/enable && \
pip3 install \
lxml \
psutil \
python-dateutil \
subprocess32 \
psutil && \
subprocess32 && \
mkdir fdb-joshua && \
cd fdb-joshua && \
git clone --branch code_pipeline https://github.com/FoundationDB/fdb-joshua . && \
pip3 install /tmp/fdb-joshua && \
cd /tmp && \
curl -Ls https://amazon-eks.s3.us-west-2.amazonaws.com/1.18.9/2020-11-02/bin/linux/amd64/kubectl -o kubectl && \
echo "3dbe69e6deb35fbd6fec95b13d20ac1527544867ae56e3dae17e8c4d638b25b9 kubectl" > kubectl.txt && \
sha256sum -c kubectl.txt && \
mv kubectl /usr/local/bin/kubectl && \
chmod 755 /usr/local/bin/kubectl && \
curl https://awscli.amazonaws.com/awscli-exe-linux-x86_64-2.0.30.zip -o "awscliv2.zip" && \
echo "7ee475f22c1b35cc9e53affbf96a9ffce91706e154a9441d0d39cbf8366b718e awscliv2.zip" > awscliv2.txt && \
sha256sum -c awscliv2.txt && \
unzip -qq awscliv2.zip && \
./aws/install && \
rm -rf /tmp/*
ARG OLD_FDB_BINARY_DIR=/app/deploy/global_data/oldBinaries/
@ -45,17 +59,23 @@ RUN mkdir -p ${OLD_FDB_BINARY_DIR} \
ln -s ${OLD_TLS_LIBRARY_DIR}/FDBGnuTLS.so /usr/lib/foundationdb/plugins/FDBGnuTLS.so
WORKDIR /root
RUN echo -en "\n"\
"source /opt/rh/devtoolset-8/enable\n"\
"source /opt/rh/rh-python36/enable\n"\
"source /opt/rh/rh-ruby24/enable\n"\
"\n"\
"function cmk() {\n"\
" cmake -S ${HOME}/src/foundationdb -B build_output -D USE_CCACHE=1 -D RocksDB_ROOT=/opt/rocksdb-6.10.1 -G Ninja && ninja -C build_output -j 84\n"\
"}\n"\
"function ct() {\n"\
" cd ${HOME}/build_output && ctest -j 32 --output-on-failure\n"\
"}\n"\
"function j() {\n"\
" python3 -m joshua.joshua --cluster-file /etc/foundationdb/cluster-file \"\${@}\"\n"\
"}\n" >> .bashrc
RUN rm -f /root/anaconda-ks.cfg && \
printf '%s\n' \
'source /opt/rh/devtoolset-8/enable' \
'source /opt/rh/rh-python36/enable' \
'source /opt/rh/rh-ruby26/enable' \
'' \
'function cmk() {' \
' cmake -S ${HOME}/src/foundationdb -B ${HOME}/build_output -D USE_CCACHE=1 -D RocksDB_ROOT=/opt/rocksdb-6.10.1 -G Ninja && ninja -C build_output -j 84' \
'}' \
'function ct() {' \
' cd ${HOME}/build_output && ctest -j 32 --output-on-failure' \
'}' \
'function j() {' \
' python3 -m joshua.joshua "${@}"' \
'}' \
'function jsd() {' \
' j start --tarball $(find ${HOME}/build_output/packages -name correctness\*.tar.gz) "${@}"' \
'}' \
'' \
>> .bashrc

36
build/docker/centos7/build/Dockerfile
View File

@ -10,6 +10,7 @@ RUN rpmkeys --import mono-project.com.rpmkey.pgp && \
epel-release \
scl-utils \
yum-utils && \
yum-config-manager --add-repo https://download.docker.com/linux/centos/docker-ce.repo && \
yum install -y \
autoconf \
automake \
@ -19,6 +20,7 @@ RUN rpmkeys --import mono-project.com.rpmkey.pgp && \
devtoolset-8 \
devtoolset-8-libubsan-devel \
devtoolset-8-valgrind-devel \
docker-ce \
dos2unix \
dpkg \
gettext-devel \
@ -32,13 +34,21 @@ RUN rpmkeys --import mono-project.com.rpmkey.pgp && \
lz4-devel \
lz4-static \
mono-devel \
rh-python36 \
rh-python36-python-devel \
rh-ruby26 \
rpm-build \
tcl-devel \
unzip \
wget && \
if [ "$(uname -p)" == "aarch64" ]; then \
yum install -y \
rh-python38 \
rh-python38-python-devel \
rh-ruby27; \
else \
yum install -y \
rh-python36 \
rh-python36-python-devel \
rh-ruby26; \
fi && \
yum clean all && \
rm -rf /var/cache/yum
@ -51,9 +61,10 @@ RUN source /opt/rh/devtoolset-8/enable && \
tar --strip-components 1 --no-same-owner --directory git -xf git.tar.gz && \
cd git && \
make configure && \
./configure \
&& make && \
./configure && \
make && \
make install && \
cd ../ && \
rm -rf /tmp/*
# build/install ninja
@ -69,8 +80,13 @@ RUN source /opt/rh/devtoolset-8/enable && \
rm -rf /tmp/*
# install cmake
RUN curl -Ls https://github.com/Kitware/CMake/releases/download/v3.13.4/cmake-3.13.4-Linux-x86_64.tar.gz -o cmake.tar.gz && \
echo "563a39e0a7c7368f81bfa1c3aff8b590a0617cdfe51177ddc808f66cc0866c76 cmake.tar.gz" > cmake-sha.txt && \
RUN if [ "$(uname -p)" == "aarch64" ]; then \
curl -Ls https://github.com/Kitware/CMake/releases/download/v3.19.6/cmake-3.19.6-Linux-aarch64.tar.gz -o cmake.tar.gz; \
echo "69ec045c6993907a4f4a77349d0a0668f1bd3ce8bc5f6fbab6dc7a7e2ffc4f80 cmake.tar.gz" > cmake-sha.txt; \
else \
curl -Ls https://github.com/Kitware/CMake/releases/download/v3.13.4/cmake-3.13.4-Linux-x86_64.tar.gz -o cmake.tar.gz; \
echo "563a39e0a7c7368f81bfa1c3aff8b590a0617cdfe51177ddc808f66cc0866c76 cmake.tar.gz" > cmake-sha.txt; \
fi && \
sha256sum -c cmake-sha.txt && \
mkdir cmake && \
tar --strip-components 1 --no-same-owner --directory cmake -xf cmake.tar.gz && \
@ -185,7 +201,11 @@ RUN source /opt/rh/devtoolset-8/enable && \
# build/install distcc
RUN source /opt/rh/devtoolset-8/enable && \
source /opt/rh/rh-python36/enable && \
if [ "$(uname -p)" == "aarch64" ]; then \
source /opt/rh/rh-python38/enable; \
else \
source /opt/rh/rh-python36/enable; \
fi && \
curl -Ls https://github.com/distcc/distcc/archive/v3.3.5.tar.gz -o distcc.tar.gz && \
echo "13a4b3ce49dfc853a3de550f6ccac583413946b3a2fa778ddf503a9edc8059b0 distcc.tar.gz" > distcc-sha256.txt && \
sha256sum -c distcc-sha256.txt && \

87
build/docker/centos7/devel/Dockerfile
View File

@ -3,15 +3,21 @@ ARG VERSION=centos7-latest
FROM ${REPOSITORY}:${VERSION}
# add vscode server
RUN yum repolist && \
RUN yum-config-manager --add-repo=https://copr.fedorainfracloud.org/coprs/carlwgeorge/ripgrep/repo/epel-7/carlwgeorge-ripgrep-epel-7.repo && \
yum repolist && \
yum -y install \
bash-completion \
byobu \
cgdb \
emacs-nox \
fish \
jq \
ripgrep \
the_silver_searcher \
tmux \
tree \
emacs-nox \
vim \
bash-completion \
jq \
cgdb && \
zsh && \
yum clean all && \
rm -rf /var/cache/yum
@ -19,14 +25,25 @@ WORKDIR /tmp
RUN source /opt/rh/devtoolset-8/enable && \
source /opt/rh/rh-python36/enable && \
pip3 install \
lxml \
psutil \
python-dateutil \
subprocess32 \
psutil && \
subprocess32 && \
mkdir fdb-joshua && \
cd fdb-joshua && \
git clone --branch code_pipeline https://github.com/FoundationDB/fdb-joshua . && \
pip3 install /tmp/fdb-joshua && \
cd /tmp && \
curl -Ls https://amazon-eks.s3.us-west-2.amazonaws.com/1.18.9/2020-11-02/bin/linux/amd64/kubectl -o kubectl && \
echo "3dbe69e6deb35fbd6fec95b13d20ac1527544867ae56e3dae17e8c4d638b25b9 kubectl" > kubectl.txt && \
sha256sum -c kubectl.txt && \
mv kubectl /usr/local/bin/kubectl && \
chmod 755 /usr/local/bin/kubectl && \
curl https://awscli.amazonaws.com/awscli-exe-linux-x86_64-2.0.30.zip -o "awscliv2.zip" && \
echo "7ee475f22c1b35cc9e53affbf96a9ffce91706e154a9441d0d39cbf8366b718e awscliv2.zip" > awscliv2.txt && \
sha256sum -c awscliv2.txt && \
unzip -qq awscliv2.zip && \
./aws/install && \
rm -rf /tmp/*
ARG OLD_FDB_BINARY_DIR=/app/deploy/global_data/oldBinaries/
@ -49,18 +66,44 @@ RUN curl -Ls https://update.code.visualstudio.com/latest/server-linux-x64/stable
mkdir -p .vscode-server/bin/latest && \
tar --strip-components 1 --no-same-owner --directory .vscode-server/bin/latest -xf /tmp/vscode-server-linux-x64.tar.gz && \
touch .vscode-server/bin/latest/0 && \
rm /tmp/*
RUN echo -en "\n"\
"source /opt/rh/devtoolset-8/enable\n"\
"source /opt/rh/rh-python36/enable\n"\
"source /opt/rh/rh-ruby26/enable\n"\
"\n"\
"function cmk() {\n"\
" cmake -S ${HOME}/src/foundationdb -B build_output -D USE_CCACHE=1 -D RocksDB_ROOT=/opt/rocksdb-6.10.1 -G Ninja && ninja -C build_output -j 84\n"\
"}\n"\
"function ct() {\n"\
" cd ${HOME}/build_output && ctest -j 32 --output-on-failure\n"\
"}\n"\
"function j() {\n"\
" python3 -m joshua.joshua --cluster-file /etc/foundationdb/cluster-file \"\${@}\"\n"\
"}\n" >> .bashrc
rm -rf /tmp/*
RUN rm -f /root/anaconda-ks.cfg && \
printf '%s\n' \
'#!/usr/bin/env bash' \
'set -Eeuo pipefail' \
'' \
'mkdir -p ~/.docker' \
'cat > ~/.docker/config.json << EOF' \
'{' \
' "proxies":' \
' {' \
' "default":' \
' {' \
' "httpProxy": "${HTTP_PROXY}",' \
' "httpsProxy": "${HTTPS_PROXY}",' \
' "noProxy": "${NO_PROXY}"' \
' }' \
' }' \
'}' \
'EOF' \
> docker_proxy.sh && \
chmod 755 docker_proxy.sh && \
printf '%s\n' \
'source /opt/rh/devtoolset-8/enable' \
'source /opt/rh/rh-python36/enable' \
'source /opt/rh/rh-ruby26/enable' \
'' \
'function cmk() {' \
' cmake -S ${HOME}/src/foundationdb -B ${HOME}/build_output -D USE_CCACHE=1 -D RocksDB_ROOT=/opt/rocksdb-6.10.1 -G Ninja && ninja -C build_output -j 84' \
'}' \
'function ct() {' \
' cd ${HOME}/build_output && ctest -j 32 --output-on-failure' \
'}' \
'function j() {' \
' python3 -m joshua.joshua "${@}"' \
'}' \
'function jsd() {' \
' j start --tarball $(find ${HOME}/build_output/packages -name correctness\*.tar.gz) "${@}"' \
'}' \
'' \
>> .bashrc

20
build/docker/centos7/ycsb/Dockerfile Normal file
View File

@ -0,0 +1,20 @@
ARG REPOSITORY=foundationdb/build
ARG VERSION=centos7-latest
FROM ${REPOSITORY}:${VERSION}
ENV YCSB_VERSION=ycsb-foundationdb-binding-0.17.0 \
PATH=${PATH}:/usr/bin
RUN cd /opt \
&& eval curl "-Ls https://github.com/brianfrankcooper/YCSB/releases/download/0.17.0/ycsb-foundationdb-binding-0.17.0.tar.gz" \
| tar -xzvf -
RUN rm -Rf /opt/${YCSB_VERSION}/lib/fdb-java-5.2.5.jar
# COPY The Appropriate fdb-java-.jar Aaron from packages
# COPY binary RPM for foundationd-db
# Install Binary
WORKDIR "/opt/${YCSB_VERSION}"
ENTRYPOINT ["bin/ycsb.sh"]

2
cmake/Jemalloc.cmake
View File

@ -3,7 +3,7 @@ add_library(jemalloc INTERFACE)
set(USE_JEMALLOC ON)
# We don't want to use jemalloc on Windows
# Nor on FreeBSD, where jemalloc is the default system allocator
if(USE_SANITIZER OR WIN32 OR (CMAKE_SYSTEM_NAME STREQUAL "FreeBSD"))
if(USE_SANITIZER OR WIN32 OR (CMAKE_SYSTEM_NAME STREQUAL "FreeBSD") OR APPLE)
set(USE_JEMALLOC OFF)
return()
endif()

6
documentation/sphinx/source/api-c.rst
View File

@ -481,7 +481,11 @@ An |database-blurb1| Modifications to a database are performed via transactions.
|length-of| ``snapshot_command``
.. note:: The function is exposing the functionality of the fdbcli command ``snapshot``. Please take a look at the documentation before using (see :ref:`disk-snapshot-backups`).
.. function:: double fdb_database_get_main_thread_busyness(FDBDatabase* database)
Returns a value where 0 indicates that the client is idle and 1 (or larger) indicates that the client is saturated. By default, this value is updated every second.
Transaction
===========

1
documentation/sphinx/source/release-notes/release-notes-620.rst
View File

@ -7,6 +7,7 @@ Release Notes
* Fixed an issue where storage servers could shutdown with ``unknown_error``. `(PR #4437) <https://github.com/apple/foundationdb/pull/4437>`_
* Fix backup agent stall when writing to local filesystem with slow metadata operations. `(PR #4428) <https://github.com/apple/foundationdb/pull/4428>`_
* Backup agent no longer uses 4k block caching layer on local output files so that write operations are larger. `(PR #4428) <https://github.com/apple/foundationdb/pull/4428>`_
* Fix accounting error that could cause commits to incorrectly fail with ``proxy_memory_limit_exceeded``. `(PR #4529) <https://github.com/apple/foundationdb/pull/4529>`_
6.2.32
======

2
documentation/sphinx/source/release-notes/release-notes-630.rst
View File

@ -5,6 +5,8 @@ Release Notes
6.3.12
======
* Change the default for --knob_tls_server_handshake_threads to 64. The previous was 1000. This avoids starting 1000 threads by default, but may adversely affect recovery time for large clusters using tls. Users with large tls clusters should consider explicitly setting this knob in their foundationdb.conf file. `(PR #4421) <https://github.com/apple/foundationdb/pull/4421>`_
* Fix accounting error that could cause commits to incorrectly fail with ``proxy_memory_limit_exceeded``. `(PR #4526) <https://github.com/apple/foundationdb/pull/4526>`_
* As an optimization, partial restore using target key ranges now filters backup log data prior to loading it into the database. `(PR #4554) <https://github.com/apple/foundationdb/pull/4554>`_
6.3.11
======

166
fdbclient/FileBackupAgent.actor.cpp
View File

@ -3192,6 +3192,154 @@ struct RestoreRangeTaskFunc : RestoreFileTaskFuncBase {
StringRef RestoreRangeTaskFunc::name = LiteralStringRef("restore_range_data");
REGISTER_TASKFUNC(RestoreRangeTaskFunc);
// Decodes a mutation log key, which contains (hash, commitVersion, chunkNumber) and
// returns (commitVersion, chunkNumber)
std::pair<Version, int32_t> decodeLogKey(const StringRef& key) {
ASSERT(key.size() == sizeof(uint8_t) + sizeof(Version) + sizeof(int32_t));
uint8_t hash;
Version version;
int32_t part;
BinaryReader rd(key, Unversioned());
rd >> hash >> version >> part;
version = bigEndian64(version);
part = bigEndian32(part);
int32_t v = version / CLIENT_KNOBS->LOG_RANGE_BLOCK_SIZE;
ASSERT(((uint8_t)hashlittle(&v, sizeof(v), 0)) == hash);
return std::make_pair(version, part);
}
// Decodes an encoded list of mutations in the format of:
// [includeVersion:uint64_t][val_length:uint32_t][mutation_1][mutation_2]...[mutation_k],
// where a mutation is encoded as:
// [type:uint32_t][keyLength:uint32_t][valueLength:uint32_t][param1][param2]
std::vector<MutationRef> decodeLogValue(const StringRef& value) {
StringRefReader reader(value, restore_corrupted_data());
Version protocolVersion = reader.consume<uint64_t>();
if (protocolVersion <= 0x0FDB00A200090001) {
throw incompatible_protocol_version();
}
uint32_t val_length = reader.consume<uint32_t>();
if (val_length != value.size() - sizeof(uint64_t) - sizeof(uint32_t)) {
TraceEvent(SevError, "FileRestoreLogValueError")
.detail("ValueLen", val_length)
.detail("ValueSize", value.size())
.detail("Value", printable(value));
}
std::vector<MutationRef> mutations;
while (1) {
if (reader.eof())
break;
// Deserialization of a MutationRef, which was packed by MutationListRef::push_back_deep()
uint32_t type, p1len, p2len;
type = reader.consume<uint32_t>();
p1len = reader.consume<uint32_t>();
p2len = reader.consume<uint32_t>();
const uint8_t* key = reader.consume(p1len);
const uint8_t* val = reader.consume(p2len);
mutations.emplace_back((MutationRef::Type)type, StringRef(key, p1len), StringRef(val, p2len));
}
return mutations;
}
// Accumulates mutation log value chunks, as both a vector of chunks and as a combined chunk,
// in chunk order, and can check the chunk set for completion or intersection with a set
// of ranges.
struct AccumulatedMutations {
AccumulatedMutations() : lastChunkNumber(-1) {}
// Add a KV pair for this mutation chunk set
// It will be accumulated onto serializedMutations if the chunk number is
// the next expected value.
void addChunk(int chunkNumber, const KeyValueRef& kv) {
if (chunkNumber == lastChunkNumber + 1) {
lastChunkNumber = chunkNumber;
serializedMutations += kv.value.toString();
} else {
lastChunkNumber = -2;
serializedMutations.clear();
}
kvs.push_back(kv);
}
// Returns true if both
// - 1 or more chunks were added to this set
// - The header of the first chunk contains a valid protocol version and a length
// that matches the bytes after the header in the combined value in serializedMutations
bool isComplete() const {
if (lastChunkNumber >= 0) {
StringRefReader reader(serializedMutations, restore_corrupted_data());
Version protocolVersion = reader.consume<uint64_t>();
if (protocolVersion <= 0x0FDB00A200090001) {
throw incompatible_protocol_version();
}
uint32_t vLen = reader.consume<uint32_t>();
return vLen == reader.remainder().size();
}
return false;
}
// Returns true if a complete chunk contains any MutationRefs which intersect with any
// range in ranges.
// It is undefined behavior to run this if isComplete() does not return true.
bool matchesAnyRange(const std::vector<KeyRange>& ranges) const {
std::vector<MutationRef> mutations = decodeLogValue(serializedMutations);
for (auto& m : mutations) {
for (auto& r : ranges) {
if (m.type == MutationRef::ClearRange) {
if (r.intersects(KeyRangeRef(m.param1, m.param2))) {
return true;
}
} else {
if (r.contains(m.param1)) {
return true;
}
}
}
}
return false;
}
std::vector<KeyValueRef> kvs;
std::string serializedMutations;
int lastChunkNumber;
};
// Returns a vector of filtered KV refs from data which are either part of incomplete mutation groups OR complete
// and have data relevant to one of the KV ranges in ranges
std::vector<KeyValueRef> filterLogMutationKVPairs(VectorRef<KeyValueRef> data, const std::vector<KeyRange>& ranges) {
std::unordered_map<Version, AccumulatedMutations> mutationBlocksByVersion;
for (auto& kv : data) {
auto versionAndChunkNumber = decodeLogKey(kv.key);
mutationBlocksByVersion[versionAndChunkNumber.first].addChunk(versionAndChunkNumber.second, kv);
}
std::vector<KeyValueRef> output;
for (auto& vb : mutationBlocksByVersion) {
AccumulatedMutations& m = vb.second;
// If the mutations are incomplete or match one of the ranges, include in results.
if (!m.isComplete() || m.matchesAnyRange(ranges)) {
output.insert(output.end(), m.kvs.begin(), m.kvs.end());
}
}
return output;
}
struct RestoreLogDataTaskFunc : RestoreFileTaskFuncBase {
static StringRef name;
static constexpr uint32_t version = 1;
@ -3223,6 +3371,7 @@ struct RestoreLogDataTaskFunc : RestoreFileTaskFuncBase {
state Reference<ReadYourWritesTransaction> tr(new ReadYourWritesTransaction(cx));
state Reference<IBackupContainer> bc;
state std::vector<KeyRange> ranges;
loop {
try {
@ -3232,6 +3381,8 @@ struct RestoreLogDataTaskFunc : RestoreFileTaskFuncBase {
Reference<IBackupContainer> _bc = wait(restore.sourceContainer().getOrThrow(tr));
bc = _bc;
wait(store(ranges, restore.getRestoreRangesOrDefault(tr)));
wait(checkTaskVersion(tr->getDatabase(), task, name, version));
wait(taskBucket->keepRunning(tr, task));
@ -3243,10 +3394,14 @@ struct RestoreLogDataTaskFunc : RestoreFileTaskFuncBase {
state Key mutationLogPrefix = restore.mutationLogPrefix();
state Reference<IAsyncFile> inFile = wait(bc->readFile(logFile.fileName));
state Standalone<VectorRef<KeyValueRef>> data = wait(decodeLogFileBlock(inFile, readOffset, readLen));
state Standalone<VectorRef<KeyValueRef>> dataOriginal = wait(decodeLogFileBlock(inFile, readOffset, readLen));
// Filter the KV pairs extracted from the log file block to remove any records known to not be needed for this
// restore based on the restore range set.
state std::vector<KeyValueRef> dataFiltered = filterLogMutationKVPairs(dataOriginal, ranges);
state int start = 0;
state int end = data.size();
state int end = dataFiltered.size();
state int dataSizeLimit =
BUGGIFY ? deterministicRandom()->randomInt(256 * 1024, 10e6) : CLIENT_KNOBS->RESTORE_WRITE_TX_SIZE;
@ -3262,8 +3417,8 @@ struct RestoreLogDataTaskFunc : RestoreFileTaskFuncBase {
state int i = start;
state int txBytes = 0;
for (; i < end && txBytes < dataSizeLimit; ++i) {
Key k = data[i].key.withPrefix(mutationLogPrefix);
ValueRef v = data[i].value;
Key k = dataFiltered[i].key.withPrefix(mutationLogPrefix);
ValueRef v = dataFiltered[i].value;
tr->set(k, v);
txBytes += k.expectedSize();
txBytes += v.expectedSize();
@ -3291,7 +3446,8 @@ struct RestoreLogDataTaskFunc : RestoreFileTaskFuncBase {
.detail("CommitVersion", tr->getCommittedVersion())
.detail("StartIndex", start)
.detail("EndIndex", i)
.detail("DataSize", data.size())
.detail("RecordCountOriginal", dataOriginal.size())
.detail("RecordCountFiltered", dataFiltered.size())
.detail("Bytes", txBytes)
.detail("TaskInstance", THIS_ADDR);

1
fdbclient/IClientApi.h
View File

@ -96,6 +96,7 @@ public:
virtual Reference<ITransaction> createTransaction() = 0;
virtual void setOption(FDBDatabaseOptions::Option option, Optional<StringRef> value = Optional<StringRef>()) = 0;
virtual double getMainThreadBusyness() = 0;
virtual void addref() = 0;
virtual void delref() = 0;

1
fdbclient/Knobs.cpp
View File

@ -38,6 +38,7 @@ void ClientKnobs::initialize(bool randomize) {
init( TOO_MANY, 1000000 );
init( SYSTEM_MONITOR_INTERVAL, 5.0 );
init( NETWORK_BUSYNESS_MONITOR_INTERVAL, 1.0 );
init( FAILURE_MAX_DELAY, 5.0 );
init( FAILURE_MIN_DELAY, 4.0 ); if( randomize && BUGGIFY ) FAILURE_MIN_DELAY = 1.0;

1
fdbclient/Knobs.h
View File

@ -30,6 +30,7 @@ public:
int TOO_MANY; // FIXME: this should really be split up so we can control these more specifically
double SYSTEM_MONITOR_INTERVAL;
double NETWORK_BUSYNESS_MONITOR_INTERVAL; // The interval in which we should update the network busyness metric
double FAILURE_MAX_DELAY;
double FAILURE_MIN_DELAY;

24
fdbclient/MultiVersionTransaction.actor.cpp
View File

@ -347,6 +347,15 @@ ThreadFuture<Void> DLDatabase::createSnapshot(const StringRef& uid, const String
return toThreadFuture<Void>(api, f, [](FdbCApi::FDBFuture* f, FdbCApi* api) { return Void(); });
}
// Get network thread busyness
double DLDatabase::getMainThreadBusyness() {
if (api->databaseGetMainThreadBusyness != nullptr) {
return api->databaseGetMainThreadBusyness(db);
}
return 0;
}
// DLApi
template <class T>
void loadClientFunction(T* fp, void* lib, std::string libPath, const char* functionName, bool requireFunction = true) {
@ -360,6 +369,7 @@ void loadClientFunction(T* fp, void* lib, std::string libPath, const char* funct
DLApi::DLApi(std::string fdbCPath, bool unlinkOnLoad)
: api(new FdbCApi()), fdbCPath(fdbCPath), unlinkOnLoad(unlinkOnLoad), networkSetup(false) {}
// Loads client API functions (definitions are in FdbCApi struct)
void DLApi::init() {
if (isLibraryLoaded(fdbCPath.c_str())) {
throw external_client_already_loaded();
@ -388,6 +398,11 @@ void DLApi::init() {
loadClientFunction(&api->databaseCreateTransaction, lib, fdbCPath, "fdb_database_create_transaction");
loadClientFunction(&api->databaseSetOption, lib, fdbCPath, "fdb_database_set_option");
loadClientFunction(&api->databaseGetMainThreadBusyness,
lib,
fdbCPath,
"fdb_database_get_main_thread_busyness",
headerVersion >= 700);
loadClientFunction(&api->databaseDestroy, lib, fdbCPath, "fdb_database_destroy");
loadClientFunction(&api->databaseRebootWorker, lib, fdbCPath, "fdb_database_reboot_worker", headerVersion >= 700);
loadClientFunction(&api->databaseForceRecoveryWithDataLoss,
@ -917,6 +932,15 @@ ThreadFuture<Void> MultiVersionDatabase::createSnapshot(const StringRef& uid, co
return abortableFuture(f, dbState->dbVar->get().onChange);
}
// Get network thread busyness
double MultiVersionDatabase::getMainThreadBusyness() {
if (dbState->db) {
return dbState->db->getMainThreadBusyness();
}
return 0;
}
void MultiVersionDatabase::Connector::connect() {
addref();
onMainThreadVoid(

3
fdbclient/MultiVersionTransaction.h
View File

@ -80,6 +80,7 @@ struct FdbCApi : public ThreadSafeReferenceCounted<FdbCApi> {
int uidLength,
uint8_t const* snapshotCommmand,
int snapshotCommandLength);
double (*databaseGetMainThreadBusyness)(FDBDatabase* database);
// Transaction
fdb_error_t (*transactionSetOption)(FDBTransaction* tr,
@ -262,6 +263,7 @@ public:
Reference<ITransaction> createTransaction() override;
void setOption(FDBDatabaseOptions::Option option, Optional<StringRef> value = Optional<StringRef>()) override;
double getMainThreadBusyness() override;
void addref() override { ThreadSafeReferenceCounted<DLDatabase>::addref(); }
void delref() override { ThreadSafeReferenceCounted<DLDatabase>::delref(); }
@ -422,6 +424,7 @@ public:
Reference<ITransaction> createTransaction() override;
void setOption(FDBDatabaseOptions::Option option, Optional<StringRef> value = Optional<StringRef>()) override;
double getMainThreadBusyness() override;
void addref() override { ThreadSafeReferenceCounted<MultiVersionDatabase>::addref(); }
void delref() override { ThreadSafeReferenceCounted<MultiVersionDatabase>::delref(); }

26
fdbclient/NativeAPI.actor.cpp
View File

@ -1743,6 +1743,30 @@ void setNetworkOption(FDBNetworkOptions::Option option, Optional<StringRef> valu
}
}
// update the network busyness on a 1s cadence
ACTOR Future<Void> monitorNetworkBusyness() {
state double prevTime = now();
loop {
wait(delay(CLIENT_KNOBS->NETWORK_BUSYNESS_MONITOR_INTERVAL, TaskPriority::FlushTrace));
double elapsed = now() - prevTime; // get elapsed time from last execution
prevTime = now();
struct NetworkMetrics::PriorityStats& tracker = g_network->networkInfo.metrics.starvationTrackerNetworkBusyness;
if (tracker.active) { // update metrics
tracker.duration += now() - tracker.windowedTimer;
tracker.maxDuration = std::max(tracker.maxDuration, now() - tracker.timer);
tracker.windowedTimer = now();
}
g_network->networkInfo.metrics.networkBusyness =
std::min(elapsed, tracker.duration) / elapsed; // average duration spent doing "work"
tracker.duration = 0;
tracker.maxDuration = 0;
}
}
// Setup g_network and start monitoring for network busyness
void setupNetwork(uint64_t transportId, bool useMetrics) {
if (g_network)
throw network_already_setup();
@ -1756,6 +1780,8 @@ void setupNetwork(uint64_t transportId, bool useMetrics) {
g_network->addStopCallback(TLS::DestroyOpenSSLGlobalState);
FlowTransport::createInstance(true, transportId);
Net2FileSystem::newFileSystem();
uncancellable(monitorNetworkBusyness());
}
void runNetwork() {

17
fdbclient/ThreadSafeTransaction.cpp
View File

@ -91,6 +91,12 @@ ThreadFuture<Void> ThreadSafeDatabase::createSnapshot(const StringRef& uid, cons
return onMainThread([db, snapUID, cmd]() -> Future<Void> { return db->createSnapshot(snapUID, cmd); });
}
// Return the main network thread busyness
double ThreadSafeDatabase::getMainThreadBusyness() {
ASSERT(g_network);
return g_network->networkInfo.metrics.networkBusyness;
}
ThreadSafeDatabase::ThreadSafeDatabase(std::string connFilename, int apiVersion) {
ClusterConnectionFile* connFile =
new ClusterConnectionFile(ClusterConnectionFile::lookupClusterFileName(connFilename).first);
@ -401,11 +407,14 @@ const char* ThreadSafeApi::getClientVersion() {
return clientVersion.c_str();
}
// Wait until a quorum of coordinators with the same protocol version are available, and then return that protocol
// version.
ThreadFuture<uint64_t> ThreadSafeApi::getServerProtocol(const char* clusterFilePath) {
auto [clusterFile, isDefault] = ClusterConnectionFile::lookupClusterFileName(std::string(clusterFilePath));
Reference<ClusterConnectionFile> f = Reference<ClusterConnectionFile>(new ClusterConnectionFile(clusterFile));
return onMainThread([f]() -> Future<uint64_t> { return getCoordinatorProtocols(f); });
return onMainThread([clusterFilePath = std::string(clusterFilePath)]() -> Future<uint64_t> {
auto [clusterFile, isDefault] = ClusterConnectionFile::lookupClusterFileName(clusterFilePath);
Reference<ClusterConnectionFile> f = Reference<ClusterConnectionFile>(new ClusterConnectionFile(clusterFile));
return getCoordinatorProtocols(f);
});
}
void ThreadSafeApi::setNetworkOption(FDBNetworkOptions::Option option, Optional<StringRef> value) {

1
fdbclient/ThreadSafeTransaction.h
View File

@ -35,6 +35,7 @@ public:
Reference<ITransaction> createTransaction() override;
void setOption(FDBDatabaseOptions::Option option, Optional<StringRef> value = Optional<StringRef>()) override;
double getMainThreadBusyness() override;
ThreadFuture<Void>
onConnected(); // Returns after a majority of coordination servers are available and have reported a leader. The

31
fdbrpc/AsyncFileNonDurable.actor.h
View File

@ -145,8 +145,17 @@ private:
// The maximum amount of time a write is delayed before being passed along to the underlying file
double maxWriteDelay;
// Modifications which haven't been pushed to file, mapped by the location in the file that is being modified
// Modifications which haven't been pushed to file, mapped by the location in the file that is being modified.
// Be sure to update minSizeAfterPendingModifications when modifying pendingModifications.
RangeMap<uint64_t, Future<Void>> pendingModifications;
// The size of the file after the set of pendingModifications completes,
// (the set pending at the time of reading this member). Must be updated in
// lockstep with any inserts into the pendingModifications map. Tracking
// this variable is necessary so that we can know the range of the file a
// truncate is modifying, so we can insert it into the pendingModifications
// map. Until minSizeAfterPendingModificationsIsExact is true, this is only a lower bound.
mutable int64_t minSizeAfterPendingModifications = 0;
mutable bool minSizeAfterPendingModificationsIsExact = false;
// Will be blocked whenever kill is running
Promise<Void> killed;
@ -437,6 +446,7 @@ private:
Future<Void> writeEnded = wait(ownFuture);
std::vector<Future<Void>> priorModifications =
self->getModificationsAndInsert(offset, length, true, writeEnded);
self->minSizeAfterPendingModifications = std::max(self->minSizeAfterPendingModifications, offset + length);
if (BUGGIFY_WITH_PROB(0.001))
priorModifications.push_back(
@ -603,9 +613,19 @@ private:
//TraceEvent("AsyncFileNonDurable_Truncate", self->id).detail("Delay", delayDuration).detail("Filename", self->filename);
wait(checkKilled(self, "Truncate"));
Future<Void> truncateEnded = wait(ownFuture);
state Future<Void> truncateEnded = wait(ownFuture);
// Need to know the size of the file directly before this truncate
// takes effect to see what range it modifies.
if (!self->minSizeAfterPendingModificationsIsExact) {
wait(success(self->size()));
}
ASSERT(self->minSizeAfterPendingModificationsIsExact);
int64_t beginModifiedRange = std::min(size, self->minSizeAfterPendingModifications);
self->minSizeAfterPendingModifications = size;
std::vector<Future<Void>> priorModifications =
self->getModificationsAndInsert(size, -1, true, truncateEnded);
self->getModificationsAndInsert(beginModifiedRange, /*through end of file*/ -1, true, truncateEnded);
if (BUGGIFY_WITH_PROB(0.001))
priorModifications.push_back(
@ -751,8 +771,9 @@ private:
wait(checkKilled(self, "SizeEnd"));
// Include any modifications which extend past the end of the file
uint64_t maxModification = self->pendingModifications.lastItem().begin();
self->approximateSize = std::max<int64_t>(sizeFuture.get(), maxModification);
self->approximateSize = self->minSizeAfterPendingModifications =
std::max<int64_t>(sizeFuture.get(), self->minSizeAfterPendingModifications);
self->minSizeAfterPendingModificationsIsExact = true;
return self->approximateSize;
}

19
fdbrpc/IAsyncFile.actor.cpp
View File

@ -182,3 +182,22 @@ TEST_CASE("/fileio/rename") {
wait(IAsyncFileSystem::filesystem()->deleteFile(renamedFile, true));
return Void();
}
// Truncating to extend size should zero the new data
TEST_CASE("/fileio/truncateAndRead") {
state std::string filename = "/tmp/__JUNK__";
state Reference<IAsyncFile> f = wait(IAsyncFileSystem::filesystem()->open(
filename, IAsyncFile::OPEN_ATOMIC_WRITE_AND_CREATE | IAsyncFile::OPEN_CREATE | IAsyncFile::OPEN_READWRITE, 0));
state std::array<char, 4096> data;
wait(f->sync());
wait(f->truncate(4096));
int length = wait(f->read(&data[0], 4096, 0));
ASSERT(length == 4096);
for (auto c : data) {
ASSERT(c == '\0');
}
// close the file by deleting the reference
f.clear();
wait(IAsyncFileSystem::filesystem()->incrementalDeleteFile(filename, true));
return Void();
}

26
fdbrpc/sim2.actor.cpp
View File

@ -712,6 +712,7 @@ private:
return Void();
}
// Simulated sync does not actually do anything besides wait a random amount of time
ACTOR static Future<Void> sync_impl(SimpleFile* self) {
state UID opId = deterministicRandom()->randomUniqueID();
if (randLog)
@ -737,7 +738,6 @@ private:
.detail("FileCount", machineCache.count(self->filename));
renameFile(sourceFilename.c_str(), self->filename.c_str());
ASSERT(!machineCache.count(self->filename));
machineCache[self->filename] = machineCache[sourceFilename];
machineCache.erase(sourceFilename);
self->actualFilename = self->filename;
@ -2436,19 +2436,19 @@ Future<Reference<class IAsyncFile>> Sim2FileSystem::open(const std::string& file
if (flags & IAsyncFile::OPEN_UNCACHED) {
auto& machineCache = g_simulator.getCurrentProcess()->machine->openFiles;
std::string actualFilename = filename;
if (machineCache.find(filename) == machineCache.end()) {
if (flags & IAsyncFile::OPEN_ATOMIC_WRITE_AND_CREATE) {
actualFilename = filename + ".part";
auto partFile = machineCache.find(actualFilename);
if (partFile != machineCache.end()) {
Future<Reference<IAsyncFile>> f = AsyncFileDetachable::open(partFile->second);
if (FLOW_KNOBS->PAGE_WRITE_CHECKSUM_HISTORY > 0)
f = map(f, [=](Reference<IAsyncFile> r) {
return Reference<IAsyncFile>(new AsyncFileWriteChecker(r));
});
return f;
}
if (flags & IAsyncFile::OPEN_ATOMIC_WRITE_AND_CREATE) {
actualFilename = filename + ".part";
auto partFile = machineCache.find(actualFilename);
if (partFile != machineCache.end()) {
Future<Reference<IAsyncFile>> f = AsyncFileDetachable::open(partFile->second);
if (FLOW_KNOBS->PAGE_WRITE_CHECKSUM_HISTORY > 0)
f = map(f, [=](Reference<IAsyncFile> r) {
return Reference<IAsyncFile>(new AsyncFileWriteChecker(r));
});
return f;
}
}
if (machineCache.find(actualFilename) == machineCache.end()) {
// Simulated disk parameters are shared by the AsyncFileNonDurable and the underlying SimpleFile.
// This way, they can both keep up with the time to start the next operation
auto diskParameters =

123
fdbserver/ClusterController.actor.cpp
View File

@ -320,6 +320,17 @@ public:
return results;
}
// Selects workers as TLogs from available workers based on input parameters.
// conf: the database configuration.
// required: the required number of TLog workers to select.
// desired: the desired number of TLog workers to select.
// policy: the TLog replication policy the selection needs to satisfy.
// id_used: keep track of process IDs of selected workers.
// checkStable: when true, only select from workers that are considered as stable worker (not rebooted more than
// twice recently).
// dcIds: the target data centers the workers are in. The selected workers must all be from these
// data centers:
// exclusionWorkerIds: the workers to be excluded from the selection.
std::vector<WorkerDetails> getWorkersForTlogs(DatabaseConfiguration const& conf,
int32_t required,
int32_t desired,
@ -335,21 +346,67 @@ public:
LocalityMap<WorkerDetails>* logServerMap;
bool bCompleted = false;
// Construct the list of DCs where the TLog recruitment is happening. This is mainly for logging purpose.
std::string dcList;
for (const auto& dc : dcIds) {
if (!dcList.empty()) {
dcList += ',';
}
dcList += printable(dc);
}
logServerSet = Reference<LocalitySet>(new LocalityMap<WorkerDetails>());
logServerMap = (LocalityMap<WorkerDetails>*)logServerSet.getPtr();
for (auto& it : id_worker) {
if (std::find(exclusionWorkerIds.begin(), exclusionWorkerIds.end(), it.second.details.interf.id()) ==
// Populate `unavailableLocals` and log the reason why the worker is considered as unavailable.
auto logWorkerUnavailable = [this, &unavailableLocals, &dcList](const std::string& reason,
const WorkerDetails& details,
ProcessClass::Fitness fitness) {
unavailableLocals.push_back(details.interf.locality);
// Note that the recruitment happens only during initial database creation and recovery. So these trace
// events should be sparse.
TraceEvent("GetTLogTeamWorkerUnavailable", id)
.detail("Reason", reason)
.detail("WorkerID", details.interf.id())
.detail("WorkerDC", details.interf.locality.dcId())
.detail("Address", details.interf.addresses().toString())
.detail("Fitness", fitness)
.detail("RecruitmentDcIds", dcList);
};
// Go through all the workers to list all the workers that can be recruited.
for (const auto& [worker_process_id, worker_info] : id_worker) {
const auto& worker_details = worker_info.details;
auto fitness = worker_details.processClass.machineClassFitness(ProcessClass::TLog);
if (std::find(exclusionWorkerIds.begin(), exclusionWorkerIds.end(), worker_details.interf.id()) !=
exclusionWorkerIds.end()) {
auto fitness = it.second.details.processClass.machineClassFitness(ProcessClass::TLog);
if (workerAvailable(it.second, checkStable) &&
!conf.isExcludedServer(it.second.details.interf.addresses()) &&
fitness != ProcessClass::NeverAssign &&
(!dcIds.size() || dcIds.count(it.second.details.interf.locality.dcId()))) {
fitness_workers[std::make_pair(fitness, it.second.details.degraded)].push_back(it.second.details);
} else {
unavailableLocals.push_back(it.second.details.interf.locality);
}
logWorkerUnavailable("Worker is excluded", worker_details, fitness);
continue;
}
if (!workerAvailable(worker_info, checkStable)) {
logWorkerUnavailable("Worker is not available", worker_details, fitness);
continue;
}
if (conf.isExcludedServer(worker_details.interf.addresses())) {
logWorkerUnavailable("Worker server is excluded from the cluster", worker_details, fitness);
continue;
}
if (fitness == ProcessClass::NeverAssign) {
logWorkerUnavailable("Worker's fitness is NeverAssign", worker_details, fitness);
continue;
}
if (!dcIds.empty() && dcIds.count(worker_details.interf.locality.dcId()) == 0) {
logWorkerUnavailable("Worker is not in the target DC", worker_details, fitness);
continue;
}
// This worker is a candidate for TLog recruitment.
fitness_workers[std::make_pair(fitness, worker_details.degraded)].push_back(worker_details);
}
results.reserve(results.size() + id_worker.size());
@ -373,6 +430,7 @@ public:
break;
}
TraceEvent(SevWarn, "GWFTADNotAcceptable", id)
.detail("DcIds", dcList)
.detail("Fitness", fitness)
.detail("Processes", logServerSet->size())
.detail("Required", required)
@ -400,6 +458,7 @@ public:
tLocalities.push_back(object->interf.locality);
}
TraceEvent("GWFTADBestResults", id)
.detail("DcIds", dcList)
.detail("Fitness", fitness)
.detail("Processes", logServerSet->size())
.detail("BestCount", bestSet.size())
@ -413,6 +472,7 @@ public:
break;
}
TraceEvent(SevWarn, "GWFTADNoBest", id)
.detail("DcIds", dcList)
.detail("Fitness", fitness)
.detail("Processes", logServerSet->size())
.detail("Required", required)
@ -431,6 +491,7 @@ public:
}
TraceEvent(SevWarn, "GetTLogTeamFailed")
.detail("DcIds", dcList)
.detail("Policy", policy->info())
.detail("Processes", logServerSet->size())
.detail("Workers", id_worker.size())
@ -457,6 +518,7 @@ public:
}
TraceEvent("GetTLogTeamDone")
.detail("DcIds", dcList)
.detail("Completed", bCompleted)
.detail("Policy", policy->info())
.detail("Results", results.size())
@ -1259,6 +1321,9 @@ public:
}
// FIXME: determine when to fail the cluster controller when a primaryDC has not been set
// This function returns true when the cluster controller determines it is worth forcing
// a master recovery in order to change the recruited processes in the transaction subsystem.
bool betterMasterExists() {
const ServerDBInfo dbi = db.serverInfo->get();
@ -1426,13 +1491,15 @@ public:
bool oldSatelliteFallback = false;
for (auto& logSet : dbi.logSystemConfig.tLogs) {
if (region.satelliteTLogPolicy.isValid() && logSet.isLocal && logSet.locality == tagLocalitySatellite) {
oldSatelliteFallback = logSet.tLogPolicy->info() != region.satelliteTLogPolicy->info();
ASSERT(!oldSatelliteFallback ||
(region.satelliteTLogPolicyFallback.isValid() &&
logSet.tLogPolicy->info() == region.satelliteTLogPolicyFallback->info()));
break;
if (region.satelliteTLogPolicyFallback.isValid()) {
for (auto& logSet : dbi.logSystemConfig.tLogs) {
if (region.satelliteTLogPolicy.isValid() && logSet.isLocal && logSet.locality == tagLocalitySatellite) {
oldSatelliteFallback = logSet.tLogPolicy->info() != region.satelliteTLogPolicy->info();
ASSERT(!oldSatelliteFallback ||
(region.satelliteTLogPolicyFallback.isValid() &&
logSet.tLogPolicy->info() == region.satelliteTLogPolicyFallback->info()));
break;
}
}
}
@ -1567,12 +1634,26 @@ public:
return false;
}
// Because a configuration with fewer proxies or resolvers does not cause this function to fail,
// we need an extra check to determine if the total number of processes has been reduced.
// This is mainly helpful in avoiding situations where killing a degraded process
// would result in a configuration with less total processes than desired.
if (oldTLogFit.count + oldInFit.proxy.count + oldInFit.grvProxy.count + oldInFit.resolver.count >
newTLogFit.count + newInFit.proxy.count + newInFit.grvProxy.count + newInFit.resolver.count) {
return false;
}
// Check backup worker fitness
RoleFitness oldBackupWorkersFit(backup_workers, ProcessClass::Backup);
const int nBackup = backup_addresses.size();
RoleFitness newBackupWorkersFit(
getWorkersForRoleInDatacenter(clusterControllerDcId, ProcessClass::Backup, nBackup, db.config, id_used),
ProcessClass::Backup);
RoleFitness newBackupWorkersFit(getWorkersForRoleInDatacenter(clusterControllerDcId,
ProcessClass::Backup,
nBackup,
db.config,
id_used,
Optional<WorkerFitnessInfo>(),
true),
ProcessClass::Backup);
if (oldTLogFit > newTLogFit || oldInFit > newInFit || oldSatelliteTLogFit > newSatelliteTLogFit ||
oldRemoteTLogFit > newRemoteTLogFit || oldLogRoutersFit > newLogRoutersFit ||

4
fdbserver/CommitProxyServer.actor.cpp
View File

@ -1407,8 +1407,10 @@ ACTOR Future<Void> commitBatch(ProxyCommitData* self,
/////// Phase 1: Pre-resolution processing (CPU bound except waiting for a version # which is separately pipelined
/// and *should* be available by now (unless empty commit); ordered; currently atomic but could yield)
wait(CommitBatch::preresolutionProcessing(&context));
if (context.rejected)
if (context.rejected) {
self->commitBatchesMemBytesCount -= currentBatchMemBytesCount;
return Void();
}
/////// Phase 2: Resolution (waiting on the network; pipelined)
wait(CommitBatch::getResolution(&context));

69
fdbserver/DataDistribution.actor.cpp
View File

@ -658,6 +658,7 @@ struct DDTeamCollection : ReferenceCounted<DDTeamCollection> {
AsyncTrigger printDetailedTeamsInfo;
PromiseStream<GetMetricsRequest> getShardMetrics;
Promise<UID> removeFailedServer;
void resetLocalitySet() {
storageServerSet = Reference<LocalitySet>(new LocalityMap<UID>());
@ -695,7 +696,8 @@ struct DDTeamCollection : ReferenceCounted<DDTeamCollection> {
Reference<AsyncVar<bool>> zeroHealthyTeams,
bool primary,
Reference<AsyncVar<bool>> processingUnhealthy,
PromiseStream<GetMetricsRequest> getShardMetrics)
PromiseStream<GetMetricsRequest> getShardMetrics,
Promise<UID> removeFailedServer)
: cx(cx), distributorId(distributorId), lock(lock), output(output),
shardsAffectedByTeamFailure(shardsAffectedByTeamFailure), doBuildTeams(true), lastBuildTeamsFailed(false),
teamBuilder(Void()), badTeamRemover(Void()), checkInvalidLocalities(Void()), wrongStoreTypeRemover(Void()),
@ -710,7 +712,7 @@ struct DDTeamCollection : ReferenceCounted<DDTeamCollection> {
zeroHealthyTeams(zeroHealthyTeams), zeroOptimalTeams(true), primary(primary),
medianAvailableSpace(SERVER_KNOBS->MIN_AVAILABLE_SPACE_RATIO), lastMedianAvailableSpaceUpdate(0),
processingUnhealthy(processingUnhealthy), lowestUtilizationTeam(0), highestUtilizationTeam(0),
getShardMetrics(getShardMetrics) {
getShardMetrics(getShardMetrics), removeFailedServer(removeFailedServer) {
if (!primary || configuration.usableRegions == 1) {
TraceEvent("DDTrackerStarting", distributorId).detail("State", "Inactive").trackLatest("DDTrackerStarting");
}
@ -718,6 +720,13 @@ struct DDTeamCollection : ReferenceCounted<DDTeamCollection> {
~DDTeamCollection() {
TraceEvent("DDTeamCollectionDestructed", distributorId).detail("Primary", primary);
// Cancel the teamBuilder to avoid creating new teams after teams are cancelled.
teamBuilder.cancel();
// TraceEvent("DDTeamCollectionDestructed", distributorId)
// .detail("Primary", primary)
// .detail("TeamBuilderDestroyed", server_info.size());
// Other teamCollections also hold pointer to this teamCollection;
// TeamTracker may access the destructed DDTeamCollection if we do not reset the pointer
for (int i = 0; i < teamCollections.size(); i++) {
@ -754,12 +763,8 @@ struct DDTeamCollection : ReferenceCounted<DDTeamCollection> {
info->collection = nullptr;
}
// TraceEvent("DDTeamCollectionDestructed", distributorId)
// .detail("Primary", primary)
// .detail("ServerTrackerDestroyed", server_info.size());
teamBuilder.cancel();
// TraceEvent("DDTeamCollectionDestructed", distributorId)
// .detail("Primary", primary)
// .detail("TeamBuilderDestroyed", server_info.size());
// .detail("Primary", primary)
// .detail("ServerTrackerDestroyed", server_info.size());
}
void addLaggingStorageServer(Key zoneId) {
@ -4145,10 +4150,14 @@ ACTOR Future<Void> storageServerTracker(
TraceEvent(SevWarn, "FailedServerRemoveKeys", self->distributorId)
.detail("Server", server->id)
.detail("Excluded", worstAddr.toString());
wait(removeKeysFromFailedServer(cx, server->id, self->lock, ddEnabledState));
if (BUGGIFY)
wait(delay(5.0));
self->shardsAffectedByTeamFailure->eraseServer(server->id);
wait(delay(0.0)); //Do not throw an error while still inside trackExcludedServers
while (!ddEnabledState->isDDEnabled()) {
wait(delay(1.0));
}
if (self->removeFailedServer.canBeSet()) {
self->removeFailedServer.send(server->id);
}
throw movekeys_conflict();
}
}
@ -4944,6 +4953,7 @@ ACTOR Future<Void> monitorBatchLimitedTime(Reference<AsyncVar<ServerDBInfo>> db,
}
}
// Runs the data distribution algorithm for FDB, including the DD Queue, DD tracker, and DD team collection
ACTOR Future<Void> dataDistribution(Reference<DataDistributorData> self,
PromiseStream<GetMetricsListRequest> getShardMetricsList,
const DDEnabledState* ddEnabledState) {
@ -4973,7 +4983,7 @@ ACTOR Future<Void> dataDistribution(Reference<DataDistributorData> self,
// Stored outside of data distribution tracker to avoid slow tasks
// when tracker is cancelled
state KeyRangeMap<ShardTrackedData> shards;
state Promise<UID> removeFailedServer;
try {
loop {
TraceEvent("DDInitTakingMoveKeysLock", self->ddId);
@ -5204,7 +5214,8 @@ ACTOR Future<Void> dataDistribution(Reference<DataDistributorData> self,
zeroHealthyTeams[0],
true,
processingUnhealthy,
getShardMetrics);
getShardMetrics,
removeFailedServer);
teamCollectionsPtrs.push_back(primaryTeamCollection.getPtr());
if (configuration.usableRegions > 1) {
remoteTeamCollection =
@ -5220,7 +5231,8 @@ ACTOR Future<Void> dataDistribution(Reference<DataDistributorData> self,
zeroHealthyTeams[1],
false,
processingUnhealthy,
getShardMetrics);
getShardMetrics,
removeFailedServer);
teamCollectionsPtrs.push_back(remoteTeamCollection.getPtr());
remoteTeamCollection->teamCollections = teamCollectionsPtrs;
actors.push_back(
@ -5252,12 +5264,21 @@ ACTOR Future<Void> dataDistribution(Reference<DataDistributorData> self,
primaryTeamCollection = Reference<DDTeamCollection>();
remoteTeamCollection = Reference<DDTeamCollection>();
wait(shards.clearAsync());
if (err.code() != error_code_movekeys_conflict)
throw err;
bool ddEnabled = wait(isDataDistributionEnabled(cx, ddEnabledState));
TraceEvent("DataDistributionMoveKeysConflict").detail("DataDistributionEnabled", ddEnabled).error(err);
if (ddEnabled)
throw err;
TraceEvent("DataDistributorTeamCollectionsDestroyed").error(err);
if (removeFailedServer.getFuture().isReady() && !removeFailedServer.getFuture().isError()) {
TraceEvent("RemoveFailedServer", removeFailedServer.getFuture().get()).error(err);
wait(removeKeysFromFailedServer(cx, removeFailedServer.getFuture().get(), lock, ddEnabledState));
wait(removeStorageServer(cx, removeFailedServer.getFuture().get(), lock, ddEnabledState));
} else {
if (err.code() != error_code_movekeys_conflict) {
throw err;
}
bool ddEnabled = wait(isDataDistributionEnabled(cx, ddEnabledState));
TraceEvent("DataDistributionMoveKeysConflict").detail("DataDistributionEnabled", ddEnabled).error(err);
if (ddEnabled) {
throw err;
}
}
}
}
}
@ -5682,7 +5703,8 @@ std::unique_ptr<DDTeamCollection> testTeamCollection(int teamSize,
makeReference<AsyncVar<bool>>(true),
true,
makeReference<AsyncVar<bool>>(false),
PromiseStream<GetMetricsRequest>()));
PromiseStream<GetMetricsRequest>(),
Promise<UID>()));
for (int id = 1; id <= processCount; ++id) {
UID uid(id, 0);
@ -5723,7 +5745,8 @@ std::unique_ptr<DDTeamCollection> testMachineTeamCollection(int teamSize,
makeReference<AsyncVar<bool>>(true),
true,
makeReference<AsyncVar<bool>>(false),
PromiseStream<GetMetricsRequest>()));
PromiseStream<GetMetricsRequest>(),
Promise<UID>()));
for (int id = 1; id <= processCount; id++) {
UID uid(id, 0);

1
fdbserver/DataDistribution.actor.h
View File

@ -178,7 +178,6 @@ public:
void moveShard(KeyRangeRef keys, std::vector<Team> destinationTeam);
void finishMove(KeyRangeRef keys);
void check();
void eraseServer(UID ssID);
private:
struct OrderByTeamKey {

4
fdbserver/DataDistributionTracker.actor.cpp
View File

@ -999,10 +999,6 @@ void ShardsAffectedByTeamFailure::erase(Team team, KeyRange const& range) {
}
}
void ShardsAffectedByTeamFailure::eraseServer(UID ssID) {
storageServerShards[ssID] = 0;
}
void ShardsAffectedByTeamFailure::insert(Team team, KeyRange const& range) {
if (team_shards.insert(std::pair<Team, KeyRange>(team, range)).second) {
for (auto uid = team.servers.begin(); uid != team.servers.end(); ++uid)

1
fdbserver/Knobs.cpp
View File

@ -492,6 +492,7 @@ void ServerKnobs::initialize(bool randomize, ClientKnobs* clientKnobs, bool isSi
init( MAX_REBOOT_TIME, 5.0 ); if( longReboots ) MAX_REBOOT_TIME = 20.0;
init( LOG_DIRECTORY, "."); // Will be set to the command line flag.
init( SERVER_MEM_LIMIT, 8LL << 30 );
init( SYSTEM_MONITOR_FREQUENCY, 5.0 );
//Ratekeeper
bool slowRatekeeper = randomize && BUGGIFY;

1
fdbserver/Knobs.h
View File

@ -416,6 +416,7 @@ public:
double MAX_REBOOT_TIME;
std::string LOG_DIRECTORY;
int64_t SERVER_MEM_LIMIT;
double SYSTEM_MONITOR_FREQUENCY;
// Ratekeeper
double SMOOTHING_AMOUNT;

126
fdbserver/SimulatedCluster.actor.cpp
View File

@ -673,12 +673,15 @@ IPAddress makeIPAddressForSim(bool isIPv6, std::array<int, 4> parts) {
#include "fdbclient/MonitorLeader.h"
// Configures the system according to the given specifications in order to run
// simulation, but with the additional consideration that it is meant to act
// like a "rebooted" machine, mostly used for restarting tests.
ACTOR Future<Void> restartSimulatedSystem(vector<Future<Void>>* systemActors,
std::string baseFolder,
int* pTesterCount,
Optional<ClusterConnectionString>* pConnString,
Standalone<StringRef>* pStartingConfiguration,
int extraDB,
TestConfig testConfig,
std::string whitelistBinPaths,
ProtocolVersion protocolVersion) {
CSimpleIni ini;
@ -698,7 +701,7 @@ ACTOR Future<Void> restartSimulatedSystem(vector<Future<Void>>* systemActors,
}
int desiredCoordinators = atoi(ini.GetValue("META", "desiredCoordinators"));
int testerCount = atoi(ini.GetValue("META", "testerCount"));
bool enableExtraDB = (extraDB == 3);
bool enableExtraDB = (testConfig.extraDB == 3);
ClusterConnectionString conn(ini.GetValue("META", "connectionString"));
if (enableExtraDB) {
g_simulator.extraDB = new ClusterConnectionString(ini.GetValue("META", "connectionString"));
@ -836,8 +839,9 @@ ACTOR Future<Void> restartSimulatedSystem(vector<Future<Void>>* systemActors,
return Void();
}
// Configuration details compiled in a structure used when setting up a simulated cluster
struct SimulationConfig {
explicit SimulationConfig(int extraDB, int minimumReplication, int minimumRegions);
explicit SimulationConfig(const TestConfig& testConfig);
int extraDB;
DatabaseConfiguration db;
@ -851,11 +855,11 @@ struct SimulationConfig {
int coordinators;
private:
void generateNormalConfig(int minimumReplication, int minimumRegions);
void generateNormalConfig(const TestConfig& testConfig);
};
SimulationConfig::SimulationConfig(int extraDB, int minimumReplication, int minimumRegions) : extraDB(extraDB) {
generateNormalConfig(minimumReplication, minimumRegions);
SimulationConfig::SimulationConfig(const TestConfig& testConfig) : extraDB(testConfig.extraDB) {
generateNormalConfig(testConfig);
}
void SimulationConfig::set_config(std::string config) {
@ -871,18 +875,18 @@ StringRef StringRefOf(const char* s) {
return StringRef((uint8_t*)s, strlen(s));
}
void SimulationConfig::generateNormalConfig(int minimumReplication, int minimumRegions) {
void SimulationConfig::generateNormalConfig(const TestConfig& testConfig) {
set_config("new");
const bool simple = false; // Set true to simplify simulation configs for easier debugging
// generateMachineTeamTestConfig set up the number of servers per machine and the number of machines such that
// if we do not remove the surplus server and machine teams, the simulation test will report error.
// This is needed to make sure the number of server (and machine) teams is no larger than the desired number.
bool generateMachineTeamTestConfig = BUGGIFY_WITH_PROB(0.1) ? true : false;
bool generateFearless = simple ? false : (minimumRegions > 1 || deterministicRandom()->random01() < 0.5);
datacenters = simple
? 1
: (generateFearless ? (minimumReplication > 0 || deterministicRandom()->random01() < 0.5 ? 4 : 6)
: deterministicRandom()->randomInt(1, 4));
bool generateFearless = simple ? false : (testConfig.minimumRegions > 1 || deterministicRandom()->random01() < 0.5);
datacenters = simple ? 1
: (generateFearless
? (testConfig.minimumReplication > 0 || deterministicRandom()->random01() < 0.5 ? 4 : 6)
: deterministicRandom()->randomInt(1, 4));
if (deterministicRandom()->random01() < 0.25)
db.desiredTLogCount = deterministicRandom()->randomInt(1, 7);
if (deterministicRandom()->random01() < 0.25)
@ -892,6 +896,10 @@ void SimulationConfig::generateNormalConfig(int minimumReplication, int minimumR
if (deterministicRandom()->random01() < 0.25)
db.resolverCount = deterministicRandom()->randomInt(1, 7);
int storage_engine_type = deterministicRandom()->randomInt(0, 4);
// Continuously re-pick the storage engine type if it's the one we want to exclude
while (storage_engine_type == testConfig.storageEngineExcludeType) {
storage_engine_type = deterministicRandom()->randomInt(0, 4);
}
switch (storage_engine_type) {
case 0: {
TEST(true); // Simulated cluster using ssd storage engine
@ -930,7 +938,7 @@ void SimulationConfig::generateNormalConfig(int minimumReplication, int minimumR
db.resolverCount = 1;
}
int replication_type = simple ? 1
: (std::max(minimumReplication,
: (std::max(testConfig.minimumReplication,
datacenters > 4 ? deterministicRandom()->randomInt(1, 3)
: std::min(deterministicRandom()->randomInt(0, 6), 3)));
switch (replication_type) {
@ -1078,7 +1086,7 @@ void SimulationConfig::generateNormalConfig(int minimumReplication, int minimumR
// We cannot run with a remote DC when MAX_READ_TRANSACTION_LIFE_VERSIONS is too small, because the log
// routers will not be able to keep up.
if (minimumRegions <= 1 &&
if (testConfig.minimumRegions <= 1 &&
(deterministicRandom()->random01() < 0.25 ||
SERVER_KNOBS->MAX_READ_TRANSACTION_LIFE_VERSIONS < SERVER_KNOBS->VERSIONS_PER_SECOND)) {
TEST(true); // Simulated cluster using one region
@ -1124,7 +1132,7 @@ void SimulationConfig::generateNormalConfig(int minimumReplication, int minimumR
db.remoteDesiredTLogCount = deterministicRandom()->randomInt(1, 7);
bool useNormalDCsAsSatellites =
datacenters > 4 && minimumRegions < 2 && deterministicRandom()->random01() < 0.3;
datacenters > 4 && testConfig.minimumRegions < 2 && deterministicRandom()->random01() < 0.3;
StatusObject primarySatelliteObj;
primarySatelliteObj["id"] = useNormalDCsAsSatellites ? "1" : "2";
primarySatelliteObj["priority"] = 1;
@ -1191,7 +1199,7 @@ void SimulationConfig::generateNormalConfig(int minimumReplication, int minimumR
}
}
if (generateFearless && minimumReplication > 1) {
if (generateFearless && testConfig.minimumReplication > 1) {
// low latency tests in fearless configurations need 4 machines per datacenter (3 for triple replication, 1 that
// is down during failures).
machine_count = 16;
@ -1216,10 +1224,11 @@ void SimulationConfig::generateNormalConfig(int minimumReplication, int minimumR
// because we protect a majority of coordinators from being killed, it is better to run with low numbers of
// coordinators to prevent too many processes from being protected
coordinators =
(minimumRegions <= 1 && BUGGIFY) ? deterministicRandom()->randomInt(1, std::max(machine_count, 2)) : 1;
coordinators = (testConfig.minimumRegions <= 1 && BUGGIFY)
? deterministicRandom()->randomInt(1, std::max(machine_count, 2))
: 1;
if (minimumReplication > 1 && datacenters == 3) {
if (testConfig.minimumReplication > 1 && datacenters == 3) {
// low latency tests in 3 data hall mode need 2 other data centers with 2 machines each to avoid waiting for
// logs to recover.
machine_count = std::max(machine_count, 6);
@ -1233,26 +1242,24 @@ void SimulationConfig::generateNormalConfig(int minimumReplication, int minimumR
}
}
// Configures the system according to the given specifications in order to run
// simulation under the correct conditions
void setupSimulatedSystem(vector<Future<Void>>* systemActors,
std::string baseFolder,
int* pTesterCount,
Optional<ClusterConnectionString>* pConnString,
Standalone<StringRef>* pStartingConfiguration,
int extraDB,
int minimumReplication,
int minimumRegions,
std::string whitelistBinPaths,
bool configureLocked,
int logAntiQuorum,
TestConfig testConfig,
ProtocolVersion protocolVersion) {
// SOMEDAY: this does not test multi-interface configurations
SimulationConfig simconfig(extraDB, minimumReplication, minimumRegions);
if (logAntiQuorum != -1) {
simconfig.db.tLogWriteAntiQuorum = logAntiQuorum;
SimulationConfig simconfig(testConfig);
if (testConfig.logAntiQuorum != -1) {
simconfig.db.tLogWriteAntiQuorum = testConfig.logAntiQuorum;
}
StatusObject startingConfigJSON = simconfig.db.toJSON(true);
std::string startingConfigString = "new";
if (configureLocked) {
if (testConfig.configureLocked) {
startingConfigString += " locked";
}
for (auto kv : startingConfigJSON) {
@ -1338,7 +1345,7 @@ void setupSimulatedSystem(vector<Future<Void>>* systemActors,
TEST(!useIPv6); // Use IPv4
vector<NetworkAddress> coordinatorAddresses;
if (minimumRegions > 1) {
if (testConfig.minimumRegions > 1) {
// do not put coordinators in the primary region so that we can kill that region safely
int nonPrimaryDcs = dataCenters / 2;
for (int dc = 1; dc < dataCenters; dc += 2) {
@ -1409,14 +1416,14 @@ void setupSimulatedSystem(vector<Future<Void>>* systemActors,
ClusterConnectionString conn(coordinatorAddresses, LiteralStringRef("TestCluster:0"));
// If extraDB==0, leave g_simulator.extraDB as null because the test does not use DR.
if (extraDB == 1) {
if (testConfig.extraDB == 1) {
// The DR database can be either a new database or itself
g_simulator.extraDB = new ClusterConnectionString(
coordinatorAddresses, BUGGIFY ? LiteralStringRef("TestCluster:0") : LiteralStringRef("ExtraCluster:0"));
} else if (extraDB == 2) {
} else if (testConfig.extraDB == 2) {
// The DR database is a new database
g_simulator.extraDB = new ClusterConnectionString(coordinatorAddresses, LiteralStringRef("ExtraCluster:0"));
} else if (extraDB == 3) {
} else if (testConfig.extraDB == 3) {
// The DR database is the same database
g_simulator.extraDB = new ClusterConnectionString(coordinatorAddresses, LiteralStringRef("TestCluster:0"));
}
@ -1427,7 +1434,7 @@ void setupSimulatedSystem(vector<Future<Void>>* systemActors,
.detail("String", conn.toString())
.detail("ConfigString", startingConfigString);
bool requiresExtraDBMachines = extraDB && g_simulator.extraDB->toString() != conn.toString();
bool requiresExtraDBMachines = testConfig.extraDB && g_simulator.extraDB->toString() != conn.toString();
int assignedMachines = 0, nonVersatileMachines = 0;
std::vector<ProcessClass::ClassType> processClassesSubSet = { ProcessClass::UnsetClass,
ProcessClass::ResolutionClass,
@ -1602,13 +1609,8 @@ void setupSimulatedSystem(vector<Future<Void>>* systemActors,
.detail("StartingConfiguration", pStartingConfiguration->toString());
}
void checkTestConf(const char* testFile,
int& extraDB,
int& minimumReplication,
int& minimumRegions,
int& configureLocked,
int& logAntiQuorum,
bool& startIncompatibleProcess) {
// Populates the TestConfig fields according to what is found in the test file.
void checkTestConf(const char* testFile, TestConfig* testConfig) {
std::ifstream ifs;
ifs.open(testFile, std::ifstream::in);
if (!ifs.good())
@ -1630,26 +1632,31 @@ void checkTestConf(const char* testFile,
std::string value = removeWhitespace(line.substr(found + 1));
if (attrib == "extraDB") {
sscanf(value.c_str(), "%d", &extraDB);
sscanf(value.c_str(), "%d", &testConfig->extraDB);
}
if (attrib == "minimumReplication") {
sscanf(value.c_str(), "%d", &minimumReplication);
sscanf(value.c_str(), "%d", &testConfig->minimumReplication);
}
if (attrib == "minimumRegions") {
sscanf(value.c_str(), "%d", &minimumRegions);
sscanf(value.c_str(), "%d", &testConfig->minimumRegions);
}
if (attrib == "configureLocked") {
sscanf(value.c_str(), "%d", &configureLocked);
sscanf(value.c_str(), "%d", &testConfig->configureLocked);
}
if (attrib == "startIncompatibleProcess") {
startIncompatibleProcess = strcmp(value.c_str(), "true") == 0;
testConfig->startIncompatibleProcess = strcmp(value.c_str(), "true") == 0;
}
if (attrib == "logAntiQuorum") {
sscanf(value.c_str(), "%d", &logAntiQuorum);
sscanf(value.c_str(), "%d", &testConfig->logAntiQuorum);
}
if (attrib == "storageEngineExcludeType") {
sscanf(value.c_str(), "%d", &testConfig->storageEngineExcludeType);
}
}
@ -1665,24 +1672,13 @@ ACTOR void setupAndRun(std::string dataFolder,
state Optional<ClusterConnectionString> connFile;
state Standalone<StringRef> startingConfiguration;
state int testerCount = 1;
state int extraDB = 0;
state int minimumReplication = 0;
state int minimumRegions = 0;
state int configureLocked = 0;
state int logAntiQuorum = -1;
state bool startIncompatibleProcess = false;
checkTestConf(testFile,
extraDB,
minimumReplication,
minimumRegions,
configureLocked,
logAntiQuorum,
startIncompatibleProcess);
g_simulator.hasDiffProtocolProcess = startIncompatibleProcess;
state TestConfig testConfig;
checkTestConf(testFile, &testConfig);
g_simulator.hasDiffProtocolProcess = testConfig.startIncompatibleProcess;
g_simulator.setDiffProtocol = false;
state ProtocolVersion protocolVersion = currentProtocolVersion;
if (startIncompatibleProcess) {
if (testConfig.startIncompatibleProcess) {
// isolates right most 1 bit of compatibleProtocolVersionMask to make this protocolVersion incompatible
uint64_t minAddToMakeIncompatible =
ProtocolVersion::compatibleProtocolVersionMask & ~(ProtocolVersion::compatibleProtocolVersionMask - 1);
@ -1717,7 +1713,7 @@ ACTOR void setupAndRun(std::string dataFolder,
&testerCount,
&connFile,
&startingConfiguration,
extraDB,
testConfig,
whitelistBinPaths,
protocolVersion),
100.0));
@ -1732,12 +1728,8 @@ ACTOR void setupAndRun(std::string dataFolder,
&testerCount,
&connFile,
&startingConfiguration,
extraDB,
minimumReplication,
minimumRegions,
whitelistBinPaths,
configureLocked,
logAntiQuorum,
testConfig,
protocolVersion);
wait(delay(1.0)); // FIXME: WHY!!! //wait for machines to boot
}

8
fdbserver/Status.actor.cpp
View File

@ -2239,6 +2239,13 @@ static JsonBuilderObject tlogFetcher(int* logFaultTolerance,
int localSetsWithNonNegativeFaultTolerance = 0;
for (const auto& tLogSet : tLogs) {
if (tLogSet.tLogs.size() == 0) {
// We can have LogSets where there are no tLogs but some LogRouters. It's the way
// recruiting is implemented for old LogRouters in TagPartitionedLogSystem, where
// it adds an empty LogSet for missing locality.
continue;
}
int failedLogs = 0;
for (auto& log : tLogSet.tLogs) {
JsonBuilderObject logObj;
@ -2255,6 +2262,7 @@ static JsonBuilderObject tlogFetcher(int* logFaultTolerance,
}
if (tLogSet.isLocal) {
ASSERT_WE_THINK(tLogSet.tLogReplicationFactor > 0);
int currentFaultTolerance = tLogSet.tLogReplicationFactor - 1 - tLogSet.tLogWriteAntiQuorum - failedLogs;
if (currentFaultTolerance >= 0) {
localSetsWithNonNegativeFaultTolerance++;

18
fdbserver/TesterInterface.actor.h
View File

@ -99,6 +99,24 @@ struct WorkloadRequest {
}
};
// Configuration details specified in workload test files that change the simulation
// environment details
struct TestConfig {
int extraDB = 0;
int minimumReplication = 0;
int minimumRegions = 0;
int configureLocked = 0;
bool startIncompatibleProcess = false;
int logAntiQuorum = -1;
// Storage Engine Types: Verify match with SimulationConfig::generateNormalConfig
// -1 = None
// 0 = "ssd"
// 1 = "memory"
// 2 = "memory-radixtree-beta"
// 3 = "ssd-redwood-experimental"
int storageEngineExcludeType = -1;
};
struct TesterInterface {
constexpr static FileIdentifier file_identifier = 4465210;
RequestStream<WorkloadRequest> recruitments;

2
fdbserver/fdbserver.actor.cpp
View File

@ -464,7 +464,7 @@ Future<Void> startSystemMonitor(std::string dataFolder,
SystemMonitorMachineState(dataFolder, dcId, zoneId, machineId, g_network->getLocalAddress().ip));
systemMonitor();
return recurring(&systemMonitor, 5.0, TaskPriority::FlushTrace);
return recurring(&systemMonitor, SERVER_KNOBS->SYSTEM_MONITOR_FREQUENCY, TaskPriority::FlushTrace);
}
void testIndexedSet();

12
fdbserver/masterserver.actor.cpp
View File

@ -1406,22 +1406,26 @@ ACTOR Future<Void> rejoinRequestHandler(Reference<MasterData> self) {
}
}
// Keeps the coordinated state (cstate) updated as the set of recruited tlogs change through recovery.
ACTOR Future<Void> trackTlogRecovery(Reference<MasterData> self,
Reference<AsyncVar<Reference<ILogSystem>>> oldLogSystems,
Future<Void> minRecoveryDuration) {
state Future<Void> rejoinRequests = Never();
state DBRecoveryCount recoverCount = self->cstate.myDBState.recoveryCount + 1;
state DatabaseConfiguration configuration =
self->configuration; // self-configuration can be changed by configurationMonitor so we need a copy
loop {
state DBCoreState newState;
self->logSystem->toCoreState(newState);
newState.recoveryCount = recoverCount;
state Future<Void> changed = self->logSystem->onCoreStateChanged();
ASSERT(newState.tLogs[0].tLogWriteAntiQuorum == self->configuration.tLogWriteAntiQuorum &&
newState.tLogs[0].tLogReplicationFactor == self->configuration.tLogReplicationFactor);
ASSERT(newState.tLogs[0].tLogWriteAntiQuorum == configuration.tLogWriteAntiQuorum &&
newState.tLogs[0].tLogReplicationFactor == configuration.tLogReplicationFactor);
state bool allLogs =
newState.tLogs.size() ==
self->configuration.expectedLogSets(self->primaryDcId.size() ? self->primaryDcId[0] : Optional<Key>());
configuration.expectedLogSets(self->primaryDcId.size() ? self->primaryDcId[0] : Optional<Key>());
state bool finalUpdate = !newState.oldTLogData.size() && allLogs;
wait(self->cstate.write(newState, finalUpdate));
wait(minRecoveryDuration);
@ -1455,7 +1459,7 @@ ACTOR Future<Void> trackTlogRecovery(Reference<MasterData> self,
.trackLatest("MasterRecoveryState");
}
if (newState.oldTLogData.size() && self->configuration.repopulateRegionAntiQuorum > 0 &&
if (newState.oldTLogData.size() && configuration.repopulateRegionAntiQuorum > 0 &&
self->logSystem->remoteStorageRecovered()) {
TraceEvent(SevWarnAlways, "RecruitmentStalled_RemoteStorageRecovered", self->dbgid);
self->recruitmentStalled->set(true);

125
fdbserver/tester.actor.cpp
View File

@ -363,6 +363,64 @@ TestWorkload* getWorkloadIface(WorkloadRequest work, Reference<AsyncVar<ServerDB
return compound;
}
/**
* Only works in simulation. This method prints all simulated processes in a human readable form to stdout. It groups
* processes by data center, data hall, zone, and machine (in this order).
*/
void printSimulatedTopology() {
if (!g_network->isSimulated()) {
return;
}
auto processes = g_simulator.getAllProcesses();
std::sort(processes.begin(), processes.end(), [](ISimulator::ProcessInfo* lhs, ISimulator::ProcessInfo* rhs) {
auto l = lhs->locality;
auto r = rhs->locality;
if (l.dcId() != r.dcId()) {
return l.dcId() < r.dcId();
}
if (l.dataHallId() != r.dataHallId()) {
return l.dataHallId() < r.dataHallId();
}
if (l.zoneId() != r.zoneId()) {
return l.zoneId() < r.zoneId();
}
if (l.machineId() != r.zoneId()) {
return l.machineId() < r.machineId();
}
return lhs->address < rhs->address;
});
printf("Simulated Cluster Topology:\n");
printf("===========================\n");
Optional<Standalone<StringRef>> dcId, dataHallId, zoneId, machineId;
for (auto p : processes) {
std::string indent = "";
if (dcId != p->locality.dcId()) {
dcId = p->locality.dcId();
printf("%sdcId: %s\n", indent.c_str(), p->locality.describeDcId().c_str());
}
indent += " ";
if (dataHallId != p->locality.dataHallId()) {
dataHallId = p->locality.dataHallId();
printf("%sdataHallId: %s\n", indent.c_str(), p->locality.describeDataHall().c_str());
}
indent += " ";
if (zoneId != p->locality.zoneId()) {
zoneId = p->locality.zoneId();
printf("%szoneId: %s\n", indent.c_str(), p->locality.describeZone().c_str());
}
indent += " ";
if (machineId != p->locality.machineId()) {
machineId = p->locality.machineId();
printf("%smachineId: %s\n", indent.c_str(), p->locality.describeMachineId().c_str());
}
indent += " ";
printf("%sAddress: %s\n", indent.c_str(), p->address.toString().c_str(), p->name);
indent += " ";
printf("%sClass: %s\n", indent.c_str(), p->startingClass.toString().c_str());
printf("%sName: %s\n", indent.c_str(), p->name);
}
}
ACTOR Future<Void> databaseWarmer(Database cx) {
loop {
state Transaction tr(cx);
@ -977,7 +1035,9 @@ std::map<std::string, std::function<void(const std::string&)>> testSpecGlobalKey
TraceEvent("TestParserTest").detail("ClientInfoLogging", value);
} },
{ "startIncompatibleProcess",
[](const std::string& value) { TraceEvent("TestParserTest").detail("ParsedStartIncompatibleProcess", value); } }
[](const std::string& value) { TraceEvent("TestParserTest").detail("ParsedStartIncompatibleProcess", value); } },
{ "storageEngineExcludeType",
[](const std::string& value) { TraceEvent("TestParserTest").detail("ParsedStorageEngineExcludeType", ""); } }
};
std::map<std::string, std::function<void(const std::string& value, TestSpec* spec)>> testSpecTestKeys = {
@ -1291,6 +1351,24 @@ ACTOR Future<Void> monitorServerDBInfo(Reference<AsyncVar<Optional<ClusterContro
}
}
/**
* \brief Test orchestrator: sends test specification to testers in the right order and collects the results.
*
* There are multiple actors in this file with similar names (runTest, runTests) and slightly different signatures.
*
* This is the actual orchestrator. It reads the test specifications (from tests), prepares the cluster (by running the
* configure command given in startingConfiguration) and then runs the workload.
*
* \param cc The cluster controller interface
* \param ci Same as cc.clientInterface
* \param testers The interfaces of the testers that should run the actual workloads
* \param tests The test specifications to run
* \param startingConfiguration If non-empty, the orchestrator will attempt to set this configuration before starting
* the tests.
* \param locality client locality (it seems this is unused?)
*
* \returns A future which will be set after all tests finished.
*/
ACTOR Future<Void> runTests(Reference<AsyncVar<Optional<struct ClusterControllerFullInterface>>> cc,
Reference<AsyncVar<Optional<struct ClusterInterface>>> ci,
vector<TesterInterface> testers,
@ -1346,6 +1424,7 @@ ACTOR Future<Void> runTests(Reference<AsyncVar<Optional<struct ClusterController
// Change the configuration (and/or create the database) if necessary
printf("startingConfiguration:%s start\n", startingConfiguration.toString().c_str());
printSimulatedTopology();
if (useDB && startingConfiguration != StringRef()) {
try {
wait(timeoutError(changeConfiguration(cx, testers, startingConfiguration), 2000.0));
@ -1402,6 +1481,24 @@ ACTOR Future<Void> runTests(Reference<AsyncVar<Optional<struct ClusterController
return Void();
}
/**
* \brief Proxy function that waits until enough testers are available and then calls into the orchestrator.
*
* There are multiple actors in this file with similar names (runTest, runTests) and slightly different signatures.
*
* This actor wraps the actual orchestrator (also called runTests). But before calling that actor, it waits for enough
* testers to come up.
*
* \param cc The cluster controller interface
* \param ci Same as cc.clientInterface
* \param tests The test specifications to run
* \param minTestersExpected The number of testers to expect. This actor will block until it can find this many testers.
* \param startingConfiguration If non-empty, the orchestrator will attempt to set this configuration before starting
* the tests.
* \param locality client locality (it seems this is unused?)
*
* \returns A future which will be set after all tests finished.
*/
ACTOR Future<Void> runTests(Reference<AsyncVar<Optional<struct ClusterControllerFullInterface>>> cc,
Reference<AsyncVar<Optional<struct ClusterInterface>>> ci,
vector<TestSpec> tests,
@ -1443,6 +1540,32 @@ ACTOR Future<Void> runTests(Reference<AsyncVar<Optional<struct ClusterController
return Void();
}
/**
* \brief Set up testing environment and run the given tests on a cluster.
*
* There are multiple actors in this file with similar names (runTest, runTests) and slightly different signatures.
*
* This actor is usually the first entry point into the test environment. It itself doesn't implement too much
* functionality. Its main purpose is to generate the test specification from passed arguments and then call into the
* correct actor which will orchestrate the actual test.
*
* \param connFile A cluster connection file. Not all tests require a functional cluster but all tests require
* a cluster file.
* \param whatToRun TEST_TYPE_FROM_FILE to read the test description from a passed toml file or
* TEST_TYPE_CONSISTENCY_CHECK to generate a test spec for consistency checking
* \param at TEST_HERE: this process will act as a test client and execute the given workload. TEST_ON_SERVERS: Run a
* test client on every worker in the cluster. TEST_ON_TESTERS: Run a test client on all servers with class Test
* \param minTestersExpected In at is not TEST_HERE, this will instruct the orchestrator until it can find at least
* minTestersExpected test-clients. This is usually passed through from a command line argument. In simulation, the
* simulator will pass the number of testers that it started.
* \param fileName The path to the toml-file containing the test description. Is ignored if whatToRun !=
* TEST_TYPE_FROM_FILE
* \param startingConfiguration Can be used to configure a cluster before running the test. If this is an empty string,
* it will be ignored, otherwise it will be passed to changeConfiguration.
* \param locality The client locality to be used. This is only used if at == TEST_HERE
*
* \returns A future which will be set after all tests finished.
*/
ACTOR Future<Void> runTests(Reference<ClusterConnectionFile> connFile,
test_type_t whatToRun,
test_location_t at,

10
fdbserver/workloads/FastTriggeredWatches.actor.cpp
View File

@ -27,6 +27,7 @@
#include "flow/actorcompiler.h" // This must be the last #include.
struct FastTriggeredWatchesWorkload : TestWorkload {
// Tests the time it takes for a watch to be fired after the value has changed in the storage server
int nodes, keyBytes;
double testDuration;
vector<Future<Void>> clients;
@ -75,6 +76,7 @@ struct FastTriggeredWatchesWorkload : TestWorkload {
ACTOR Future<Version> setter(Database cx, Key key, Optional<Value> value) {
state ReadYourWritesTransaction tr(cx);
// set the value of key and return the commit version
wait(delay(deterministicRandom()->random01()));
loop {
try {
@ -105,22 +107,24 @@ struct FastTriggeredWatchesWorkload : TestWorkload {
state Optional<Value> setValue;
if (deterministicRandom()->random01() > 0.5)
setValue = StringRef(format("%010d", deterministicRandom()->randomInt(0, 1000)));
// Set the value at setKey to something random
state Future<Version> setFuture = self->setter(cx, setKey, setValue);
wait(delay(deterministicRandom()->random01()));
loop {
state ReadYourWritesTransaction tr(cx);
try {
Optional<Value> val = wait(tr.get(setKey));
if (!first) {
getDuration = now() - watchEnd;
}
lastReadVersion = tr.getReadVersion().get();
//TraceEvent("FTWGet").detail("Key", printable(setKey)).detail("Value", printable(val)).detail("Ver", tr.getReadVersion().get());
// if the value is already setValue then there is no point setting a watch so break out of the loop
if (val == setValue)
break;
ASSERT(first);
// set a watch and wait for it to be triggered (i.e for self->setter to set the value)
state Future<Void> watchFuture = tr.watch(setKey);
wait(tr.commit());
//TraceEvent("FTWStartWatch").detail("Key", printable(setKey));
@ -134,8 +138,10 @@ struct FastTriggeredWatchesWorkload : TestWorkload {
}
Version ver = wait(setFuture);
//TraceEvent("FTWWatchDone").detail("Key", printable(setKey));
// Assert that the time from setting the key to triggering the watch is no greater than 25s
// TODO: This assertion can cause flaky behaviour since sometimes a watch can take longer to fire
ASSERT(lastReadVersion - ver >= SERVER_KNOBS->MAX_VERSIONS_IN_FLIGHT ||
lastReadVersion - ver < SERVER_KNOBS->VERSIONS_PER_SECOND * (12 + getDuration));
lastReadVersion - ver < SERVER_KNOBS->VERSIONS_PER_SECOND * (25 + getDuration));
if (now() - testStart > self->testDuration)
break;

8
fdbserver/workloads/RemoveServersSafely.actor.cpp
View File

@ -502,6 +502,8 @@ struct RemoveServersSafelyWorkload : TestWorkload {
return killProcArray;
}
// Attempts to exclude a set of processes, and once the exclusion is successful it kills them.
// If markExcludeAsFailed is true, then it is an error if we cannot complete the exclusion.
ACTOR static Future<Void> removeAndKill(RemoveServersSafelyWorkload* self,
Database cx,
std::set<AddressExclusion> toKill,
@ -556,7 +558,11 @@ struct RemoveServersSafelyWorkload : TestWorkload {
.detail("Step", "SafetyCheck")
.detail("Exclusions", describe(toKillMarkFailedArray));
choose {
when(bool _safe = wait(checkSafeExclusions(cx, toKillMarkFailedArray))) { safe = _safe; }
when(bool _safe = wait(checkSafeExclusions(cx, toKillMarkFailedArray))) {
safe = _safe && self->protectServers(std::set<AddressExclusion>(toKillMarkFailedArray.begin(),
toKillMarkFailedArray.end()))
.size() == toKillMarkFailedArray.size();
}
when(wait(delay(5.0))) {
TraceEvent("RemoveAndKill", functionId)
.detail("Step", "SafetyCheckTimedOut")

49
flow/Net2.actor.cpp
View File

@ -135,6 +135,12 @@ thread_local INetwork* thread_network = 0;
class Net2 final : public INetwork, public INetworkConnections {
private:
void updateStarvationTracker(struct NetworkMetrics::PriorityStats& binStats,
TaskPriority priority,
TaskPriority lastPriority,
double now);
public:
Net2(const TLSConfig& tlsConfig, bool useThreadPool, bool useMetrics);
void initTLS(ETLSInitState targetState) override;
@ -221,7 +227,8 @@ public:
uint64_t tasksIssued;
TDMetricCollection tdmetrics;
double currentTime;
bool stopped;
// May be accessed off the network thread, e.g. by onMainThread
std::atomic<bool> stopped;
mutable std::map<IPAddress, bool> addressOnHostCache;
std::atomic<bool> started;
@ -1581,6 +1588,28 @@ void Net2::run() {
#endif
}
// Updates the PriorityStats found in NetworkMetrics
void Net2::updateStarvationTracker(struct NetworkMetrics::PriorityStats& binStats,
TaskPriority priority,
TaskPriority lastPriority,
double now) {
// Busy -> idle at binStats.priority
if (binStats.priority > priority && binStats.priority <= lastPriority) {
binStats.active = false;
binStats.duration += now - binStats.windowedTimer;
binStats.maxDuration = std::max(binStats.maxDuration, now - binStats.timer);
}
// Idle -> busy at binStats.priority
else if (binStats.priority <= priority && binStats.priority > lastPriority) {
binStats.active = true;
binStats.timer = now;
binStats.windowedTimer = now;
}
}
// Update both vectors of starvation trackers (one that updates every 5s and the other every 1s)
void Net2::trackAtPriority(TaskPriority priority, double now) {
if (lastPriorityStats == nullptr || priority != lastPriorityStats->priority) {
// Start tracking current priority
@ -1600,22 +1629,12 @@ void Net2::trackAtPriority(TaskPriority priority, double now) {
if (binStats.priority > lastPriority && binStats.priority > priority) {
break;
}
// Busy -> idle at binStats.priority
if (binStats.priority > priority && binStats.priority <= lastPriority) {
binStats.active = false;
binStats.duration += now - binStats.windowedTimer;
binStats.maxDuration = std::max(binStats.maxDuration, now - binStats.timer);
}
// Idle -> busy at binStats.priority
else if (binStats.priority <= priority && binStats.priority > lastPriority) {
binStats.active = true;
binStats.timer = now;
binStats.windowedTimer = now;
}
updateStarvationTracker(binStats, priority, lastPriority, now);
}
// Update starvation trackers for network busyness
updateStarvationTracker(networkInfo.metrics.starvationTrackerNetworkBusyness, priority, lastPriority, now);
lastPriorityStats = &activeStatsItr.first->second;
}
}

29
flow/network.h
View File

@ -27,6 +27,7 @@
#include <string>
#include <stdint.h>
#include <variant>
#include <atomic>
#include "boost/asio.hpp"
#ifndef TLS_DISABLED
#include "boost/asio/ssl.hpp"
@ -320,6 +321,7 @@ class Future;
template <class T>
class Promise;
// Metrics which represent various network properties
struct NetworkMetrics {
enum { SLOW_EVENT_BINS = 16 };
uint64_t countSlowEvents[SLOW_EVENT_BINS] = {};
@ -340,16 +342,37 @@ struct NetworkMetrics {
};
std::unordered_map<TaskPriority, struct PriorityStats> activeTrackers;
double lastRunLoopBusyness;
double lastRunLoopBusyness; // network thread busyness (measured every 5s by default)
std::atomic<double> networkBusyness; // network thread busyness which is returned to the the client (measured every 1s by default)
// starvation trackers which keeps track of different task priorities
std::vector<struct PriorityStats> starvationTrackers;
struct PriorityStats starvationTrackerNetworkBusyness;
static const std::vector<int> starvationBins;
NetworkMetrics() : lastRunLoopBusyness(0) {
for (int priority : starvationBins) {
NetworkMetrics()
: lastRunLoopBusyness(0), networkBusyness(0),
starvationTrackerNetworkBusyness(PriorityStats(static_cast<TaskPriority>(starvationBins.at(0)))) {
for (int priority : starvationBins) { // initalize starvation trackers with given priorities
starvationTrackers.emplace_back(static_cast<TaskPriority>(priority));
}
}
// Since networkBusyness is atomic we need to redefine copy assignment operator
NetworkMetrics& operator=(const NetworkMetrics& rhs) {
for (int i = 0; i < SLOW_EVENT_BINS; i++) {
countSlowEvents[i] = rhs.countSlowEvents[i];
}
secSquaredSubmit = rhs.secSquaredSubmit;
secSquaredDiskStall = rhs.secSquaredDiskStall;
activeTrackers = rhs.activeTrackers;
lastRunLoopBusyness = rhs.lastRunLoopBusyness;
networkBusyness = rhs.networkBusyness.load();
starvationTrackers = rhs.starvationTrackers;
starvationTrackerNetworkBusyness = rhs.starvationTrackerNetworkBusyness;
return *this;
}
};
struct FlowLock;

8
packaging/docker/samples/golang/app/Dockerfile
View File

@ -21,14 +21,14 @@
# see https://github.com/moby/moby/issues/34129
ARG FDB_VERSION
FROM foundationdb/foundationdb:${FDB_VERSION} as fdb
FROM golang:1.13.4-stretch
FROM golang:1.16.2-stretch
ARG FDB_VERSION
WORKDIR /tmp
RUN apt update
RUN apt-get update
# dnsutils is needed to have dig installed to create cluster file
RUN apt install -y dnsutils
RUN apt-get install -y --no-install-recommends ca-certificates dnsutils
ARG FDB_WEBSITE=https://foundationdb.org
RUN wget "${FDB_WEBSITE}/downloads/${FDB_VERSION}/ubuntu/installers/foundationdb-clients_${FDB_VERSION}-1_amd64.deb"
@ -44,4 +44,4 @@ COPY start.bash /start.bash
RUN go get -d -v ./...
RUN go install -v ./...
CMD ["/start.bash"]
CMD ["/start.bash"]

1
tests/restarting/to_6.3.10/CycleTestRestart-1.txt
View File

@ -1,3 +1,4 @@
storageEngineExcludeType=-1
testTitle=Clogged
clearAfterTest=false
testName=Cycle

1
tests/restarting/to_6.3.10/CycleTestRestart-2.txt
View File

@ -1,3 +1,4 @@
storageEngineExcludeType=-1
testTitle=Clogged
runSetup=false
testName=Cycle