Merge branch 'master' into fdb_cache_wo_allocator

This commit is contained in:
sfc-gh-ngoyal 2020-06-09 15:09:58 -07:00 committed by GitHub
commit 693d9e8b89
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
135 changed files with 3668 additions and 820 deletions

View File

@ -536,3 +536,51 @@ sse2neon Authors (sse2neon)
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
rte_memcpy.h (from DPDK):
SPDX-License-Identifier: BSD-3-Clause
Copyright(c) 2010-2014 Intel Corporation
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
1. Redistributions of source code must retain the above copyright notice,
this list of conditions and the following disclaimer.
2. Redistributions in binary form must reproduce the above copyright
notice, this list of conditions and the following disclaimer in the
documentation and/or other materials provided with the distribution.
3. Neither the name of the copyright holder nor the names of its
contributors may be used to endorse or promote products derived from this
software without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
POSSIBILITY OF SUCH DAMAGE.
folly_memcpy:
Copyright (c) Facebook, Inc. and its affiliates.
Author: Bin Liu <binliu@fb.com>
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

View File

@ -18,7 +18,7 @@
# limitations under the License.
cmake_minimum_required(VERSION 3.13)
project(foundationdb
VERSION 6.3.1
VERSION 7.0.0
DESCRIPTION "FoundationDB is a scalable, fault-tolerant, ordered key-value store with full ACID transactions."
HOMEPAGE_URL "http://www.foundationdb.org/"
LANGUAGES C CXX ASM)

View File

@ -26,7 +26,7 @@ sys.path[:0] = [os.path.join(os.path.dirname(__file__), '..', '..', 'bindings',
import util
FDB_API_VERSION = 630
FDB_API_VERSION = 700
LOGGING = {
'version': 1,

View File

@ -157,7 +157,7 @@ def choose_api_version(selected_api_version, tester_min_version, tester_max_vers
api_version = min_version
elif random.random() < 0.9:
api_version = random.choice([v for v in [13, 14, 16, 21, 22, 23, 100, 200, 300, 400, 410, 420, 430,
440, 450, 460, 500, 510, 520, 600, 610, 620, 630] if v >= min_version and v <= max_version])
440, 450, 460, 500, 510, 520, 600, 610, 620, 630, 700] if v >= min_version and v <= max_version])
else:
api_version = random.randint(min_version, max_version)

View File

@ -20,7 +20,7 @@
import os
MAX_API_VERSION = 630
MAX_API_VERSION = 700
COMMON_TYPES = ['null', 'bytes', 'string', 'int', 'uuid', 'bool', 'float', 'double', 'tuple']
ALL_TYPES = COMMON_TYPES + ['versionstamp']

View File

@ -34,7 +34,7 @@ fdb.api_version(FDB_API_VERSION)
class ScriptedTest(Test):
TEST_API_VERSION = 630
TEST_API_VERSION = 700
def __init__(self, subspace):
super(ScriptedTest, self).__init__(subspace, ScriptedTest.TEST_API_VERSION, ScriptedTest.TEST_API_VERSION)

View File

@ -18,7 +18,7 @@
* limitations under the License.
*/
#define FDB_API_VERSION 630
#define FDB_API_VERSION 700
#define FDB_INCLUDE_LEGACY_TYPES
#include "fdbclient/MultiVersionTransaction.h"

View File

@ -28,10 +28,10 @@
#endif
#if !defined(FDB_API_VERSION)
#error You must #define FDB_API_VERSION prior to including fdb_c.h (current version is 630)
#error You must #define FDB_API_VERSION prior to including fdb_c.h (current version is 700)
#elif FDB_API_VERSION < 13
#error API version no longer supported (upgrade to 13)
#elif FDB_API_VERSION > 630
#elif FDB_API_VERSION > 700
#error Requested API version requires a newer version of this header
#endif
@ -91,7 +91,7 @@ extern "C" {
DLLEXPORT WARN_UNUSED_RESULT fdb_error_t fdb_add_network_thread_completion_hook(void (*hook)(void*), void *hook_parameter);
#pragma pack(push, 4)
#if FDB_API_VERSION >= 630
#if FDB_API_VERSION >= 700
typedef struct keyvalue {
const uint8_t* key;
int key_length;

View File

@ -3,7 +3,7 @@
#pragma once
#ifndef FDB_API_VERSION
#define FDB_API_VERSION 630
#define FDB_API_VERSION 700
#endif
#include <foundationdb/fdb_c.h>

View File

@ -603,7 +603,7 @@ void runTests(struct ResultSet *rs) {
int main(int argc, char **argv) {
srand(time(NULL));
struct ResultSet *rs = newResultSet();
checkError(fdb_select_api_version(630), "select API version", rs);
checkError(fdb_select_api_version(700), "select API version", rs);
printf("Running performance test at client version: %s\n", fdb_get_client_version());
valueStr = (uint8_t*)malloc((sizeof(uint8_t))*valueSize);

View File

@ -244,7 +244,7 @@ void runTests(struct ResultSet *rs) {
int main(int argc, char **argv) {
srand(time(NULL));
struct ResultSet *rs = newResultSet();
checkError(fdb_select_api_version(630), "select API version", rs);
checkError(fdb_select_api_version(700), "select API version", rs);
printf("Running RYW Benchmark test at client version: %s\n", fdb_get_client_version());
keys = generateKeys(numKeys, keySize);

View File

@ -29,7 +29,7 @@
#include <inttypes.h>
#ifndef FDB_API_VERSION
#define FDB_API_VERSION 630
#define FDB_API_VERSION 700
#endif
#include <foundationdb/fdb_c.h>

View File

@ -97,7 +97,7 @@ void runTests(struct ResultSet *rs) {
int main(int argc, char **argv) {
srand(time(NULL));
struct ResultSet *rs = newResultSet();
checkError(fdb_select_api_version(630), "select API version", rs);
checkError(fdb_select_api_version(700), "select API version", rs);
printf("Running performance test at client version: %s\n", fdb_get_client_version());
keys = generateKeys(numKeys, KEY_SIZE);

View File

@ -18,7 +18,7 @@
* limitations under the License.
*/
#define FDB_API_VERSION 630
#define FDB_API_VERSION 700
#include "foundationdb/fdb_c.h"
#undef DLLEXPORT
#include "workloads.h"
@ -258,7 +258,7 @@ struct SimpleWorkload : FDBWorkload {
insertsPerTx = context->getOption("insertsPerTx", 100ul);
opsPerTx = context->getOption("opsPerTx", 100ul);
runFor = context->getOption("runFor", 10.0);
auto err = fdb_select_api_version(630);
auto err = fdb_select_api_version(700);
if (err) {
context->trace(FDBSeverity::Info, "SelectAPIVersionFailed",
{ { "Error", std::string(fdb_get_error(err)) } });

View File

@ -36,7 +36,7 @@ THREAD_FUNC networkThread(void* fdb) {
}
ACTOR Future<Void> _test() {
API *fdb = FDB::API::selectAPIVersion(630);
API *fdb = FDB::API::selectAPIVersion(700);
auto db = fdb->createDatabase();
state Reference<Transaction> tr = db->createTransaction();
@ -79,7 +79,7 @@ ACTOR Future<Void> _test() {
}
void fdb_flow_test() {
API *fdb = FDB::API::selectAPIVersion(630);
API *fdb = FDB::API::selectAPIVersion(700);
fdb->setupNetwork();
startThread(networkThread, fdb);

View File

@ -23,7 +23,7 @@
#include <flow/flow.h>
#define FDB_API_VERSION 630
#define FDB_API_VERSION 700
#include <bindings/c/foundationdb/fdb_c.h>
#undef DLLEXPORT

View File

@ -1817,7 +1817,7 @@ ACTOR void _test_versionstamp() {
try {
g_network = newNet2(TLSConfig());
API *fdb = FDB::API::selectAPIVersion(630);
API *fdb = FDB::API::selectAPIVersion(700);
fdb->setupNetwork();
startThread(networkThread, fdb);

View File

@ -9,7 +9,7 @@ This package requires:
- [Mono](http://www.mono-project.com/) (macOS or Linux) or [Visual Studio](https://www.visualstudio.com/) (Windows) (build-time only)
- FoundationDB C API 2.0.x-6.1.x (part of the [FoundationDB client packages](https://apple.github.io/foundationdb/downloads.html#c))
Use of this package requires the selection of a FoundationDB API version at runtime. This package currently supports FoundationDB API versions 200-630.
Use of this package requires the selection of a FoundationDB API version at runtime. This package currently supports FoundationDB API versions 200-700.
To install this package, you can run the "fdb-go-install.sh" script (for versions 5.0.x and greater):

View File

@ -22,7 +22,7 @@
package fdb
// #define FDB_API_VERSION 630
// #define FDB_API_VERSION 700
// #include <foundationdb/fdb_c.h>
import "C"

View File

@ -22,7 +22,7 @@
package fdb
// #define FDB_API_VERSION 630
// #define FDB_API_VERSION 700
// #include <foundationdb/fdb_c.h>
import "C"

View File

@ -46,7 +46,7 @@ A basic interaction with the FoundationDB API is demonstrated below:
func main() {
// Different API versions may expose different runtime behaviors.
fdb.MustAPIVersion(630)
fdb.MustAPIVersion(700)
// Open the default database from the system cluster
db := fdb.MustOpenDefault()

View File

@ -22,7 +22,7 @@
package fdb
// #define FDB_API_VERSION 630
// #define FDB_API_VERSION 700
// #include <foundationdb/fdb_c.h>
import "C"

View File

@ -22,7 +22,7 @@
package fdb
// #define FDB_API_VERSION 630
// #define FDB_API_VERSION 700
// #include <foundationdb/fdb_c.h>
// #include <stdlib.h>
import "C"
@ -108,7 +108,7 @@ func (opt NetworkOptions) setOpt(code int, param []byte) error {
// library, an error will be returned. APIVersion must be called prior to any
// other functions in the fdb package.
//
// Currently, this package supports API versions 200 through 630.
// Currently, this package supports API versions 200 through 700.
//
// Warning: When using the multi-version client API, setting an API version that
// is not supported by a particular client library will prevent that client from
@ -116,7 +116,7 @@ func (opt NetworkOptions) setOpt(code int, param []byte) error {
// the API version of your application after upgrading your client until the
// cluster has also been upgraded.
func APIVersion(version int) error {
headerVersion := 630
headerVersion := 700
networkMutex.Lock()
defer networkMutex.Unlock()
@ -128,7 +128,7 @@ func APIVersion(version int) error {
return errAPIVersionAlreadySet
}
if version < 200 || version > 630 {
if version < 200 || version > 700 {
return errAPIVersionNotSupported
}

View File

@ -32,7 +32,7 @@ import (
func ExampleOpenDefault() {
var e error
e = fdb.APIVersion(630)
e = fdb.APIVersion(700)
if e != nil {
fmt.Printf("Unable to set API version: %v\n", e)
return
@ -52,7 +52,7 @@ func ExampleOpenDefault() {
}
func TestVersionstamp(t *testing.T) {
fdb.MustAPIVersion(630)
fdb.MustAPIVersion(700)
db := fdb.MustOpenDefault()
setVs := func(t fdb.Transactor, key fdb.Key) (fdb.FutureKey, error) {
@ -98,7 +98,7 @@ func TestVersionstamp(t *testing.T) {
}
func ExampleTransactor() {
fdb.MustAPIVersion(630)
fdb.MustAPIVersion(700)
db := fdb.MustOpenDefault()
setOne := func(t fdb.Transactor, key fdb.Key, value []byte) error {
@ -149,7 +149,7 @@ func ExampleTransactor() {
}
func ExampleReadTransactor() {
fdb.MustAPIVersion(630)
fdb.MustAPIVersion(700)
db := fdb.MustOpenDefault()
getOne := func(rt fdb.ReadTransactor, key fdb.Key) ([]byte, error) {
@ -202,7 +202,7 @@ func ExampleReadTransactor() {
}
func ExamplePrefixRange() {
fdb.MustAPIVersion(630)
fdb.MustAPIVersion(700)
db := fdb.MustOpenDefault()
tr, e := db.CreateTransaction()
@ -241,7 +241,7 @@ func ExamplePrefixRange() {
}
func ExampleRangeIterator() {
fdb.MustAPIVersion(630)
fdb.MustAPIVersion(700)
db := fdb.MustOpenDefault()
tr, e := db.CreateTransaction()

View File

@ -23,7 +23,7 @@
package fdb
// #cgo LDFLAGS: -lfdb_c -lm
// #define FDB_API_VERSION 630
// #define FDB_API_VERSION 700
// #include <foundationdb/fdb_c.h>
// #include <string.h>
//

View File

@ -22,7 +22,7 @@
package fdb
// #define FDB_API_VERSION 630
// #define FDB_API_VERSION 700
// #include <foundationdb/fdb_c.h>
import "C"

View File

@ -22,7 +22,7 @@
package fdb
// #define FDB_API_VERSION 630
// #define FDB_API_VERSION 700
// #include <foundationdb/fdb_c.h>
import "C"

View File

@ -19,7 +19,7 @@
*/
#include <foundationdb/ClientWorkload.h>
#define FDB_API_VERSION 630
#define FDB_API_VERSION 700
#include <foundationdb/fdb_c.h>
#include <jni.h>
@ -370,7 +370,7 @@ struct JVM {
jmethodID selectMethod =
env->GetStaticMethodID(fdbClass, "selectAPIVersion", "(I)Lcom/apple/foundationdb/FDB;");
checkException();
auto fdbInstance = env->CallStaticObjectMethod(fdbClass, selectMethod, jint(630));
auto fdbInstance = env->CallStaticObjectMethod(fdbClass, selectMethod, jint(700));
checkException();
env->CallObjectMethod(fdbInstance, getMethod(fdbClass, "disableShutdownHook", "()V"));
checkException();

View File

@ -21,7 +21,7 @@
#include <jni.h>
#include <string.h>
#define FDB_API_VERSION 630
#define FDB_API_VERSION 700
#include <foundationdb/fdb_c.h>

View File

@ -35,7 +35,7 @@ import java.util.concurrent.atomic.AtomicInteger;
* This call is required before using any other part of the API. The call allows
* an error to be thrown at this point to prevent client code from accessing a later library
* with incorrect assumptions from the current version. The API version documented here is version
* {@code 630}.<br><br>
* {@code 700}.<br><br>
* FoundationDB encapsulates multiple versions of its interface by requiring
* the client to explicitly specify the version of the API it uses. The purpose
* of this design is to allow you to upgrade the server, client libraries, or
@ -181,8 +181,8 @@ public class FDB {
}
if(version < 510)
throw new IllegalArgumentException("API version not supported (minimum 510)");
if(version > 630)
throw new IllegalArgumentException("API version not supported (maximum 630)");
if(version > 700)
throw new IllegalArgumentException("API version not supported (maximum 700)");
Select_API_version(version);
singleton = new FDB(version);

View File

@ -13,7 +13,7 @@ and then added to your classpath.<br>
<h1>Getting started</h1>
To start using FoundationDB from Java, create an instance of the
{@link com.apple.foundationdb.FDB FoundationDB API interface} with the version of the
API that you want to use (this release of the FoundationDB Java API supports versions between {@code 510} and {@code 630}).
API that you want to use (this release of the FoundationDB Java API supports versions between {@code 510} and {@code 700}).
With this API object you can then open {@link com.apple.foundationdb.Cluster Cluster}s and
{@link com.apple.foundationdb.Database Database}s and start using
{@link com.apple.foundationdb.Transaction Transaction}s.
@ -29,7 +29,7 @@ import com.apple.foundationdb.tuple.Tuple;
public class Example {
public static void main(String[] args) {
FDB fdb = FDB.selectAPIVersion(630);
FDB fdb = FDB.selectAPIVersion(700);
try(Database db = fdb.open()) {
// Run an operation on the database

View File

@ -27,7 +27,7 @@ import com.apple.foundationdb.Database;
import com.apple.foundationdb.FDB;
public abstract class AbstractTester {
public static final int API_VERSION = 630;
public static final int API_VERSION = 700;
protected static final int NUM_RUNS = 25;
protected static final Charset ASCII = Charset.forName("ASCII");

View File

@ -33,7 +33,7 @@ public class BlockingBenchmark {
private static final int PARALLEL = 100;
public static void main(String[] args) throws InterruptedException {
FDB fdb = FDB.selectAPIVersion(630);
FDB fdb = FDB.selectAPIVersion(700);
// The cluster file DOES NOT need to be valid, although it must exist.
// This is because the database is never really contacted in this test.

View File

@ -48,7 +48,7 @@ public class ConcurrentGetSetGet {
}
public static void main(String[] args) {
try(Database database = FDB.selectAPIVersion(630).open()) {
try(Database database = FDB.selectAPIVersion(700).open()) {
new ConcurrentGetSetGet().apply(database);
}
}

View File

@ -33,7 +33,7 @@ import com.apple.foundationdb.directory.DirectorySubspace;
public class DirectoryTest {
public static void main(String[] args) throws Exception {
try {
FDB fdb = FDB.selectAPIVersion(630);
FDB fdb = FDB.selectAPIVersion(700);
try(Database db = fdb.open()) {
runTests(db);
}

View File

@ -26,7 +26,7 @@ import com.apple.foundationdb.tuple.Tuple;
public class Example {
public static void main(String[] args) {
FDB fdb = FDB.selectAPIVersion(630);
FDB fdb = FDB.selectAPIVersion(700);
try(Database db = fdb.open()) {
// Run an operation on the database

View File

@ -31,7 +31,7 @@ public class IterableTest {
public static void main(String[] args) throws InterruptedException {
final int reps = 1000;
try {
FDB fdb = FDB.selectAPIVersion(630);
FDB fdb = FDB.selectAPIVersion(700);
try(Database db = fdb.open()) {
runTests(reps, db);
}

View File

@ -34,7 +34,7 @@ import com.apple.foundationdb.tuple.ByteArrayUtil;
public class LocalityTests {
public static void main(String[] args) {
FDB fdb = FDB.selectAPIVersion(630);
FDB fdb = FDB.selectAPIVersion(700);
try(Database database = fdb.open(args[0])) {
try(Transaction tr = database.createTransaction()) {
String[] keyAddresses = LocalityUtil.getAddressesForKey(tr, "a".getBytes()).join();

View File

@ -43,7 +43,7 @@ public class ParallelRandomScan {
private static final int PARALLELISM_STEP = 5;
public static void main(String[] args) throws InterruptedException {
FDB api = FDB.selectAPIVersion(630);
FDB api = FDB.selectAPIVersion(700);
try(Database database = api.open(args[0])) {
for(int i = PARALLELISM_MIN; i <= PARALLELISM_MAX; i += PARALLELISM_STEP) {
runTest(database, i, ROWS, DURATION_MS);

View File

@ -34,7 +34,7 @@ import com.apple.foundationdb.Transaction;
import com.apple.foundationdb.async.AsyncIterable;
public class RangeTest {
private static final int API_VERSION = 630;
private static final int API_VERSION = 700;
public static void main(String[] args) {
System.out.println("About to use version " + API_VERSION);

View File

@ -34,7 +34,7 @@ public class SerialInsertion {
private static final int NODES = 1000000;
public static void main(String[] args) {
FDB api = FDB.selectAPIVersion(630);
FDB api = FDB.selectAPIVersion(700);
try(Database database = api.open()) {
long start = System.currentTimeMillis();

View File

@ -39,7 +39,7 @@ public class SerialIteration {
private static final int THREAD_COUNT = 1;
public static void main(String[] args) throws InterruptedException {
FDB api = FDB.selectAPIVersion(630);
FDB api = FDB.selectAPIVersion(700);
try(Database database = api.open(args[0])) {
for(int i = 1; i <= THREAD_COUNT; i++) {
runThreadedTest(database, i);

View File

@ -30,7 +30,7 @@ public class SerialTest {
public static void main(String[] args) throws InterruptedException {
final int reps = 1000;
try {
FDB fdb = FDB.selectAPIVersion(630);
FDB fdb = FDB.selectAPIVersion(700);
try(Database db = fdb.open()) {
runTests(reps, db);
}

View File

@ -39,7 +39,7 @@ public class SnapshotTransactionTest {
private static final Subspace SUBSPACE = new Subspace(Tuple.from("test", "conflict_ranges"));
public static void main(String[] args) {
FDB fdb = FDB.selectAPIVersion(630);
FDB fdb = FDB.selectAPIVersion(700);
try(Database db = fdb.open()) {
snapshotReadShouldNotConflict(db);
snapshotShouldNotAddConflictRange(db);

View File

@ -50,7 +50,7 @@ public class TupleTest {
public static void main(String[] args) throws NoSuchFieldException {
final int reps = 1000;
try {
FDB fdb = FDB.selectAPIVersion(630);
FDB fdb = FDB.selectAPIVersion(700);
addMethods();
comparisons();
emptyTuple();

View File

@ -32,7 +32,7 @@ import com.apple.foundationdb.tuple.Versionstamp;
public class VersionstampSmokeTest {
public static void main(String[] args) {
FDB fdb = FDB.selectAPIVersion(630);
FDB fdb = FDB.selectAPIVersion(700);
try(Database db = fdb.open()) {
db.run(tr -> {
tr.clear(Tuple.from("prefix").range());

View File

@ -34,7 +34,7 @@ import com.apple.foundationdb.Transaction;
public class WatchTest {
public static void main(String[] args) {
FDB fdb = FDB.selectAPIVersion(630);
FDB fdb = FDB.selectAPIVersion(700);
try(Database database = fdb.open(args[0])) {
database.options().setLocationCacheSize(42);
try(Transaction tr = database.createTransaction()) {

View File

@ -52,7 +52,7 @@ def get_api_version():
def api_version(ver):
header_version = 630
header_version = 700
if '_version' in globals():
if globals()['_version'] != ver:

View File

@ -253,7 +253,7 @@ def transactional(*tr_args, **tr_kwargs):
@functools.wraps(func)
def wrapper(*args, **kwargs):
# We can't throw this from the decorator, as when a user runs
# >>> import fdb ; fdb.api_version(630)
# >>> import fdb ; fdb.api_version(700)
# the code above uses @transactional before the API version is set
if fdb.get_api_version() >= 630 and inspect.isgeneratorfunction(func):
raise ValueError("Generators can not be wrapped with fdb.transactional")

View File

@ -22,7 +22,7 @@ import fdb
import sys
if __name__ == '__main__':
fdb.api_version(630)
fdb.api_version(700)
@fdb.transactional
def setValue(tr, key, value):

View File

@ -36,7 +36,7 @@ module FDB
end
end
def self.api_version(version)
header_version = 630
header_version = 700
if self.is_api_version_selected?()
if @@chosen_version != version
raise "FDB API already loaded at version #{@@chosen_version}."

View File

@ -1,7 +1,7 @@
#define FDB_API_VERSION 630
#define FDB_API_VERSION 700
#include <foundationdb/fdb_c.h>
int main(int argc, char* argv[]) {
fdb_select_api_version(630);
fdb_select_api_version(700);
return 0;
}

View File

@ -65,7 +65,7 @@ then
python setup.py install
successOr "Installing python bindings failed"
popd
python -c 'import fdb; fdb.api_version(630)'
python -c 'import fdb; fdb.api_version(700)'
successOr "Loading python bindings failed"
# Test cmake and pkg-config integration: https://github.com/apple/foundationdb/issues/1483

View File

@ -209,6 +209,25 @@ else()
# -mavx
# -msse4.2)
# Tentatively re-enabling vector instructions
set(USE_AVX512F OFF CACHE BOOL "Enable AVX 512F instructions")
if (USE_AVX512F)
add_compile_options(-mavx512f)
endif()
set(USE_AVX ON CACHE BOOL "Enable AVX instructions")
if (USE_AVX)
add_compile_options(-mavx)
endif()
# Intentionally using builtin memcpy. G++ does a good job on small memcpy's when the size is known at runtime.
# If the size is not known, then it falls back on the memcpy that's available at runtime (rte_memcpy, as of this
# writing; see flow.cpp).
#
# The downside of the builtin memcpy is that it's slower at large copies, so if we spend a lot of time on large
# copies of sizes that are known at compile time, this might not be a win. See the output of performance/memcpy
# for more information.
#add_compile_options(-fno-builtin-memcpy)
if (USE_VALGRIND)
add_compile_options(-DVALGRIND -DUSE_VALGRIND)
endif()
@ -241,7 +260,8 @@ else()
-Wno-delete-non-virtual-dtor
-Wno-undefined-var-template
-Wno-tautological-pointer-compare
-Wno-format)
-Wno-format
-Woverloaded-virtual)
if (USE_CCACHE)
add_compile_options(
-Wno-register
@ -253,7 +273,6 @@ else()
endif()
if (GCC)
add_compile_options(-Wno-pragmas)
# Otherwise `state [[maybe_unused]] int x;` will issue a warning.
# https://stackoverflow.com/questions/50646334/maybe-unused-on-member-variable-gcc-warns-incorrectly-that-attribute-is
add_compile_options(-Wno-attributes)
@ -267,6 +286,7 @@ else()
-fvisibility=hidden
-Wreturn-type
-fPIC)
add_compile_options($<$<COMPILE_LANGUAGE:CXX>:-Wclass-memaccess>)
if (GPERFTOOLS_FOUND AND GCC)
add_compile_options(
-fno-builtin-malloc

View File

@ -39,7 +39,9 @@ from json import JSONEncoder
import logging
import struct
from bisect import bisect_left
from bisect import bisect_right
import time
import datetime
PROTOCOL_VERSION_5_2 = 0x0FDB00A552000001
PROTOCOL_VERSION_6_0 = 0x0FDB00A570010001
@ -414,7 +416,7 @@ class TransactionInfoLoader(object):
else:
end_key = self.client_latency_end_key_selector
valid_transaction_infos = 0
transaction_infos = 0
invalid_transaction_infos = 0
def build_client_transaction_info(v):
@ -446,11 +448,12 @@ class TransactionInfoLoader(object):
info = build_client_transaction_info(v)
if info.has_types():
buffer.append(info)
valid_transaction_infos += 1
except UnsupportedProtocolVersionError as e:
invalid_transaction_infos += 1
except ValueError:
invalid_transaction_infos += 1
transaction_infos += 1
else:
if chunk_num == 1:
# first chunk
@ -476,14 +479,15 @@ class TransactionInfoLoader(object):
info = build_client_transaction_info(b''.join([chunk.value for chunk in c_list]))
if info.has_types():
buffer.append(info)
valid_transaction_infos += 1
except UnsupportedProtocolVersionError as e:
invalid_transaction_infos += 1
except ValueError:
invalid_transaction_infos += 1
transaction_infos += 1
self._check_and_adjust_chunk_cache_size()
if (valid_transaction_infos + invalid_transaction_infos) % 1000 == 0:
print("Processed valid: %d, invalid: %d" % (valid_transaction_infos, invalid_transaction_infos))
if transaction_infos % 1000 == 0:
print("Processed %d transactions, %d invalid" % (transaction_infos, invalid_transaction_infos))
if found == 0:
more = False
except fdb.FDBError as e:
@ -495,13 +499,15 @@ class TransactionInfoLoader(object):
for item in buffer:
yield item
print("Processed %d transactions, %d invalid\n" % (transaction_infos, invalid_transaction_infos))
def has_sortedcontainers():
try:
import sortedcontainers
return True
except ImportError:
logger.warn("Can't find sortedcontainers so disabling RangeCounter")
logger.warn("Can't find sortedcontainers so disabling ReadCounter")
return False
@ -513,155 +519,197 @@ def has_dateparser():
logger.warn("Can't find dateparser so disabling human date parsing")
return False
class RangeCounter(object):
def __init__(self, k):
self.k = k
class ReadCounter(object):
def __init__(self):
from sortedcontainers import SortedDict
self.ranges = SortedDict()
self.reads = SortedDict()
self.reads[b''] = [0, 0]
self.read_counts = {}
self.hit_count=0
def process(self, transaction_info):
for get in transaction_info.gets:
self._insert_read(get.key, None)
for get_range in transaction_info.get_ranges:
self._insert_range(get_range.key_range.start_key, get_range.key_range.end_key)
self._insert_read(get_range.key_range.start_key, get_range.key_range.end_key)
def _insert_range(self, start_key, end_key):
keys = self.ranges.keys()
if len(keys) == 0:
self.ranges[start_key] = end_key, 1
return
def _insert_read(self, start_key, end_key):
self.read_counts.setdefault((start_key, end_key), 0)
self.read_counts[(start_key, end_key)] += 1
start_pos = bisect_left(keys, start_key)
end_pos = bisect_left(keys, end_key)
#print("start_pos=%d, end_pos=%d" % (start_pos, end_pos))
self.reads.setdefault(start_key, [0, 0])[0] += 1
if end_key is not None:
self.reads.setdefault(end_key, [0, 0])[1] += 1
else:
self.reads.setdefault(start_key+b'\x00', [0, 0])[1] += 1
possible_intersection_keys = keys[max(0, start_pos - 1):min(len(keys), end_pos+1)]
def get_total_reads(self):
return sum([v for v in self.read_counts.values()])
def matches_filter(addresses, required_addresses):
for addr in required_addresses:
if addr not in addresses:
return False
return True
start_range_left = start_key
def get_top_k_reads(self, num, filter_addresses, shard_finder=None):
count_pairs = sorted([(v, k) for (k, v) in self.read_counts.items()], reverse=True, key=lambda item: item[0])
if not filter_addresses:
count_pairs = count_pairs[0:num]
for key in possible_intersection_keys:
cur_end_key, cur_count = self.ranges[key]
#logger.debug("key=%s, cur_end_key=%s, cur_count=%d, start_range_left=%s" % (key, cur_end_key, cur_count, start_range_left))
if start_range_left < key:
if end_key <= key:
self.ranges[start_range_left] = end_key, 1
return
self.ranges[start_range_left] = key, 1
start_range_left = key
assert start_range_left >= key
if start_range_left >= cur_end_key:
continue
if shard_finder:
results = []
for (count, (start, end)) in count_pairs:
results.append((start, end, count, shard_finder.get_addresses_for_key(start)))
# [key, start_range_left) = cur_count
# if key == start_range_left this will get overwritten below
self.ranges[key] = start_range_left, cur_count
shard_finder.wait_for_shard_addresses(results, 0, 3)
if end_key <= cur_end_key:
# [start_range_left, end_key) = cur_count+1
# [end_key, cur_end_key) = cur_count
self.ranges[start_range_left] = end_key, cur_count + 1
if end_key != cur_end_key:
self.ranges[end_key] = cur_end_key, cur_count
start_range_left = end_key
break
else:
# [start_range_left, cur_end_key) = cur_count+1
self.ranges[start_range_left] = cur_end_key, cur_count+1
start_range_left = cur_end_key
assert start_range_left <= end_key
if filter_addresses:
filter_addresses = set(filter_addresses)
results = [r for r in results if filter_addresses.issubset(set(r[3]))][0:num]
else:
results = [(start, end, count) for (count, (start, end)) in count_pairs[0:num]]
# there may be some range left
if start_range_left < end_key:
self.ranges[start_range_left] = end_key, 1
return results
def get_count_for_key(self, key):
if key in self.ranges:
return self.ranges[key][1]
keys = self.ranges.keys()
index = bisect_left(keys, key)
if index == 0:
return 0
index_key = keys[index-1]
if index_key <= key < self.ranges[index_key][0]:
return self.ranges[index_key][1]
return 0
def get_range_boundaries(self, shard_finder=None):
total = sum([count for _, (_, count) in self.ranges.items()])
range_size = total // self.k
def get_range_boundaries(self, num_buckets, shard_finder=None):
total = sum([start_count for (start_count, end_count) in self.reads.values()])
range_size = total // num_buckets
output_range_counts = []
def add_boundary(start, end, count):
def add_boundary(start, end, started_count, total_count):
if shard_finder:
shard_count = shard_finder.get_shard_count(start, end)
if shard_count == 1:
addresses = shard_finder.get_addresses_for_key(start)
else:
addresses = None
output_range_counts.append((start, end, count, shard_count, addresses))
output_range_counts.append((start, end, started_count, total_count, shard_count, addresses))
else:
output_range_counts.append((start, end, count, None, None))
output_range_counts.append((start, end, started_count, total_count, None, None))
this_range_start_key = None
last_end = None
open_count = 0
opened_this_range = 0
count_this_range = 0
for (start_key, (end_key, count)) in self.ranges.items():
if not this_range_start_key:
this_range_start_key = start_key
count_this_range += count
if count_this_range >= range_size:
add_boundary(this_range_start_key, end_key, count_this_range)
count_this_range = 0
this_range_start_key = None
if count_this_range > 0:
add_boundary(this_range_start_key, end_key, count_this_range)
for (start_key, (start_count, end_count)) in self.reads.items():
open_count -= end_count
if opened_this_range >= range_size:
add_boundary(this_range_start_key, start_key, opened_this_range, count_this_range)
count_this_range = open_count
opened_this_range = 0
this_range_start_key = None
count_this_range += start_count
opened_this_range += start_count
open_count += start_count
if count_this_range > 0 and this_range_start_key is None:
this_range_start_key = start_key
if end_count > 0:
last_end = start_key
if last_end is None:
last_end = b'\xff'
if count_this_range > 0:
add_boundary(this_range_start_key, last_end, opened_this_range, count_this_range)
shard_finder.wait_for_shard_addresses(output_range_counts, 0, 5)
return output_range_counts
class ShardFinder(object):
def __init__(self, db):
def __init__(self, db, exclude_ports):
self.db = db
self.exclude_ports = exclude_ports
self.tr = db.create_transaction()
self.refresh_tr()
self.outstanding = []
self.boundary_keys = list(fdb.locality.get_boundary_keys(db, b'', b'\xff\xff'))
self.shard_cache = {}
def _get_boundary_keys(self, begin, end):
start_pos = max(0, bisect_right(self.boundary_keys, begin)-1)
end_pos = max(0, bisect_right(self.boundary_keys, end)-1)
return self.boundary_keys[start_pos:end_pos]
def refresh_tr(self):
self.tr.options.set_read_lock_aware()
if not self.exclude_ports:
self.tr.options.set_include_port_in_address()
@staticmethod
@fdb.transactional
def _get_boundary_keys(tr, begin, end):
tr.options.set_read_lock_aware()
return fdb.locality.get_boundary_keys(tr, begin, end)
@staticmethod
@fdb.transactional
def _get_addresses_for_key(tr, key):
tr.options.set_read_lock_aware()
return fdb.locality.get_addresses_for_key(tr, key)
def get_shard_count(self, start_key, end_key):
return len(list(self._get_boundary_keys(self.db, start_key, end_key))) + 1
return len(self._get_boundary_keys(start_key, end_key)) + 1
def get_addresses_for_key(self, key):
return [a.decode('ascii') for a in self._get_addresses_for_key(self.db, key).wait()]
shard = self.boundary_keys[max(0, bisect_right(self.boundary_keys, key)-1)]
do_load = False
if not shard in self.shard_cache:
do_load = True
elif self.shard_cache[shard].is_ready():
try:
self.shard_cache[shard].wait()
except fdb.FDBError as e:
self.tr.on_error(e).wait()
self.refresh_tr()
do_load = True
if do_load:
if len(self.outstanding) > 1000:
for f in self.outstanding:
try:
f.wait()
except fdb.FDBError as e:
pass
class TopKeysCounter(object):
self.outstanding = []
self.tr.reset()
self.refresh_tr()
self.outstanding.append(self._get_addresses_for_key(self.tr, shard))
self.shard_cache[shard] = self.outstanding[-1]
return self.shard_cache[shard]
def wait_for_shard_addresses(self, ranges, key_idx, addr_idx):
for index in range(len(ranges)):
item = ranges[index]
if item[addr_idx] is not None:
while True:
try:
ranges[index] = item[0:addr_idx] + ([a.decode('ascii') for a in item[addr_idx].wait()],) + item[addr_idx+1:]
break
except fdb.FDBError as e:
ranges[index] = item[0:addr_idx] + (self.get_addresses_for_key(item[key_idx]),) + item[addr_idx+1:]
class WriteCounter(object):
mutation_types_to_consider = frozenset([MutationType.SET_VALUE, MutationType.ADD_VALUE])
def __init__(self, k):
self.k = k
self.reads = defaultdict(lambda: 0)
def __init__(self):
self.writes = defaultdict(lambda: 0)
def process(self, transaction_info):
for get in transaction_info.gets:
self.reads[get.key] += 1
if transaction_info.commit:
for mutation in transaction_info.commit.mutations:
if mutation.code in self.mutation_types_to_consider:
self.writes[mutation.param_one] += 1
def _get_range_boundaries(self, counts, shard_finder=None):
total = sum([v for (k, v) in counts.items()])
range_size = total // self.k
key_counts_sorted = sorted(counts.items())
def get_range_boundaries(self, num_buckets, shard_finder=None):
total = sum([v for (k, v) in self.writes.items()])
range_size = total // num_buckets
key_counts_sorted = sorted(self.writes.items())
output_range_counts = []
def add_boundary(start, end, count):
@ -671,9 +719,9 @@ class TopKeysCounter(object):
addresses = shard_finder.get_addresses_for_key(start)
else:
addresses = None
output_range_counts.append((start, end, count, shard_count, addresses))
output_range_counts.append((start, end, count, None, shard_count, addresses))
else:
output_range_counts.append((start, end, count, None, None))
output_range_counts.append((start, end, count, None, None, None))
start_key = None
count_this_range = 0
@ -688,24 +736,31 @@ class TopKeysCounter(object):
if count_this_range > 0:
add_boundary(start_key, k, count_this_range)
shard_finder.wait_for_shard_addresses(output_range_counts, 0, 5)
return output_range_counts
def _get_top_k(self, counts):
count_key_pairs = sorted([(v, k) for (k, v) in counts.items()], reverse=True)
return count_key_pairs[0:self.k]
def get_total_writes(self):
return sum([v for v in self.writes.values()])
def get_top_k_reads(self):
return self._get_top_k(self.reads)
def get_top_k_writes(self, num, filter_addresses, shard_finder=None):
count_pairs = sorted([(v, k) for (k, v) in self.writes.items()], reverse=True)
if not filter_addresses:
count_pairs = count_pairs[0:num]
def get_top_k_writes(self):
return self._get_top_k(self.writes)
if shard_finder:
results = []
for (count, key) in count_pairs:
results.append((key, None, count, shard_finder.get_addresses_for_key(key)))
def get_k_read_range_boundaries(self, shard_finder=None):
return self._get_range_boundaries(self.reads, shard_finder)
shard_finder.wait_for_shard_addresses(results, 0, 3)
def get_k_write_range_boundaries(self, shard_finder=None):
return self._get_range_boundaries(self.writes, shard_finder)
if filter_addresses:
filter_addresses = set(filter_addresses)
results = [r for r in results if filter_addresses.issubset(set(r[3]))][0:num]
else:
results = [(key, end, count) for (count, key) in count_pairs[0:num]]
return results
def connect(cluster_file=None):
db = fdb.open(cluster_file=cluster_file)
@ -722,6 +777,8 @@ def main():
help="Include get type. If no filter args are given all will be returned.")
parser.add_argument("--filter-get-range", action="store_true",
help="Include get_range type. If no filter args are given all will be returned.")
parser.add_argument("--filter-reads", action="store_true",
help="Include get and get_range type. If no filter args are given all will be returned.")
parser.add_argument("--filter-commit", action="store_true",
help="Include commit type. If no filter args are given all will be returned.")
parser.add_argument("--filter-error-get", action="store_true",
@ -737,21 +794,34 @@ def main():
end_time_group = parser.add_mutually_exclusive_group()
end_time_group.add_argument("--max-timestamp", type=int, help="Don't return events newer than this epoch time")
end_time_group.add_argument("-e", "--end-time", type=str, help="Don't return events older than this parsed time")
parser.add_argument("--top-keys", type=int, help="If specified will output this many top keys for reads or writes", default=0)
parser.add_argument("--num-buckets", type=int, help="The number of buckets to partition the key-space into for operation counts", default=100)
parser.add_argument("--top-requests", type=int, help="If specified will output this many top keys for reads or writes", default=0)
parser.add_argument("--exclude-ports", action="store_true", help="Print addresses without the port number. Only works in versions older than 6.3, and is required in versions older than 6.2.")
parser.add_argument("--single-shard-ranges-only", action="store_true", help="Only print range boundaries that exist in a single shard")
parser.add_argument("-a", "--filter-address", action="append", help="Only print range boundaries that include the given address. This option can used multiple times to include more than one address in the filter, in which case all addresses must match.")
args = parser.parse_args()
type_filter = set()
if args.filter_get_version: type_filter.add("get_version")
if args.filter_get: type_filter.add("get")
if args.filter_get_range: type_filter.add("get_range")
if args.filter_get or args.filter_reads: type_filter.add("get")
if args.filter_get_range or args.filter_reads: type_filter.add("get_range")
if args.filter_commit: type_filter.add("commit")
if args.filter_error_get: type_filter.add("error_get")
if args.filter_error_get_range: type_filter.add("error_get_range")
if args.filter_error_commit: type_filter.add("error_commit")
top_keys = args.top_keys
key_counter = TopKeysCounter(top_keys) if top_keys else None
range_counter = RangeCounter(top_keys) if (has_sortedcontainers() and top_keys) else None
full_output = args.full_output or (top_keys is not None)
if (not type_filter or "commit" in type_filter):
write_counter = WriteCounter() if args.num_buckets else None
else:
write_counter = None
if (not type_filter or "get" in type_filter or "get_range" in type_filter):
read_counter = ReadCounter() if (has_sortedcontainers() and args.num_buckets) else None
else:
read_counter = None
full_output = args.full_output or (args.num_buckets is not None)
if args.min_timestamp:
min_timestamp = args.min_timestamp
@ -784,48 +854,128 @@ def main():
db = connect(cluster_file=args.cluster_file)
loader = TransactionInfoLoader(db, full_output=full_output, type_filter=type_filter,
min_timestamp=min_timestamp, max_timestamp=max_timestamp)
for info in loader.fetch_transaction_info():
if info.has_types():
if not key_counter and not range_counter:
if not write_counter and not read_counter:
print(info.to_json())
else:
if key_counter:
key_counter.process(info)
if range_counter:
range_counter.process(info)
if write_counter:
write_counter.process(info)
if read_counter:
read_counter.process(info)
if key_counter:
def print_top(top):
for (count, key) in top:
print("%s %d" % (key, count))
def print_range_boundaries(range_boundaries):
for (start, end, count, shard_count, addresses) in range_boundaries:
if not shard_count:
print("[%s, %s] %d" % (start, end, count))
def print_top(top, total, context):
if top:
running_count = 0
for (idx, (start, end, count, addresses)) in enumerate(top):
running_count += count
if end is not None:
op_str = 'Range %r - %r' % (start, end)
else:
addresses_string = "addresses=%s" % ','.join(addresses) if addresses else ''
print("[%s, %s] %d shards=%d %s" % (start, end, count, shard_count, addresses_string))
op_str = 'Key %r' % start
print(" %d. %s\n %d sampled %s (%.2f%%, %.2f%% cumulative)" % (idx+1, op_str, count, context, 100*count/total, 100*running_count/total))
print(" shard addresses: %s\n" % ", ".join(addresses))
else:
print(" No %s found" % context)
def print_range_boundaries(range_boundaries, context):
omit_start = None
for (idx, (start, end, start_count, total_count, shard_count, addresses)) in enumerate(range_boundaries):
omit = args.single_shard_ranges_only and shard_count is not None and shard_count > 1
if args.filter_address:
if not addresses:
omit = True
else:
for addr in args.filter_address:
if addr not in addresses:
omit = True
break
if not omit:
if omit_start is not None:
if omit_start == idx-1:
print(" %d. Omitted\n" % (idx))
else:
print(" %d - %d. Omitted\n" % (omit_start+1, idx))
omit_start = None
if total_count is None:
count_str = '%d sampled %s' % (start_count, context)
else:
count_str = '%d sampled %s (%d intersecting)' % (start_count, context, total_count)
if not shard_count:
print(" %d. [%s, %s]\n %d sampled %s\n" % (idx+1, start, end, count, context))
else:
addresses_string = "; addresses=%s" % ', '.join(addresses) if addresses else ''
print(" %d. [%s, %s]\n %s spanning %d shard(s)%s\n" % (idx+1, start, end, count_str, shard_count, addresses_string))
elif omit_start is None:
omit_start = idx
if omit_start is not None:
if omit_start == len(range_boundaries)-1:
print(" %d. Omitted\n" % len(range_boundaries))
else:
print(" %d - %d. Omitted\n" % (omit_start+1, len(range_boundaries)))
shard_finder = ShardFinder(db, args.exclude_ports)
print("NOTE: shard locations are current and may not reflect where an operation was performed in the past\n")
if write_counter:
if args.top_requests:
top_writes = write_counter.get_top_k_writes(args.top_requests, args.filter_address, shard_finder=shard_finder)
range_boundaries = write_counter.get_range_boundaries(args.num_buckets, shard_finder=shard_finder)
num_writes = write_counter.get_total_writes()
if args.top_requests or range_boundaries:
print("WRITES")
print("------\n")
print("Processed %d total writes\n" % num_writes)
if args.top_requests:
suffix = ""
if args.filter_address:
suffix = " (%s)" % ", ".join(args.filter_address)
print("Top %d writes%s:\n" % (args.top_requests, suffix))
print_top(top_writes, write_counter.get_total_writes(), "writes")
print("")
shard_finder = ShardFinder(db)
top_reads = key_counter.get_top_k_reads()
if top_reads:
print("Top %d reads:" % min(top_keys, len(top_reads)))
print_top(top_reads)
print("Approx equal sized gets range boundaries:")
print_range_boundaries(key_counter.get_k_read_range_boundaries(shard_finder=shard_finder))
top_writes = key_counter.get_top_k_writes()
if top_writes:
print("Top %d writes:" % min(top_keys, len(top_writes)))
print_top(top_writes)
print("Approx equal sized commits range boundaries:")
print_range_boundaries(key_counter.get_k_write_range_boundaries(shard_finder=shard_finder))
if range_counter:
range_boundaries = range_counter.get_range_boundaries(shard_finder=shard_finder)
if range_boundaries:
print("Approx equal sized get_ranges boundaries:")
print_range_boundaries(range_boundaries)
print("Key-space boundaries with approximately equal mutation counts:\n")
print_range_boundaries(range_boundaries, "writes")
if args.top_requests or range_boundaries:
print("")
if read_counter:
if args.top_requests:
top_reads = read_counter.get_top_k_reads(args.top_requests, args.filter_address, shard_finder=shard_finder)
range_boundaries = read_counter.get_range_boundaries(args.num_buckets, shard_finder=shard_finder)
num_reads = read_counter.get_total_reads()
if args.top_requests or range_boundaries:
print("READS")
print("-----\n")
print("Processed %d total reads\n" % num_reads)
if args.top_requests:
suffix = ""
if args.filter_address:
suffix = " (%s)" % ", ".join(args.filter_address)
print("Top %d reads%s:\n" % (args.top_requests, suffix))
print_top(top_reads, num_reads, "reads")
print("")
if range_boundaries:
print("Key-space boundaries with approximately equal read counts:\n")
print_range_boundaries(range_boundaries, "reads")
if __name__ == "__main__":
main()

View File

@ -493,7 +493,7 @@ If a process has had more than 10 TCP segments retransmitted in the last 5 secon
10.0.4.1:4500 ( 3% cpu; 2% machine; 0.004 Gbps; 0% disk; REXMIT! 2.5 GB / 4.1 GB RAM )
Machine-readable status
--------------------------------
-----------------------
The status command can provide a complete summary of statistics about the cluster and the database with the ``json`` argument. Full documentation for ``status json`` output can be found :doc:`here <mr-status>`.
From the output of ``status json``, operators can find useful health metrics to determine whether or not their cluster is hitting performance limits.
@ -505,6 +505,72 @@ Durable version lag ``cluster.qos.worst_durability_lag_storage_server`` cont
Transaction log queue ``cluster.qos.worst_queue_bytes_log_server`` contains the maximum size in bytes of the mutations stored on a transaction log that have not yet been popped by storage servers. A large transaction log queue size can potentially cause the ratekeeper to increase throttling.
====================== ==============================================================================================================
Server-side latency band tracking
---------------------------------
As part of the status document, ``status json`` provides some sampled latency metrics obtained by running probe transactions internally. While this can often be useful, it does not necessarily reflect the distribution of latencies for requests originated by clients.
FoundationDB additionally provides optional functionality to measure the latencies of all incoming get read version (GRV), read, and commit requests and report some basic details about those requests. The latencies are measured from the time the server receives the request to the point when it replies, and will therefore not include time spent in transit between the client and server or delays in the client process itself.
The latency band tracking works by configuring various latency thresholds and counting the number of requests that occur in each band (i.e. between two consecutive thresholds). For example, if you wanted to define a service-level objective (SLO) for your cluster where 99.9% of read requests were answered within N seconds, you could set a read latency threshold at N. You could then count the number of requests below and above the threshold and determine whether the required percentage of requests are answered sufficiently quickly.
Configuration of server-side latency bands is performed by setting the ``\xff\x02/latencyBandConfig`` key to a string encoding the following JSON document::
{
"get_read_version" : {
"bands" : [ 0.01, 0.1]
},
"read" : {
"bands" : [ 0.01, 0.1],
"max_key_selector_offset" : 1000,
"max_read_bytes" : 1000000
},
"commit" : {
"bands" : [ 0.01, 0.1],
"max_commit_bytes" : 1000000
}
}
Every field in this configuration is optional, and any missing fields will be left unset (i.e. no bands will be tracked or limits will not apply). The configuration takes the following arguments:
* ``bands`` - a list of thresholds (in seconds) to be measured for the given request type (``get_read_version``, ``read``, or ``commit``)
* ``max_key_selector_offset`` - an integer specifying the maximum key selector offset a read request can have and still be counted
* ``max_read_bytes`` - an integer specifying the maximum size in bytes of a read response that will be counted
* ``max_commit_bytes`` - an integer specifying the maximum size in bytes of a commit request that will be counted
Setting this configuration key to a value that changes the configuration will result in the cluster controller server process logging a ``LatencyBandConfigChanged`` event. This event will indicate whether a configuration is present or not using its ``Present`` field. Specifying an invalid configuration will result in the latency band feature being unconfigured, and the server process running the cluster controller will log a ``InvalidLatencyBandConfiguration`` trace event.
.. note:: GRV requests are counted only at default and immediate priority. Batch priority GRV requests are ignored for the purposes of latency band tracking.
When configured, the ``status json`` output will include additional fields to report the number of requests in each latency band located at ``cluster.processes.<ID>.roles[N].*_latency_bands``::
"grv_latency_bands" : {
0.01: 10,
0.1: 0,
inf: 1,
filtered: 0
},
"read_latency_bands" : {
0.01: 12,
0.1: 1,
inf: 0,
filtered: 0
},
"commit_latency_bands" : {
0.01: 5,
0.1: 5,
inf: 2,
filtered: 1
}
The ``grv_latency_bands`` and ``commit_latency_bands`` objects will only be logged for ``proxy`` roles, and ``read_latency_bands`` will only be logged for storage roles. Each threshold is represented as a key in the map, and its associated value will be the total number of requests in the lifetime of the process with a latency smaller than the threshold but larger than the next smaller threshold.
For example, ``0.1: 1`` in ``read_latency_bands`` indicates that there has been 1 read request with a latency in the range ``[0.01, 0.1)``. For the smallest specified threshold, the lower bound is 0 (e.g. ``[0, 0.01)`` in the example above). Requests that took longer than any defined latency band will be reported in the ``inf`` (infinity) band. Requests that were filtered by the configuration (e.g. using ``max_read_bytes``) are reported in the ``filtered`` category.
Because each threshold reports latencies strictly in the range between the next lower threshold and itself, it may be necessary to sum up the counts for multiple bands to determine the total number of requests below a certain threshold.
.. note:: No history of request counts is recorded for processes that ran in the past. This includes the history prior to restart for a process that has been restarted, for which the counts get reset to 0. For this reason, it is recommended that you collect this information periodically if you need to be able to track requests from such processes.
.. _administration_fdbmonitor:
``fdbmonitor`` and ``fdbserver``

View File

@ -133,7 +133,7 @@ API versioning
Prior to including ``fdb_c.h``, you must define the ``FDB_API_VERSION`` macro. This, together with the :func:`fdb_select_api_version()` function, allows programs written against an older version of the API to compile and run with newer versions of the C library. The current version of the FoundationDB C API is |api-version|. ::
#define FDB_API_VERSION 630
#define FDB_API_VERSION 700
#include <foundationdb/fdb_c.h>
.. function:: fdb_error_t fdb_select_api_version(int version)

View File

@ -147,7 +147,7 @@
.. |atomic-versionstamps-tuple-warning-value| replace::
At this time, versionstamped values are not compatible with the Tuple layer except in Java, Python, and Go. Note that this implies versionstamped values may not be used with the Subspace and Directory layers except in those languages.
.. |api-version| replace:: 630
.. |api-version| replace:: 700
.. |streaming-mode-blurb1| replace::
When using |get-range-func| and similar interfaces, API clients can request large ranges of the database to iterate over. Making such a request doesn't necessarily mean that the client will consume all of the data in the range - sometimes the client doesn't know how far it intends to iterate in advance. FoundationDB tries to balance latency and bandwidth by requesting data for iteration in batches.

View File

@ -108,7 +108,7 @@ Opening a database
After importing the ``fdb`` module and selecting an API version, you probably want to open a :class:`Database` using :func:`open`::
import fdb
fdb.api_version(630)
fdb.api_version(700)
db = fdb.open()
.. function:: open( cluster_file=None, event_model=None )

View File

@ -93,7 +93,7 @@ Opening a database
After requiring the ``FDB`` gem and selecting an API version, you probably want to open a :class:`Database` using :func:`open`::
require 'fdb'
FDB.api_version 630
FDB.api_version 700
db = FDB.open
.. function:: open( cluster_file=nil ) -> Database

View File

@ -29,7 +29,7 @@ Before using the API, we need to specify the API version. This allows programs t
.. code-block:: go
fdb.MustAPIVersion(630)
fdb.MustAPIVersion(700)
Next, we open a FoundationDB database. The API will connect to the FoundationDB cluster indicated by the :ref:`default cluster file <default-cluster-file>`.
@ -78,7 +78,7 @@ If this is all working, it looks like we are ready to start building a real appl
func main() {
// Different API versions may expose different runtime behaviors.
fdb.MustAPIVersion(630)
fdb.MustAPIVersion(700)
// Open the default database from the system cluster
db := fdb.MustOpenDefault()
@ -666,7 +666,7 @@ Here's the code for the scheduling tutorial:
}
func main() {
fdb.MustAPIVersion(630)
fdb.MustAPIVersion(700)
db := fdb.MustOpenDefault()
db.Options().SetTransactionTimeout(60000) // 60,000 ms = 1 minute
db.Options().SetTransactionRetryLimit(100)

View File

@ -30,7 +30,7 @@ Before using the API, we need to specify the API version. This allows programs t
private static final Database db;
static {
fdb = FDB.selectAPIVersion(630);
fdb = FDB.selectAPIVersion(700);
db = fdb.open();
}
@ -66,7 +66,7 @@ If this is all working, it looks like we are ready to start building a real appl
private static final Database db;
static {
fdb = FDB.selectAPIVersion(630);
fdb = FDB.selectAPIVersion(700);
db = fdb.open();
}
@ -441,7 +441,7 @@ Here's the code for the scheduling tutorial:
private static final Database db;
static {
fdb = FDB.selectAPIVersion(630);
fdb = FDB.selectAPIVersion(700);
db = fdb.open();
db.options().setTransactionTimeout(60000); // 60,000 ms = 1 minute
db.options().setTransactionRetryLimit(100);

View File

@ -23,7 +23,7 @@ Open a Ruby interactive interpreter and import the FoundationDB API module::
Before using the API, we need to specify the API version. This allows programs to maintain compatibility even if the API is modified in future versions::
> FDB.api_version 630
> FDB.api_version 700
=> nil
Next, we open a FoundationDB database. The API will connect to the FoundationDB cluster indicated by the :ref:`default cluster file <default-cluster-file>`. ::
@ -46,7 +46,7 @@ If this is all working, it looks like we are ready to start building a real appl
.. code-block:: ruby
require 'fdb'
FDB.api_version 630
FDB.api_version 700
@db = FDB.open
@db['hello'] = 'world'
print 'hello ', @db['hello']
@ -373,7 +373,7 @@ Here's the code for the scheduling tutorial:
require 'fdb'
FDB.api_version 630
FDB.api_version 700
####################################
## Initialization ##

View File

@ -30,7 +30,7 @@ Open a Python interactive interpreter and import the FoundationDB API module::
Before using the API, we need to specify the API version. This allows programs to maintain compatibility even if the API is modified in future versions::
>>> fdb.api_version(630)
>>> fdb.api_version(700)
Next, we open a FoundationDB database. The API will connect to the FoundationDB cluster indicated by the :ref:`default cluster file <default-cluster-file>`. ::
@ -48,7 +48,7 @@ When this command returns without exception, the modification is durably stored
If this is all working, it looks like we are ready to start building a real application. For reference, here's the full code for "hello world"::
import fdb
fdb.api_version(630)
fdb.api_version(700)
db = fdb.open()
db[b'hello'] = b'world'
print 'hello', db[b'hello']
@ -91,7 +91,7 @@ FoundationDB includes a few tools that make it easy to model data using this app
opening a :ref:`directory <developer-guide-directories>` in the database::
import fdb
fdb.api_version(630)
fdb.api_version(700)
db = fdb.open()
scheduling = fdb.directory.create_or_open(db, ('scheduling',))
@ -337,7 +337,7 @@ Here's the code for the scheduling tutorial::
import fdb
import fdb.tuple
fdb.api_version(630)
fdb.api_version(700)
####################################

View File

@ -10,38 +10,38 @@ macOS
The macOS installation package is supported on macOS 10.7+. It includes the client and (optionally) the server.
* `FoundationDB-6.3.0.pkg <https://www.foundationdb.org/downloads/6.3.0/macOS/installers/FoundationDB-6.3.0.pkg>`_
* `FoundationDB-6.3.1.pkg <https://www.foundationdb.org/downloads/6.3.1/macOS/installers/FoundationDB-6.3.1.pkg>`_
Ubuntu
------
The Ubuntu packages are supported on 64-bit Ubuntu 12.04+, but beware of the Linux kernel bug in Ubuntu 12.x.
* `foundationdb-clients-6.3.0-1_amd64.deb <https://www.foundationdb.org/downloads/6.3.0/ubuntu/installers/foundationdb-clients_6.3.0-1_amd64.deb>`_
* `foundationdb-server-6.3.0-1_amd64.deb <https://www.foundationdb.org/downloads/6.3.0/ubuntu/installers/foundationdb-server_6.3.0-1_amd64.deb>`_ (depends on the clients package)
* `foundationdb-clients-6.3.1-1_amd64.deb <https://www.foundationdb.org/downloads/6.3.1/ubuntu/installers/foundationdb-clients_6.3.1-1_amd64.deb>`_
* `foundationdb-server-6.3.1-1_amd64.deb <https://www.foundationdb.org/downloads/6.3.1/ubuntu/installers/foundationdb-server_6.3.1-1_amd64.deb>`_ (depends on the clients package)
RHEL/CentOS EL6
---------------
The RHEL/CentOS EL6 packages are supported on 64-bit RHEL/CentOS 6.x.
* `foundationdb-clients-6.3.0-1.el6.x86_64.rpm <https://www.foundationdb.org/downloads/6.3.0/rhel6/installers/foundationdb-clients-6.3.0-1.el6.x86_64.rpm>`_
* `foundationdb-server-6.3.0-1.el6.x86_64.rpm <https://www.foundationdb.org/downloads/6.3.0/rhel6/installers/foundationdb-server-6.3.0-1.el6.x86_64.rpm>`_ (depends on the clients package)
* `foundationdb-clients-6.3.1-1.el6.x86_64.rpm <https://www.foundationdb.org/downloads/6.3.1/rhel6/installers/foundationdb-clients-6.3.1-1.el6.x86_64.rpm>`_
* `foundationdb-server-6.3.1-1.el6.x86_64.rpm <https://www.foundationdb.org/downloads/6.3.1/rhel6/installers/foundationdb-server-6.3.1-1.el6.x86_64.rpm>`_ (depends on the clients package)
RHEL/CentOS EL7
---------------
The RHEL/CentOS EL7 packages are supported on 64-bit RHEL/CentOS 7.x.
* `foundationdb-clients-6.3.0-1.el7.x86_64.rpm <https://www.foundationdb.org/downloads/6.3.0/rhel7/installers/foundationdb-clients-6.3.0-1.el7.x86_64.rpm>`_
* `foundationdb-server-6.3.0-1.el7.x86_64.rpm <https://www.foundationdb.org/downloads/6.3.0/rhel7/installers/foundationdb-server-6.3.0-1.el7.x86_64.rpm>`_ (depends on the clients package)
* `foundationdb-clients-6.3.1-1.el7.x86_64.rpm <https://www.foundationdb.org/downloads/6.3.1/rhel7/installers/foundationdb-clients-6.3.1-1.el7.x86_64.rpm>`_
* `foundationdb-server-6.3.1-1.el7.x86_64.rpm <https://www.foundationdb.org/downloads/6.3.1/rhel7/installers/foundationdb-server-6.3.1-1.el7.x86_64.rpm>`_ (depends on the clients package)
Windows
-------
The Windows installer is supported on 64-bit Windows XP and later. It includes the client and (optionally) the server.
* `foundationdb-6.3.0-x64.msi <https://www.foundationdb.org/downloads/6.3.0/windows/installers/foundationdb-6.3.0-x64.msi>`_
* `foundationdb-6.3.1-x64.msi <https://www.foundationdb.org/downloads/6.3.1/windows/installers/foundationdb-6.3.1-x64.msi>`_
API Language Bindings
=====================
@ -58,18 +58,18 @@ On macOS and Windows, the FoundationDB Python API bindings are installed as part
If you need to use the FoundationDB Python API from other Python installations or paths, use the Python package manager ``pip`` (``pip install foundationdb``) or download the Python package:
* `foundationdb-6.3.0.tar.gz <https://www.foundationdb.org/downloads/6.3.0/bindings/python/foundationdb-6.3.0.tar.gz>`_
* `foundationdb-6.3.1.tar.gz <https://www.foundationdb.org/downloads/6.3.1/bindings/python/foundationdb-6.3.1.tar.gz>`_
Ruby 1.9.3/2.0.0+
-----------------
* `fdb-6.3.0.gem <https://www.foundationdb.org/downloads/6.3.0/bindings/ruby/fdb-6.3.0.gem>`_
* `fdb-6.3.1.gem <https://www.foundationdb.org/downloads/6.3.1/bindings/ruby/fdb-6.3.1.gem>`_
Java 8+
-------
* `fdb-java-6.3.0.jar <https://www.foundationdb.org/downloads/6.3.0/bindings/java/fdb-java-6.3.0.jar>`_
* `fdb-java-6.3.0-javadoc.jar <https://www.foundationdb.org/downloads/6.3.0/bindings/java/fdb-java-6.3.0-javadoc.jar>`_
* `fdb-java-6.3.1.jar <https://www.foundationdb.org/downloads/6.3.1/bindings/java/fdb-java-6.3.1.jar>`_
* `fdb-java-6.3.1-javadoc.jar <https://www.foundationdb.org/downloads/6.3.1/bindings/java/fdb-java-6.3.1-javadoc.jar>`_
Go 1.11+
--------

View File

@ -69,7 +69,7 @@ Heres a basic implementation of the recipe.
private static final long EMPTY_ARRAY = -1;
static {
fdb = FDB.selectAPIVersion(630);
fdb = FDB.selectAPIVersion(700);
db = fdb.open();
docSpace = new Subspace(Tuple.from("D"));
}

View File

@ -74,7 +74,7 @@ Heres a simple implementation of multimaps with multisets as described:
private static final int N = 100;
static {
fdb = FDB.selectAPIVersion(630);
fdb = FDB.selectAPIVersion(700);
db = fdb.open();
multi = new Subspace(Tuple.from("M"));
}

View File

@ -2,6 +2,15 @@
Release Notes
#############
6.2.22
======
Fixes
-----
* Coordinator class processes could be recruited as the cluster controller. `(PR #3282) <https://github.com/apple/foundationdb/pull/3282>`_
* HTTPS requests made by backup would fail (introduced in 6.2.21). `(PR #3284) <https://github.com/apple/foundationdb/pull/3284>`_
6.2.21
======

View File

@ -0,0 +1,123 @@
#############
Release Notes
#############
6.3.1
=====
Features
--------
* Added the ability to set arbitrary tags on transactions. Tags can be specifically throttled using ``fdbcli``, and certain types of tags can be automatically throttled by ratekeeper. `(PR #2942) <https://github.com/apple/foundationdb/pull/2942>`_
* Add an option for transactions to report conflicting keys by calling ``getRange`` with the special key prefix ``\xff\xff/transaction/conflicting_keys/``. `(PR 2257) <https://github.com/apple/foundationdb/pull/2257>`_
* Added the ``exclude failed`` command to ``fdbcli``. This command designates that a process is dead and will never come back, so the transaction logs can forget about mutations sent to that process. `(PR #1955) <https://github.com/apple/foundationdb/pull/1955>`_
* A new fast restore system that can restore a database to a point in time from backup files. It is a Spark-like parallel processing framework that processes backup data asynchronously, in parallel and in pipeline. `(Fast Restore Project) <https://github.com/apple/foundationdb/projects/7>`_
* Added backup workers for pulling mutations from transaction logs and uploading them to blob storage. Switching from the previous backup implementation will double a cluster's maximum write bandwidth. `(PR #1625) <https://github.com/apple/foundationdb/pull/1625>`_ `(PR #2588) <https://github.com/apple/foundationdb/pull/2588>`_ `(PR #2642) <https://github.com/apple/foundationdb/pull/2642>`_
* Added a new API in all bindings that can be used to query the estimated byte size of a given range. `(PR #2537) <https://github.com/apple/foundationdb/pull/2537>`_
* Added the ``lock`` and ``unlock`` commands to ``fdbcli`` which lock or unlock a cluster. `(PR #2890) <https://github.com/apple/foundationdb/pull/2890>`_
* Add a framework which helps to add client functions using special keys (keys within ``[\xff\xff, \xff\xff\xff)``). `(PR #2662) <https://github.com/apple/foundationdb/pull/2662>`_
Performance
-----------
* Improved the client's load balancing algorithm so that each proxy processes an equal number of requests. `(PR #2520) <https://github.com/apple/foundationdb/pull/2520>`_
* Significantly reduced the amount of work done on the cluster controller by removing the centralized failure monitoring. `(PR #2518) <https://github.com/apple/foundationdb/pull/2518>`_
* Improved master recovery speeds by more efficiently broadcasting the recovery state between processes. `(PR #2941) <https://github.com/apple/foundationdb/pull/2941>`_
* Significantly reduced the number of network connections opened to the coordinators. `(PR #3069) <https://github.com/apple/foundationdb/pull/3069>`_
* Improve GRV tail latencies, particularly as the transaction rate gets nearer the ratekeeper limit. `(PR #2735) <https://github.com/apple/foundationdb/pull/2735>`_
* The proxies are now more responsive to changes in workload when unthrottling lower priority transactions. `(PR #2735) <https://github.com/apple/foundationdb/pull/2735>`_
* Removed a lot of unnecessary copying across the codebase. `(PR #2986) <https://github.com/apple/foundationdb/pull/2986>`_ `(PR #2915) <https://github.com/apple/foundationdb/pull/2915>`_ `(PR #3024) <https://github.com/apple/foundationdb/pull/3024>`_ `(PR #2999) <https://github.com/apple/foundationdb/pull/2999>`_
* Optimized the performance of the storage server. `(PR #1988) <https://github.com/apple/foundationdb/pull/1988>`_ `(PR #3103) <https://github.com/apple/foundationdb/pull/3103>`_
* Optimized the performance of the resolver. `(PR #2648) <https://github.com/apple/foundationdb/pull/2648>`_
* Replaced most uses of hashlittle2 with crc32 for better performance. `(PR #2538) <https://github.com/apple/foundationdb/pull/2538>`_
* Significantly reduced the serialized size of conflict ranges and single key clears. `(PR #2513) <https://github.com/apple/foundationdb/pull/2513>`_
* Improved range read performance when the reads overlap recently cleared key ranges. `(PR #2028) <https://github.com/apple/foundationdb/pull/2028>`_
* Reduced the number of comparisons used by various map implementations. `(PR #2882) <https://github.com/apple/foundationdb/pull/2882>`_
* Reduced the serialized size of empty strings. `(PR #3063) <https://github.com/apple/foundationdb/pull/3063>`_
* Reduced the serialized size of various interfaces by 10x. `(PR #3068) <https://github.com/apple/foundationdb/pull/3068>`_
Reliability
-----------
* Connections that disconnect frequently are not immediately marked available. `(PR #2932) <https://github.com/apple/foundationdb/pull/2932>`_
* The data distributor will consider storage servers that are continually lagging behind as if they were failed. `(PR #2917) <https://github.com/apple/foundationdb/pull/2917>`_
* Changing the storage engine type of a cluster will no longer cause the cluster to run out of memory. Instead, the cluster will gracefully migrate storage server processes to the new storage engine one by one. `(PR #1985) <https://github.com/apple/foundationdb/pull/1985>`_
* Batch priority transactions which are being throttled by ratekeeper will get a ``batch_transaction_throttled`` error instead of hanging indefinitely. `(PR #1868) <https://github.com/apple/foundationdb/pull/1868>`_
* Avoid using too much memory on the transaction logs when multiple types of transaction logs exist in the same process. `(PR #2213) <https://github.com/apple/foundationdb/pull/2213>`_
Fixes
-----
* The ``SetVersionstampedKey`` atomic operation no longer conflicts with versions smaller than the current read version of the transaction. `(PR #2557) <https://github.com/apple/foundationdb/pull/2557>`_
* Ratekeeper would measure durability lag a few seconds higher than reality. `(PR #2499) <https://github.com/apple/foundationdb/pull/2499>`_
* In very rare scenarios, the data distributor process could get stuck in an infinite loop. `(PR #2228) <https://github.com/apple/foundationdb/pull/2228>`_
* If the number of configured transaction logs were reduced at the exact same time a change to the system keyspace took place, it was possible for the transaction state store to become corrupted. `(PR #3051) <https://github.com/apple/foundationdb/pull/3051>`_
* Fix multiple data races between threads on the client. `(PR #3026) <https://github.com/apple/foundationdb/pull/3026>`_
* Transaction logs configured to spill by reference had an unintended delay between each spilled batch. `(PR #3153) <https://github.com/apple/foundationdb/pull/3153>`_
* Added guards to honor ``DISABLE_POSIX_KERNEL_AIO``. `(PR #2888) <https://github.com/apple/foundationdb/pull/2888>`_
Status
------
* A process's ``memory.available_bytes`` can no longer exceed the memory limit of the process. For purposes of this statistic, processes on the same machine will be allocated memory proportionally based on the size of their memory limits. `(PR #3174) <https://github.com/apple/foundationdb/pull/3174>`_
* Replaced ``cluster.database_locked`` status field with ``cluster.database_lock_state``, which contains two subfields: ``locked`` (boolean) and ``lock_uid`` (which contains the database lock uid if the database is locked). `(PR #2058) <https://github.com/apple/foundationdb/pull/2058>`_
* Removed fields ``worst_version_lag_storage_server`` and ``limiting_version_lag_storage_server`` from the ``cluster.qos`` section. The ``worst_data_lag_storage_server`` and ``limiting_data_lag_storage_server`` objects can be used instead. `(PR #3196) <https://github.com/apple/foundationdb/pull/3196>`_
* If a process is unable to flush trace logs to disk, the problem will now be reported via the output of ``status`` command inside ``fdbcli``. `(PR #2605) <https://github.com/apple/foundationdb/pull/2605>`_ `(PR #2820) <https://github.com/apple/foundationdb/pull/2820>`_
Bindings
--------
* API version updated to 630. See the :ref:`API version upgrade guide <api-version-upgrade-guide-630>` for upgrade details.
* Python: The ``@fdb.transactional`` decorator will now throw an error if the decorated function returns a generator. `(PR #1724) <https://github.com/apple/foundationdb/pull/1724>`_
* Java: Add caching for various JNI objects to improve performance. `(PR #2809) <https://github.com/apple/foundationdb/pull/2809>`_
* Java: Optimize byte array comparisons in ``ByteArrayUtil``. `(PR #2823) <https://github.com/apple/foundationdb/pull/2823>`_
* Java: Add ``FDB.disableShutdownHook`` that can be used to prevent the default shutdown hook from running. Users of this new function should make sure to call ``stopNetwork`` before terminating a client process. `(PR #2635) <https://github.com/apple/foundationdb/pull/2635>`_
* Java: Introduced ``keyAfter`` utility function that can be used to create the immediate next key for a given byte array. `(PR #2458) <https://github.com/apple/foundationdb/pull/2458>`_
* Golang: The ``Transact`` function will unwrap errors that have been wrapped using ``xerrors`` to determine if a retryable FoundationDB error is in the error chain. `(PR #3131) <https://github.com/apple/foundationdb/pull/3131>`_
* Golang: Added ``Subspace.PackWithVersionstamp`` that can be used to pack a ``Tuple`` that contains a versionstamp. `(PR #2243) <https://github.com/apple/foundationdb/pull/2243>`_
* Golang: Implement ``Stringer`` interface for ``Tuple``, ``Subspace``, ``UUID``, and ``Versionstamp``. `(PR #3032) <https://github.com/apple/foundationdb/pull/3032>`_
* C: The ``FDBKeyValue`` struct's ``key`` and ``value`` members have changed type from ``void*`` to ``uint8_t*``. `(PR #2622) <https://github.com/apple/foundationdb/pull/2622>`_
* Deprecated ``enable_slow_task_profiling`` network option and replaced it with ``enable_run_loop_profiling``. `(PR #2608) <https://github.com/apple/foundationdb/pull/2608>`_
Other Changes
-------------
* Small key ranges which are being heavily read will be reported in the logs using the trace event ``ReadHotRangeLog``. `(PR #2046) <https://github.com/apple/foundationdb/pull/2046>`_ `(PR #2378) <https://github.com/apple/foundationdb/pull/2378>`_ `(PR #2532) <https://github.com/apple/foundationdb/pull/2532>`_
* Added the read version, commit version, and datacenter locality to the client transaction information. `(PR #3079) <https://github.com/apple/foundationdb/pull/3079>`_ `(PR #3205) <https://github.com/apple/foundationdb/pull/3205>`_
* Added a network option ``TRACE_FILE_IDENTIFIER`` that can be used to provide a custom identifier string that will be part of the file name for all trace log files created on the client. `(PR #2869) <https://github.com/apple/foundationdb/pull/2869>`_
* It is now possible to use the ``TRACE_LOG_GROUP`` option on a client process after the database has been created. `(PR #2862) <https://github.com/apple/foundationdb/pull/2862>`_
* Added a network option ``TRACE_CLOCK_SOURCE`` that can be used to switch the trace event timestamps to use a realtime clock source. `(PR #2329) <https://github.com/apple/foundationdb/pull/2329>`_
* The ``INCLUDE_PORT_IN_ADDRESS`` transaction option is now on by default. This means ``get_addresses_for_key`` will always return ports in the address strings. `(PR #2639) <https://github.com/apple/foundationdb/pull/2639>`_
* Added the ``getversion`` command to ``fdbcli`` which returns the current read version of the cluster. `(PR #2882) <https://github.com/apple/foundationdb/pull/2882>`_
* Added the ``advanceversion`` command to ``fdbcli`` which increases the current version of a cluster. `(PR #2965) <https://github.com/apple/foundationdb/pull/2965>`_
* Improved the slow task profiler to also report backtraces for periods when the run loop is saturated. `(PR #2608) <https://github.com/apple/foundationdb/pull/2608>`_
* Double the number of shard locations that the client will cache locally. `(PR #2198) <https://github.com/apple/foundationdb/pull/2198>`_
* Replaced the ``-add_prefix`` and ``-remove_prefix`` options with ``--add_prefix`` and ``--remove_prefix`` in ``fdbrestore`` `(PR 3206) <https://github.com/apple/foundationdb/pull/3206>`_
* Data distribution metrics can now be read using the special keyspace ``\xff\xff/metrics/data_distribution_stats``. `(PR #2547) <https://github.com/apple/foundationdb/pull/2547>`_
* The ``\xff\xff/worker_interfaces/`` keyspace now begins at a key which includes a trailing ``/`` (previously ``\xff\xff/worker_interfaces``). Range reads to this range now respect the end key passed into the range and include the keyspace prefix in the resulting keys. `(PR #3095) <https://github.com/apple/foundationdb/pull/3095>`_
* Added FreeBSD support. `(PR #2634) <https://github.com/apple/foundationdb/pull/2634>`_
* Updated boost to 1.72. `(PR #2684) <https://github.com/apple/foundationdb/pull/2684>`_
Earlier release notes
---------------------
* :doc:`6.2 (API Version 620) </old-release-notes/release-notes-620>`
* :doc:`6.1 (API Version 610) </old-release-notes/release-notes-610>`
* :doc:`6.0 (API Version 600) </old-release-notes/release-notes-600>`
* :doc:`5.2 (API Version 520) </old-release-notes/release-notes-520>`
* :doc:`5.1 (API Version 510) </old-release-notes/release-notes-510>`
* :doc:`5.0 (API Version 500) </old-release-notes/release-notes-500>`
* :doc:`4.6 (API Version 460) </old-release-notes/release-notes-460>`
* :doc:`4.5 (API Version 450) </old-release-notes/release-notes-450>`
* :doc:`4.4 (API Version 440) </old-release-notes/release-notes-440>`
* :doc:`4.3 (API Version 430) </old-release-notes/release-notes-430>`
* :doc:`4.2 (API Version 420) </old-release-notes/release-notes-420>`
* :doc:`4.1 (API Version 410) </old-release-notes/release-notes-410>`
* :doc:`4.0 (API Version 400) </old-release-notes/release-notes-400>`
* :doc:`3.0 (API Version 300) </old-release-notes/release-notes-300>`
* :doc:`2.0 (API Version 200) </old-release-notes/release-notes-200>`
* :doc:`1.0 (API Version 100) </old-release-notes/release-notes-100>`
* :doc:`Beta 3 (API Version 23) </old-release-notes/release-notes-023>`
* :doc:`Beta 2 (API Version 22) </old-release-notes/release-notes-022>`
* :doc:`Beta 1 (API Version 21) </old-release-notes/release-notes-021>`
* :doc:`Alpha 6 (API Version 16) </old-release-notes/release-notes-016>`
* :doc:`Alpha 5 (API Version 14) </old-release-notes/release-notes-014>`

View File

@ -74,7 +74,7 @@ Here's a basic implementation of the model:
private static final Random randno;
static{
fdb = FDB.selectAPIVersion(630);
fdb = FDB.selectAPIVersion(700);
db = fdb.open();
pq = new Subspace(Tuple.from("P"));

View File

@ -73,7 +73,7 @@ The following is a simple implementation of the basic pattern:
private static final Random randno;
static{
fdb = FDB.selectAPIVersion(630);
fdb = FDB.selectAPIVersion(700);
db = fdb.open();
queue = new Subspace(Tuple.from("Q"));
randno = new Random();

View File

@ -2,7 +2,7 @@
Release Notes
#############
6.3.0
6.3.1
=====
Features
@ -100,6 +100,7 @@ Other Changes
Earlier release notes
---------------------
* :doc:`6.3 (API Version 630) </old-release-notes/release-notes-630>`
* :doc:`6.2 (API Version 620) </old-release-notes/release-notes-620>`
* :doc:`6.1 (API Version 610) </old-release-notes/release-notes-610>`
* :doc:`6.0 (API Version 600) </old-release-notes/release-notes-600>`

View File

@ -87,7 +87,7 @@ In this example, were storing user data based on user ID but sometimes need t
private static final Subspace index;
static {
fdb = FDB.selectAPIVersion(630);
fdb = FDB.selectAPIVersion(700);
db = fdb.open();
main = new Subspace(Tuple.from("user"));
index = new Subspace(Tuple.from("zipcode_index"));

View File

@ -62,7 +62,7 @@ Heres a simple implementation of the basic table pattern:
private static final Subspace colIndex;
static {
fdb = FDB.selectAPIVersion(630);
fdb = FDB.selectAPIVersion(700);
db = fdb.open();
table = new Subspace(Tuple.from("T"));
rowIndex = table.subspace(Tuple.from("R"));

View File

@ -77,7 +77,7 @@ Heres the basic pattern:
private static final Subspace vector;
static {
fdb = FDB.selectAPIVersion(630);
fdb = FDB.selectAPIVersion(700);
db = fdb.open();
vector = new Subspace(Tuple.from("V"));
}

View File

@ -3465,16 +3465,6 @@ int main(int argc, char* argv[]) {
std::set_new_handler( &platform::outOfMemory );
setMemoryQuota( memLimit );
int total = 0;
for(auto i = Error::errorCounts().begin(); i != Error::errorCounts().end(); ++i)
total += i->second;
if (total)
printf("%d errors:\n", total);
for(auto i = Error::errorCounts().begin(); i != Error::errorCounts().end(); ++i)
if (i->second > 0)
printf(" %d: %d %s\n", i->second, i->first, Error::fromCode(i->first).what());
Reference<ClusterConnectionFile> ccf;
Database db;
Reference<ClusterConnectionFile> sourceCcf;

View File

@ -132,6 +132,13 @@ struct MutationRef {
};
};
template<>
struct Traceable<MutationRef> : std::true_type {
static std::string toString(MutationRef const& value) {
return value.toString();
}
};
static inline std::string getTypeString(MutationRef::Type type) {
return type < MutationRef::MAX_ATOMIC_OP ? typeString[(int)type] : "Unset";
}
@ -206,7 +213,4 @@ struct CommitTransactionRef {
}
};
bool debugMutation( const char* context, Version version, MutationRef const& m );
bool debugKeyRange( const char* context, Version version, KeyRangeRef const& keyRange );
#endif

View File

@ -26,6 +26,7 @@
#include <string>
#include <vector>
#include "flow/Arena.h"
#include "flow/flow.h"
#include "fdbclient/Knobs.h"
@ -77,6 +78,10 @@ struct Tag {
serializer(ar, locality, id);
}
};
template <>
struct flow_ref<Tag> : std::integral_constant<bool, false> {};
#pragma pack(pop)
template <class Ar> void load( Ar& ar, Tag& tag ) { tag.serialize_unversioned(ar); }
@ -108,6 +113,13 @@ struct struct_like_traits<Tag> : std::true_type {
}
};
template<>
struct Traceable<Tag> : std::true_type {
static std::string toString(const Tag& value) {
return value.toString();
}
};
static const Tag invalidTag {tagLocalitySpecial, 0};
static const Tag txsTag {tagLocalitySpecial, 1};
static const Tag cacheTag {tagLocalitySpecial, 2};
@ -222,11 +234,25 @@ std::string describe( std::vector<T> const& items, int max_items = -1 ) {
return describeList(items, max_items);
}
template<typename T>
struct Traceable<std::vector<T>> : std::true_type {
static std::string toString(const std::vector<T>& value) {
return describe(value);
}
};
template <class T>
std::string describe( std::set<T> const& items, int max_items = -1 ) {
return describeList(items, max_items);
}
template<typename T>
struct Traceable<std::set<T>> : std::true_type {
static std::string toString(const std::set<T>& value) {
return describe(value);
}
};
std::string printable( const StringRef& val );
std::string printable( const std::string& val );
std::string printable( const KeyRangeRef& range );

View File

@ -352,6 +352,9 @@ namespace HTTP {
send_start = timer();
loop {
wait(conn->onWritable());
wait( delay( 0, TaskPriority::WriteSocket ) );
// If we already got a response, before finishing sending the request, then close the connection,
// set the Connection header to "close" as a hint to the caller that this connection can't be used
// again, and break out of the send loop.
@ -372,11 +375,6 @@ namespace HTTP {
pContent->sent(len);
if(pContent->empty())
break;
if(len == 0) {
wait(conn->onWritable());
wait( delay( 0, TaskPriority::WriteSocket ) );
}
}
wait(responseReading);

View File

@ -246,20 +246,13 @@ public:
// stream.send( request )
// Unreliable at most once delivery: Delivers request unless there is a connection failure (zero or one times)
void send(const T& value) const {
template<class U>
void send(U && value) const {
if (queue->isRemoteEndpoint()) {
FlowTransport::transport().sendUnreliable(SerializeSource<T>(value), getEndpoint(), true);
FlowTransport::transport().sendUnreliable(SerializeSource<T>(std::forward<U>(value)), getEndpoint(), true);
}
else
queue->send(value);
}
void send(T&& value) const {
if (queue->isRemoteEndpoint()) {
FlowTransport::transport().sendUnreliable(SerializeSource<T>(std::move(value)), getEndpoint(), true);
}
else
queue->send(std::move(value));
queue->send(std::forward<U>(value));
}
/*void sendError(const Error& error) const {

View File

@ -1654,15 +1654,8 @@ public:
this->currentProcess = t.machine;
try {
//auto before = getCPUTicks();
t.action.send(Void());
ASSERT( this->currentProcess == t.machine );
/*auto elapsed = getCPUTicks() - before;
currentProcess->cpuTicks += elapsed;
if (deterministicRandom()->random01() < 0.01){
TraceEvent("TaskDuration").detail("CpuTicks", currentProcess->cpuTicks);
currentProcess->cpuTicks = 0;
}*/
} catch (Error& e) {
TraceEvent(SevError, "UnhandledSimulationEventError").error(e, true);
killProcess(t.machine, KillInstantly);

View File

@ -58,7 +58,6 @@ public:
bool failed;
bool excluded;
bool cleared;
int64_t cpuTicks;
bool rebooting;
std::vector<flowGlobalType> globals;
@ -68,12 +67,11 @@ public:
double fault_injection_p1, fault_injection_p2;
ProcessInfo(const char* name, LocalityData locality, ProcessClass startingClass, NetworkAddressList addresses,
INetworkConnections *net, const char* dataFolder, const char* coordinationFolder )
: name(name), locality(locality), startingClass(startingClass),
addresses(addresses), address(addresses.address), dataFolder(dataFolder),
network(net), coordinationFolder(coordinationFolder), failed(false), excluded(false), cpuTicks(0),
rebooting(false), fault_injection_p1(0), fault_injection_p2(0),
fault_injection_r(0), machine(0), cleared(false) {}
INetworkConnections* net, const char* dataFolder, const char* coordinationFolder)
: name(name), locality(locality), startingClass(startingClass), addresses(addresses),
address(addresses.address), dataFolder(dataFolder), network(net), coordinationFolder(coordinationFolder),
failed(false), excluded(false), rebooting(false), fault_injection_p1(0), fault_injection_p2(0),
fault_injection_r(0), machine(0), cleared(false) {}
Future<KillType> onShutdown() { return shutdownSignal.getFuture(); }

View File

@ -115,7 +115,7 @@ std::map<std::tuple<LogEpoch, Version, int>, std::map<Tag, Version>> BackupProgr
// ASSERT(info.logRouterTags == epochTags[rit->first]);
updateTagVersions(&tagVersions, &tags, rit->second, info.epochEnd, adjustedBeginVersion, epoch);
break;
if (tags.empty()) break;
}
rit++;
}

View File

@ -734,13 +734,11 @@ ACTOR Future<Void> saveMutationsToFile(BackupData* self, Version popVersion, int
MutationRef m;
if (!message.isBackupMessage(&m)) continue;
if (debugMutation("addMutation", message.version.version, m)) {
TraceEvent("BackupWorkerDebug", self->myId)
DEBUG_MUTATION("addMutation", message.version.version, m)
.detail("Version", message.version.toString())
.detail("Mutation", m.toString())
.detail("Mutation", m)
.detail("KCV", self->minKnownCommittedVersion)
.detail("SavedVersion", self->savedVersion);
}
std::vector<Future<Void>> adds;
if (m.type != MutationRef::Type::ClearRange) {

View File

@ -46,8 +46,10 @@ set(FDBSERVER_SRCS
MasterInterface.h
MasterProxyServer.actor.cpp
masterserver.actor.cpp
MoveKeys.actor.cpp
MutationTracking.h
MutationTracking.cpp
MoveKeys.actor.h
MoveKeys.actor.cpp
networktest.actor.cpp
NetworkTest.h
OldTLogServer_4_6.actor.cpp

View File

@ -1013,7 +1013,7 @@ private:
ASSERT( nextPageSeq%sizeof(Page)==0 );
auto& p = backPage();
memset(&p, 0, sizeof(Page)); // FIXME: unnecessary?
memset(static_cast<void*>(&p), 0, sizeof(Page)); // FIXME: unnecessary?
p.magic = 0xFDB;
switch (diskQueueVersion) {
case DiskQueueVersion::V0:

View File

@ -27,6 +27,7 @@
#include "fdbserver/TLogInterface.h"
#include "fdbserver/WorkerInterface.actor.h"
#include "fdbclient/DatabaseConfiguration.h"
#include "fdbserver/MutationTracking.h"
#include "flow/IndexedSet.h"
#include "fdbrpc/ReplicationPolicy.h"
#include "fdbrpc/Locality.h"
@ -877,17 +878,27 @@ struct LogPushData : NonCopyable {
msg_locations.clear();
logSystem->getPushLocations(prev_tags, msg_locations, allLocations);
BinaryWriter bw(AssumeVersion(currentProtocolVersion));
uint32_t subseq = this->subsequence++;
bool first = true;
int firstOffset=-1, firstLength=-1;
for(int loc : msg_locations) {
// FIXME: memcpy after the first time
BinaryWriter& wr = messagesWriter[loc];
int offset = wr.getLength();
wr << uint32_t(0) << subseq << uint16_t(prev_tags.size());
for(auto& tag : prev_tags) {
wr << tag;
if (first) {
BinaryWriter& wr = messagesWriter[loc];
firstOffset = wr.getLength();
wr << uint32_t(0) << subseq << uint16_t(prev_tags.size());
for(auto& tag : prev_tags)
wr << tag;
wr << item;
firstLength = wr.getLength() - firstOffset;
*(uint32_t*)((uint8_t*)wr.getData() + firstOffset) = firstLength - sizeof(uint32_t);
DEBUG_TAGS_AND_MESSAGE("ProxyPushLocations", invalidVersion, StringRef(((uint8_t*)wr.getData() + firstOffset), firstLength)).detail("PushLocations", msg_locations);
first = false;
} else {
BinaryWriter& wr = messagesWriter[loc];
BinaryWriter& from = messagesWriter[msg_locations[0]];
wr.serializeBytes( (uint8_t*)from.getData() + firstOffset, firstLength );
}
wr << item;
*(uint32_t*)((uint8_t*)wr.getData() + offset) = wr.getLength() - offset - sizeof(uint32_t);
}
next_message_tags.clear();
}

View File

@ -21,6 +21,7 @@
#include "fdbserver/LogSystem.h"
#include "fdbrpc/FailureMonitor.h"
#include "fdbserver/Knobs.h"
#include "fdbserver/MutationTracking.h"
#include "fdbrpc/ReplicationUtils.h"
#include "flow/actorcompiler.h" // has to be last include
@ -90,6 +91,7 @@ void ILogSystem::ServerPeekCursor::nextMessage() {
}
messageAndTags.loadFromArena(&rd, &messageVersion.sub);
DEBUG_TAGS_AND_MESSAGE("ServerPeekCursor", messageVersion.version, messageAndTags.getRawMessage()).detail("CursorID", this->randomID);
// Rewind and consume the header so that reader() starts from the message.
rd.rewind();
rd.readBytes(messageAndTags.getHeaderSize());

View File

@ -38,6 +38,7 @@
#include "fdbserver/LogSystem.h"
#include "fdbserver/LogSystemDiskQueueAdapter.h"
#include "fdbserver/MasterInterface.h"
#include "fdbserver/MutationTracking.h"
#include "fdbserver/RecoveryState.h"
#include "fdbserver/ServerDBInfo.h"
#include "fdbserver/WaitFailure.h"
@ -759,7 +760,7 @@ ACTOR Future<Void> addBackupMutations(ProxyCommitData* self, std::map<Key, Mutat
toCommit->addTags(tags);
toCommit->addTypedMessage(backupMutation);
// if (debugMutation("BackupProxyCommit", commitVersion, backupMutation)) {
// if (DEBUG_MUTATION("BackupProxyCommit", commitVersion, backupMutation)) {
// TraceEvent("BackupProxyCommitTo", self->dbgid).detail("To", describe(tags)).detail("BackupMutation", backupMutation.toString())
// .detail("BackupMutationSize", val.size()).detail("Version", commitVersion).detail("DestPath", logRangeMutation.first)
// .detail("PartIndex", part).detail("PartIndexEndian", bigEndian32(part)).detail("PartData", backupMutation.param1);
@ -1079,8 +1080,7 @@ ACTOR Future<Void> commitBatch(
self->singleKeyMutationEvent->log();
}
if (debugMutation("ProxyCommit", commitVersion, m))
TraceEvent("ProxyCommitTo", self->dbgid).detail("To", describe(tags)).detail("Mutation", m.toString()).detail("Version", commitVersion);
DEBUG_MUTATION("ProxyCommit", commitVersion, m).detail("Dbgid", self->dbgid).detail("To", tags).detail("Mutation", m);
toCommit.addTags(tags);
if(self->cacheInfo[m.param1]) {
@ -1095,8 +1095,7 @@ ACTOR Future<Void> commitBatch(
++firstRange;
if (firstRange == ranges.end()) {
// Fast path
if (debugMutation("ProxyCommit", commitVersion, m))
TraceEvent("ProxyCommitTo", self->dbgid).detail("To", describe(ranges.begin().value().tags)).detail("Mutation", m.toString()).detail("Version", commitVersion);
DEBUG_MUTATION("ProxyCommit", commitVersion, m).detail("Dbgid", self->dbgid).detail("To", ranges.begin().value().tags).detail("Mutation", m);
ranges.begin().value().populateTags();
toCommit.addTags(ranges.begin().value().tags);
@ -1108,8 +1107,7 @@ ACTOR Future<Void> commitBatch(
r.value().populateTags();
allSources.insert(r.value().tags.begin(), r.value().tags.end());
}
if (debugMutation("ProxyCommit", commitVersion, m))
TraceEvent("ProxyCommitTo", self->dbgid).detail("To", describe(allSources)).detail("Mutation", m.toString()).detail("Version", commitVersion);
DEBUG_MUTATION("ProxyCommit", commitVersion, m).detail("Dbgid", self->dbgid).detail("To", allSources).detail("Mutation", m);
toCommit.addTags(allSources);
}

View File

@ -0,0 +1,101 @@
/*
* MutationTracking.cpp
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2020 Apple Inc. and the FoundationDB project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <vector>
#include "fdbserver/MutationTracking.h"
#include "fdbserver/LogProtocolMessage.h"
#if defined(FDB_CLEAN_BUILD) && MUTATION_TRACKING_ENABLED
#error "You cannot use mutation tracking in a clean/release build."
#endif
// Track up to 2 keys in simulation via enabling MUTATION_TRACKING_ENABLED and setting the keys here.
StringRef debugKey = LiteralStringRef( "" );
StringRef debugKey2 = LiteralStringRef( "\xff\xff\xff\xff" );
TraceEvent debugMutationEnabled( const char* context, Version version, MutationRef const& mutation ) {
if ((mutation.type == mutation.ClearRange || mutation.type == mutation.DebugKeyRange) &&
((mutation.param1<=debugKey && mutation.param2>debugKey) || (mutation.param1<=debugKey2 && mutation.param2>debugKey2))) {
return std::move(TraceEvent("MutationTracking").detail("At", context).detail("Version", version).detail("MutationType", typeString[mutation.type]).detail("KeyBegin", mutation.param1).detail("KeyEnd", mutation.param2));
} else if (mutation.param1 == debugKey || mutation.param1 == debugKey2) {
return std::move(TraceEvent("MutationTracking").detail("At", context).detail("Version", version).detail("MutationType", typeString[mutation.type]).detail("Key", mutation.param1).detail("Value", mutation.param2));
} else {
return std::move(TraceEvent());
}
}
TraceEvent debugKeyRangeEnabled( const char* context, Version version, KeyRangeRef const& keys ) {
if (keys.contains(debugKey) || keys.contains(debugKey2)) {
return std::move(debugMutation(context, version, MutationRef(MutationRef::DebugKeyRange, keys.begin, keys.end) ));
} else {
return std::move(TraceEvent());
}
}
TraceEvent debugTagsAndMessageEnabled( const char* context, Version version, StringRef commitBlob ) {
BinaryReader rdr(commitBlob, AssumeVersion(currentProtocolVersion));
while (!rdr.empty()) {
if (*(int32_t*)rdr.peekBytes(4) == VERSION_HEADER) {
int32_t dummy;
rdr >> dummy >> version;
continue;
}
TagsAndMessage msg;
msg.loadFromArena(&rdr, nullptr);
bool logAdapterMessage = std::any_of(
msg.tags.begin(), msg.tags.end(), [](const Tag& t) { return t == txsTag || t.locality == tagLocalityTxs; });
StringRef mutationData = msg.getMessageWithoutTags();
uint8_t mutationType = *mutationData.begin();
if (logAdapterMessage) {
// Skip the message, as there will always be an idential non-logAdapterMessage mutation
// that we can match against in the same commit.
} else if (LogProtocolMessage::startsLogProtocolMessage(mutationType)) {
BinaryReader br(mutationData, AssumeVersion(rdr.protocolVersion()));
LogProtocolMessage lpm;
br >> lpm;
rdr.setProtocolVersion(br.protocolVersion());
} else {
MutationRef m;
BinaryReader br(mutationData, AssumeVersion(rdr.protocolVersion()));
br >> m;
TraceEvent&& event = debugMutation(context, version, m);
if (event.isEnabled()) {
return std::move(event.detail("MessageTags", msg.tags));
}
}
}
return std::move(TraceEvent());
}
#if MUTATION_TRACKING_ENABLED
TraceEvent debugMutation( const char* context, Version version, MutationRef const& mutation ) {
return debugMutationEnabled( context, version, mutation );
}
TraceEvent debugKeyRange( const char* context, Version version, KeyRangeRef const& keys ) {
return debugKeyRangeEnabled( context, version, keys );
}
TraceEvent debugTagsAndMessage( const char* context, Version version, StringRef commitBlob ) {
return debugTagsAndMessageEnabled( context, version, commitBlob );
}
#else
TraceEvent debugMutation( const char* context, Version version, MutationRef const& mutation ) { return std::move(TraceEvent()); }
TraceEvent debugKeyRange( const char* context, Version version, KeyRangeRef const& keys ) { return std::move(TraceEvent()); }
TraceEvent debugTagsAndMessage( const char* context, Version version, StringRef commitBlob ) { return std::move(TraceEvent()); }
#endif

View File

@ -0,0 +1,49 @@
/*
* MutationTracking.h
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2020 Apple Inc. and the FoundationDB project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#ifndef _FDBSERVER_MUTATIONTRACKING_H_
#define _FDBSERVER_MUTATIONTRACKING_H_
#pragma once
#include "fdbclient/FDBTypes.h"
#include "fdbclient/CommitTransaction.h"
#define MUTATION_TRACKING_ENABLED 0
// The keys to track are defined in the .cpp file to limit recompilation.
#define DEBUG_MUTATION(context, version, mutation) MUTATION_TRACKING_ENABLED && debugMutation(context, version, mutation)
TraceEvent debugMutation( const char* context, Version version, MutationRef const& mutation );
// debugKeyRange and debugTagsAndMessage only log the *first* occurrence of a key in their range/commit.
// TODO: Create a TraceEventGroup that forwards all calls to each element of a vector<TraceEvent>,
// to allow "multiple" TraceEvents to be returned.
#define DEBUG_KEY_RANGE(context, version, keys) MUTATION_TRACKING_ENABLED && debugKeyRange(context, version, keys)
TraceEvent debugKeyRange( const char* context, Version version, KeyRangeRef const& keys );
#define DEBUG_TAGS_AND_MESSAGE(context, version, commitBlob) MUTATION_TRACKING_ENABLED && debugTagsAndMessage(context, version, commitBlob)
TraceEvent debugTagsAndMessage( const char* context, Version version, StringRef commitBlob );
// TODO: Version Tracking. If the bug is in handling a version rather than a key, then it'd be good to be able to log each time
// that version is handled within simulation. A similar set of functions should be implemented.
#endif

View File

@ -83,6 +83,7 @@ ACTOR Future<Void> restoreApplierCore(RestoreApplierInterface applierInterf, int
updateProcessStats(self);
updateProcessStatsTimer = delay(SERVER_KNOBS->FASTRESTORE_UPDATE_PROCESS_STATS_INTERVAL);
}
when(wait(actors.getResult())) {}
when(wait(exitRole)) {
TraceEvent("RestoreApplierCoreExitRole", self->id());
break;
@ -92,6 +93,7 @@ ACTOR Future<Void> restoreApplierCore(RestoreApplierInterface applierInterf, int
TraceEvent(SevWarn, "FastRestoreApplierError", self->id())
.detail("RequestType", requestTypeStr)
.error(e, true);
actors.clear(false);
break;
}
}
@ -211,52 +213,75 @@ ACTOR static Future<Void> applyClearRangeMutations(Standalone<VectorRef<KeyRange
return Void();
}
ACTOR Future<Optional<Value>> getValue(Reference<ReadYourWritesTransaction> tr, Key key, int i,
std::set<int>* keysNotFound) {
try {
Optional<Value> v = wait(tr->get(key));
return v;
} catch (Error& e) {
if (e.code() == error_code_key_not_found) {
keysNotFound->insert(i);
return Optional<Value>();
} else {
throw;
}
}
}
// Get keys in incompleteStagingKeys and precompute the stagingKey which is stored in batchData->stagingKeys
ACTOR static Future<Void> getAndComputeStagingKeys(
std::map<Key, std::map<Key, StagingKey>::iterator> incompleteStagingKeys, double delayTime, Database cx,
UID applierID, int batchIndex) {
state Reference<ReadYourWritesTransaction> tr(new ReadYourWritesTransaction(cx));
state std::vector<Future<Optional<Value>>> fValues;
state std::vector<Future<Optional<Value>>> fValues(incompleteStagingKeys.size(), Never());
state int retries = 0;
state UID randomID = deterministicRandom()->randomUniqueID();
wait(delay(delayTime + deterministicRandom()->random01() * delayTime));
TraceEvent("FastRestoreApplierGetAndComputeStagingKeysStart", applierID)
.detail("RandomUID", randomID)
.detail("BatchIndex", batchIndex)
.detail("GetKeys", incompleteStagingKeys.size())
.detail("DelayTime", delayTime);
state std::set<int> keysNotFound;
state int i = 0;
loop {
try {
tr->reset();
tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
tr->setOption(FDBTransactionOptions::LOCK_AWARE);
i = 0;
for (auto& key : incompleteStagingKeys) {
fValues.push_back(tr->get(key.first));
if (!keysNotFound.count(i)) { // only get exist-keys
fValues[i] = getValue(tr, key.first, i, &keysNotFound);
}
++i;
}
wait(waitForAll(fValues));
break;
} catch (Error& e) {
if (retries++ > 10) { // TODO: Can we stop retry at the first error?
TraceEvent(SevWarn, "FastRestoreApplierGetAndComputeStagingKeysGetKeysStuck", applierID)
bool ok = (e.code() != error_code_key_not_found);
if (!ok || retries++ > incompleteStagingKeys.size()) {
TraceEvent(!ok ? SevError : SevWarnAlways, "GetAndComputeStagingKeys", applierID)
.detail("BatchIndex", batchIndex)
.detail("GetKeys", incompleteStagingKeys.size())
.detail("KeyIndex", i)
.error(e);
break;
}
wait(tr->onError(e));
fValues.clear();
}
}
ASSERT(fValues.size() == incompleteStagingKeys.size());
int i = 0;
for (auto& key : incompleteStagingKeys) {
if (!fValues[i].get().present()) { // Debug info to understand which key does not exist in DB
if (keysNotFound.count(i) || (!fValues[i].get().present())) { // Key not exist in DB
// if condition: fValues[i].Valid() && fValues[i].isReady() && !fValues[i].isError() &&
TraceEvent(SevWarn, "FastRestoreApplierGetAndComputeStagingKeysNoBaseValueInDB", applierID)
.detail("BatchIndex", batchIndex)
.detail("Key", key.first)
.detail("Reason", "Not found in DB")
.detail("IsReady", fValues[i].isReady())
.detail("PendingMutations", key.second->second.pendingMutations.size())
.detail("StagingKeyType", (int)key.second->second.type);
.detail("StagingKeyType", getTypeString(key.second->second.type));
for (auto& vm : key.second->second.pendingMutations) {
TraceEvent(SevWarn, "FastRestoreApplierGetAndComputeStagingKeysNoBaseValueInDB")
.detail("PendingMutationVersion", vm.first.toString())
@ -274,8 +299,10 @@ ACTOR static Future<Void> getAndComputeStagingKeys(
}
TraceEvent("FastRestoreApplierGetAndComputeStagingKeysDone", applierID)
.detail("RandomUID", randomID)
.detail("BatchIndex", batchIndex)
.detail("GetKeys", incompleteStagingKeys.size());
.detail("GetKeys", incompleteStagingKeys.size())
.detail("DelayTime", delayTime);
return Void();
}
@ -502,6 +529,7 @@ ACTOR static Future<Void> handleApplyToDBRequest(RestoreVersionBatchRequest req,
.detail("FinishedBatch", self->finishedBatch.get());
// Ensure batch (i-1) is applied before batch i
// TODO: Add a counter to warn when too many requests are waiting on the actor
wait(self->finishedBatch.whenAtLeast(req.batchIndex - 1));
state bool isDuplicated = true;
@ -523,6 +551,8 @@ ACTOR static Future<Void> handleApplyToDBRequest(RestoreVersionBatchRequest req,
}
ASSERT(batchData->dbApplier.present());
ASSERT(!batchData->dbApplier.get().isError()); // writeMutationsToDB actor cannot have error.
// We cannot blindly retry because it is not idempodent
wait(batchData->dbApplier.get());
@ -578,4 +608,4 @@ Value applyAtomicOp(Optional<StringRef> existingValue, Value value, MutationRef:
ASSERT(false);
}
return Value();
}
}

View File

@ -36,6 +36,7 @@
#include "fdbrpc/Locality.h"
#include "fdbserver/CoordinationInterface.h"
#include "fdbclient/RestoreWorkerInterface.actor.h"
#include "fdbserver/MutationTracking.h"
#include "fdbserver/RestoreUtil.h"
#include "fdbserver/RestoreRoleCommon.actor.h"
@ -60,19 +61,17 @@ struct StagingKey {
// Assume: SetVersionstampedKey and SetVersionstampedValue have been converted to set
void add(const MutationRef& m, LogMessageVersion newVersion) {
ASSERT(m.type != MutationRef::SetVersionstampedKey && m.type != MutationRef::SetVersionstampedValue);
if (debugMutation("StagingKeyAdd", newVersion.version, m)) {
TraceEvent("StagingKeyAdd")
.detail("Version", version.toString())
.detail("NewVersion", newVersion.toString())
.detail("Mutation", m.toString());
}
DEBUG_MUTATION("StagingKeyAdd", newVersion.version, m)
.detail("Version", version.toString())
.detail("NewVersion", newVersion.toString())
.detail("Mutation", m);
if (version == newVersion) {
// This could happen because the same mutation can be present in
// overlapping mutation logs, because new TLogs can copy mutations
// from old generation TLogs (or backup worker is recruited without
// knowning previously saved progress).
ASSERT(type == m.type && key == m.param1 && val == m.param2);
TraceEvent("SameVersion").detail("Version", version.toString()).detail("Mutation", m.toString());
TraceEvent("SameVersion").detail("Version", version.toString()).detail("Mutation", m);
return;
}
@ -84,15 +83,13 @@ struct StagingKey {
ASSERT(m.param1 == m.param2);
}
if (version < newVersion) {
if (debugMutation("StagingKeyAdd", newVersion.version, m)) {
TraceEvent("StagingKeyAdd")
DEBUG_MUTATION("StagingKeyAdd", newVersion.version, m)
.detail("Version", version.toString())
.detail("NewVersion", newVersion.toString())
.detail("MType", getTypeString(type))
.detail("Key", key)
.detail("Val", val)
.detail("NewMutation", m.toString());
}
key = m.param1;
val = m.param2;
type = (MutationRef::Type)m.type;
@ -108,8 +105,8 @@ struct StagingKey {
TraceEvent("SameVersion")
.detail("Version", version.toString())
.detail("NewVersion", newVersion.toString())
.detail("OldMutation", it->second.toString())
.detail("NewMutation", m.toString());
.detail("OldMutation", it->second)
.detail("NewMutation", m);
ASSERT(it->second.type == m.type && it->second.param1 == m.param1 && it->second.param2 == m.param2);
}
}
@ -126,7 +123,8 @@ struct StagingKey {
.detail("Value", val)
.detail("MType", type < MutationRef::MAX_ATOMIC_OP ? getTypeString(type) : "[Unset]")
.detail("LargestPendingVersion",
(pendingMutations.empty() ? "[none]" : pendingMutations.rbegin()->first.toString()));
(pendingMutations.empty() ? "[none]" : pendingMutations.rbegin()->first.toString()))
.detail("PendingMutations", pendingMutations.size());
std::map<LogMessageVersion, Standalone<MutationRef>>::iterator lb = pendingMutations.lower_bound(version);
if (lb == pendingMutations.end()) {
return;

View File

@ -26,6 +26,7 @@
#include "fdbclient/BackupAgent.actor.h"
#include "fdbserver/RestoreLoader.actor.h"
#include "fdbserver/RestoreRoleCommon.actor.h"
#include "fdbserver/MutationTracking.h"
#include "flow/actorcompiler.h" // This must be the last #include.
@ -110,13 +111,17 @@ ACTOR Future<Void> restoreLoaderCore(RestoreLoaderInterface loaderInterf, int no
updateProcessStats(self);
updateProcessStatsTimer = delay(SERVER_KNOBS->FASTRESTORE_UPDATE_PROCESS_STATS_INTERVAL);
}
when(wait(actors.getResult())) {}
when(wait(exitRole)) {
TraceEvent("FastRestoreLoaderCoreExitRole", self->id());
break;
}
}
} catch (Error& e) {
TraceEvent(SevWarn, "FastRestoreLoader", self->id()).detail("RequestType", requestTypeStr).error(e, true);
TraceEvent(SevWarn, "FastRestoreLoaderError", self->id())
.detail("RequestType", requestTypeStr)
.error(e, true);
actors.clear(false);
break;
}
}
@ -508,23 +513,23 @@ ACTOR Future<Void> sendMutationsToApplier(VersionedMutationsMap* pkvOps, int bat
nodeIDs.contents());
ASSERT(mvector.size() == nodeIDs.size());
if (debugMutation("RestoreLoader", commitVersion.version, kvm)) {
TraceEvent e("DebugSplit");
int i = 0;
for (auto& [key, uid] : *pRangeToApplier) {
e.detail(format("Range%d", i).c_str(), printable(key))
.detail(format("UID%d", i).c_str(), uid.toString());
i++;
if (MUTATION_TRACKING_ENABLED) {
TraceEvent&& e = debugMutation("RestoreLoaderDebugSplit", commitVersion.version, kvm);
if (e.isEnabled()) {
int i = 0;
for (auto& [key, uid] : *pRangeToApplier) {
e.detail(format("Range%d", i).c_str(), printable(key))
.detail(format("UID%d", i).c_str(), uid.toString());
i++;
}
}
}
for (splitMutationIndex = 0; splitMutationIndex < mvector.size(); splitMutationIndex++) {
MutationRef mutation = mvector[splitMutationIndex];
UID applierID = nodeIDs[splitMutationIndex];
if (debugMutation("RestoreLoader", commitVersion.version, mutation)) {
TraceEvent("SplittedMutation")
.detail("Version", commitVersion.toString())
.detail("Mutation", mutation.toString());
}
DEBUG_MUTATION("RestoreLoaderSplittedMutation", commitVersion.version, mutation)
.detail("Version", commitVersion.toString())
.detail("Mutation", mutation);
// CAREFUL: The splitted mutations' lifetime is shorter than the for-loop
// Must use deep copy for splitted mutations
applierVersionedMutationsBuffer[applierID].push_back_deep(
@ -540,12 +545,10 @@ ACTOR Future<Void> sendMutationsToApplier(VersionedMutationsMap* pkvOps, int bat
UID applierID = itlow->second;
kvCount++;
if (debugMutation("RestoreLoader", commitVersion.version, kvm)) {
TraceEvent("SendMutation")
.detail("Applier", applierID)
.detail("Version", commitVersion.toString())
.detail("Mutation", kvm.toString());
}
DEBUG_MUTATION("RestoreLoaderSendMutation", commitVersion.version, kvm)
.detail("Applier", applierID)
.detail("Version", commitVersion.toString())
.detail("Mutation", kvm);
// kvm data is saved in pkvOps in batchData, so shallow copy is ok here.
applierVersionedMutationsBuffer[applierID].push_back(applierVersionedMutationsBuffer[applierID].arena(),
VersionedMutation(kvm, commitVersion));
@ -1057,4 +1060,4 @@ TEST_CASE("/FastRestore/RestoreLoader/splitMutation") {
}
return Void();
}
}

View File

@ -866,6 +866,7 @@ ACTOR static Future<Void> notifyApplierToApplyMutations(Reference<MasterBatchDat
}
ASSERT(batchData->applyToDB.present());
ASSERT(!batchData->applyToDB.get().isError());
wait(batchData->applyToDB.get());
// Sanity check all appliers have applied data to destination DB

View File

@ -76,4 +76,4 @@ bool isRangeMutation(MutationRef m) {
ASSERT(m.type == MutationRef::Type::SetValue || isAtomicOp((MutationRef::Type)m.type));
return false;
}
}
}

View File

@ -31,6 +31,7 @@
#include "fdbclient/Notified.h"
#include "fdbserver/LogProtocolMessage.h"
#include "fdbserver/LogSystem.h"
#include "fdbserver/MutationTracking.h"
#include "fdbserver/WaitFailure.h"
#include "fdbserver/WorkerInterface.actor.h"
#include "fdbclient/DatabaseContext.h"
@ -488,9 +489,8 @@ ACTOR Future<Void> getValueQ( StorageCacheData* data, GetValueRequest req ) {
data->checkChangeCounter(changeCounter, req.key);
}
// FIXME: enable when debugMutation is active
//debugMutation("CacheGetValue", version, MutationRef(MutationRef::DebugKey, req.key, v.present()?v.get():LiteralStringRef("<null>")));
//debugMutation("CacheGetPath", version, MutationRef(MutationRef::DebugKey, req.key, path==0?LiteralStringRef("0"):path==1?LiteralStringRef("1"):LiteralStringRef("2")));
//DEBUG_MUTATION("CacheGetValue", version, MutationRef(MutationRef::DebugKey, req.key, v.present()?v.get():LiteralStringRef("<null>")));
//DEBUG_MUTATION("CacheGetPath", version, MutationRef(MutationRef::DebugKey, req.key, path==0?LiteralStringRef("0"):path==1?LiteralStringRef("1"):LiteralStringRef("2")));
if (v.present()) {
++data->counters.rowsQueried;
@ -1022,23 +1022,11 @@ void StorageCacheData::addMutation(KeyRangeRef const& cachedKeyRange, Version ve
return;
}
expanded = addMutationToMutationLog(mLog, expanded);
// FIXME: enable when debugMutation is active
if (false && debugMutation("expandedMutation", version, expanded)) {
const char* type =
mutation.type == MutationRef::SetValue ? "SetValue" :
mutation.type == MutationRef::ClearRange ? "ClearRange" :
mutation.type == MutationRef::DebugKeyRange ? "DebugKeyRange" :
mutation.type == MutationRef::DebugKey ? "DebugKey" :
"UnknownMutation";
printf("DEBUGMUTATION:\t%.6f\t%s\t%s\t%s\t%s\t%s\n",
now(), g_network->getLocalAddress().toString().c_str(), "originalMutation",
type, printable(mutation.param1).c_str(), printable(mutation.param2).c_str());
printf(" Cached Key-range: %s - %s\n", printable(cachedKeyRange.begin).c_str(), printable(cachedKeyRange.end).c_str());
}
applyMutation( expanded, mLog.arena(), mutableData() );
DEBUG_MUTATION("expandedMutation", version, expanded).detail("Begin", cachedKeyRange.begin).detail("End", cachedKeyRange.end);
applyMutation( this, expanded, mLog.arena(), mutableData() );
//printf("\nSCUpdate: Printing versioned tree after applying mutation\n");
//mutableData().printTree(version);
}
void removeDataRange( StorageCacheData *sc, Standalone<VersionUpdateRef> &mLV, KeyRangeMap<Reference<CacheRangeInfo>>& cacheRanges, KeyRangeRef range ) {
@ -1596,15 +1584,11 @@ public:
data->mutableData().createNewVersion(ver);
}
DEBUG_MUTATION("SCUpdateMutation", ver, m);
if (m.param1.startsWith( systemKeys.end )) {
//TraceEvent("SCPrivateData", data->thisServerID).detail("Mutation", m.toString()).detail("Version", ver);
applyPrivateCacheData( data, m );
} else {
// FIXME: enable when debugMutation is active
//for(auto m = changes[c].mutations.begin(); m; ++m) {
// debugMutation("SCUpdateMutation", changes[c].version, *m);
//}
splitMutation(data, data->cachedRangeMap, m, ver);
}
@ -1623,7 +1607,8 @@ private:
// that this cache server is responsible for
// TODO Revisit during failure handling. Might we loose some private mutations?
void applyPrivateCacheData( StorageCacheData* data, MutationRef const& m ) {
//TraceEvent(SevDebug, "SCPrivateCacheMutation", data->thisServerID).detail("Mutation", m.toString());
//TraceEvent(SevDebug, "SCPrivateCacheMutation", data->thisServerID).detail("Mutation", m);
if (processedCacheStartKey) {
// we expect changes in pairs, [begin,end). This mutation is for end key of the range
@ -1896,9 +1881,8 @@ ACTOR Future<Void> pullAsyncData( StorageCacheData *data ) {
data->debug_inApplyUpdate = false;
if(ver != invalidVersion && ver > data->version.get()) {
// FIXME: enable when debugKeyRange is active
//debugKeyRange("SCUpdate", ver, allKeys);
if(ver != invalidVersion && ver > data->version.get()) {
DEBUG_KEY_RANGE("SCUpdate", ver, allKeys);
data->mutableData().createNewVersion(ver);

View File

@ -32,6 +32,7 @@
#include "fdbserver/TLogInterface.h"
#include "fdbserver/Knobs.h"
#include "fdbserver/IKeyValueStore.h"
#include "fdbserver/MutationTracking.h"
#include "flow/ActorCollection.h"
#include "fdbrpc/FailureMonitor.h"
#include "fdbserver/IDiskQueue.h"
@ -1261,6 +1262,7 @@ void commitMessages( TLogData* self, Reference<LogData> logData, Version version
block.reserve(block.arena(), std::max<int64_t>(SERVER_KNOBS->TLOG_MESSAGE_BLOCK_BYTES, msgSize));
}
DEBUG_TAGS_AND_MESSAGE("TLogCommitMessages", version, msg.getRawMessage()).detail("UID", self->dbgid).detail("LogId", logData->logId);
block.append(block.arena(), msg.message.begin(), msg.message.size());
for(auto tag : msg.tags) {
if(logData->locality == tagLocalitySatellite) {
@ -1375,7 +1377,12 @@ void peekMessagesFromMemory( Reference<LogData> self, TLogPeekRequest const& req
messages << VERSION_HEADER << currentVersion;
}
// We need the 4 byte length prefix to be a TagsAndMessage format, but that prefix is added as part of StringRef serialization.
int offset = messages.getLength();
messages << it->second.toStringRef();
void* data = messages.getData();
DEBUG_TAGS_AND_MESSAGE("TLogPeek", currentVersion, StringRef((uint8_t*)data+offset, messages.getLength()-offset))
.detail("LogId", self->logId).detail("PeekTag", req.tag);
}
}
@ -1637,6 +1644,7 @@ ACTOR Future<Void> tLogPeekMessages( TLogData* self, TLogPeekRequest req, Refere
wait(parseMessagesForTag(entry.messages, req.tag, logData->logRouterTags));
for (const StringRef& msg : rawMessages) {
messages.serializeBytes(msg);
DEBUG_TAGS_AND_MESSAGE("TLogPeekFromDisk", entry.version, msg).detail("UID", self->dbgid).detail("LogId", logData->logId).detail("PeekTag", req.tag);
}
lastRefMessageVersion = entry.version;

View File

@ -531,7 +531,7 @@ static int asyncOpen(
if (flags & SQLITE_OPEN_WAL) oflags |= IAsyncFile::OPEN_LARGE_PAGES;
oflags |= IAsyncFile::OPEN_LOCK;
memset(p, 0, sizeof(VFSAsyncFile));
memset(static_cast<void*>(p), 0, sizeof(VFSAsyncFile));
new (p) VFSAsyncFile(zName, flags);
try {
// Note that SQLiteDB::open also opens the db file, so its flags and modes are important, too

View File

@ -2204,6 +2204,7 @@ struct SplitStringRef {
// A BTree "page id" is actually a list of LogicalPageID's whose contents should be concatenated together.
// NOTE: Uses host byte order
typedef VectorRef<LogicalPageID> BTreePageIDRef;
constexpr LogicalPageID maxPageID = (LogicalPageID)-1;
std::string toString(BTreePageIDRef id) {
return std::string("BTreePageID") + toString(id.begin(), id.end());
@ -2246,6 +2247,10 @@ struct RedwoodRecordRef {
inline RedwoodRecordRef withoutValue() const { return RedwoodRecordRef(key, version); }
inline RedwoodRecordRef withMaxPageID() const {
return RedwoodRecordRef(key, version, StringRef((uint8_t *)&maxPageID, sizeof(maxPageID)));
}
// Truncate (key, version, part) tuple to len bytes.
void truncate(int len) {
ASSERT(len <= key.size());
@ -3872,7 +3877,7 @@ private:
// If the decode upper boundary is the subtree upper boundary the pointers will be the same
// For the lower boundary, if the pointers are not the same there is still a possibility
// that the keys are the same. This happens for the first remaining subtree of an internal page
// after the previous first subtree was cleared.
// after the prior subtree(s) were cleared.
return (decodeUpperBound == subtreeUpperBound) &&
(decodeLowerBound == subtreeLowerBound || decodeLowerBound->sameExceptValue(*subtreeLowerBound));
}
@ -4984,6 +4989,246 @@ public:
Future<bool> moveLast() { return move_end(this, false); }
};
// Cursor designed for short lifespans.
// Holds references to all pages touched.
// All record references returned from it are valid until the cursor is destroyed.
class BTreeCursor {
Arena arena;
Reference<IPagerSnapshot> pager;
std::unordered_map<LogicalPageID, Reference<const IPage>> pages;
VersionedBTree* btree;
bool valid;
struct PathEntry {
BTreePage* btPage;
BTreePage::BinaryTree::Cursor cursor;
};
VectorRef<PathEntry> path;
public:
BTreeCursor() {}
bool isValid() const { return valid; }
std::string toString() const {
std::string r;
for (int i = 0; i < path.size(); ++i) {
r += format("[%d/%d: %s] ", i + 1, path.size(),
path[i].cursor.valid() ? path[i].cursor.get().toString(path[i].btPage->isLeaf()).c_str()
: "<invalid>");
}
if (!valid) {
r += " (invalid) ";
}
return r;
}
const RedwoodRecordRef& get() { return path.back().cursor.get(); }
bool inRoot() const { return path.size() == 1; }
// Pop and return the page cursor at the end of the path.
// This is meant to enable range scans to consume the contents of a leaf page more efficiently.
// Can only be used when inRoot() is true.
BTreePage::BinaryTree::Cursor popPath() {
BTreePage::BinaryTree::Cursor c = path.back().cursor;
path.pop_back();
return c;
}
Future<Void> pushPage(BTreePageIDRef id, const RedwoodRecordRef& lowerBound,
const RedwoodRecordRef& upperBound) {
Reference<const IPage>& page = pages[id.front()];
if (page.isValid()) {
path.push_back(arena, { (BTreePage*)page->begin(), getCursor(page) });
return Void();
}
return map(readPage(pager, id, &lowerBound, &upperBound), [this, &page, id](Reference<const IPage> p) {
page = p;
path.push_back(arena, { (BTreePage*)p->begin(), getCursor(p) });
return Void();
});
}
Future<Void> pushPage(BTreePage::BinaryTree::Cursor c) {
const RedwoodRecordRef& rec = c.get();
auto next = c;
next.moveNext();
BTreePageIDRef id = rec.getChildPage();
return pushPage(id, rec, next.getOrUpperBound());
}
Future<Void> init(VersionedBTree* btree_in, Reference<IPagerSnapshot> pager_in, BTreePageIDRef root) {
btree = btree_in;
pager = pager_in;
path.reserve(arena, 6);
valid = false;
return pushPage(root, dbBegin, dbEnd);
}
// Seeks cursor to query if it exists, the record before or after it, or an undefined and invalid
// position between those records
// If 0 is returned, then
// If the cursor is valid then it points to query
// If the cursor is not valid then the cursor points to some place in the btree such that
// If there is a record in the tree < query then movePrev() will move to it, and
// If there is a record in the tree > query then moveNext() will move to it.
// If non-zero is returned then the cursor is valid and the return value is logically equivalent
// to query.compare(cursor.get())
ACTOR Future<int> seek_impl(BTreeCursor* self, RedwoodRecordRef query, int prefetchBytes) {
state RedwoodRecordRef internalPageQuery = query.withMaxPageID();
self->path = self->path.slice(0, 1);
debug_printf("seek(%s, %d) start cursor = %s\n", query.toString().c_str(), prefetchBytes,
self->toString().c_str());
loop {
auto& entry = self->path.back();
if (entry.btPage->isLeaf()) {
int cmp = entry.cursor.seek(query);
self->valid = entry.cursor.valid() && !entry.cursor.node->isDeleted();
debug_printf("seek(%s, %d) loop exit cmp=%d cursor=%s\n", query.toString().c_str(), prefetchBytes,
cmp, self->toString().c_str());
return self->valid ? cmp : 0;
}
// Internal page, so seek to the branch where query must be
// Currently, after a subtree deletion internal page boundaries are still strictly adhered
// to and will be updated if anything is inserted into the cleared range, so if the seek fails
// or it finds an entry with a null child page then query does not exist in the BTree.
if (entry.cursor.seekLessThan(internalPageQuery) && entry.cursor.get().value.present()) {
debug_printf("seek(%s, %d) loop seek success cursor=%s\n", query.toString().c_str(), prefetchBytes,
self->toString().c_str());
Future<Void> f = self->pushPage(entry.cursor);
// Prefetch siblings, at least prefetchBytes, at level 2 but without jumping to another level 2
// sibling
if (prefetchBytes != 0 && entry.btPage->height == 2) {
auto c = entry.cursor;
bool fwd = prefetchBytes > 0;
prefetchBytes = abs(prefetchBytes);
// While we should still preload more bytes and a move in the target direction is successful
while (prefetchBytes > 0 && (fwd ? c.moveNext() : c.movePrev())) {
// If there is a page link, preload it.
if (c.get().value.present()) {
BTreePageIDRef childPage = c.get().getChildPage();
preLoadPage(self->pager.getPtr(), childPage);
prefetchBytes -= self->btree->m_blockSize * childPage.size();
}
}
}
wait(f);
} else {
self->valid = false;
debug_printf("seek(%s, %d) loop exit cmp=0 cursor=%s\n", query.toString().c_str(), prefetchBytes,
self->toString().c_str());
return 0;
}
}
}
Future<int> seek(RedwoodRecordRef query, int prefetchBytes) { return seek_impl(this, query, prefetchBytes); }
ACTOR Future<Void> seekGTE_impl(BTreeCursor* self, RedwoodRecordRef query, int prefetchBytes) {
debug_printf("seekGTE(%s, %d) start\n", query.toString().c_str(), prefetchBytes);
int cmp = wait(self->seek(query, prefetchBytes));
if (cmp > 0 || (cmp == 0 && !self->isValid())) {
wait(self->moveNext());
}
return Void();
}
Future<Void> seekGTE(RedwoodRecordRef query, int prefetchBytes) {
return seekGTE_impl(this, query, prefetchBytes);
}
ACTOR Future<Void> seekLT_impl(BTreeCursor* self, RedwoodRecordRef query, int prefetchBytes) {
debug_printf("seekLT(%s, %d) start\n", query.toString().c_str(), prefetchBytes);
int cmp = wait(self->seek(query, prefetchBytes));
if (cmp <= 0) {
wait(self->movePrev());
}
return Void();
}
Future<Void> seekLT(RedwoodRecordRef query, int prefetchBytes) {
return seekLT_impl(this, query, -prefetchBytes);
}
ACTOR Future<Void> move_impl(BTreeCursor* self, bool forward) {
// Try to the move cursor at the end of the path in the correct direction
debug_printf("move%s() start cursor=%s\n", forward ? "Next" : "Prev", self->toString().c_str());
while (1) {
debug_printf("move%s() first loop cursor=%s\n", forward ? "Next" : "Prev", self->toString().c_str());
auto& entry = self->path.back();
bool success;
if(entry.cursor.valid()) {
success = forward ? entry.cursor.moveNext() : entry.cursor.movePrev();
} else {
success = forward ? entry.cursor.moveFirst() : false;
}
// Skip over internal page entries that do not link to child pages. There should never be two in a row.
if (success && !entry.btPage->isLeaf() && !entry.cursor.get().value.present()) {
success = forward ? entry.cursor.moveNext() : entry.cursor.movePrev();
ASSERT(!success || entry.cursor.get().value.present());
}
// Stop if successful
if (success) {
break;
}
if (self->path.size() == 1) {
self->valid = false;
return Void();
}
// Move to parent
self->path = self->path.slice(0, self->path.size() - 1);
}
// While not on a leaf page, move down to get to one.
while (1) {
debug_printf("move%s() second loop cursor=%s\n", forward ? "Next" : "Prev", self->toString().c_str());
auto& entry = self->path.back();
if (entry.btPage->isLeaf()) {
break;
}
// The last entry in an internal page could be a null link, if so move back
if (!forward && !entry.cursor.get().value.present()) {
ASSERT(entry.cursor.movePrev());
ASSERT(entry.cursor.get().value.present());
}
wait(self->pushPage(entry.cursor));
auto& newEntry = self->path.back();
ASSERT(forward ? newEntry.cursor.moveFirst() : newEntry.cursor.moveLast());
}
self->valid = true;
debug_printf("move%s() exit cursor=%s\n", forward ? "Next" : "Prev", self->toString().c_str());
return Void();
}
Future<Void> moveNext() { return move_impl(this, true); }
Future<Void> movePrev() { return move_impl(this, false); }
};
Future<Void> initBTreeCursor(BTreeCursor* cursor, Version snapshotVersion) {
// Only committed versions can be read.
ASSERT(snapshotVersion <= m_lastCommittedVersion);
Reference<IPagerSnapshot> snapshot = m_pager->getReadSnapshot(snapshotVersion);
// This is a ref because snapshot will continue to hold the metakey value memory
KeyRef m = snapshot->getMetaKey();
return cursor->init(this, snapshot, ((MetaKey*)m.begin())->root.get());
}
// Cursor is for reading and interating over user visible KV pairs at a specific version
// KeyValueRefs returned become invalid once the cursor is moved
class Cursor : public IStoreCursor, public ReferenceCounted<Cursor>, public FastAllocated<Cursor>, NonCopyable {
@ -5264,10 +5509,13 @@ public:
ACTOR static Future<Standalone<RangeResultRef>> readRange_impl(KeyValueStoreRedwoodUnversioned* self, KeyRange keys,
int rowLimit, int byteLimit) {
state VersionedBTree::BTreeCursor cur;
wait(self->m_tree->initBTreeCursor(&cur, self->m_tree->getLastCommittedVersion()));
wait(self->m_concurrentReads.take());
state FlowLock::Releaser releaser(self->m_concurrentReads);
++g_redwoodMetrics.opGetRange;
state Standalone<RangeResultRef> result;
state int accumulatedBytes = 0;
ASSERT(byteLimit > 0);
@ -5276,33 +5524,58 @@ public:
return result;
}
state Reference<IStoreCursor> cur = self->m_tree->readAtVersion(self->m_tree->getLastCommittedVersion());
// Prefetch is currently only done in the forward direction
state int prefetchBytes = rowLimit > 1 ? byteLimit : 0;
// Prefetch is disabled for now pending some decent logic for deciding how much to fetch
state int prefetchBytes = 0;
if (rowLimit > 0) {
wait(cur->findFirstEqualOrGreater(keys.begin, prefetchBytes));
while (cur->isValid() && cur->getKey() < keys.end) {
KeyValueRef kv(KeyRef(result.arena(), cur->getKey()), ValueRef(result.arena(), cur->getValue()));
accumulatedBytes += kv.expectedSize();
result.push_back(result.arena(), kv);
if (--rowLimit == 0 || accumulatedBytes >= byteLimit) {
wait(cur.seekGTE(keys.begin, prefetchBytes));
while (cur.isValid()) {
// Read page contents without using waits
bool isRoot = cur.inRoot();
BTreePage::BinaryTree::Cursor leafCursor = cur.popPath();
while(leafCursor.valid()) {
KeyValueRef kv = leafCursor.get().toKeyValueRef();
if(kv.key >= keys.end) {
break;
}
accumulatedBytes += kv.expectedSize();
result.push_back_deep(result.arena(), kv);
if (--rowLimit == 0 || accumulatedBytes >= byteLimit) {
break;
}
leafCursor.moveNext();
}
// Stop if the leaf cursor is still valid which means we hit a key or size limit or
// if we started in the root page
if(leafCursor.valid() || isRoot) {
break;
}
wait(cur->next());
wait(cur.moveNext());
}
} else {
wait(cur->findLastLessOrEqual(keys.end));
if (cur->isValid() && cur->getKey() == keys.end) wait(cur->prev());
while (cur->isValid() && cur->getKey() >= keys.begin) {
KeyValueRef kv(KeyRef(result.arena(), cur->getKey()), ValueRef(result.arena(), cur->getValue()));
accumulatedBytes += kv.expectedSize();
result.push_back(result.arena(), kv);
if (++rowLimit == 0 || accumulatedBytes >= byteLimit) {
wait(cur.seekLT(keys.end, prefetchBytes));
while (cur.isValid()) {
// Read page contents without using waits
bool isRoot = cur.inRoot();
BTreePage::BinaryTree::Cursor leafCursor = cur.popPath();
while(leafCursor.valid()) {
KeyValueRef kv = leafCursor.get().toKeyValueRef();
if(kv.key < keys.begin) {
break;
}
accumulatedBytes += kv.expectedSize();
result.push_back_deep(result.arena(), kv);
if (++rowLimit == 0 || accumulatedBytes >= byteLimit) {
break;
}
leafCursor.movePrev();
}
// Stop if the leaf cursor is still valid which means we hit a key or size limit or
// if we started in the root page
if(leafCursor.valid() || isRoot) {
break;
}
wait(cur->prev());
wait(cur.movePrev());
}
}
@ -5316,15 +5589,16 @@ public:
ACTOR static Future<Optional<Value>> readValue_impl(KeyValueStoreRedwoodUnversioned* self, Key key,
Optional<UID> debugID) {
state VersionedBTree::BTreeCursor cur;
wait(self->m_tree->initBTreeCursor(&cur, self->m_tree->getLastCommittedVersion()));
wait(self->m_concurrentReads.take());
state FlowLock::Releaser releaser(self->m_concurrentReads);
++g_redwoodMetrics.opGet;
state Reference<IStoreCursor> cur = self->m_tree->readAtVersion(self->m_tree->getLastCommittedVersion());
wait(cur->findEqual(key));
if (cur->isValid()) {
return cur->getValue();
wait(cur.seekGTE(key, 0));
if (cur.isValid() && cur.get().key == key) {
return cur.get().value.get();
}
return Optional<Value>();
}
@ -5335,18 +5609,20 @@ public:
ACTOR static Future<Optional<Value>> readValuePrefix_impl(KeyValueStoreRedwoodUnversioned* self, Key key,
int maxLength, Optional<UID> debugID) {
state VersionedBTree::BTreeCursor cur;
wait(self->m_tree->initBTreeCursor(&cur, self->m_tree->getLastCommittedVersion()));
wait(self->m_concurrentReads.take());
state FlowLock::Releaser releaser(self->m_concurrentReads);
++g_redwoodMetrics.opGet;
state Reference<IStoreCursor> cur = self->m_tree->readAtVersion(self->m_tree->getLastCommittedVersion());
wait(cur->findEqual(key));
if (cur->isValid()) {
Value v = cur->getValue();
wait(cur.seekGTE(key, 0));
if (cur.isValid() && cur.get().key == key) {
Value v = cur.get().value.get();
int len = std::min(v.size(), maxLength);
return Value(cur->getValue().substr(0, len));
return Value(v.substr(0, len));
}
return Optional<Value>();
}
@ -5411,6 +5687,157 @@ KeyValue randomKV(int maxKeySize = 10, int maxValueSize = 5) {
return kv;
}
// Verify a range using a BTreeCursor.
// Assumes that the BTree holds a single data version and the version is 0.
ACTOR Future<int> verifyRangeBTreeCursor(VersionedBTree* btree, Key start, Key end, Version v,
std::map<std::pair<std::string, Version>, Optional<std::string>>* written,
int* pErrorCount) {
state int errors = 0;
if (end <= start) end = keyAfter(start);
state std::map<std::pair<std::string, Version>, Optional<std::string>>::const_iterator i =
written->lower_bound(std::make_pair(start.toString(), 0));
state std::map<std::pair<std::string, Version>, Optional<std::string>>::const_iterator iEnd =
written->upper_bound(std::make_pair(end.toString(), 0));
state std::map<std::pair<std::string, Version>, Optional<std::string>>::const_iterator iLast;
state VersionedBTree::BTreeCursor cur;
wait(btree->initBTreeCursor(&cur, v));
debug_printf("VerifyRange(@%" PRId64 ", %s, %s): Start\n", v, start.printable().c_str(), end.printable().c_str());
// Randomly use the cursor for something else first.
if (deterministicRandom()->coinflip()) {
state Key randomKey = randomKV().key;
debug_printf("VerifyRange(@%" PRId64 ", %s, %s): Dummy seek to '%s'\n", v, start.printable().c_str(),
end.printable().c_str(), randomKey.toString().c_str());
wait(success(cur.seek(randomKey, 0)));
}
debug_printf("VerifyRange(@%" PRId64 ", %s, %s): Actual seek\n", v, start.printable().c_str(),
end.printable().c_str());
wait(cur.seekGTE(start, 0));
state std::vector<KeyValue> results;
while (cur.isValid() && cur.get().key < end) {
// Find the next written kv pair that would be present at this version
while (1) {
iLast = i;
if (i == iEnd) break;
++i;
if (iLast->first.second <= v && iLast->second.present() &&
(i == iEnd || i->first.first != iLast->first.first || i->first.second > v)) {
debug_printf("VerifyRange(@%" PRId64 ", %s, %s) Found key in written map: %s\n", v,
start.printable().c_str(), end.printable().c_str(), iLast->first.first.c_str());
break;
}
}
if (iLast == iEnd) {
++errors;
++*pErrorCount;
printf("VerifyRange(@%" PRId64 ", %s, %s) ERROR: Tree key '%s' vs nothing in written map.\n", v,
start.printable().c_str(), end.printable().c_str(), cur.get().key.toString().c_str());
break;
}
if (cur.get().key != iLast->first.first) {
++errors;
++*pErrorCount;
printf("VerifyRange(@%" PRId64 ", %s, %s) ERROR: Tree key '%s' but expected '%s'\n", v,
start.printable().c_str(), end.printable().c_str(), cur.get().key.toString().c_str(),
iLast->first.first.c_str());
break;
}
if (cur.get().value.get() != iLast->second.get()) {
++errors;
++*pErrorCount;
printf("VerifyRange(@%" PRId64 ", %s, %s) ERROR: Tree key '%s' has tree value '%s' but expected '%s'\n", v,
start.printable().c_str(), end.printable().c_str(), cur.get().key.toString().c_str(),
cur.get().value.get().toString().c_str(), iLast->second.get().c_str());
break;
}
ASSERT(errors == 0);
results.push_back(KeyValue(KeyValueRef(cur.get().key, cur.get().value.get())));
wait(cur.moveNext());
}
// Make sure there are no further written kv pairs that would be present at this version.
while (1) {
iLast = i;
if (i == iEnd) break;
++i;
if (iLast->first.second <= v && iLast->second.present() &&
(i == iEnd || i->first.first != iLast->first.first || i->first.second > v))
break;
}
if (iLast != iEnd) {
++errors;
++*pErrorCount;
printf("VerifyRange(@%" PRId64 ", %s, %s) ERROR: Tree range ended but written has @%" PRId64 " '%s'\n", v,
start.printable().c_str(), end.printable().c_str(), iLast->first.second, iLast->first.first.c_str());
}
debug_printf("VerifyRangeReverse(@%" PRId64 ", %s, %s): start\n", v, start.printable().c_str(),
end.printable().c_str());
// Randomly use a new cursor at the same version for the reverse range read, if the version is still available for
// opening new cursors
if (v >= btree->getOldestVersion() && deterministicRandom()->coinflip()) {
cur = VersionedBTree::BTreeCursor();
wait(btree->initBTreeCursor(&cur, v));
}
// Now read the range from the tree in reverse order and compare to the saved results
wait(cur.seekLT(end, 0));
state std::vector<KeyValue>::const_reverse_iterator r = results.rbegin();
while (cur.isValid() && cur.get().key >= start) {
if (r == results.rend()) {
++errors;
++*pErrorCount;
printf("VerifyRangeReverse(@%" PRId64 ", %s, %s) ERROR: Tree key '%s' vs nothing in written map.\n", v,
start.printable().c_str(), end.printable().c_str(), cur.get().key.toString().c_str());
break;
}
if (cur.get().key != r->key) {
++errors;
++*pErrorCount;
printf("VerifyRangeReverse(@%" PRId64 ", %s, %s) ERROR: Tree key '%s' but expected '%s'\n", v,
start.printable().c_str(), end.printable().c_str(), cur.get().key.toString().c_str(),
r->key.toString().c_str());
break;
}
if (cur.get().value.get() != r->value) {
++errors;
++*pErrorCount;
printf("VerifyRangeReverse(@%" PRId64
", %s, %s) ERROR: Tree key '%s' has tree value '%s' but expected '%s'\n",
v, start.printable().c_str(), end.printable().c_str(), cur.get().key.toString().c_str(),
cur.get().value.get().toString().c_str(), r->value.toString().c_str());
break;
}
++r;
wait(cur.movePrev());
}
if (r != results.rend()) {
++errors;
++*pErrorCount;
printf("VerifyRangeReverse(@%" PRId64 ", %s, %s) ERROR: Tree range ended but written has '%s'\n", v,
start.printable().c_str(), end.printable().c_str(), r->key.toString().c_str());
}
return errors;
}
ACTOR Future<int> verifyRange(VersionedBTree* btree, Key start, Key end, Version v,
std::map<std::pair<std::string, Version>, Optional<std::string>>* written,
int* pErrorCount) {
@ -5607,6 +6034,58 @@ ACTOR Future<int> seekAll(VersionedBTree* btree, Version v,
return errors;
}
// Verify the result of point reads for every set or cleared key at the given version
ACTOR Future<int> seekAllBTreeCursor(VersionedBTree* btree, Version v,
std::map<std::pair<std::string, Version>, Optional<std::string>>* written, int* pErrorCount) {
state std::map<std::pair<std::string, Version>, Optional<std::string>>::const_iterator i = written->cbegin();
state std::map<std::pair<std::string, Version>, Optional<std::string>>::const_iterator iEnd = written->cend();
state int errors = 0;
state VersionedBTree::BTreeCursor cur;
wait(btree->initBTreeCursor(&cur, v));
while (i != iEnd) {
state std::string key = i->first.first;
state Version ver = i->first.second;
if (ver == v) {
state Optional<std::string> val = i->second;
debug_printf("Verifying @%" PRId64 " '%s'\n", ver, key.c_str());
state Arena arena;
wait(cur.seekGTE(RedwoodRecordRef(KeyRef(arena, key), 0), 0));
bool foundKey = cur.isValid() && cur.get().key == key;
bool hasValue = foundKey && cur.get().value.present();
if (val.present()) {
bool valueMatch = hasValue && cur.get().value.get() == val.get();
if (!foundKey || !hasValue || !valueMatch) {
++errors;
++*pErrorCount;
if (!foundKey) {
printf("Verify ERROR: key_not_found: '%s' -> '%s' @%" PRId64 "\n", key.c_str(),
val.get().c_str(), ver);
}
else if (!hasValue) {
printf("Verify ERROR: value_not_found: '%s' -> '%s' @%" PRId64 "\n", key.c_str(),
val.get().c_str(), ver);
}
else if (!valueMatch) {
printf("Verify ERROR: value_incorrect: for '%s' found '%s' expected '%s' @%" PRId64 "\n",
key.c_str(), cur.get().value.get().toString().c_str(), val.get().c_str(),
ver);
}
}
} else if (foundKey && hasValue) {
++errors;
++*pErrorCount;
printf("Verify ERROR: cleared_key_found: '%s' -> '%s' @%" PRId64 "\n", key.c_str(),
cur.get().value.get().toString().c_str(), ver);
}
}
++i;
}
return errors;
}
ACTOR Future<Void> verify(VersionedBTree* btree, FutureStream<Version> vStream,
std::map<std::pair<std::string, Version>, Optional<std::string>>* written, int* pErrorCount,
bool serial) {
@ -5637,7 +6116,13 @@ ACTOR Future<Void> verify(VersionedBTree* btree, FutureStream<Version> vStream,
state Reference<IStoreCursor> cur = btree->readAtVersion(v);
debug_printf("Verifying entire key range at version %" PRId64 "\n", v);
fRangeAll = verifyRange(btree, LiteralStringRef(""), LiteralStringRef("\xff\xff"), v, written, pErrorCount);
if(deterministicRandom()->coinflip()) {
fRangeAll = verifyRange(btree, LiteralStringRef(""), LiteralStringRef("\xff\xff"), v, written,
pErrorCount);
} else {
fRangeAll = verifyRangeBTreeCursor(btree, LiteralStringRef(""), LiteralStringRef("\xff\xff"), v, written,
pErrorCount);
}
if (serial) {
wait(success(fRangeAll));
}
@ -5646,13 +6131,21 @@ ACTOR Future<Void> verify(VersionedBTree* btree, FutureStream<Version> vStream,
Key end = randomKV().key;
debug_printf("Verifying range (%s, %s) at version %" PRId64 "\n", toString(begin).c_str(),
toString(end).c_str(), v);
fRangeRandom = verifyRange(btree, begin, end, v, written, pErrorCount);
if(deterministicRandom()->coinflip()) {
fRangeRandom = verifyRange(btree, begin, end, v, written, pErrorCount);
} else {
fRangeRandom = verifyRangeBTreeCursor(btree, begin, end, v, written, pErrorCount);
}
if (serial) {
wait(success(fRangeRandom));
}
debug_printf("Verifying seeks to each changed key at version %" PRId64 "\n", v);
fSeekAll = seekAll(btree, v, written, pErrorCount);
if(deterministicRandom()->coinflip()) {
fSeekAll = seekAll(btree, v, written, pErrorCount);
} else {
fSeekAll = seekAllBTreeCursor(btree, v, written, pErrorCount);
}
if (serial) {
wait(success(fSeekAll));
}
@ -6485,11 +6978,11 @@ TEST_CASE("!/redwood/correctness/btree") {
state int maxKeySize = deterministicRandom()->randomInt(1, pageSize * 2);
state int maxValueSize = randomSize(pageSize * 25);
state int maxCommitSize = shortTest ? 1000 : randomSize(std::min<int>((maxKeySize + maxValueSize) * 20000, 10e6));
state int mutationBytesTarget = shortTest ? 100000 : randomSize(std::min<int>(maxCommitSize * 100, 100e6));
state int mutationBytesTarget = shortTest ? 100000 : randomSize(std::min<int>(maxCommitSize * 100, pageSize * 100000));
state double clearProbability = deterministicRandom()->random01() * .1;
state double clearSingleKeyProbability = deterministicRandom()->random01();
state double clearPostSetProbability = deterministicRandom()->random01() * .1;
state double coldStartProbability = pagerMemoryOnly ? 0 : deterministicRandom()->random01();
state double coldStartProbability = pagerMemoryOnly ? 0 : (deterministicRandom()->random01() * 0.3);
state double advanceOldVersionProbability = deterministicRandom()->random01();
state double maxDuration = 60;
state int64_t cacheSizeBytes =

Some files were not shown because too many files have changed in this diff Show More