Compare commits

..
Author SHA1 Message Date
sdong d9c289caa4 Bump up patch number to 5.13.4 2018-06-12 17:53:37 -07:00
sdong 3c6bb75133 Update HISTORY.md for 5.13.4 2018-06-12 17:41:26 -07:00
sdong 0c54d2609c Fix regression bug of Prev() with upper bound (#3989)
Summary:
A recent change pushed down the upper bound checking to child iterators. However, this causes the logic of following sequence wrong:
  Seek(key);
  if (!Valid()) SeekToLast();
Because !Valid() may be caused by upper bounds, rather than the end of the iterator. In this case SeekToLast() points to totally wrong places. This can cause wrong results, infinite loops, or segfault in some cases.
This sequence is called when changing direction from forward to backward. And this by itself also implicitly happen during reseeking optimization in Prev().

Fix this bug by using SeekForPrev() rather than this sequence, as what is already done in prefix extrator case.
Closes https://github.com/facebook/rocksdb/pull/3989

Differential Revision: D8385422

Pulled By: siying

fbshipit-source-id: 429e869990cfd2dc389421e0836fc496bed67bb4
2018-06-12 17:06:24 -07:00
Andrew Kryczka f9e0a06512 update history and version 2018-06-06 11:16:32 -07:00
Huachao Huang 8d7396418c Ignore empty filter block when data block is empty
Summary:
Close https://github.com/facebook/rocksdb/issues/3592
Closes https://github.com/facebook/rocksdb/pull/3614

Differential Revision: D7291706

Pulled By: ajkr

fbshipit-source-id: 9dd8f40bd7716588e1e3fd6be0c2bc2766861f8c
2018-06-06 11:13:18 -07:00
Fosco Marotto d25ca5fd49 Update history for release of 5.13.2 2018-05-25 16:41:35 -07:00
Adam Retter 2da2286dd0 Fix an issue with unnecessary capture in lambda expressions
Summary:
Closes https://github.com/facebook/rocksdb/issues/3900
Replaces https://github.com/facebook/rocksdb/pull/3901

I needed this to build v5.12.4 on Mac OS X (10.13.3).
Closes https://github.com/facebook/rocksdb/pull/3904

Differential Revision: D8169357

Pulled By: sagar0

fbshipit-source-id: 85faac42168796e7def9250d0c221a9a03b84476
2018-05-25 16:26:04 -07:00
Yanqin Jin 83973ddd39 Fix segfault caused by object premature destruction
Summary:
Please refer to earlier discussion in [issue 3609](https://github.com/facebook/rocksdb/issues/3609).
There was also an alternative fix in [PR 3888](https://github.com/facebook/rocksdb/pull/3888), but the proposed solution requires complex change.

To summarize the cause of the problem. Upon creation of a column family, a `BlockBasedTableFactory` object is `new`ed and encapsulated by a `std::shared_ptr`. Since there is no other `std::shared_ptr` pointing to this `BlockBasedTableFactory`, when the column family is dropped, the `ColumnFamilyData` is `delete`d, causing the destructor of `std::shared_ptr`. Since there is no other `std::shared_ptr`, the underlying memory is also freed.
Later when the db exits, it releases all the table readers, including the table readers that have been operating on the dropped column family. This needs to access the `table_options` owned by `BlockBasedTableFactory` that has already been deleted. Therefore, a segfault is raised.
Previous workaround is to purge all obsolete files upon `ColumnFamilyData` destruction, which leads to a force release of table readers of the dropped column family. However this does not work when the user disables file deletion.

Our solution in this PR is making a copy of `table_options` in `BlockBasedTable::Rep`. This solution increases memory copy and usage, but is much simpler.

Test plan
```
$ make -j16
$ ./column_family_test --gtest_filter=ColumnFamilyTest.CreateDropAndDestroy:ColumnFamilyTest.CreateDropAndDestroyWithoutFileDeletion
```

Expected behavior:
All tests should pass.
Closes https://github.com/facebook/rocksdb/pull/3898

Differential Revision: D8149421

Pulled By: riversand963

fbshipit-source-id: eaecc2e064057ef607fbdd4cc275874f866c3438
2018-05-25 15:54:14 -07:00
Andrew Kryczka 1c18482bbf bump version to 5.13.2 and update HISTORY 2018-05-23 21:01:48 -07:00
Andrew Kryczka b245e51955 Introduce library-independent default compression level
Summary:
Previously we were using -1 as the default for every library, which was legacy from our zlib options. That worked for a while, but after zstd introduced https://github.com/facebook/zstd/commit/a146ee04ae5866b948be0c1911418e0436d80cb4, it started giving poor compression ratios by default in zstd.

This PR adds a constant to RocksDB public API, `CompressionOptions::kDefaultCompressionLevel`, which will get translated to the default value specific to the compression library being used in "util/compression.h". The constant uses a number that appears to be larger than any library's maximum compression level.
Closes https://github.com/facebook/rocksdb/pull/3895

Differential Revision: D8125780

Pulled By: ajkr

fbshipit-source-id: 2db157a89118cd4f94577c2f4a0a5ff31c8391c6
2018-05-23 20:59:31 -07:00
Andrew Kryczka c60df9d9e7 update history and version 2018-04-30 18:19:02 -07:00
Andrew Kryczka 747c853203 Avoid directory renames in BackupEngine
Summary:
We used to name private directories like "1.tmp" while BackupEngine populated them, and then rename without the ".tmp" suffix (i.e., rename "1.tmp" to "1") after all files were copied. On glusterfs, directory renames like this require operations across many hosts, and partial failures have caused operational problems.

Fortunately we don't need to rename private directories. We already have a meta-file that uses the tempfile-rename pattern to commit a backup atomically after all its files have been successfully copied. So we can copy private files directly to their final location, so now there's no directory rename.
Closes https://github.com/facebook/rocksdb/pull/3749

Differential Revision: D7705610

Pulled By: ajkr

fbshipit-source-id: fd724a28dd2bf993ce323a5f2cb7e7d6980cc346
2018-04-30 17:50:24 -07:00
Gabriel Wicke f5ee207c64 Support lowering CPU priority of background threads
Summary:
Background activities like compaction can negatively affect
latency of higher-priority tasks like request processing. To avoid this,
rocksdb already lowers the IO priority of background threads on Linux
systems. While this takes care of typical IO-bound systems, it does not
help much when CPU (temporarily) becomes the bottleneck. This is
especially likely when using more expensive compression settings.

This patch adds an API to allow for lowering the CPU priority of
background threads, modeled on the IO priority API. Benchmarks (see
below) show significant latency and throughput improvements when CPU
bound. As a result, workloads with some CPU usage bursts should benefit
from lower latencies at a given utilization, or should be able to push
utilization higher at a given request latency target.

A useful side effect is that compaction CPU usage is now easily visible
in common tools, allowing for an easier estimation of the contribution
of compaction vs. request processing threads.

As with IO priority, the implementation is limited to Linux, degrading
to a no-op on other systems.
Closes https://github.com/facebook/rocksdb/pull/3763

Differential Revision: D7740096

Pulled By: gwicke

fbshipit-source-id: e5d32373e8dc403a7b0c2227023f9ce4f22b413c
2018-04-24 18:18:32 -07:00
Zhongyi Xie f2fd21fa6f fix memory leak in two_level_iterator
Summary:
this PR fixes a few failed contbuild:
1. ASAN memory leak in Block::NewIterator (table/block.cc:429). the proper destruction of first_level_iter_ and second_level_iter_ of two_level_iterator.cc is missing from the code after the refactoring in https://github.com/facebook/rocksdb/pull/3406
2. various unused param errors introduced by https://github.com/facebook/rocksdb/pull/3662
3. updated comment for `ForceReleaseCachedEntry` to emphasize the use of `force_erase` flag.
Closes https://github.com/facebook/rocksdb/pull/3718

Reviewed By: maysamyabandeh

Differential Revision: D7621192

Pulled By: miasantreble

fbshipit-source-id: 476c94264083a0730ded957c29de7807e4f5b146
2018-04-16 14:23:11 -07:00
Maysam Yabandeh 758527812c Fix the memory leak with pinned partitioned filters
Summary:
The existing unit test did not set the level so the check for pinned partitioned filter/index being properly released from the block cache was not properly exercised as they only take effect in level 0. As a result a memory leak in pinned partitioned filters was hidden. The patch fix the test as well as the bug.
Closes https://github.com/facebook/rocksdb/pull/3692

Differential Revision: D7559763

Pulled By: maysamyabandeh

fbshipit-source-id: 55eff274945838af983c764a7d71e8daff092e4a
2018-04-16 14:12:49 -07:00
Sagar Vemuri 9d2e34ed4c Fix History 2018-03-23 14:58:51 -07:00
Sagar Vemuri 1f5103d583 Add Java-API-Changes section to History
Summary:
We have not been updating our HISTORY.md change log with the RocksJava changes. Going forward, lets add Java changes also to HISTORY.md.
There is an old java/HISTORY-JAVA.md, but it hasn't been updated in years. It is much easier to remember to update the change log in a single file, HISTORY.md.

I added information about shared block cache here, which was introduced in #3623.
Closes https://github.com/facebook/rocksdb/pull/3647

Differential Revision: D7384448

Pulled By: sagar0

fbshipit-source-id: 9b6e569f44e6df5cb7ba06413d9975df0b517d20
2018-03-23 14:56:44 -07:00
Sagar Vemuri 163dd4b81b Shared block cache in RocksJava
Summary:
Changes to support sharing block cache using the Java API.

Previously DB instances could share the block cache only when the same Options instance is passed to all the DB instances. But now, with this change, it is possible to explicitly create a cache and pass it to multiple options instances, to share the block cache.

Implementing this for [Rocksandra](https://github.com/instagram/cassandra/tree/rocks_3.0), but this feature has been requested by many java api users over the years.
Closes https://github.com/facebook/rocksdb/pull/3623

Differential Revision: D7305794

Pulled By: sagar0

fbshipit-source-id: 03e4e8ed7aeee6f88bada4a8365d4279ede2ad71
2018-03-23 14:56:05 -07:00
Sagar Vemuri 1642582559 Fsync after writing global seq number in ExternalSstFileIngestionJob
Summary:
Fsync after writing global sequence number to the ingestion file in ExternalSstFileIngestionJob. Otherwise the file metadata could be incorrect.
Closes https://github.com/facebook/rocksdb/pull/3644

Differential Revision: D7373813

Pulled By: sagar0

fbshipit-source-id: 4da2c9e71a8beb5c08b4ac955f288ee1576358b8
2018-03-23 14:52:21 -07:00
Fosco Marotto 8d28083264 Update history for future 5.13 release 2018-03-20 16:17:53 -07:00
118 changed files with 740 additions and 1221 deletions

No files matched your search

-1
View File
@@ -569,7 +569,6 @@ set(SOURCES
util/status_message.cc
util/string_util.cc
util/sync_point.cc
util/sync_point_impl.cc
util/testutil.cc
util/thread_local.cc
util/threadpool_imp.cc
+21 -7
View File
@@ -1,14 +1,24 @@
# Rocksdb Change Log
## Unreleased
### Public API Change
## 5.13.4 (6/12/2018)
### Bug Fixes
* Fix regression bug of Prev() with ReadOptions.iterate_upper_bound.
### New Features
## 5.13.3 (6/6/2018)
### Bug Fixes
* Fix assertion when reading bloom filter of SST files containing range deletions but no data
## 5.13.2 (5/25/2018)
### Public API Change
* Introduced `CompressionOptions::kDefaultCompressionLevel`, which is a generic way to tell RocksDB to use the compression library's default level. It is now the default value for `CompressionOptions::level`. Previously the level defaulted to -1, which gave poor compression ratios in ZSTD.
### Bug Fixes
* Fsync after writing global seq number to the ingestion file in ExternalSstFileIngestionJob.
* Fix segfault caused by object premature destruction (PR #3898)
* Fix an issue with unnecessary capture in lambda expressions (PR #3904)
### Java API Changes
* Add `BlockBasedTableConfig.setBlockCache` to allow sharing a block cache across DB instances.
## 5.13.1 (4/30/2018)
### New Features
* Add `Env::LowerThreadPoolCPUPriority(Priority)` method, which lowers the CPU priority of background (esp. compaction) threads to minimize interference with foreground tasks.
* Eliminate use of temporary directories in BackupEngine to improve reliability on distributed file systems.
## 5.13.0 (3/20/2018)
### Public API Change
@@ -20,11 +30,15 @@
* Avoid unnecessarily flushing in `CompactRange()` when the range specified by the user does not overlap unflushed memtables.
* If `ColumnFamilyOptions::max_subcompactions` is set greater than one, we now parallelize large manual level-based compactions.
* Add "rocksdb.live-sst-files-size" DB property to return total bytes of all SST files belong to the latest LSM tree.
* NewSstFileManager to add an argument bytes_max_delete_chunk with default 64MB. With this argument, a file larger than 64MB will be ftruncated multiple times based on this size.
### Bug Fixes
* Fix a leak in prepared_section_completed_ where the zeroed entries would not removed from the map.
* Fix WAL corruption caused by race condition between user write thread and backup/checkpoint thread.
* Fsync after writing global seq number to the ingestion file in ExternalSstFileIngestionJob.
* Fix memory leak when pin_l0_filter_and_index_blocks_in_cache is used with partitioned filters
### Java API Changes
* Add `BlockBasedTableConfig.setBlockCache` to allow sharing a block cache across DB instances.
## 5.12.0 (2/14/2018)
### Public API Change
-1
View File
@@ -215,7 +215,6 @@ cpp_library(
"util/status_message.cc",
"util/string_util.cc",
"util/sync_point.cc",
"util/sync_point_impl.cc",
"util/thread_local.cc",
"util/threadpool_imp.cc",
"util/transaction_test_util.cc",
+6 -14
View File
@@ -1,4 +1,4 @@
#!/usr/bin/env bash
#!/bin/sh
#
# Detects OS we're compiling on and outputs a file specified by the first
# argument, which in turn gets read while processing Makefile.
@@ -66,19 +66,11 @@ rm -f "$OUTPUT"
touch "$OUTPUT"
if test -z "$CC"; then
if [ -x "$(command -v clang)" ]; then
CC=clang
else
CC=cc
fi
CC=cc
fi
if test -z "$CXX"; then
if [ -x "$(command -v clang++)" ]; then
CXX=clang++
else
CXX=g++
fi
CXX=g++
fi
# Detect OS
@@ -95,7 +87,7 @@ if test -z "$CLANG_SCAN_BUILD"; then
fi
if test -z "$CLANG_ANALYZER"; then
CLANG_ANALYZER=$(command -v clang++ 2> /dev/null)
CLANG_ANALYZER=$(which clang++ 2> /dev/null)
fi
if test -z "$FIND"; then
@@ -506,9 +498,9 @@ if test -z "$PORTABLE"; then
elif test -n "`echo $TARGET_ARCHITECTURE | grep ^arm`"; then
# TODO: Handle this with approprite options.
COMMON_FLAGS="$COMMON_FLAGS"
elif [ "$TARGET_OS" == "IOS" ]; then
elif [ "$TARGET_OS" == IOS ]; then
COMMON_FLAGS="$COMMON_FLAGS"
elif [ "$TARGET_OS" != "AIX" ] && [ "$TARGET_OS" != "SunOS" ]; then
elif [ "$TARGET_OS" != AIX ] && [ "$TARGET_OS" != SunOS ]; then
COMMON_FLAGS="$COMMON_FLAGS -march=native "
elif test "$USE_SSE"; then
COMMON_FLAGS="$COMMON_FLAGS -msse4.2 -mpclmul"
+1 -1
View File
@@ -1,4 +1,4 @@
#/usr/bin/env bash
#!/bin/sh
if [ "$#" = "0" ]; then
echo "Usage: $0 major|minor|patch|full"
exit 1
-174
View File
@@ -38,7 +38,6 @@
#include "rocksdb/utilities/transaction_db.h"
#include "rocksdb/utilities/write_batch_with_index.h"
#include "rocksdb/write_batch.h"
#include "rocksdb/perf_context.h"
#include "utilities/merge_operators.h"
using rocksdb::BytewiseComparator;
@@ -102,8 +101,6 @@ using rocksdb::OptimisticTransactionDB;
using rocksdb::OptimisticTransactionOptions;
using rocksdb::Transaction;
using rocksdb::Checkpoint;
using rocksdb::PerfLevel;
using rocksdb::PerfContext;
using std::shared_ptr;
@@ -144,7 +141,6 @@ struct rocksdb_envoptions_t { EnvOptions rep; };
struct rocksdb_ingestexternalfileoptions_t { IngestExternalFileOptions rep; };
struct rocksdb_sstfilewriter_t { SstFileWriter* rep; };
struct rocksdb_ratelimiter_t { RateLimiter* rep; };
struct rocksdb_perfcontext_t { PerfContext* rep; };
struct rocksdb_pinnableslice_t {
PinnableSlice rep;
};
@@ -2556,176 +2552,6 @@ void rocksdb_ratelimiter_destroy(rocksdb_ratelimiter_t *limiter) {
delete limiter;
}
void rocksdb_set_perf_level(int v) {
PerfLevel level = static_cast<PerfLevel>(v);
SetPerfLevel(level);
}
rocksdb_perfcontext_t* rocksdb_perfcontext_create() {
rocksdb_perfcontext_t* context = new rocksdb_perfcontext_t;
context->rep = rocksdb::get_perf_context();
return context;
}
void rocksdb_perfcontext_reset(rocksdb_perfcontext_t* context) {
context->rep->Reset();
}
char* rocksdb_perfcontext_report(rocksdb_perfcontext_t* context,
unsigned char exclude_zero_counters) {
return strdup(context->rep->ToString(exclude_zero_counters).c_str());
}
uint64_t rocksdb_perfcontext_metric(rocksdb_perfcontext_t* context,
int metric) {
PerfContext* rep = context->rep;
switch (metric) {
case rocksdb_user_key_comparison_count:
return rep->user_key_comparison_count;
case rocksdb_block_cache_hit_count:
return rep->block_cache_hit_count;
case rocksdb_block_read_count:
return rep->block_read_count;
case rocksdb_block_read_byte:
return rep->block_read_byte;
case rocksdb_block_read_time:
return rep->block_read_time;
case rocksdb_block_checksum_time:
return rep->block_checksum_time;
case rocksdb_block_decompress_time:
return rep->block_decompress_time;
case rocksdb_get_read_bytes:
return rep->get_read_bytes;
case rocksdb_multiget_read_bytes:
return rep->multiget_read_bytes;
case rocksdb_iter_read_bytes:
return rep->iter_read_bytes;
case rocksdb_internal_key_skipped_count:
return rep->internal_key_skipped_count;
case rocksdb_internal_delete_skipped_count:
return rep->internal_delete_skipped_count;
case rocksdb_internal_recent_skipped_count:
return rep->internal_recent_skipped_count;
case rocksdb_internal_merge_count:
return rep->internal_merge_count;
case rocksdb_get_snapshot_time:
return rep->get_snapshot_time;
case rocksdb_get_from_memtable_time:
return rep->get_from_memtable_time;
case rocksdb_get_from_memtable_count:
return rep->get_from_memtable_count;
case rocksdb_get_post_process_time:
return rep->get_post_process_time;
case rocksdb_get_from_output_files_time:
return rep->get_from_output_files_time;
case rocksdb_seek_on_memtable_time:
return rep->seek_on_memtable_time;
case rocksdb_seek_on_memtable_count:
return rep->seek_on_memtable_count;
case rocksdb_next_on_memtable_count:
return rep->next_on_memtable_count;
case rocksdb_prev_on_memtable_count:
return rep->prev_on_memtable_count;
case rocksdb_seek_child_seek_time:
return rep->seek_child_seek_time;
case rocksdb_seek_child_seek_count:
return rep->seek_child_seek_count;
case rocksdb_seek_min_heap_time:
return rep->seek_min_heap_time;
case rocksdb_seek_max_heap_time:
return rep->seek_max_heap_time;
case rocksdb_seek_internal_seek_time:
return rep->seek_internal_seek_time;
case rocksdb_find_next_user_entry_time:
return rep->find_next_user_entry_time;
case rocksdb_write_wal_time:
return rep->write_wal_time;
case rocksdb_write_memtable_time:
return rep->write_memtable_time;
case rocksdb_write_delay_time:
return rep->write_delay_time;
case rocksdb_write_pre_and_post_process_time:
return rep->write_pre_and_post_process_time;
case rocksdb_db_mutex_lock_nanos:
return rep->db_mutex_lock_nanos;
case rocksdb_db_condition_wait_nanos:
return rep->db_condition_wait_nanos;
case rocksdb_merge_operator_time_nanos:
return rep->merge_operator_time_nanos;
case rocksdb_read_index_block_nanos:
return rep->read_index_block_nanos;
case rocksdb_read_filter_block_nanos:
return rep->read_filter_block_nanos;
case rocksdb_new_table_block_iter_nanos:
return rep->new_table_block_iter_nanos;
case rocksdb_new_table_iterator_nanos:
return rep->new_table_iterator_nanos;
case rocksdb_block_seek_nanos:
return rep->block_seek_nanos;
case rocksdb_find_table_nanos:
return rep->find_table_nanos;
case rocksdb_bloom_memtable_hit_count:
return rep->bloom_memtable_hit_count;
case rocksdb_bloom_memtable_miss_count:
return rep->bloom_memtable_miss_count;
case rocksdb_bloom_sst_hit_count:
return rep->bloom_sst_hit_count;
case rocksdb_bloom_sst_miss_count:
return rep->bloom_sst_miss_count;
case rocksdb_key_lock_wait_time:
return rep->key_lock_wait_time;
case rocksdb_key_lock_wait_count:
return rep->key_lock_wait_count;
case rocksdb_env_new_sequential_file_nanos:
return rep->env_new_sequential_file_nanos;
case rocksdb_env_new_random_access_file_nanos:
return rep->env_new_random_access_file_nanos;
case rocksdb_env_new_writable_file_nanos:
return rep->env_new_writable_file_nanos;
case rocksdb_env_reuse_writable_file_nanos:
return rep->env_reuse_writable_file_nanos;
case rocksdb_env_new_random_rw_file_nanos:
return rep->env_new_random_rw_file_nanos;
case rocksdb_env_new_directory_nanos:
return rep->env_new_directory_nanos;
case rocksdb_env_file_exists_nanos:
return rep->env_file_exists_nanos;
case rocksdb_env_get_children_nanos:
return rep->env_get_children_nanos;
case rocksdb_env_get_children_file_attributes_nanos:
return rep->env_get_children_file_attributes_nanos;
case rocksdb_env_delete_file_nanos:
return rep->env_delete_file_nanos;
case rocksdb_env_create_dir_nanos:
return rep->env_create_dir_nanos;
case rocksdb_env_create_dir_if_missing_nanos:
return rep->env_create_dir_if_missing_nanos;
case rocksdb_env_delete_dir_nanos:
return rep->env_delete_dir_nanos;
case rocksdb_env_get_file_size_nanos:
return rep->env_get_file_size_nanos;
case rocksdb_env_get_file_modification_time_nanos:
return rep->env_get_file_modification_time_nanos;
case rocksdb_env_rename_file_nanos:
return rep->env_rename_file_nanos;
case rocksdb_env_link_file_nanos:
return rep->env_link_file_nanos;
case rocksdb_env_lock_file_nanos:
return rep->env_lock_file_nanos;
case rocksdb_env_unlock_file_nanos:
return rep->env_unlock_file_nanos;
case rocksdb_env_new_logger_nanos:
return rep->env_new_logger_nanos;
default:
break;
}
return 0;
}
void rocksdb_perfcontext_destroy(rocksdb_perfcontext_t* context) {
delete context;
}
/*
TODO:
DB::OpenForReadOnly
+1 -4
View File
@@ -54,9 +54,6 @@ ColumnFamilyHandleImpl::~ColumnFamilyHandleImpl() {
#endif // ROCKSDB_LITE
// Job id == 0 means that this is not our background process, but rather
// user thread
// Need to hold some shared pointers owned by the initial_cf_options
// before final cleaning up finishes.
ColumnFamilyOptions initial_cf_options_copy = cfd_->initial_cf_options();
JobContext job_context(0);
mutex_->Lock();
if (cfd_->Unref()) {
@@ -386,7 +383,7 @@ ColumnFamilyData::ColumnFamilyData(
next_(nullptr),
prev_(nullptr),
log_number_(0),
flush_reason_(FlushReason::kOthers),
flush_reason_(FlushReason::kUnknown),
column_family_set_(column_family_set),
pending_flush_(false),
pending_compaction_(false),
-4
View File
@@ -370,10 +370,6 @@ class ColumnFamilyData {
bool initialized() const { return initialized_.load(); }
const ColumnFamilyOptions& initial_cf_options() {
return initial_cf_options_;
}
Env::WriteLifeTimeHint CalculateSSTWriteHint(int level);
private:
+28
View File
@@ -383,6 +383,9 @@ class ColumnFamilyTest : public testing::Test {
void AssertFilesPerLevel(const std::string& value, int cf) {
#ifndef ROCKSDB_LITE
ASSERT_EQ(value, FilesPerLevel(cf));
#else
(void) value;
(void) cf;
#endif
}
@@ -397,6 +400,8 @@ class ColumnFamilyTest : public testing::Test {
void AssertCountLiveFiles(int expected_value) {
#ifndef ROCKSDB_LITE
ASSERT_EQ(expected_value, CountLiveFiles());
#else
(void) expected_value;
#endif
}
@@ -445,6 +450,8 @@ class ColumnFamilyTest : public testing::Test {
void AssertCountLiveLogFiles(int value) {
#ifndef ROCKSDB_LITE // GetSortedWalFiles is not supported
ASSERT_EQ(value, CountLiveLogFiles());
#else
(void) value;
#endif // !ROCKSDB_LITE
}
@@ -2803,6 +2810,27 @@ TEST_F(ColumnFamilyTest, CreateAndDestoryOptions) {
ASSERT_OK(db_->DestroyColumnFamilyHandle(cfh));
}
TEST_F(ColumnFamilyTest, CreateDropAndDestroy) {
ColumnFamilyHandle* cfh;
Open();
ASSERT_OK(db_->CreateColumnFamily(ColumnFamilyOptions(), "yoyo", &cfh));
ASSERT_OK(db_->Put(WriteOptions(), cfh, "foo", "bar"));
ASSERT_OK(db_->Flush(FlushOptions(), cfh));
ASSERT_OK(db_->DropColumnFamily(cfh));
ASSERT_OK(db_->DestroyColumnFamilyHandle(cfh));
}
TEST_F(ColumnFamilyTest, CreateDropAndDestroyWithoutFileDeletion) {
ColumnFamilyHandle* cfh;
Open();
ASSERT_OK(db_->CreateColumnFamily(ColumnFamilyOptions(), "yoyo", &cfh));
ASSERT_OK(db_->Put(WriteOptions(), cfh, "foo", "bar"));
ASSERT_OK(db_->Flush(FlushOptions(), cfh));
ASSERT_OK(db_->DisableFileDeletions());
ASSERT_OK(db_->DropColumnFamily(cfh));
ASSERT_OK(db_->DestroyColumnFamilyHandle(cfh));
}
#ifndef ROCKSDB_LITE
TEST_F(ColumnFamilyTest, FlushCloseWALFiles) {
SpecialEnv env(Env::Default());
+1 -1
View File
@@ -318,7 +318,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr,
"SKIPPED as DBImpl::CompactFiles is not supported in ROCKSDB_LITE\n");
return 0;
+2 -2
View File
@@ -1034,7 +1034,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr, "SKIPPED, not supported in ROCKSDB_LITE\n");
return 0;
}
@@ -1043,5 +1043,5 @@ int main(int argc, char** argv) {
#else
int main(int argc, char** argv) { return 0; }
int main(int /*argc*/, char** /*argv*/) { return 0; }
#endif // !defined(IOS_CROSS_COMPILE)
+1 -1
View File
@@ -946,7 +946,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr,
"SKIPPED as CompactionJobStats is not supported in ROCKSDB_LITE\n");
return 0;
+1 -1
View File
@@ -510,7 +510,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr, "SKIPPED as RepairDB() is not supported in ROCKSDB_LITE\n");
return 0;
}
+1 -1
View File
@@ -333,7 +333,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr, "SKIPPED as Cuckoo table is not supported in ROCKSDB_LITE\n");
return 0;
}
+2
View File
@@ -3482,6 +3482,8 @@ int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
#else
(void) argc;
(void) argv;
return 0;
#endif
}
+2
View File
@@ -501,6 +501,8 @@ int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
#else
(void) argc;
(void) argv;
return 0;
#endif
}
+1 -1
View File
@@ -74,7 +74,7 @@ Status DBImpl::EnableFileDeletions(bool force) {
}
int DBImpl::IsFileDeletionsEnabled() const {
return !disable_delete_obsolete_files_;
return disable_delete_obsolete_files_;
}
Status DBImpl::GetLiveFiles(std::vector<std::string>& ret,
+8 -8
View File
@@ -2145,9 +2145,9 @@ Status DBImpl::DeleteFile(std::string name) {
status = versions_->LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(),
&edit, &mutex_, directories_.GetDbDir());
if (status.ok()) {
InstallSuperVersionAndScheduleWork(cfd, &job_context.superversion_context,
*cfd->GetLatestMutableCFOptions(),
FlushReason::kDeleteFiles);
InstallSuperVersionAndScheduleWork(
cfd, &job_context.superversion_context,
*cfd->GetLatestMutableCFOptions());
}
FindObsoleteFiles(&job_context, false);
} // lock released here
@@ -2229,9 +2229,9 @@ Status DBImpl::DeleteFilesInRanges(ColumnFamilyHandle* column_family,
status = versions_->LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(),
&edit, &mutex_, directories_.GetDbDir());
if (status.ok()) {
InstallSuperVersionAndScheduleWork(cfd, &job_context.superversion_context,
*cfd->GetLatestMutableCFOptions(),
FlushReason::kDeleteFiles);
InstallSuperVersionAndScheduleWork(
cfd, &job_context.superversion_context,
*cfd->GetLatestMutableCFOptions());
}
for (auto* deleted_file : deleted_files) {
deleted_file->being_compacted = false;
@@ -2859,8 +2859,8 @@ Status DBImpl::IngestExternalFile(
&mutex_, directories_.GetDbDir());
}
if (status.ok()) {
InstallSuperVersionAndScheduleWork(cfd, &sv_context, *mutable_cf_options,
FlushReason::kExternalFileIngestion);
InstallSuperVersionAndScheduleWork(cfd, &sv_context,
*mutable_cf_options);
}
// Resume writes to the DB
+3 -7
View File
@@ -223,8 +223,6 @@ class DBImpl : public DB {
virtual Status SyncWAL() override;
virtual SequenceNumber GetLatestSequenceNumber() const override;
// REQUIRES: joined the main write queue if two_write_queues is disabled, and
// the second write queue otherwise.
virtual void SetLastPublishedSequence(SequenceNumber seq);
// Returns LastSequence in last_seq_same_as_publish_seq_
// mode and LastAllocatedSequence otherwise. This is useful when visiblility
@@ -835,8 +833,7 @@ class DBImpl : public DB {
Status ScheduleFlushes(WriteContext* context);
Status SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context,
FlushReason flush_reason = FlushReason::kOthers);
Status SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context);
// Force current memtable contents to be flushed.
Status FlushMemTable(ColumnFamilyData* cfd, const FlushOptions& options,
@@ -876,7 +873,7 @@ class DBImpl : public DB {
size_t seq_inc);
// Used by WriteImpl to update bg_error_ if paranoid check is enabled.
void WriteStatusCheck(const Status& status);
void WriteCallbackStatusCheck(const Status& status);
// Used by WriteImpl to update bg_error_ in case of memtable insert error.
void MemTableInsertStatusCheck(const Status& memtable_insert_status);
@@ -1362,8 +1359,7 @@ class DBImpl : public DB {
// state needs flush or compaction.
void InstallSuperVersionAndScheduleWork(
ColumnFamilyData* cfd, SuperVersionContext* sv_context,
const MutableCFOptions& mutable_cf_options,
FlushReason flush_reason = FlushReason::kOthers);
const MutableCFOptions& mutable_cf_options);
#ifndef ROCKSDB_LITE
using DB::GetPropertiesOfAllTables;
+7 -7
View File
@@ -655,7 +655,7 @@ Status DBImpl::CompactFilesImpl(
if (status.ok()) {
InstallSuperVersionAndScheduleWork(
c->column_family_data(), &job_context->superversion_context,
*c->mutable_cf_options(), FlushReason::kManualCompaction);
*c->mutable_cf_options());
}
c->ReleaseCompactionFiles(s);
@@ -891,7 +891,7 @@ Status DBImpl::Flush(const FlushOptions& flush_options,
ROCKS_LOG_INFO(immutable_db_options_.info_log, "[%s] Manual flush start.",
cfh->GetName().c_str());
Status s =
FlushMemTable(cfh->cfd(), flush_options, FlushReason::kManualFlush);
FlushMemTable(cfh->cfd(), flush_options, FlushReason::kManualCompaction);
ROCKS_LOG_INFO(immutable_db_options_.info_log,
"[%s] Manual flush finished, status: %s\n",
cfh->GetName().c_str(), s.ToString().c_str());
@@ -1727,7 +1727,7 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
&mutex_, directories_.GetDbDir());
InstallSuperVersionAndScheduleWork(
c->column_family_data(), &job_context->superversion_context,
*c->mutable_cf_options(), FlushReason::kAutoCompaction);
*c->mutable_cf_options());
ROCKS_LOG_BUFFER(log_buffer, "[%s] Deleted %d files\n",
c->column_family_data()->GetName().c_str(),
c->num_input_files(0));
@@ -1774,7 +1774,7 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
// Use latest MutableCFOptions
InstallSuperVersionAndScheduleWork(
c->column_family_data(), &job_context->superversion_context,
*c->mutable_cf_options(), FlushReason::kAutoCompaction);
*c->mutable_cf_options());
VersionStorageInfo::LevelSummaryStorage tmp;
c->column_family_data()->internal_stats()->IncBytesMoved(c->output_level(),
@@ -1853,7 +1853,7 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
if (status.ok()) {
InstallSuperVersionAndScheduleWork(
c->column_family_data(), &job_context->superversion_context,
*c->mutable_cf_options(), FlushReason::kAutoCompaction);
*c->mutable_cf_options());
}
*made_progress = true;
}
@@ -2044,7 +2044,7 @@ bool DBImpl::MCOverlap(ManualCompactionState* m, ManualCompactionState* m1) {
void DBImpl::InstallSuperVersionAndScheduleWork(
ColumnFamilyData* cfd, SuperVersionContext* sv_context,
const MutableCFOptions& mutable_cf_options, FlushReason flush_reason) {
const MutableCFOptions& mutable_cf_options) {
mutex_.AssertHeld();
// Update max_total_in_memory_state_
@@ -2063,7 +2063,7 @@ void DBImpl::InstallSuperVersionAndScheduleWork(
// Whenever we install new SuperVersion, we might need to issue new flushes or
// compactions.
SchedulePendingFlush(cfd, flush_reason);
SchedulePendingFlush(cfd, FlushReason::kSuperVersionChange);
SchedulePendingCompaction(cfd);
MaybeScheduleFlushOrCompaction();
+11 -17
View File
@@ -133,8 +133,7 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
for (auto* writer : *(w.write_group)) {
if (!writer->CallbackFailed() && writer->pre_release_callback) {
assert(writer->sequence != kMaxSequenceNumber);
Status ws = writer->pre_release_callback->Callback(writer->sequence,
disable_memtable);
Status ws = writer->pre_release_callback->Callback(writer->sequence);
if (!ws.ok()) {
status = ws;
break;
@@ -340,7 +339,7 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
PERF_TIMER_START(write_pre_and_post_process_time);
if (!w.CallbackFailed()) {
WriteStatusCheck(status);
WriteCallbackStatusCheck(status);
}
if (need_log_sync) {
@@ -369,8 +368,7 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
for (auto* writer : write_group) {
if (!writer->CallbackFailed() && writer->pre_release_callback) {
assert(writer->sequence != kMaxSequenceNumber);
Status ws = writer->pre_release_callback->Callback(writer->sequence,
disable_memtable);
Status ws = writer->pre_release_callback->Callback(writer->sequence);
if (!ws.ok()) {
status = ws;
break;
@@ -464,7 +462,7 @@ Status DBImpl::PipelinedWriteImpl(const WriteOptions& write_options,
}
if (!w.CallbackFailed()) {
WriteStatusCheck(w.status);
WriteCallbackStatusCheck(w.status);
}
if (need_log_sync) {
@@ -625,15 +623,13 @@ Status DBImpl::WriteImplWALOnly(const WriteOptions& write_options,
PERF_TIMER_START(write_pre_and_post_process_time);
if (!w.CallbackFailed()) {
WriteStatusCheck(status);
WriteCallbackStatusCheck(status);
}
if (status.ok()) {
for (auto* writer : write_group) {
if (!writer->CallbackFailed() && writer->pre_release_callback) {
assert(writer->sequence != kMaxSequenceNumber);
const bool DISABLE_MEMTABLE = true;
Status ws = writer->pre_release_callback->Callback(writer->sequence,
DISABLE_MEMTABLE);
Status ws = writer->pre_release_callback->Callback(writer->sequence);
if (!ws.ok()) {
status = ws;
break;
@@ -651,7 +647,7 @@ Status DBImpl::WriteImplWALOnly(const WriteOptions& write_options,
return status;
}
void DBImpl::WriteStatusCheck(const Status& status) {
void DBImpl::WriteCallbackStatusCheck(const Status& status) {
// Is setting bg_error_ enough here? This will at least stop
// compaction and fail any further writes.
if (immutable_db_options_.paranoid_checks && !status.ok() &&
@@ -1066,8 +1062,7 @@ Status DBImpl::HandleWriteBufferFull(WriteContext* write_context) {
}
}
if (cfd_picked != nullptr) {
status = SwitchMemtable(cfd_picked, write_context,
FlushReason::kWriteBufferFull);
status = SwitchMemtable(cfd_picked, write_context);
if (status.ok()) {
cfd_picked->imm()->FlushRequested();
SchedulePendingFlush(cfd_picked, FlushReason::kWriteBufferFull);
@@ -1168,7 +1163,7 @@ Status DBImpl::ThrottleLowPriWritesIfNeeded(const WriteOptions& write_options,
Status DBImpl::ScheduleFlushes(WriteContext* context) {
ColumnFamilyData* cfd;
while ((cfd = flush_scheduler_.TakeNextColumnFamily()) != nullptr) {
auto status = SwitchMemtable(cfd, context, FlushReason::kWriteBufferFull);
auto status = SwitchMemtable(cfd, context);
if (cfd->Unref()) {
delete cfd;
}
@@ -1197,8 +1192,7 @@ void DBImpl::NotifyOnMemTableSealed(ColumnFamilyData* /*cfd*/,
// REQUIRES: mutex_ is held
// REQUIRES: this thread is currently at the front of the writer queue
Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context,
FlushReason flush_reason) {
Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) {
mutex_.AssertHeld();
WriteThread::Writer nonmem_w;
if (two_write_queues_) {
@@ -1366,7 +1360,7 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context,
new_mem->Ref();
cfd->SetMemtable(new_mem);
InstallSuperVersionAndScheduleWork(cfd, &context->superversion_context,
mutable_cf_options, flush_reason);
mutable_cf_options);
if (two_write_queues_) {
nonmem_write_thread_.ExitUnbatched(&nonmem_w);
}
+6 -4
View File
@@ -36,7 +36,9 @@ class TestIterator : public InternalIterator {
valid_(false),
sequence_number_(0),
iter_(0),
cmp(comparator) {}
cmp(comparator) {
data_.reserve(16);
}
void AddPut(std::string argkey, std::string argvalue) {
Add(argkey, kTypeValue, argvalue);
@@ -2549,7 +2551,7 @@ TEST_F(DBIterWithMergeIterTest, InnerMergeIteratorDataRace1) {
// MergeIterator::Prev() realized the mem table iterator is at its end
// and before an SeekToLast() is called.
rocksdb::SyncPoint::GetInstance()->SetCallBack(
"MergeIterator::Prev:BeforeSeekToLast",
"MergeIterator::Prev:BeforePrev",
[&](void* arg) { internal_iter2_->Add("z", kTypeValue, "7", 12u); });
rocksdb::SyncPoint::GetInstance()->EnableProcessing();
@@ -2585,7 +2587,7 @@ TEST_F(DBIterWithMergeIterTest, InnerMergeIteratorDataRace2) {
// mem table after MergeIterator::Prev() realized the mem tableiterator is at
// its end and before an SeekToLast() is called.
rocksdb::SyncPoint::GetInstance()->SetCallBack(
"MergeIterator::Prev:BeforeSeekToLast", [&](void* arg) {
"MergeIterator::Prev:BeforePrev", [&](void* arg) {
internal_iter2_->Add("z", kTypeValue, "7", 12u);
internal_iter2_->Add("z", kTypeValue, "7", 11u);
});
@@ -2623,7 +2625,7 @@ TEST_F(DBIterWithMergeIterTest, InnerMergeIteratorDataRace3) {
// mem table after MergeIterator::Prev() realized the mem table iterator is at
// its end and before an SeekToLast() is called.
rocksdb::SyncPoint::GetInstance()->SetCallBack(
"MergeIterator::Prev:BeforeSeekToLast", [&](void* arg) {
"MergeIterator::Prev:BeforePrev", [&](void* arg) {
internal_iter2_->Add("z", kTypeValue, "7", 16u, true);
internal_iter2_->Add("z", kTypeValue, "7", 15u, true);
internal_iter2_->Add("z", kTypeValue, "7", 14u, true);
+78
View File
@@ -2043,6 +2043,43 @@ TEST_P(DBIteratorTest, CreationFailure) {
delete iter;
}
TEST_P(DBIteratorTest, UpperBoundWithChangeDirection) {
Options options = CurrentOptions();
options.max_sequential_skip_in_iterations = 3;
DestroyAndReopen(options);
// write a bunch of kvs to the database.
ASSERT_OK(Put("a", "1"));
ASSERT_OK(Put("y", "1"));
ASSERT_OK(Put("y1", "1"));
ASSERT_OK(Put("y2", "1"));
ASSERT_OK(Put("y3", "1"));
ASSERT_OK(Put("z", "1"));
ASSERT_OK(Flush());
ASSERT_OK(Put("a", "1"));
ASSERT_OK(Put("z", "1"));
ASSERT_OK(Put("bar", "1"));
ASSERT_OK(Put("foo", "1"));
std::string upper_bound = "x";
Slice ub_slice(upper_bound);
ReadOptions ro;
ro.iterate_upper_bound = &ub_slice;
ro.max_skippable_internal_keys = 1000;
Iterator* iter = NewIterator(ro);
iter->Seek("foo");
ASSERT_TRUE(iter->Valid());
ASSERT_EQ("foo", iter->key().ToString());
iter->Prev();
ASSERT_TRUE(iter->Valid());
ASSERT_OK(iter->status());
ASSERT_EQ("bar", iter->key().ToString());
delete iter;
}
TEST_P(DBIteratorTest, TableFilter) {
ASSERT_OK(Put("a", "1"));
dbfull()->Flush(FlushOptions());
@@ -2109,6 +2146,47 @@ TEST_P(DBIteratorTest, TableFilter) {
}
}
TEST_P(DBIteratorTest, UpperBoundWithPrevReseek) {
Options options = CurrentOptions();
options.max_sequential_skip_in_iterations = 3;
DestroyAndReopen(options);
// write a bunch of kvs to the database.
ASSERT_OK(Put("a", "1"));
ASSERT_OK(Put("y", "1"));
ASSERT_OK(Put("z", "1"));
ASSERT_OK(Flush());
ASSERT_OK(Put("a", "1"));
ASSERT_OK(Put("z", "1"));
ASSERT_OK(Put("bar", "1"));
ASSERT_OK(Put("foo", "1"));
ASSERT_OK(Put("foo", "2"));
ASSERT_OK(Put("foo", "3"));
ASSERT_OK(Put("foo", "4"));
ASSERT_OK(Put("foo", "5"));
const Snapshot* snapshot = db_->GetSnapshot();
ASSERT_OK(Put("foo", "6"));
std::string upper_bound = "x";
Slice ub_slice(upper_bound);
ReadOptions ro;
ro.snapshot = snapshot;
ro.iterate_upper_bound = &ub_slice;
Iterator* iter = NewIterator(ro);
iter->SeekForPrev("goo");
ASSERT_TRUE(iter->Valid());
ASSERT_EQ("foo", iter->key().ToString());
iter->Prev();
ASSERT_TRUE(iter->Valid());
ASSERT_EQ("bar", iter->key().ToString());
delete iter;
db_->ReleaseSnapshot(snapshot);
}
TEST_P(DBIteratorTest, SkipStatistics) {
Options options = CurrentOptions();
options.statistics = rocksdb::CreateDBStatistics();
+2
View File
@@ -289,6 +289,8 @@ int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
#else
(void) argc;
(void) argv;
return 0;
#endif
}
+5 -5
View File
@@ -69,27 +69,27 @@ TEST_F(DBPropertiesTest, Empty) {
ASSERT_OK(db_->DisableFileDeletions());
ASSERT_TRUE(
dbfull()->GetProperty("rocksdb.is-file-deletions-enabled", &num));
ASSERT_EQ("0", num);
ASSERT_EQ("1", num);
ASSERT_OK(db_->DisableFileDeletions());
ASSERT_TRUE(
dbfull()->GetProperty("rocksdb.is-file-deletions-enabled", &num));
ASSERT_EQ("0", num);
ASSERT_EQ("2", num);
ASSERT_OK(db_->DisableFileDeletions());
ASSERT_TRUE(
dbfull()->GetProperty("rocksdb.is-file-deletions-enabled", &num));
ASSERT_EQ("0", num);
ASSERT_EQ("3", num);
ASSERT_OK(db_->EnableFileDeletions(false));
ASSERT_TRUE(
dbfull()->GetProperty("rocksdb.is-file-deletions-enabled", &num));
ASSERT_EQ("0", num);
ASSERT_EQ("2", num);
ASSERT_OK(db_->EnableFileDeletions());
ASSERT_TRUE(
dbfull()->GetProperty("rocksdb.is-file-deletions-enabled", &num));
ASSERT_EQ("1", num);
ASSERT_EQ("0", num);
} while (ChangeOptions());
}
+2
View File
@@ -809,6 +809,8 @@ int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
#else
(void) argc;
(void) argv;
return 0;
#endif
}
+2
View File
@@ -1759,6 +1759,8 @@ int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
#else
(void) argc;
(void) argv;
return 0;
#endif
}
-16
View File
@@ -80,22 +80,6 @@ TEST_P(DBWriteTest, IOErrorOnWALWritePropagateToWriteThreadFollower) {
Close();
}
TEST_P(DBWriteTest, IOErrorOnWALWriteTriggersReadOnlyMode) {
std::unique_ptr<FaultInjectionTestEnv> mock_env(
new FaultInjectionTestEnv(Env::Default()));
Options options = GetOptions();
options.env = mock_env.get();
Reopen(options);
for (int i = 0; i < 2; i++) {
// Forcibly fail WAL write for the first Put only. Subsequent Puts should
// fail due to read-only mode
mock_env->SetFilesystemActive(i != 0);
ASSERT_FALSE(Put("key" + ToString(i), "value").ok());
}
// Close before mock_env destruct.
Close();
}
INSTANTIATE_TEST_CASE_P(DBWriteTestInstance, DBWriteTest,
testing::Values(DBTestBase::kDefault,
DBTestBase::kConcurrentWALWrites,
+39
View File
@@ -106,6 +106,45 @@ const char* InternalKeyComparator::Name() const {
return name_.c_str();
}
int InternalKeyComparator::Compare(const Slice& akey, const Slice& bkey) const {
// Order by:
// increasing user key (according to user-supplied comparator)
// decreasing sequence number
// decreasing type (though sequence# should be enough to disambiguate)
int r = user_comparator_->Compare(ExtractUserKey(akey), ExtractUserKey(bkey));
PERF_COUNTER_ADD(user_key_comparison_count, 1);
if (r == 0) {
const uint64_t anum = DecodeFixed64(akey.data() + akey.size() - 8);
const uint64_t bnum = DecodeFixed64(bkey.data() + bkey.size() - 8);
if (anum > bnum) {
r = -1;
} else if (anum < bnum) {
r = +1;
}
}
return r;
}
int InternalKeyComparator::CompareKeySeq(const Slice& akey,
const Slice& bkey) const {
// Order by:
// increasing user key (according to user-supplied comparator)
// decreasing sequence number
int r = user_comparator_->Compare(ExtractUserKey(akey), ExtractUserKey(bkey));
PERF_COUNTER_ADD(user_key_comparison_count, 1);
if (r == 0) {
// Shift the number to exclude the last byte which contains the value type
const uint64_t anum = DecodeFixed64(akey.data() + akey.size() - 8) >> 8;
const uint64_t bnum = DecodeFixed64(bkey.data() + bkey.size() - 8) >> 8;
if (anum > bnum) {
r = -1;
} else if (anum < bnum) {
r = +1;
}
}
return r;
}
int InternalKeyComparator::Compare(const ParsedInternalKey& a,
const ParsedInternalKey& b) const {
// Order by:
-43
View File
@@ -11,7 +11,6 @@
#include <stdio.h>
#include <string>
#include <utility>
#include "monitoring/perf_context_imp.h"
#include "rocksdb/comparator.h"
#include "rocksdb/db.h"
#include "rocksdb/filter_policy.h"
@@ -608,46 +607,4 @@ struct RangeTombstone {
}
};
inline
int InternalKeyComparator::Compare(const Slice& akey, const Slice& bkey) const {
// Order by:
// increasing user key (according to user-supplied comparator)
// decreasing sequence number
// decreasing type (though sequence# should be enough to disambiguate)
int r = user_comparator_->Compare(ExtractUserKey(akey), ExtractUserKey(bkey));
PERF_COUNTER_ADD(user_key_comparison_count, 1);
if (r == 0) {
const uint64_t anum = DecodeFixed64(akey.data() + akey.size() - 8);
const uint64_t bnum = DecodeFixed64(bkey.data() + bkey.size() - 8);
if (anum > bnum) {
r = -1;
} else if (anum < bnum) {
r = +1;
}
}
return r;
}
inline
int InternalKeyComparator::CompareKeySeq(const Slice& akey,
const Slice& bkey) const {
// Order by:
// increasing user key (according to user-supplied comparator)
// decreasing sequence number
int r = user_comparator_->Compare(ExtractUserKey(akey), ExtractUserKey(bkey));
PERF_COUNTER_ADD(user_key_comparison_count, 1);
if (r == 0) {
// Shift the number to exclude the last byte which contains the value type
const uint64_t anum = DecodeFixed64(akey.data() + akey.size() - 8) >> 8;
const uint64_t bnum = DecodeFixed64(bkey.data() + bkey.size() - 8) >> 8;
if (anum > bnum) {
r = -1;
} else if (anum < bnum) {
r = +1;
}
}
return r;
}
} // namespace rocksdb
+1 -1
View File
@@ -500,7 +500,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr,
"SKIPPED as DBImpl::DeleteFile is not supported in ROCKSDB_LITE\n");
return 0;
+1 -1
View File
@@ -2002,7 +2002,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr,
"SKIPPED as External SST File Writer and Ingestion are not supported "
"in ROCKSDB_LITE\n");
+4 -8
View File
@@ -56,8 +56,8 @@ namespace rocksdb {
const char* GetFlushReasonString (FlushReason flush_reason) {
switch (flush_reason) {
case FlushReason::kOthers:
return "Other Reasons";
case FlushReason::kUnknown:
return "Unknown";
case FlushReason::kGetLiveFiles:
return "Get Live Files";
case FlushReason::kShutDown:
@@ -72,12 +72,8 @@ const char* GetFlushReasonString (FlushReason flush_reason) {
return "Write Buffer Full";
case FlushReason::kTest:
return "Test";
case FlushReason::kDeleteFiles:
return "Delete Files";
case FlushReason::kAutoCompaction:
return "Auto Compaction";
case FlushReason::kManualFlush:
return "Manual Flush";
case FlushReason::kSuperVersionChange:
return "SuperVersion Change";
default:
return "Invalid";
}
+1 -1
View File
@@ -229,7 +229,7 @@ int MemTable::KeyComparator::operator()(const char* prefix_len_key1,
}
int MemTable::KeyComparator::operator()(const char* prefix_len_key,
const KeyComparator::DecodedType& key)
const Slice& key)
const {
// Internal keys are encoded as length-prefixed strings.
Slice a = GetLengthPrefixedSlice(prefix_len_key);
+2 -2
View File
@@ -64,7 +64,7 @@ struct MemTablePostProcessInfo {
};
// Note: Many of the methods in this class have comments indicating that
// external synchronization is required as these methods are not thread-safe.
// external synchromization is required as these methods are not thread-safe.
// It is up to higher layers of code to decide how to prevent concurrent
// invokation of these methods. This is usually done by acquiring either
// the db mutex or the single writer thread.
@@ -84,7 +84,7 @@ class MemTable {
virtual int operator()(const char* prefix_len_key1,
const char* prefix_len_key2) const override;
virtual int operator()(const char* prefix_len_key,
const DecodedType& key) const override;
const Slice& key) const override;
};
// MemTables are reference counted. The initial reference count
+1 -1
View File
@@ -112,7 +112,7 @@ int main(int argc, char** argv) {
#include <cstdio>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
printf("Skipped as Options file is not supported in RocksDBLite.\n");
return 0;
}
+1 -1
View File
@@ -1170,7 +1170,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr, "SKIPPED as plain table is not supported in ROCKSDB_LITE\n");
return 0;
}
+1 -3
View File
@@ -24,9 +24,7 @@ class PreReleaseCallback {
// propagated to all the writers in the write group.
// seq is the sequence number that is used for this write and will be
// released.
// is_mem_disabled is currently used for debugging purposes to assert that
// the callback is done from the right write queue.
virtual Status Callback(SequenceNumber seq, const bool is_mem_disabled) = 0;
virtual Status Callback(SequenceNumber seq) = 0;
};
} // namespace rocksdb
+1 -1
View File
@@ -887,7 +887,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr,
"SKIPPED as HashSkipList and HashLinkList are not supported in "
"ROCKSDB_LITE\n");
-7
View File
@@ -536,11 +536,4 @@ bool RangeDelAggregator::IsEmpty() {
return true;
}
bool RangeDelAggregator::AddFile(uint64_t file_number) {
if (added_files_ == nullptr) {
added_files_.reset(new std::set<uint64_t>());
}
return added_files_->emplace(file_number).second;
}
} // namespace rocksdb
-6
View File
@@ -6,7 +6,6 @@
#pragma once
#include <map>
#include <set>
#include <string>
#include <vector>
@@ -141,7 +140,6 @@ class RangeDelAggregator {
CompactionIterationStats* range_del_out_stats = nullptr,
bool bottommost_level = false);
bool IsEmpty();
bool AddFile(uint64_t file_number);
private:
// Maps tombstone user start key -> tombstone object
@@ -182,10 +180,6 @@ class RangeDelAggregator {
const InternalKeyComparator& icmp_;
// collapse range deletions so they're binary searchable
const bool collapse_deletions_;
// Record files whose tombstones have been added, to avoid duplicate adding.
// Same as rep_, we initializes it lazily.
std::unique_ptr<std::set<uint64_t>> added_files_;
};
} // namespace rocksdb
+1 -1
View File
@@ -356,7 +356,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr, "SKIPPED as RepairDB is not supported in ROCKSDB_LITE\n");
return 0;
}
+7 -9
View File
@@ -247,15 +247,13 @@ InternalIterator* TableCache::NewIterator(
}
}
if (s.ok() && range_del_agg != nullptr && !options.ignore_range_deletions) {
if (range_del_agg->AddFile(fd.GetNumber())) {
std::unique_ptr<InternalIterator> range_del_iter(
table_reader->NewRangeTombstoneIterator(options));
if (range_del_iter != nullptr) {
s = range_del_iter->status();
}
if (s.ok()) {
s = range_del_agg->AddTombstones(std::move(range_del_iter));
}
std::unique_ptr<InternalIterator> range_del_iter(
table_reader->NewRangeTombstoneIterator(options));
if (range_del_iter != nullptr) {
s = range_del_iter->status();
}
if (s.ok()) {
s = range_del_agg->AddTombstones(std::move(range_del_iter));
}
}
+1 -1
View File
@@ -303,7 +303,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr, "SKIPPED as WalManager is not supported in ROCKSDB_LITE\n");
return 0;
}
+2 -3
View File
@@ -294,8 +294,7 @@ TEST_F(WriteCallbackTest, WriteWithCallbackTest) {
public:
PublishSeqCallback(DBImpl* db_impl_in)
: db_impl_(db_impl_in) {}
virtual Status Callback(SequenceNumber last_seq,
const bool /*not used*/) override {
virtual Status Callback(SequenceNumber last_seq) {
db_impl_->SetLastPublishedSequence(last_seq);
return Status::OK();
}
@@ -432,7 +431,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr,
"SKIPPED as WriteWithCallback is not supported in ROCKSDB_LITE\n");
return 0;
+9
View File
@@ -802,6 +802,15 @@ class PosixEnv : public Env {
#endif
}
virtual void LowerThreadPoolCPUPriority(Priority pool = LOW) override {
assert(pool >= Priority::BOTTOM && pool <= Priority::HIGH);
#ifdef OS_LINUX
thread_pools_[pool].LowerCPUPriority();
#else
(void)pool;
#endif
}
virtual std::string TimeToString(uint64_t secondsSince1970) override {
const time_t seconds = (time_t)secondsSince1970;
struct tm t;
+6 -2
View File
@@ -873,8 +873,12 @@ void PosixWritableFile::SetWriteLifeTimeHint(Env::WriteLifeTimeHint hint) {
if (fcntl(fd_, F_SET_RW_HINT, &hint) == 0) {
write_hint_ = hint;
}
#endif
#endif
#else
(void)hint;
#endif // ROCKSDB_VALGRIND_RUN
#else
(void)hint;
#endif // OS_LINUX
}
Status PosixWritableFile::InvalidateCache(size_t offset, size_t length) {
+9 -1
View File
@@ -87,6 +87,14 @@ struct CompactionOptionsFIFO {
// Compression options for different compression algorithms like Zlib
struct CompressionOptions {
// RocksDB's generic default compression level. Internally it'll be translated
// to the default compression level specific to the library being used (see
// comment above `ColumnFamilyOptions::compression`).
//
// The default value is the max 16-bit int as it'll be written out in OPTIONS
// file, which should be portable.
const static int kDefaultCompressionLevel = 32767;
int window_bits;
int level;
int strategy;
@@ -120,7 +128,7 @@ struct CompressionOptions {
CompressionOptions()
: window_bits(-14),
level(-1),
level(kDefaultCompressionLevel),
strategy(0),
max_dict_bytes(0),
zstd_max_train_bytes(0) {}
-92
View File
@@ -113,7 +113,6 @@ typedef struct rocksdb_envoptions_t rocksdb_envoptions_t;
typedef struct rocksdb_ingestexternalfileoptions_t rocksdb_ingestexternalfileoptions_t;
typedef struct rocksdb_sstfilewriter_t rocksdb_sstfilewriter_t;
typedef struct rocksdb_ratelimiter_t rocksdb_ratelimiter_t;
typedef struct rocksdb_perfcontext_t rocksdb_perfcontext_t;
typedef struct rocksdb_pinnableslice_t rocksdb_pinnableslice_t;
typedef struct rocksdb_transactiondb_options_t rocksdb_transactiondb_options_t;
typedef struct rocksdb_transactiondb_t rocksdb_transactiondb_t;
@@ -970,97 +969,6 @@ extern ROCKSDB_LIBRARY_API rocksdb_ratelimiter_t* rocksdb_ratelimiter_create(
int64_t rate_bytes_per_sec, int64_t refill_period_us, int32_t fairness);
extern ROCKSDB_LIBRARY_API void rocksdb_ratelimiter_destroy(rocksdb_ratelimiter_t*);
/* PerfContext */
enum {
rocksdb_uninitialized = 0,
rocksdb_disable = 1,
rocksdb_enable_count = 2,
rocksdb_enable_time_except_for_mutex = 3,
rocksdb_enable_time = 4,
rocksdb_out_of_bounds = 5
};
enum {
rocksdb_user_key_comparison_count = 0,
rocksdb_block_cache_hit_count,
rocksdb_block_read_count,
rocksdb_block_read_byte,
rocksdb_block_read_time,
rocksdb_block_checksum_time,
rocksdb_block_decompress_time,
rocksdb_get_read_bytes,
rocksdb_multiget_read_bytes,
rocksdb_iter_read_bytes,
rocksdb_internal_key_skipped_count,
rocksdb_internal_delete_skipped_count,
rocksdb_internal_recent_skipped_count,
rocksdb_internal_merge_count,
rocksdb_get_snapshot_time,
rocksdb_get_from_memtable_time,
rocksdb_get_from_memtable_count,
rocksdb_get_post_process_time,
rocksdb_get_from_output_files_time,
rocksdb_seek_on_memtable_time,
rocksdb_seek_on_memtable_count,
rocksdb_next_on_memtable_count,
rocksdb_prev_on_memtable_count,
rocksdb_seek_child_seek_time,
rocksdb_seek_child_seek_count,
rocksdb_seek_min_heap_time,
rocksdb_seek_max_heap_time,
rocksdb_seek_internal_seek_time,
rocksdb_find_next_user_entry_time,
rocksdb_write_wal_time,
rocksdb_write_memtable_time,
rocksdb_write_delay_time,
rocksdb_write_pre_and_post_process_time,
rocksdb_db_mutex_lock_nanos,
rocksdb_db_condition_wait_nanos,
rocksdb_merge_operator_time_nanos,
rocksdb_read_index_block_nanos,
rocksdb_read_filter_block_nanos,
rocksdb_new_table_block_iter_nanos,
rocksdb_new_table_iterator_nanos,
rocksdb_block_seek_nanos,
rocksdb_find_table_nanos,
rocksdb_bloom_memtable_hit_count,
rocksdb_bloom_memtable_miss_count,
rocksdb_bloom_sst_hit_count,
rocksdb_bloom_sst_miss_count,
rocksdb_key_lock_wait_time,
rocksdb_key_lock_wait_count,
rocksdb_env_new_sequential_file_nanos,
rocksdb_env_new_random_access_file_nanos,
rocksdb_env_new_writable_file_nanos,
rocksdb_env_reuse_writable_file_nanos,
rocksdb_env_new_random_rw_file_nanos,
rocksdb_env_new_directory_nanos,
rocksdb_env_file_exists_nanos,
rocksdb_env_get_children_nanos,
rocksdb_env_get_children_file_attributes_nanos,
rocksdb_env_delete_file_nanos,
rocksdb_env_create_dir_nanos,
rocksdb_env_create_dir_if_missing_nanos,
rocksdb_env_delete_dir_nanos,
rocksdb_env_get_file_size_nanos,
rocksdb_env_get_file_modification_time_nanos,
rocksdb_env_rename_file_nanos,
rocksdb_env_link_file_nanos,
rocksdb_env_lock_file_nanos,
rocksdb_env_unlock_file_nanos,
rocksdb_env_new_logger_nanos,
rocksdb_total_metric_count = 68
};
extern ROCKSDB_LIBRARY_API void rocksdb_set_perf_level(int);
extern ROCKSDB_LIBRARY_API rocksdb_perfcontext_t* rocksdb_perfcontext_create();
extern ROCKSDB_LIBRARY_API void rocksdb_perfcontext_reset(
rocksdb_perfcontext_t* context);
extern ROCKSDB_LIBRARY_API char* rocksdb_perfcontext_report(
rocksdb_perfcontext_t* context, unsigned char exclude_zero_counters);
extern ROCKSDB_LIBRARY_API uint64_t rocksdb_perfcontext_metric(
rocksdb_perfcontext_t* context, int metric);
/* Compaction Filter */
extern ROCKSDB_LIBRARY_API rocksdb_compactionfilter_t*
+7
View File
@@ -391,6 +391,9 @@ class Env {
// Lower IO priority for threads from the specified pool.
virtual void LowerThreadPoolIOPriority(Priority /*pool*/ = LOW) {}
// Lower CPU priority for threads from the specified pool.
virtual void LowerThreadPoolCPUPriority(Priority /*pool*/ = LOW) {}
// Converts seconds-since-Jan-01-1970 to a printable string
virtual std::string TimeToString(uint64_t time) = 0;
@@ -1082,6 +1085,10 @@ class EnvWrapper : public Env {
target_->LowerThreadPoolIOPriority(pool);
}
void LowerThreadPoolCPUPriority(Priority pool = LOW) override {
target_->LowerThreadPoolCPUPriority(pool);
}
std::string TimeToString(uint64_t time) override {
return target_->TimeToString(time);
}
+2 -4
View File
@@ -83,7 +83,7 @@ enum class CompactionReason {
};
enum class FlushReason : int {
kOthers = 0x00,
kUnknown = 0x00,
kGetLiveFiles = 0x01,
kShutDown = 0x02,
kExternalFileIngestion = 0x03,
@@ -91,9 +91,7 @@ enum class FlushReason : int {
kWriteBufferManager = 0x05,
kWriteBufferFull = 0x06,
kTest = 0x07,
kDeleteFiles = 0x08,
kAutoCompaction = 0x09,
kManualFlush = 0x0a,
kSuperVersionChange = 0x08,
};
enum class BackgroundErrorReason {
+1 -11
View File
@@ -39,34 +39,24 @@
#include <stdexcept>
#include <stdint.h>
#include <stdlib.h>
#include <rocksdb/slice.h>
namespace rocksdb {
class Arena;
class Allocator;
class LookupKey;
class Slice;
class SliceTransform;
class Logger;
typedef void* KeyHandle;
extern Slice GetLengthPrefixedSlice(const char* data);
class MemTableRep {
public:
// KeyComparator provides a means to compare keys, which are internal keys
// concatenated with values.
class KeyComparator {
public:
typedef rocksdb::Slice DecodedType;
virtual DecodedType decode_key(const char* key) const {
// The format of key is frozen and can be terated as a part of the API
// contract. Refer to MemTable::Add for details.
return GetLengthPrefixedSlice(key);
}
// Compare a and b. Return a negative value if a is less than b, 0 if they
// are equal, and a positive value if a is greater than b
virtual int operator()(const char* prefix_len_key1,
+10
View File
@@ -197,11 +197,21 @@ struct ColumnFamilyOptions : public AdvancedColumnFamilyOptions {
// Typical speeds of kSnappyCompression on an Intel(R) Core(TM)2 2.4GHz:
// ~200-500MB/s compression
// ~400-800MB/s decompression
//
// Note that these speeds are significantly faster than most
// persistent storage speeds, and therefore it is typically never
// worth switching to kNoCompression. Even if the input data is
// incompressible, the kSnappyCompression implementation will
// efficiently detect that and will switch to uncompressed mode.
//
// If you do not set `compression_opts.level`, or set it to
// `CompressionOptions::kDefaultCompressionLevel`, we will attempt to pick the
// default corresponding to `compression` as follows:
//
// - kZSTD: 3
// - kZlibCompression: Z_DEFAULT_COMPRESSION (currently -1)
// - kLZ4HCCompression: 0
// - For all others, we do not specify a compression level
CompressionType compression;
// Compression algorithm that will be used for the bottommost level that
+1 -5
View File
@@ -96,14 +96,10 @@ class SstFileManager {
// @param max_trash_db_ratio: If the trash size constitutes for more than this
// fraction of the total DB size we will start deleting new files passed to
// DeleteScheduler immediately
// @param bytes_max_delete_chunk: if a single file is larger than delete chunk,
// ftruncate the file by this size each time, rather than dropping the whole
// file. 0 means to always delete the whole file.
extern SstFileManager* NewSstFileManager(
Env* env, std::shared_ptr<Logger> info_log = nullptr,
std::string trash_dir = "", int64_t rate_bytes_per_sec = 0,
bool delete_existing_trash = true, Status* status = nullptr,
double max_trash_db_ratio = 0.25,
uint64_t bytes_max_delete_chunk = 64 * 1024 * 1024);
double max_trash_db_ratio = 0.25);
} // namespace rocksdb
+1 -1
View File
@@ -6,7 +6,7 @@
#define ROCKSDB_MAJOR 5
#define ROCKSDB_MINOR 13
#define ROCKSDB_PATCH 0
#define ROCKSDB_PATCH 4
// Do not use these. We made the mistake of declaring macros starting with
// double underscore. Now we have to live with our choice. We'll deprecate these
+8 -5
View File
@@ -304,11 +304,14 @@ jobject Java_org_rocksdb_TransactionDB_getLockStatusData(
return nullptr;
}
const rocksdb::HashMapJni::FnMapKV<const int32_t, const rocksdb::KeyLockInfo> fn_map_kv =
[env, txn_db, &lock_status_data](const std::pair<const int32_t, const rocksdb::KeyLockInfo>& pair) {
const jobject jlong_column_family_id =
rocksdb::LongJni::valueOf(env, pair.first);
if (jlong_column_family_id == nullptr) {
const rocksdb::HashMapJni::FnMapKV<const int32_t, const rocksdb::KeyLockInfo>
fn_map_kv =
[env](
const std::pair<const int32_t, const rocksdb::KeyLockInfo>&
pair) {
const jobject jlong_column_family_id =
rocksdb::LongJni::valueOf(env, pair.first);
if (jlong_column_family_id == nullptr) {
// an error occurred
return std::unique_ptr<std::pair<jobject, jobject>>(nullptr);
}
+17 -38
View File
@@ -45,12 +45,9 @@
#include <stdlib.h>
#include <algorithm>
#include <atomic>
#include <type_traits>
#include "port/likely.h"
#include "port/port.h"
#include "rocksdb/slice.h"
#include "util/allocator.h"
#include "util/coding.h"
#include "util/random.h"
namespace rocksdb {
@@ -62,9 +59,6 @@ class InlineSkipList {
struct Splice;
public:
using DecodedKey = \
typename std::remove_reference<Comparator>::type::DecodedType;
static const uint16_t kMaxPossibleHeight = 32;
// Create a new InlineSkipList object that will use "cmp" for comparing
@@ -218,7 +212,6 @@ class InlineSkipList {
// Return true if key is greater than the data stored in "n". Null n
// is considered infinite. n should not be head_.
bool KeyIsAfterNode(const char* key, Node* n) const;
bool KeyIsAfterNode(const DecodedKey& key, Node* n) const;
// Returns the earliest node with a key >= key.
// Return nullptr if there is no such node.
@@ -248,12 +241,12 @@ class InlineSkipList {
// a node that is after the key. after should be nullptr if a good after
// node isn't conveniently available.
template<bool prefetch_before>
void FindSpliceForLevel(const DecodedKey& key, Node* before, Node* after, int level,
void FindSpliceForLevel(const char* key, Node* before, Node* after, int level,
Node** out_prev, Node** out_next);
// Recomputes Splice levels from highest_level (inclusive) down to
// lowest_level (inclusive).
void RecomputeSpliceLevels(const DecodedKey& key, Splice* splice,
void RecomputeSpliceLevels(const char* key, Splice* splice,
int recompute_level);
// No copying allowed
@@ -442,14 +435,6 @@ bool InlineSkipList<Comparator>::KeyIsAfterNode(const char* key,
return (n != nullptr) && (compare_(n->Key(), key) < 0);
}
template <class Comparator>
bool InlineSkipList<Comparator>::KeyIsAfterNode(const DecodedKey& key,
Node* n) const {
// nullptr n is considered infinite
assert(n != head_);
return (n != nullptr) && (compare_(n->Key(), key) < 0);
}
template <class Comparator>
typename InlineSkipList<Comparator>::Node*
InlineSkipList<Comparator>::FindGreaterOrEqual(const char* key) const {
@@ -461,7 +446,6 @@ InlineSkipList<Comparator>::FindGreaterOrEqual(const char* key) const {
Node* x = head_;
int level = GetMaxHeight() - 1;
Node* last_bigger = nullptr;
const DecodedKey key_decoded = compare_.decode_key(key);
while (true) {
Node* next = x->Next(level);
if (next != nullptr) {
@@ -470,10 +454,10 @@ InlineSkipList<Comparator>::FindGreaterOrEqual(const char* key) const {
// Make sure the lists are sorted
assert(x == head_ || next == nullptr || KeyIsAfterNode(next->Key(), x));
// Make sure we haven't overshot during our search
assert(x == head_ || KeyIsAfterNode(key_decoded, x));
assert(x == head_ || KeyIsAfterNode(key, x));
int cmp = (next == nullptr || next == last_bigger)
? 1
: compare_(next->Key(), key_decoded);
: compare_(next->Key(), key);
if (cmp == 0 || (cmp > 0 && level == 0)) {
return next;
} else if (cmp < 0) {
@@ -503,7 +487,6 @@ InlineSkipList<Comparator>::FindLessThan(const char* key, Node** prev,
Node* x = root;
// KeyIsAfter(key, last_not_after) is definitely false
Node* last_not_after = nullptr;
const DecodedKey key_decoded = compare_.decode_key(key);
while (true) {
assert(x != nullptr);
Node* next = x->Next(level);
@@ -511,8 +494,8 @@ InlineSkipList<Comparator>::FindLessThan(const char* key, Node** prev,
PREFETCH(next->Next(level), 0, 1);
}
assert(x == head_ || next == nullptr || KeyIsAfterNode(next->Key(), x));
assert(x == head_ || KeyIsAfterNode(key_decoded, x));
if (next != last_not_after && KeyIsAfterNode(key_decoded, next)) {
assert(x == head_ || KeyIsAfterNode(key, x));
if (next != last_not_after && KeyIsAfterNode(key, next)) {
// Keep searching in this list
assert(next != nullptr);
x = next;
@@ -557,14 +540,13 @@ uint64_t InlineSkipList<Comparator>::EstimateCount(const char* key) const {
Node* x = head_;
int level = GetMaxHeight() - 1;
const DecodedKey key_decoded = compare_.decode_key(key);
while (true) {
assert(x == head_ || compare_(x->Key(), key_decoded) < 0);
assert(x == head_ || compare_(x->Key(), key) < 0);
Node* next = x->Next(level);
if (next != nullptr) {
PREFETCH(next->Next(level), 0, 1);
}
if (next == nullptr || compare_(next->Key(), key_decoded) >= 0) {
if (next == nullptr || compare_(next->Key(), key) >= 0) {
if (level == 0) {
return count;
} else {
@@ -672,7 +654,7 @@ bool InlineSkipList<Comparator>::InsertWithHint(const char* key, void** hint) {
template <class Comparator>
template <bool prefetch_before>
void InlineSkipList<Comparator>::FindSpliceForLevel(const DecodedKey& key,
void InlineSkipList<Comparator>::FindSpliceForLevel(const char* key,
Node* before, Node* after,
int level, Node** out_prev,
Node** out_next) {
@@ -700,7 +682,7 @@ void InlineSkipList<Comparator>::FindSpliceForLevel(const DecodedKey& key,
}
template <class Comparator>
void InlineSkipList<Comparator>::RecomputeSpliceLevels(const DecodedKey& key,
void InlineSkipList<Comparator>::RecomputeSpliceLevels(const char* key,
Splice* splice,
int recompute_level) {
assert(recompute_level > 0);
@@ -716,7 +698,6 @@ template <bool UseCAS>
bool InlineSkipList<Comparator>::Insert(const char* key, Splice* splice,
bool allow_partial_splice_fix) {
Node* x = reinterpret_cast<Node*>(const_cast<char*>(key)) - 1;
const DecodedKey key_decoded = compare_.decode_key(key);
int height = x->UnstashHeight();
assert(height >= 1 && height <= kMaxHeight_);
@@ -784,8 +765,7 @@ bool InlineSkipList<Comparator>::Insert(const char* key, Splice* splice,
// our chances of success.
++recompute_height;
} else if (splice->prev_[recompute_height] != head_ &&
!KeyIsAfterNode(key_decoded,
splice->prev_[recompute_height])) {
!KeyIsAfterNode(key, splice->prev_[recompute_height])) {
// key is from before splice
if (allow_partial_splice_fix) {
// skip all levels with the same node without more comparisons
@@ -797,8 +777,7 @@ bool InlineSkipList<Comparator>::Insert(const char* key, Splice* splice,
// we're pessimistic, recompute everything
recompute_height = max_height;
}
} else if (KeyIsAfterNode(key_decoded,
splice->next_[recompute_height])) {
} else if (KeyIsAfterNode(key, splice->next_[recompute_height])) {
// key is from after splice
if (allow_partial_splice_fix) {
Node* bad = splice->next_[recompute_height];
@@ -816,7 +795,7 @@ bool InlineSkipList<Comparator>::Insert(const char* key, Splice* splice,
}
assert(recompute_height <= max_height);
if (recompute_height > 0) {
RecomputeSpliceLevels(key_decoded, splice, recompute_height);
RecomputeSpliceLevels(key, splice, recompute_height);
}
bool splice_is_valid = true;
@@ -848,8 +827,8 @@ bool InlineSkipList<Comparator>::Insert(const char* key, Splice* splice,
// search, because it should be unlikely that lots of nodes have
// been inserted between prev[i] and next[i]. No point in using
// next[i] as the after hint, because we know it is stale.
FindSpliceForLevel<false>(key_decoded, splice->prev_[i], nullptr, i,
&splice->prev_[i], &splice->next_[i]);
FindSpliceForLevel<false>(key, splice->prev_[i], nullptr, i, &splice->prev_[i],
&splice->next_[i]);
// Since we've narrowed the bracket for level i, we might have
// violated the Splice constraint between i and i-1. Make sure
@@ -863,8 +842,8 @@ bool InlineSkipList<Comparator>::Insert(const char* key, Splice* splice,
for (int i = 0; i < height; ++i) {
if (i >= recompute_height &&
splice->prev_[i]->Next(i) != splice->next_[i]) {
FindSpliceForLevel<false>(key_decoded, splice->prev_[i], nullptr, i,
&splice->prev_[i], &splice->next_[i]);
FindSpliceForLevel<false>(key, splice->prev_[i], nullptr, i, &splice->prev_[i],
&splice->next_[i]);
}
// Checking for duplicate keys on the level 0 is sufficient
if (UNLIKELY(i == 0 && splice->next_[i] != nullptr &&
-16
View File
@@ -32,12 +32,6 @@ static Key Decode(const char* key) {
}
struct TestComparator {
typedef Key DecodedType;
static DecodedType decode_key(const char* b) {
return Decode(b);
}
int operator()(const char* a, const char* b) const {
if (Decode(a) < Decode(b)) {
return -1;
@@ -47,16 +41,6 @@ struct TestComparator {
return 0;
}
}
int operator()(const char* a, const DecodedType b) const {
if (Decode(a) < b) {
return -1;
} else if (Decode(a) > b) {
return +1;
} else {
return 0;
}
}
};
typedef InlineSkipList<TestComparator> TestInlineSkipList;
-4
View File
@@ -168,10 +168,6 @@ void ColumnFamilyOptions::Dump(Logger* log) const {
log,
" Options.compression_opts.max_dict_bytes: %" ROCKSDB_PRIszt,
compression_opts.max_dict_bytes);
ROCKS_LOG_HEADER(log,
" Options.compression_opts.zstd_max_train_bytes: "
"%" ROCKSDB_PRIszt,
compression_opts.zstd_max_train_bytes);
ROCKS_LOG_HEADER(log, " Options.level0_file_num_compaction_trigger: %d",
level0_file_num_compaction_trigger);
ROCKS_LOG_HEADER(log, " Options.level0_slowdown_writes_trigger: %d",
-11
View File
@@ -919,17 +919,6 @@ Status ParseColumnFamilyOption(const std::string& name,
}
new_options->compression_opts.max_dict_bytes =
ParseInt(value.substr(start, value.size() - start));
end = value.find(':', start);
}
// zstd_max_train_bytes is optional for backwards compatibility
if (end != std::string::npos) {
start = end + 1;
if (start >= value.size()) {
return Status::InvalidArgument(
"unable to parse the specified CF option " + name);
}
new_options->compression_opts.zstd_max_train_bytes =
ParseInt(value.substr(start, value.size() - start));
}
} else {
auto iter = cf_options_type_info.find(name);
+2 -4
View File
@@ -138,9 +138,7 @@ void WindowsThread::join() {
"WaitForSingleObjectFailed: thread join");
}
BOOL rc;
rc = CloseHandle(reinterpret_cast<HANDLE>(data_->handle_));
assert(rc != 0);
CloseHandle(reinterpret_cast<HANDLE>(data_->handle_));
data_->handle_ = 0;
}
@@ -156,7 +154,7 @@ bool WindowsThread::detach() {
BOOL ret = CloseHandle(reinterpret_cast<HANDLE>(data_->handle_));
data_->handle_ = 0;
return (ret != 0);
return (ret == TRUE);
}
void WindowsThread::swap(WindowsThread& o) {
-1
View File
@@ -148,7 +148,6 @@ LIB_SOURCES = \
util/status_message.cc \
util/string_util.cc \
util/sync_point.cc \
util/sync_point_impl.cc \
util/thread_local.cc \
util/threadpool_imp.cc \
util/transaction_test_util.cc \
+1 -1
View File
@@ -120,7 +120,7 @@ void ReleaseCachedEntry(void* arg, void* h) {
void ForceReleaseCachedEntry(void* arg, void* h) {
Cache* cache = reinterpret_cast<Cache*>(arg);
Cache::Handle* handle = reinterpret_cast<Cache::Handle*>(h);
cache->Release(handle, true);
cache->Release(handle, true /* force_erase */);
}
Slice GetCacheKeyFromOffset(const char* cache_key_prefix,
+1 -1
View File
@@ -419,7 +419,7 @@ struct BlockBasedTable::Rep {
const ImmutableCFOptions& ioptions;
const EnvOptions& env_options;
const BlockBasedTableOptions& table_options;
const BlockBasedTableOptions table_options;
const FilterPolicy* const filter_policy;
const InternalKeyComparator& internal_comparator;
Status status;
+3 -10
View File
@@ -30,7 +30,6 @@
namespace rocksdb {
inline
void BlockFetcher::CheckBlockChecksum() {
// Check the crc of the type and the block contents
if (read_options_.verify_checksums) {
@@ -63,7 +62,6 @@ void BlockFetcher::CheckBlockChecksum() {
}
}
inline
bool BlockFetcher::TryGetUncompressBlockFromPersistentCache() {
if (cache_options_.persistent_cache &&
!cache_options_.persistent_cache->IsCompressed()) {
@@ -85,7 +83,6 @@ bool BlockFetcher::TryGetUncompressBlockFromPersistentCache() {
return false;
}
inline
bool BlockFetcher::TryGetFromPrefetchBuffer() {
if (prefetch_buffer_ != nullptr &&
prefetch_buffer_->TryReadFromCache(
@@ -102,7 +99,6 @@ bool BlockFetcher::TryGetFromPrefetchBuffer() {
return got_from_prefetch_buffer_;
}
inline
bool BlockFetcher::TryGetCompressedBlockFromPersistentCache() {
if (cache_options_.persistent_cache &&
cache_options_.persistent_cache->IsCompressed()) {
@@ -123,7 +119,6 @@ bool BlockFetcher::TryGetCompressedBlockFromPersistentCache() {
return false;
}
inline
void BlockFetcher::PrepareBufferForBlockFromFile() {
// cache miss read from device
if (do_uncompress_ &&
@@ -132,12 +127,12 @@ void BlockFetcher::PrepareBufferForBlockFromFile() {
// trivially allocated stack buffer instead of needing a full malloc()
used_buf_ = &stack_buf_[0];
} else {
heap_buf_.reset(new char[block_size_ + kBlockTrailerSize]);
heap_buf_ =
std::unique_ptr<char[]>(new char[block_size_ + kBlockTrailerSize]);
used_buf_ = heap_buf_.get();
}
}
inline
void BlockFetcher::InsertCompressedBlockToPersistentCacheIfNeeded() {
if (status_.ok() && read_options_.fill_cache &&
cache_options_.persistent_cache &&
@@ -148,7 +143,6 @@ void BlockFetcher::InsertCompressedBlockToPersistentCacheIfNeeded() {
}
}
inline
void BlockFetcher::InsertUncompressedBlockToPersistentCacheIfNeeded() {
if (status_.ok() && !got_from_prefetch_buffer_ && read_options_.fill_cache &&
cache_options_.persistent_cache &&
@@ -159,7 +153,6 @@ void BlockFetcher::InsertUncompressedBlockToPersistentCacheIfNeeded() {
}
}
inline
void BlockFetcher::GetBlockContents() {
if (slice_.data() != used_buf_) {
// the slice content is not the buffer provided
@@ -168,7 +161,7 @@ void BlockFetcher::GetBlockContents() {
} else {
// page is uncompressed, the buffer either stack or heap provided
if (got_from_prefetch_buffer_ || used_buf_ == &stack_buf_[0]) {
heap_buf_.reset(new char[block_size_]);
heap_buf_ = std::unique_ptr<char[]>(new char[block_size_]);
memcpy(heap_buf_.get(), used_buf_, block_size_);
}
*contents_ = BlockContents(std::move(heap_buf_), block_size_, true,
+1 -1
View File
@@ -625,7 +625,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr, "SKIPPED as Cuckoo table is not supported in ROCKSDB_LITE\n");
return 0;
}
+1 -1
View File
@@ -560,7 +560,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr, "SKIPPED as Cuckoo table is not supported in ROCKSDB_LITE\n");
return 0;
}
+5 -17
View File
@@ -193,23 +193,11 @@ class MergingIterator : public InternalIterator {
InitMaxHeap();
for (auto& child : children_) {
if (&child != current_) {
if (!prefix_seek_mode_) {
child.Seek(key());
if (child.Valid()) {
// Child is at first entry >= key(). Step back one to be < key()
TEST_SYNC_POINT_CALLBACK("MergeIterator::Prev:BeforePrev",
&child);
child.Prev();
} else {
// Child has no entries >= key(). Position at last entry.
TEST_SYNC_POINT("MergeIterator::Prev:BeforeSeekToLast");
child.SeekToLast();
}
} else {
child.SeekForPrev(key());
if (child.Valid() && comparator_->Equal(key(), child.key())) {
child.Prev();
}
child.SeekForPrev(key());
// Child is at first entry >= key(). Step back one to be < key()
TEST_SYNC_POINT_CALLBACK("MergeIterator::Prev:BeforePrev", &child);
if (child.Valid() && comparator_->Equal(key(), child.key())) {
child.Prev();
}
}
if (child.Valid()) {
+9
View File
@@ -246,6 +246,13 @@ size_t PartitionedFilterBlockReader::ApproximateMemoryUsage() const {
return idx_on_fltr_blk_->size();
}
// Release the cached entry and decrement its ref count.
void ReleaseFilterCachedEntry(void* arg, void* h) {
Cache* cache = reinterpret_cast<Cache*>(arg);
Cache::Handle* handle = reinterpret_cast<Cache::Handle*>(h);
cache->Release(handle);
}
// TODO(myabandeh): merge this with the same function in IndexReader
void PartitionedFilterBlockReader::CacheDependencies(bool pin) {
// Before read partitions, prefetch them to avoid lots of IOs
@@ -304,6 +311,8 @@ void PartitionedFilterBlockReader::CacheDependencies(bool pin) {
if (LIKELY(filter.IsSet())) {
if (pin) {
filter_map_[handle.offset()] = std::move(filter);
RegisterCleanup(&ReleaseFilterCachedEntry, block_cache,
filter.cache_handle);
} else {
block_cache->Release(filter.cache_handle);
}
+2 -1
View File
@@ -65,7 +65,8 @@ class PartitionedFilterBlockBuilder : public FullFilterBlockBuilder {
size_t num_added_;
};
class PartitionedFilterBlockReader : public FilterBlockReader {
class PartitionedFilterBlockReader : public FilterBlockReader,
public Cleanable {
public:
explicit PartitionedFilterBlockReader(const SliceTransform* prefix_extractor,
bool whole_key_filtering,
+18 -11
View File
@@ -302,9 +302,11 @@ class KeyConvertingIterator : public InternalIterator {
class TableConstructor: public Constructor {
public:
explicit TableConstructor(const Comparator* cmp,
bool convert_to_internal_key = false)
bool convert_to_internal_key = false,
int level = -1)
: Constructor(cmp),
convert_to_internal_key_(convert_to_internal_key) {}
convert_to_internal_key_(convert_to_internal_key),
level_(level) {}
~TableConstructor() { Reset(); }
virtual Status FinishImpl(const Options& options,
@@ -319,14 +321,12 @@ class TableConstructor: public Constructor {
std::vector<std::unique_ptr<IntTblPropCollectorFactory>>
int_tbl_prop_collector_factories;
std::string column_family_name;
int unknown_level = -1;
builder.reset(ioptions.table_factory->NewTableBuilder(
TableBuilderOptions(ioptions, internal_comparator,
&int_tbl_prop_collector_factories,
options.compression, CompressionOptions(),
nullptr /* compression_dict */,
false /* skip_filters */, column_family_name,
unknown_level),
TableBuilderOptions(
ioptions, internal_comparator, &int_tbl_prop_collector_factories,
options.compression, CompressionOptions(),
nullptr /* compression_dict */, false /* skip_filters */,
column_family_name, level_),
TablePropertiesCollectorFactory::Context::kUnknownColumnFamily,
file_writer_.get()));
@@ -351,8 +351,10 @@ class TableConstructor: public Constructor {
uniq_id_ = cur_uniq_id_++;
file_reader_.reset(test::GetRandomAccessFileReader(new test::StringSource(
GetSink()->contents(), uniq_id_, ioptions.allow_mmap_reads)));
const bool skip_filters = false;
return ioptions.table_factory->NewTableReader(
TableReaderOptions(ioptions, soptions, internal_comparator),
TableReaderOptions(ioptions, soptions, internal_comparator,
skip_filters, level_),
std::move(file_reader_), GetSink()->contents().size(), &table_reader_);
}
@@ -412,6 +414,7 @@ class TableConstructor: public Constructor {
unique_ptr<RandomAccessFileReader> file_reader_;
unique_ptr<TableReader> table_reader_;
bool convert_to_internal_key_;
int level_;
TableConstructor();
@@ -2249,6 +2252,7 @@ std::map<std::string, size_t> MockCache::marked_data_in_cache_;
// table is closed. This test makes sure that the only items remains in the
// cache after the table is closed are raw data blocks.
TEST_F(BlockBasedTableTest, NoObjectInCacheAfterTableClose) {
for (int level: {-1, 0, 1, 10}) {
for (auto index_type :
{BlockBasedTableOptions::IndexType::kBinarySearch,
BlockBasedTableOptions::IndexType::kTwoLevelIndexSearch}) {
@@ -2285,7 +2289,9 @@ TEST_F(BlockBasedTableTest, NoObjectInCacheAfterTableClose) {
rocksdb::NewBloomFilterPolicy(10, block_based_filter));
opt.table_factory.reset(NewBlockBasedTableFactory(table_options));
TableConstructor c(BytewiseComparator());
bool convert_to_internal_key = false;
TableConstructor c(BytewiseComparator(), convert_to_internal_key,
level);
std::string user_key = "k01";
std::string key =
InternalKey(user_key, 0, kTypeValue).Encode().ToString();
@@ -2326,6 +2332,7 @@ TEST_F(BlockBasedTableTest, NoObjectInCacheAfterTableClose) {
}
}
}
} // level
}
TEST_F(BlockBasedTableTest, BlockCacheLeak) {
+5 -1
View File
@@ -24,7 +24,11 @@ class TwoLevelIterator : public InternalIterator {
explicit TwoLevelIterator(TwoLevelIteratorState* state,
InternalIterator* first_level_iter);
virtual ~TwoLevelIterator() { delete state_; }
virtual ~TwoLevelIterator() {
first_level_iter_.DeleteIter(false /* is_arena_mode */);
second_level_iter_.DeleteIter(false /* is_arena_mode */);
delete state_;
}
virtual void Seek(const Slice& target) override;
virtual void SeekForPrev(const Slice& target) override;
+2 -6
View File
@@ -16,8 +16,8 @@ namespace rocksdb {
struct ReadOptions;
class InternalKeyComparator;
class Arena;
// TwoLevelIteratorState expects iterators are not created using the arena
struct TwoLevelIteratorState {
TwoLevelIteratorState() {}
@@ -35,11 +35,7 @@ struct TwoLevelIteratorState {
//
// Uses a supplied function to convert an index_iter value into
// an iterator over the contents of the corresponding block.
// arena: If not null, the arena is used to allocate the Iterator.
// When destroying the iterator, the destructor will destroy
// all the states but those allocated in arena.
// need_free_iter_and_state: free `state` and `first_level_iter` if
// true. Otherwise, just call destructor.
// Note: this function expects first_level_iter was not created using the arena
extern InternalIterator* NewTwoLevelIterator(
TwoLevelIteratorState* state, InternalIterator* first_level_iter);
+10 -8
View File
@@ -108,12 +108,14 @@ fi
params_w="$const_params \
$l0_config \
--max_background_jobs=20 \
--max_write_buffer_number=8"
--max_background_compactions=16 \
--max_write_buffer_number=8 \
--max_background_flushes=7"
params_bulkload="$const_params \
--max_background_jobs=20 \
--max_background_compactions=16 \
--max_write_buffer_number=8 \
--max_background_flushes=7 \
--level0_file_num_compaction_trigger=$((10 * M)) \
--level0_slowdown_writes_trigger=$((10 * M)) \
--level0_stop_writes_trigger=$((10 * M))"
@@ -124,14 +126,14 @@ params_bulkload="$const_params \
# LSM. In level-based compaction, it means number of L0 files.
#
params_level_compact="$const_params \
--max_background_jobs=16 \
--max_background_flushes=4 \
--max_write_buffer_number=4 \
--level0_file_num_compaction_trigger=4 \
--level0_slowdown_writes_trigger=16 \
--level0_stop_writes_trigger=20"
params_univ_compact="$const_params \
--max_background_jobs=20 \
--max_background_flushes=4 \
--max_write_buffer_number=4 \
--level0_file_num_compaction_trigger=8 \
--level0_slowdown_writes_trigger=16 \
@@ -230,7 +232,7 @@ function run_manual_compaction_worker {
--subcompactions=$3 \
--memtablerep=vector \
--disable_wal=1 \
--max_background_jobs=$4 \
--max_background_compactions=$4 \
--seed=$( date +%s ) \
2>&1 | tee -a $fillrandom_output_file"
@@ -274,7 +276,7 @@ function run_univ_compaction {
# Define a set of benchmarks.
subcompactions=(1 2 4 8 16)
max_background_jobs=(20 20 10 5 4)
max_background_compactions=(16 16 8 4 2)
i=0
total=${#subcompactions[@]}
@@ -283,7 +285,7 @@ function run_univ_compaction {
while [ "$i" -lt "$total" ]
do
run_manual_compaction_worker $io_stats $compaction_style ${subcompactions[$i]} \
${max_background_jobs[$i]}
${max_background_compactions[$i]}
((i++))
done
}
+1 -1
View File
@@ -42,7 +42,7 @@ EOF
done
declare -a backward_compatible_checkout_objs=("2.2.fb.branch" "2.3.fb.branch" "2.4.fb.branch" "2.5.fb.branch" "2.6.fb.branch" "2.7.fb.branch" "2.8.1.fb" "3.0.fb.branch" "3.1.fb" "3.2.fb" "3.3.fb" "3.4.fb" "3.5.fb" "3.6.fb" "3.7.fb" "3.8.fb" "3.9.fb")
declare -a forward_compatible_checkout_objs=("3.10.fb" "3.11.fb" "3.12.fb" "3.13.fb" "4.0.fb" "4.1.fb" "4.2.fb" "4.3.fb" "4.4.fb" "4.5.fb" "4.6.fb" "4.7.fb" "4.8.fb" "4.9.fb" "4.10.fb" "4.11.fb" "4.12.fb" "4.13.fb" "5.0.fb" "5.1.fb" "5.2.fb" "5.3.fb" "5.4.fb" "5.5.fb" "5.6.fb" "5.7.fb" "5.8.fb" "5.9.fb" "5.10.fb" "5.11.fb" "5.12.fb")
declare -a forward_compatible_checkout_objs=("3.10.fb" "3.11.fb" "3.12.fb" "3.13.fb" "4.0.fb" "4.1.fb" "4.2.fb" "4.3.fb" "4.4.fb" "4.5.fb" "4.6.fb" "4.7.fb" "4.8.fb" "4.9.fb" "4.10.fb" "4.11.fb" "4.12.fb" "4.13.fb" "5.0.fb" "5.1.fb" "5.2.fb" "5.3.fb" "5.4.fb" "5.5.fb" "5.6.fb" "5.7.fb" "5.8.fb" "5.9.fb" "5.10.fb")
declare -a checkout_objs=(${backward_compatible_checkout_objs[@]} ${forward_compatible_checkout_objs[@]})
generate_db()
+11
View File
@@ -989,6 +989,8 @@ DEFINE_int32(memtable_insert_with_hint_prefix_size, 0,
"memtable insert with hint with the given prefix size.");
DEFINE_bool(enable_io_prio, false, "Lower the background flush/compaction "
"threads' IO priority");
DEFINE_bool(enable_cpu_prio, false, "Lower the background flush/compaction "
"threads' CPU priority");
DEFINE_bool(identity_as_first_hash, false, "the first hash function of cuckoo "
"table becomes an identity function. This is only valid when key "
"is 8 bytes");
@@ -2941,6 +2943,8 @@ void VerifyDBFromDB(std::string& truth_db_name) {
FLAGS_options_file.c_str(), s.ToString().c_str());
exit(1);
}
#else
(void)opts;
#endif
return false;
}
@@ -3315,6 +3319,10 @@ void VerifyDBFromDB(std::string& truth_db_name) {
FLAGS_env->LowerThreadPoolIOPriority(Env::LOW);
FLAGS_env->LowerThreadPoolIOPriority(Env::HIGH);
}
if (FLAGS_enable_cpu_prio) {
FLAGS_env->LowerThreadPoolCPUPriority(Env::LOW);
FLAGS_env->LowerThreadPoolCPUPriority(Env::HIGH);
}
options.env = FLAGS_env;
if (FLAGS_rate_limiter_bytes_per_sec > 0) {
@@ -3996,6 +4004,9 @@ void VerifyDBFromDB(std::string& truth_db_name) {
}
return Status::OK();
#else
(void)thread;
(void)compaction_style;
(void)write_mode;
fprintf(stderr, "Rocksdb Lite doesn't support filldeterministic\n");
return Status::NotSupported(
"Rocksdb Lite doesn't support filldeterministic");
+9 -6
View File
@@ -1021,9 +1021,7 @@ class DbStressListener : public EventListener {
: db_name_(db_name), db_paths_(db_paths) {}
virtual ~DbStressListener() {}
#ifndef ROCKSDB_LITE
virtual void OnFlushCompleted(DB* db, const FlushJobInfo& info) override {
assert(db);
assert(db->GetName() == db_name_);
virtual void OnFlushCompleted(DB* /*db*/, const FlushJobInfo& info) override {
assert(IsValidColumnFamilyName(info.cf_name));
VerifyFilePath(info.file_path);
// pretending doing some work here
@@ -1031,10 +1029,8 @@ class DbStressListener : public EventListener {
std::chrono::microseconds(Random::GetTLSInstance()->Uniform(5000)));
}
virtual void OnCompactionCompleted(DB* db,
virtual void OnCompactionCompleted(DB* /*db*/,
const CompactionJobInfo& ci) override {
assert(db);
assert(db->GetName() == db_name_);
assert(IsValidColumnFamilyName(ci.cf_name));
assert(ci.input_files.size() + ci.output_files.size() > 0U);
for (const auto& file_path : ci.input_files) {
@@ -1086,6 +1082,8 @@ class DbStressListener : public EventListener {
}
}
assert(false);
#else
(void)file_dir;
#endif // !NDEBUG
}
@@ -1096,6 +1094,8 @@ class DbStressListener : public EventListener {
bool result = ParseFileName(file_name, &file_number, &file_type);
assert(result);
assert(file_type == kTableFile);
#else
(void)file_name;
#endif // !NDEBUG
}
@@ -1110,6 +1110,8 @@ class DbStressListener : public EventListener {
}
VerifyFileName(file_path.substr(pos));
}
#else
(void)file_path;
#endif // !NDEBUG
}
#endif // !ROCKSDB_LITE
@@ -2289,6 +2291,7 @@ class StressTest {
size_t value_sz =
((rand % kRandomValueMaxFactor) + 1) * FLAGS_value_size_mult;
assert(value_sz <= max_sz && value_sz >= sizeof(uint32_t));
(void) max_sz;
*((uint32_t*)v) = rand;
for (size_t i=sizeof(uint32_t); i < value_sz; i++) {
v[i] = (char)(rand ^ i);
+1 -1
View File
@@ -54,7 +54,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr, "SKIPPED as LDBCommand is not supported in ROCKSDB_LITE\n");
return 0;
}
+1 -1
View File
@@ -210,7 +210,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr, "SKIPPED as LDBCommand is not supported in ROCKSDB_LITE\n");
return 0;
}
+1 -1
View File
@@ -215,7 +215,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr, "SKIPPED as SSTDumpTool is not supported in ROCKSDB_LITE\n");
return 0;
}
+1 -1
View File
@@ -520,7 +520,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr,
"SKIPPED as AutoRollLogger is not supported in ROCKSDB_LITE\n");
return 0;
+3
View File
@@ -27,6 +27,9 @@ template <class T>
void AssertAutoVectorOnlyInStack(autovector<T, kSize>* vec, bool result) {
#ifndef ROCKSDB_LITE
ASSERT_EQ(vec->only_in_stack(), result);
#else
(void) vec;
(void) result;
#endif // !ROCKSDB_LITE
}
} // namespace
+25 -5
View File
@@ -246,9 +246,15 @@ inline bool Zlib_Compress(const CompressionOptions& opts,
// memLevel=9 uses maximum memory for optimal speed.
// The default value is 8. See zconf.h for more details.
static const int memLevel = 8;
int level;
if (opts.level == CompressionOptions::kDefaultCompressionLevel) {
level = Z_DEFAULT_COMPRESSION;
} else {
level = opts.level;
}
z_stream _stream;
memset(&_stream, 0, sizeof(z_stream));
int st = deflateInit2(&_stream, opts.level, Z_DEFLATED, opts.window_bits,
int st = deflateInit2(&_stream, level, Z_DEFLATED, opts.window_bits,
memLevel, opts.strategy);
if (st != Z_OK) {
return false;
@@ -682,9 +688,15 @@ inline bool LZ4HC_Compress(const CompressionOptions& opts,
output->resize(static_cast<size_t>(output_header_len + compress_bound));
int outlen;
int level;
if (opts.level == CompressionOptions::kDefaultCompressionLevel) {
level = 0; // lz4hc.h says any value < 1 will be sanitized to default
} else {
level = opts.level;
}
#if LZ4_VERSION_NUMBER >= 10400 // r124+
LZ4_streamHC_t* stream = LZ4_createStreamHC();
LZ4_resetStreamHC(stream, opts.level);
LZ4_resetStreamHC(stream, level);
const char* compression_dict_data =
compression_dict.size() > 0 ? compression_dict.data() : nullptr;
size_t compression_dict_size = compression_dict.size();
@@ -705,7 +717,7 @@ inline bool LZ4HC_Compress(const CompressionOptions& opts,
#elif LZ4_VERSION_MAJOR // r113-r123
outlen = LZ4_compressHC2_limitedOutput(input, &(*output)[output_header_len],
static_cast<int>(length),
compress_bound, opts.level);
compress_bound, level);
#else // up to r112
outlen =
LZ4_compressHC_limitedOutput(input, &(*output)[output_header_len],
@@ -766,15 +778,23 @@ inline bool ZSTD_Compress(const CompressionOptions& opts, const char* input,
size_t compressBound = ZSTD_compressBound(length);
output->resize(static_cast<size_t>(output_header_len + compressBound));
size_t outlen;
int level;
if (opts.level == CompressionOptions::kDefaultCompressionLevel) {
// 3 is the value of ZSTD_CLEVEL_DEFAULT (not exposed publicly), see
// https://github.com/facebook/zstd/issues/1148
level = 3;
} else {
level = opts.level;
}
#if ZSTD_VERSION_NUMBER >= 500 // v0.5.0+
ZSTD_CCtx* context = ZSTD_createCCtx();
outlen = ZSTD_compress_usingDict(
context, &(*output)[output_header_len], compressBound, input, length,
compression_dict.data(), compression_dict.size(), opts.level);
compression_dict.data(), compression_dict.size(), level);
ZSTD_freeCCtx(context);
#else // up to v0.4.x
outlen = ZSTD_compress(&(*output)[output_header_len], compressBound, input,
length, opts.level);
length, level);
#endif // ZSTD_VERSION_NUMBER >= 500
if (outlen == 0) {
return false;
+11 -44
View File
@@ -22,13 +22,11 @@ namespace rocksdb {
DeleteScheduler::DeleteScheduler(Env* env, int64_t rate_bytes_per_sec,
Logger* info_log,
SstFileManagerImpl* sst_file_manager,
double max_trash_db_ratio,
uint64_t bytes_max_delete_chunk)
double max_trash_db_ratio)
: env_(env),
total_trash_size_(0),
rate_bytes_per_sec_(rate_bytes_per_sec),
pending_files_(0),
bytes_max_delete_chunk_(bytes_max_delete_chunk),
closing_(false),
cv_(&mu_),
info_log_(info_log),
@@ -210,18 +208,15 @@ void DeleteScheduler::BackgroundEmptyTrash() {
// Get new file to delete
std::string path_in_trash = queue_.front();
queue_.pop();
// We dont need to hold the lock while deleting the file
mu_.Unlock();
uint64_t deleted_bytes = 0;
bool is_complete = true;
// Delete file from trash and update total_penlty value
Status s = DeleteTrashFile(path_in_trash, &deleted_bytes, &is_complete);
Status s = DeleteTrashFile(path_in_trash, &deleted_bytes);
total_deleted_bytes += deleted_bytes;
mu_.Lock();
if (is_complete) {
queue_.pop();
}
if (!s.ok()) {
bg_errors_[path_in_trash] = s;
@@ -241,9 +236,7 @@ void DeleteScheduler::BackgroundEmptyTrash() {
TEST_SYNC_POINT_CALLBACK("DeleteScheduler::BackgroundEmptyTrash:Wait",
&total_penlty);
if (is_complete) {
pending_files_--;
}
pending_files_--;
if (pending_files_ == 0) {
// Unblock WaitForEmptyTrash since there are no more files waiting
// to be deleted
@@ -254,49 +247,23 @@ void DeleteScheduler::BackgroundEmptyTrash() {
}
Status DeleteScheduler::DeleteTrashFile(const std::string& path_in_trash,
uint64_t* deleted_bytes,
bool* is_complete) {
uint64_t* deleted_bytes) {
uint64_t file_size;
Status s = env_->GetFileSize(path_in_trash, &file_size);
*is_complete = true;
TEST_SYNC_POINT("DeleteScheduler::DeleteTrashFile:DeleteFile");
if (s.ok()) {
bool need_full_delete = true;
if (bytes_max_delete_chunk_ != 0 && file_size > bytes_max_delete_chunk_) {
unique_ptr<WritableFile> wf;
Status my_status =
env_->ReopenWritableFile(path_in_trash, &wf, EnvOptions());
if (my_status.ok()) {
my_status = wf->Truncate(file_size - bytes_max_delete_chunk_);
if (my_status.ok()) {
TEST_SYNC_POINT("DeleteScheduler::DeleteTrashFile:Fsync");
my_status = wf->Fsync();
}
}
if (my_status.ok()) {
*deleted_bytes = bytes_max_delete_chunk_;
need_full_delete = false;
*is_complete = false;
} else {
ROCKS_LOG_WARN(info_log_,
"Failed to partially delete %s from trash -- %s",
path_in_trash.c_str(), my_status.ToString().c_str());
}
}
if (need_full_delete) {
s = env_->DeleteFile(path_in_trash);
*deleted_bytes = file_size;
sst_file_manager_->OnDeleteFile(path_in_trash);
}
TEST_SYNC_POINT("DeleteScheduler::DeleteTrashFile:DeleteFile");
s = env_->DeleteFile(path_in_trash);
}
if (!s.ok()) {
// Error while getting file size or while deleting
ROCKS_LOG_ERROR(info_log_, "Failed to delete %s from trash -- %s",
path_in_trash.c_str(), s.ToString().c_str());
*deleted_bytes = 0;
} else {
total_trash_size_.fetch_sub(*deleted_bytes);
*deleted_bytes = file_size;
total_trash_size_.fetch_sub(file_size);
sst_file_manager_->OnDeleteFile(path_in_trash);
}
return s;
+2 -3
View File
@@ -34,7 +34,7 @@ class DeleteScheduler {
public:
DeleteScheduler(Env* env, int64_t rate_bytes_per_sec, Logger* info_log,
SstFileManagerImpl* sst_file_manager,
double max_trash_db_ratio, uint64_t bytes_max_delete_chunk);
double max_trash_db_ratio);
~DeleteScheduler();
@@ -82,7 +82,7 @@ class DeleteScheduler {
Status MarkAsTrash(const std::string& file_path, std::string* path_in_trash);
Status DeleteTrashFile(const std::string& path_in_trash,
uint64_t* deleted_bytes, bool* is_complete);
uint64_t* deleted_bytes);
void BackgroundEmptyTrash();
@@ -97,7 +97,6 @@ class DeleteScheduler {
std::queue<std::string> queue_;
// Number of trash files that are waiting to be deleted
int32_t pending_files_;
uint64_t bytes_max_delete_chunk_;
// Errors that happened in BackgroundEmptyTrash (file_path => error)
std::map<std::string, Status> bg_errors_;
// Set to true in ~DeleteScheduler() to force BackgroundEmptyTrash to stop
+2 -30
View File
@@ -99,7 +99,7 @@ class DeleteSchedulerTest : public testing::Test {
// 25%)
sst_file_mgr_.reset(
new SstFileManagerImpl(env_, nullptr, rate_bytes_per_sec_,
/* max_trash_db_ratio= */ 1.1, 128 * 1024));
/* max_trash_db_ratio= */ 1.1));
delete_scheduler_ = sst_file_mgr_->delete_scheduler();
}
@@ -436,34 +436,6 @@ TEST_F(DeleteSchedulerTest, StartBGEmptyTrashMultipleTimes) {
rocksdb::SyncPoint::GetInstance()->EnableProcessing();
}
TEST_F(DeleteSchedulerTest, DeletePartialFile) {
int bg_delete_file = 0;
int bg_fsync = 0;
rocksdb::SyncPoint::GetInstance()->SetCallBack(
"DeleteScheduler::DeleteTrashFile:DeleteFile",
[&](void*) { bg_delete_file++; });
rocksdb::SyncPoint::GetInstance()->SetCallBack(
"DeleteScheduler::DeleteTrashFile:Fsync", [&](void*) { bg_fsync++; });
rocksdb::SyncPoint::GetInstance()->EnableProcessing();
rate_bytes_per_sec_ = 1024 * 1024; // 1 MB / sec
NewDeleteScheduler();
// Should delete in 4 batch
ASSERT_OK(delete_scheduler_->DeleteFile(NewDummyFile("data_1", 500 * 1024)));
ASSERT_OK(delete_scheduler_->DeleteFile(NewDummyFile("data_2", 100 * 1024)));
// Should delete in 2 batch
ASSERT_OK(delete_scheduler_->DeleteFile(NewDummyFile("data_2", 200 * 1024)));
delete_scheduler_->WaitForEmptyTrash();
auto bg_errors = delete_scheduler_->GetBackgroundErrors();
ASSERT_EQ(bg_errors.size(), 0);
ASSERT_EQ(7, bg_delete_file);
ASSERT_EQ(4, bg_fsync);
rocksdb::SyncPoint::GetInstance()->EnableProcessing();
}
// 1- Create a DeleteScheduler with very slow rate limit (1 Byte / sec)
// 2- Delete 100 files using DeleteScheduler
// 3- Delete the DeleteScheduler (call the destructor while queue is not empty)
@@ -641,7 +613,7 @@ int main(int argc, char** argv) {
}
#else
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
printf("DeleteScheduler is not supported in ROCKSDB_LITE\n");
return 0;
}
+5 -8
View File
@@ -18,8 +18,7 @@ namespace rocksdb {
#ifndef ROCKSDB_LITE
SstFileManagerImpl::SstFileManagerImpl(Env* env, std::shared_ptr<Logger> logger,
int64_t rate_bytes_per_sec,
double max_trash_db_ratio,
uint64_t bytes_max_delete_chunk)
double max_trash_db_ratio)
: env_(env),
logger_(logger),
total_files_size_(0),
@@ -27,7 +26,7 @@ SstFileManagerImpl::SstFileManagerImpl(Env* env, std::shared_ptr<Logger> logger,
cur_compactions_reserved_size_(0),
max_allowed_space_(0),
delete_scheduler_(env, rate_bytes_per_sec, logger.get(), this,
max_trash_db_ratio, bytes_max_delete_chunk) {}
max_trash_db_ratio) {}
SstFileManagerImpl::~SstFileManagerImpl() {}
@@ -197,11 +196,10 @@ SstFileManager* NewSstFileManager(Env* env, std::shared_ptr<Logger> info_log,
std::string trash_dir,
int64_t rate_bytes_per_sec,
bool delete_existing_trash, Status* status,
double max_trash_db_ratio,
uint64_t bytes_max_delete_chunk) {
double max_trash_db_ratio) {
SstFileManagerImpl* res =
new SstFileManagerImpl(env, info_log, rate_bytes_per_sec,
max_trash_db_ratio, bytes_max_delete_chunk);
max_trash_db_ratio);
// trash_dir is deprecated and not needed anymore, but if user passed it
// we will still remove files in it.
@@ -238,8 +236,7 @@ SstFileManager* NewSstFileManager(Env* env, std::shared_ptr<Logger> info_log,
std::string trash_dir,
int64_t rate_bytes_per_sec,
bool delete_existing_trash, Status* status,
double max_trash_db_ratio,
uint64_t bytes_max_delete_chunk) {
double max_trash_db_ratio) {
if (status) {
*status =
Status::NotSupported("SstFileManager is not supported in ROCKSDB_LITE");
+1 -2
View File
@@ -27,8 +27,7 @@ class SstFileManagerImpl : public SstFileManager {
public:
explicit SstFileManagerImpl(Env* env, std::shared_ptr<Logger> logger,
int64_t rate_bytes_per_sec,
double max_trash_db_ratio,
uint64_t bytes_max_delete_chunk);
double max_trash_db_ratio);
~SstFileManagerImpl();
+126 -24
View File
@@ -4,7 +4,10 @@
// (found in the LICENSE.Apache file in the root directory).
#include "util/sync_point.h"
#include "util/sync_point_impl.h"
#include <functional>
#include <thread>
#include "port/port.h"
#include "util/random.h"
int rocksdb_kill_odds = 0;
std::vector<std::string> rocksdb_kill_prefix_blacklist;
@@ -12,57 +15,156 @@ std::vector<std::string> rocksdb_kill_prefix_blacklist;
#ifndef NDEBUG
namespace rocksdb {
void TestKillRandom(std::string kill_point, int odds,
const std::string& srcfile, int srcline) {
for (auto& p : rocksdb_kill_prefix_blacklist) {
if (kill_point.substr(0, p.length()) == p) {
return;
}
}
assert(odds > 0);
if (odds % 7 == 0) {
// class Random uses multiplier 16807, which is 7^5. If odds are
// multiplier of 7, there might be limited values generated.
odds++;
}
auto* r = Random::GetTLSInstance();
bool crash = r->OneIn(odds);
if (crash) {
port::Crash(srcfile, srcline);
}
}
SyncPoint* SyncPoint::GetInstance() {
static SyncPoint sync_point;
return &sync_point;
}
SyncPoint::SyncPoint() :
impl_(new Data) {
}
SyncPoint:: ~SyncPoint() {
delete impl_;
}
void SyncPoint::LoadDependency(const std::vector<SyncPointPair>& dependencies) {
impl_->LoadDependency(dependencies);
std::unique_lock<std::mutex> lock(mutex_);
successors_.clear();
predecessors_.clear();
cleared_points_.clear();
for (const auto& dependency : dependencies) {
successors_[dependency.predecessor].push_back(dependency.successor);
predecessors_[dependency.successor].push_back(dependency.predecessor);
}
cv_.notify_all();
}
void SyncPoint::LoadDependencyAndMarkers(
const std::vector<SyncPointPair>& dependencies,
const std::vector<SyncPointPair>& markers) {
impl_->LoadDependencyAndMarkers(dependencies, markers);
const std::vector<SyncPointPair>& dependencies,
const std::vector<SyncPointPair>& markers) {
std::unique_lock<std::mutex> lock(mutex_);
successors_.clear();
predecessors_.clear();
cleared_points_.clear();
markers_.clear();
marked_thread_id_.clear();
for (const auto& dependency : dependencies) {
successors_[dependency.predecessor].push_back(dependency.successor);
predecessors_[dependency.successor].push_back(dependency.predecessor);
}
for (const auto& marker : markers) {
successors_[marker.predecessor].push_back(marker.successor);
predecessors_[marker.successor].push_back(marker.predecessor);
markers_[marker.predecessor].push_back(marker.successor);
}
cv_.notify_all();
}
void SyncPoint::SetCallBack(const std::string& point,
const std::function<void(void*)>& callback) {
impl_->SetCallBack(point, callback);
bool SyncPoint::PredecessorsAllCleared(const std::string& point) {
for (const auto& pred : predecessors_[point]) {
if (cleared_points_.count(pred) == 0) {
return false;
}
}
return true;
}
void SyncPoint::ClearCallBack(const std::string& point) {
impl_->ClearCallBack(point);
void SyncPoint::SetCallBack(const std::string point,
std::function<void(void*)> callback) {
std::unique_lock<std::mutex> lock(mutex_);
callbacks_[point] = callback;
}
void SyncPoint::ClearCallBack(const std::string point) {
std::unique_lock<std::mutex> lock(mutex_);
while (num_callbacks_running_ > 0) {
cv_.wait(lock);
}
callbacks_.erase(point);
}
void SyncPoint::ClearAllCallBacks() {
impl_->ClearAllCallBacks();
std::unique_lock<std::mutex> lock(mutex_);
while (num_callbacks_running_ > 0) {
cv_.wait(lock);
}
callbacks_.clear();
}
void SyncPoint::EnableProcessing() {
impl_->EnableProcessing();
std::unique_lock<std::mutex> lock(mutex_);
enabled_ = true;
}
void SyncPoint::DisableProcessing() {
impl_->DisableProcessing();
std::unique_lock<std::mutex> lock(mutex_);
enabled_ = false;
}
void SyncPoint::ClearTrace() {
impl_->ClearTrace();
std::unique_lock<std::mutex> lock(mutex_);
cleared_points_.clear();
}
bool SyncPoint::DisabledByMarker(const std::string& point,
std::thread::id thread_id) {
auto marked_point_iter = marked_thread_id_.find(point);
return marked_point_iter != marked_thread_id_.end() &&
thread_id != marked_point_iter->second;
}
void SyncPoint::Process(const std::string& point, void* cb_arg) {
impl_->Process(point, cb_arg);
}
std::unique_lock<std::mutex> lock(mutex_);
if (!enabled_) {
return;
}
auto thread_id = std::this_thread::get_id();
auto marker_iter = markers_.find(point);
if (marker_iter != markers_.end()) {
for (auto marked_point : marker_iter->second) {
marked_thread_id_.insert(std::make_pair(marked_point, thread_id));
}
}
if (DisabledByMarker(point, thread_id)) {
return;
}
while (!PredecessorsAllCleared(point)) {
cv_.wait(lock);
if (DisabledByMarker(point, thread_id)) {
return;
}
}
auto callback_pair = callbacks_.find(point);
if (callback_pair != callbacks_.end()) {
num_callbacks_running_++;
mutex_.unlock();
callback_pair->second(cb_arg);
mutex_.lock();
num_callbacks_running_--;
cv_.notify_all();
}
cleared_points_.insert(point);
cv_.notify_all();
}
} // namespace rocksdb
#endif // NDEBUG
+24 -15
View File
@@ -5,10 +5,13 @@
#pragma once
#include <assert.h>
#include <condition_variable>
#include <functional>
#include <mutex>
#include <string>
#include <thread>
#include <unordered_map>
#include <unordered_set>
#include <vector>
// This is only set from db_stress.cc and for testing only.
@@ -62,10 +65,6 @@ class SyncPoint {
public:
static SyncPoint* GetInstance();
SyncPoint(const SyncPoint&) = delete;
SyncPoint& operator=(const SyncPoint&) = delete;
~SyncPoint();
struct SyncPointPair {
std::string predecessor;
std::string successor;
@@ -82,14 +81,15 @@ class SyncPoint {
void LoadDependencyAndMarkers(const std::vector<SyncPointPair>& dependencies,
const std::vector<SyncPointPair>& markers);
// Set up a call back function in sync point.
// The argument to the callback is passed through from
// TEST_SYNC_POINT_CALLBACK(); nullptr if TEST_SYNC_POINT or
// TEST_IDX_SYNC_POINT was used.
void SetCallBack(const std::string& point,
const std::function<void(void*)>& callback);
void SetCallBack(const std::string point,
std::function<void(void*)> callback);
// Clear callback function by point
void ClearCallBack(const std::string& point);
void ClearCallBack(const std::string point);
// Clear all call back functions.
void ClearAllCallBacks();
@@ -111,20 +111,29 @@ class SyncPoint {
// TODO: it might be useful to provide a function that blocks until all
// sync points are cleared.
// We want this to be public so we can
// subclass the implementation
struct Data;
private:
// Singleton
SyncPoint();
Data* impl_;
bool PredecessorsAllCleared(const std::string& point);
bool DisabledByMarker(const std::string& point, std::thread::id thread_id);
// successor/predecessor map loaded from LoadDependency
std::unordered_map<std::string, std::vector<std::string>> successors_;
std::unordered_map<std::string, std::vector<std::string>> predecessors_;
std::unordered_map<std::string, std::function<void(void*)> > callbacks_;
std::unordered_map<std::string, std::vector<std::string> > markers_;
std::unordered_map<std::string, std::thread::id> marked_thread_id_;
std::mutex mutex_;
std::condition_variable cv_;
// sync points that have been passed through
std::unordered_set<std::string> cleared_points_;
bool enabled_ = false;
int num_callbacks_running_ = 0;
};
} // namespace rocksdb
// Use TEST_SYNC_POINT to specify sync points inside code base.
// Sync points can have happens-after depedency on other sync points,
// Sync points can have happens-after dependency on other sync points,
// configured at runtime via SyncPoint::LoadDependency. This could be
// utilized to re-produce race conditions between threads.
// See TransactionLogIteratorRace in db_test.cc for an example use case.
-129
View File
@@ -1,129 +0,0 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
#include "util/sync_point_impl.h"
#ifndef NDEBUG
namespace rocksdb {
void TestKillRandom(std::string kill_point, int odds,
const std::string& srcfile, int srcline) {
for (auto& p : rocksdb_kill_prefix_blacklist) {
if (kill_point.substr(0, p.length()) == p) {
return;
}
}
assert(odds > 0);
if (odds % 7 == 0) {
// class Random uses multiplier 16807, which is 7^5. If odds are
// multiplier of 7, there might be limited values generated.
odds++;
}
auto* r = Random::GetTLSInstance();
bool crash = r->OneIn(odds);
if (crash) {
port::Crash(srcfile, srcline);
}
}
void SyncPoint::Data::LoadDependency(const std::vector<SyncPointPair>& dependencies) {
std::lock_guard<std::mutex> lock(mutex_);
successors_.clear();
predecessors_.clear();
cleared_points_.clear();
for (const auto& dependency : dependencies) {
successors_[dependency.predecessor].push_back(dependency.successor);
predecessors_[dependency.successor].push_back(dependency.predecessor);
}
cv_.notify_all();
}
void SyncPoint::Data::LoadDependencyAndMarkers(
const std::vector<SyncPointPair>& dependencies,
const std::vector<SyncPointPair>& markers) {
std::lock_guard<std::mutex> lock(mutex_);
successors_.clear();
predecessors_.clear();
cleared_points_.clear();
markers_.clear();
marked_thread_id_.clear();
for (const auto& dependency : dependencies) {
successors_[dependency.predecessor].push_back(dependency.successor);
predecessors_[dependency.successor].push_back(dependency.predecessor);
}
for (const auto& marker : markers) {
successors_[marker.predecessor].push_back(marker.successor);
predecessors_[marker.successor].push_back(marker.predecessor);
markers_[marker.predecessor].push_back(marker.successor);
}
cv_.notify_all();
}
bool SyncPoint::Data::PredecessorsAllCleared(const std::string& point) {
for (const auto& pred : predecessors_[point]) {
if (cleared_points_.count(pred) == 0) {
return false;
}
}
return true;
}
void SyncPoint::Data::ClearCallBack(const std::string& point) {
std::unique_lock<std::mutex> lock(mutex_);
while (num_callbacks_running_ > 0) {
cv_.wait(lock);
}
callbacks_.erase(point);
}
void SyncPoint::Data::ClearAllCallBacks() {
std::unique_lock<std::mutex> lock(mutex_);
while (num_callbacks_running_ > 0) {
cv_.wait(lock);
}
callbacks_.clear();
}
void SyncPoint::Data::Process(const std::string& point, void* cb_arg) {
std::unique_lock<std::mutex> lock(mutex_);
if (!enabled_) {
return;
}
auto thread_id = std::this_thread::get_id();
auto marker_iter = markers_.find(point);
if (marker_iter != markers_.end()) {
for (auto& marked_point : marker_iter->second) {
marked_thread_id_.emplace(marked_point, thread_id);
}
}
if (DisabledByMarker(point, thread_id)) {
return;
}
while (!PredecessorsAllCleared(point)) {
cv_.wait(lock);
if (DisabledByMarker(point, thread_id)) {
return;
}
}
auto callback_pair = callbacks_.find(point);
if (callback_pair != callbacks_.end()) {
num_callbacks_running_++;
mutex_.unlock();
callback_pair->second(cb_arg);
mutex_.lock();
num_callbacks_running_--;
}
cleared_points_.insert(point);
cv_.notify_all();
}
} // rocksdb
#endif
-74
View File
@@ -1,74 +0,0 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
#include "util/sync_point.h"
#include <assert.h>
#include <condition_variable>
#include <functional>
#include <mutex>
#include <string>
#include <thread>
#include <unordered_map>
#include <unordered_set>
#include "port/port.h"
#include "util/random.h"
#pragma once
#ifndef NDEBUG
namespace rocksdb {
struct SyncPoint::Data {
// Enable proper deletion by subclasses
virtual ~Data() {}
// successor/predecessor map loaded from LoadDependency
std::unordered_map<std::string, std::vector<std::string>> successors_;
std::unordered_map<std::string, std::vector<std::string>> predecessors_;
std::unordered_map<std::string, std::function<void(void*)> > callbacks_;
std::unordered_map<std::string, std::vector<std::string> > markers_;
std::unordered_map<std::string, std::thread::id> marked_thread_id_;
std::mutex mutex_;
std::condition_variable cv_;
// sync points that have been passed through
std::unordered_set<std::string> cleared_points_;
bool enabled_ = false;
int num_callbacks_running_ = 0;
void LoadDependency(const std::vector<SyncPointPair>& dependencies);
void LoadDependencyAndMarkers(const std::vector<SyncPointPair>& dependencies,
const std::vector<SyncPointPair>& markers);
bool PredecessorsAllCleared(const std::string& point);
void SetCallBack(const std::string& point,
const std::function<void(void*)>& callback) {
std::lock_guard<std::mutex> lock(mutex_);
callbacks_[point] = callback;
}
void ClearCallBack(const std::string& point);
void ClearAllCallBacks();
void EnableProcessing() {
std::lock_guard<std::mutex> lock(mutex_);
enabled_ = true;
}
void DisableProcessing() {
std::lock_guard<std::mutex> lock(mutex_);
enabled_ = false;
}
void ClearTrace() {
std::lock_guard<std::mutex> lock(mutex_);
cleared_points_.clear();
}
bool DisabledByMarker(const std::string& point,
std::thread::id thread_id) {
auto marked_point_iter = marked_thread_id_.find(point);
return marked_point_iter != marked_thread_id_.end() &&
thread_id != marked_point_iter->second;
}
void Process(const std::string& point, void* cb_arg);
};
}
#endif // NDEBUG
+28
View File
@@ -18,6 +18,7 @@
#ifdef OS_LINUX
# include <sys/syscall.h>
# include <sys/resource.h>
#endif
#include <algorithm>
@@ -53,6 +54,8 @@ struct ThreadPoolImpl::Impl {
void LowerIOPriority();
void LowerCPUPriority();
void WakeUpAllThreads() {
bgsignal_.notify_all();
}
@@ -97,6 +100,7 @@ private:
static void* BGThreadWrapper(void* arg);
bool low_io_priority_;
bool low_cpu_priority_;
Env::Priority priority_;
Env* env_;
@@ -125,6 +129,7 @@ inline
ThreadPoolImpl::Impl::Impl()
:
low_io_priority_(false),
low_cpu_priority_(false),
priority_(Env::LOW),
env_(nullptr),
total_threads_limit_(0),
@@ -171,9 +176,16 @@ void ThreadPoolImpl::Impl::LowerIOPriority() {
low_io_priority_ = true;
}
inline
void ThreadPoolImpl::Impl::LowerCPUPriority() {
std::lock_guard<std::mutex> lock(mu_);
low_cpu_priority_ = true;
}
void ThreadPoolImpl::Impl::BGThread(size_t thread_id) {
bool low_io_priority = false;
bool low_cpu_priority = false;
while (true) {
// Wait until there is an item that is ready to run
std::unique_lock<std::mutex> lock(mu_);
@@ -213,9 +225,20 @@ void ThreadPoolImpl::Impl::BGThread(size_t thread_id) {
std::memory_order_relaxed);
bool decrease_io_priority = (low_io_priority != low_io_priority_);
bool decrease_cpu_priority = (low_cpu_priority != low_cpu_priority_);
lock.unlock();
#ifdef OS_LINUX
if (decrease_cpu_priority) {
setpriority(
PRIO_PROCESS,
// Current thread.
0,
// Lowest priority possible.
19);
low_cpu_priority = true;
}
if (decrease_io_priority) {
#define IOPRIO_CLASS_SHIFT (13)
#define IOPRIO_PRIO_VALUE(class, data) (((class) << IOPRIO_CLASS_SHIFT) | data)
@@ -236,6 +259,7 @@ void ThreadPoolImpl::Impl::BGThread(size_t thread_id) {
}
#else
(void)decrease_io_priority; // avoid 'unused variable' error
(void)decrease_cpu_priority;
#endif
func();
}
@@ -421,6 +445,10 @@ void ThreadPoolImpl::LowerIOPriority() {
impl_->LowerIOPriority();
}
void ThreadPoolImpl::LowerCPUPriority() {
impl_->LowerCPUPriority();
}
void ThreadPoolImpl::IncBackgroundThreadsIfNeeded(int num) {
impl_->SetBackgroundThreadsInternal(num, false);
}
+5 -1
View File
@@ -46,10 +46,14 @@ class ThreadPoolImpl : public ThreadPool {
// start yet
void WaitForJobsAndJoinAllThreads() override;
// Make threads to run at a lower kernel priority
// Make threads to run at a lower kernel IO priority
// Currently only has effect on Linux
void LowerIOPriority();
// Make threads to run at a lower kernel CPU priority
// Currently only has effect on Linux
void LowerCPUPriority();
// Ensure there is at aleast num threads in the pool
// but do not kill threads if there are more
void IncBackgroundThreadsIfNeeded(int num);
+51 -75
View File
@@ -747,6 +747,19 @@ Status BackupEngineImpl::CreateNewBackupWithMetadata(
BackupID new_backup_id = latest_backup_id_ + 1;
assert(backups_.find(new_backup_id) == backups_.end());
auto private_dir = GetAbsolutePath(GetPrivateFileRel(new_backup_id));
Status s = backup_env_->FileExists(private_dir);
if (s.ok()) {
// maybe last backup failed and left partial state behind, clean it up.
// need to do this before updating backups_ such that a private dir
// named after new_backup_id will be cleaned up
s = GarbageCollect();
} else if (s.IsNotFound()) {
// normal case, the new backup's private dir doesn't exist yet
s = Status::OK();
}
auto ret = backups_.insert(std::make_pair(
new_backup_id, unique_ptr<BackupMeta>(new BackupMeta(
GetBackupMetaFile(new_backup_id, false /* tmp */),
@@ -757,23 +770,13 @@ Status BackupEngineImpl::CreateNewBackupWithMetadata(
new_backup->RecordTimestamp();
new_backup->SetAppMetadata(app_metadata);
auto start_backup = backup_env_-> NowMicros();
auto start_backup = backup_env_->NowMicros();
ROCKS_LOG_INFO(options_.info_log,
"Started the backup process -- creating backup %u",
new_backup_id);
auto private_tmp_dir = GetAbsolutePath(GetPrivateFileRel(new_backup_id, true));
Status s = backup_env_->FileExists(private_tmp_dir);
if (s.ok()) {
// maybe last backup failed and left partial state behind, clean it up
s = GarbageCollect();
} else if (s.IsNotFound()) {
// normal case, the new backup's private dir doesn't exist yet
s = Status::OK();
}
if (s.ok()) {
s = backup_env_->CreateDir(private_tmp_dir);
s = backup_env_->CreateDir(private_dir);
}
RateLimiter* rate_limiter = options_.backup_rate_limiter.get();
@@ -859,18 +862,6 @@ Status BackupEngineImpl::CreateNewBackupWithMetadata(
// we copied all the files, enable file deletions
db->EnableFileDeletions(false);
if (s.ok()) {
// move tmp private backup to real backup folder
ROCKS_LOG_INFO(
options_.info_log,
"Moving tmp backup directory to the real one: %s -> %s\n",
GetAbsolutePath(GetPrivateFileRel(new_backup_id, true)).c_str(),
GetAbsolutePath(GetPrivateFileRel(new_backup_id, false)).c_str());
s = backup_env_->RenameFile(
GetAbsolutePath(GetPrivateFileRel(new_backup_id, true)), // tmp
GetAbsolutePath(GetPrivateFileRel(new_backup_id, false)));
}
auto backup_time = backup_env_->NowMicros() - start_backup;
if (s.ok()) {
@@ -1307,22 +1298,33 @@ Status BackupEngineImpl::AddBackupFileWorkItem(
dst_relative_tmp = GetSharedFileRel(dst_relative, true);
dst_relative = GetSharedFileRel(dst_relative, false);
} else {
dst_relative_tmp = GetPrivateFileRel(backup_id, true, dst_relative);
dst_relative = GetPrivateFileRel(backup_id, false, dst_relative);
}
std::string dst_path = GetAbsolutePath(dst_relative);
std::string dst_path_tmp = GetAbsolutePath(dst_relative_tmp);
// We copy into `temp_dest_path` and, once finished, rename it to
// `final_dest_path`. This allows files to atomically appear at
// `final_dest_path`. We can copy directly to the final path when atomicity
// is unnecessary, like for files in private backup directories.
const std::string* copy_dest_path;
std::string temp_dest_path;
std::string final_dest_path = GetAbsolutePath(dst_relative);
if (!dst_relative_tmp.empty()) {
temp_dest_path = GetAbsolutePath(dst_relative_tmp);
copy_dest_path = &temp_dest_path;
} else {
copy_dest_path = &final_dest_path;
}
// if it's shared, we also need to check if it exists -- if it does, no need
// to copy it again.
bool need_to_copy = true;
// true if dst_path is the same path as another live file
// true if final_dest_path is the same path as another live file
const bool same_path =
live_dst_paths.find(dst_path) != live_dst_paths.end();
live_dst_paths.find(final_dest_path) != live_dst_paths.end();
bool file_exists = false;
if (shared && !same_path) {
Status exist = backup_env_->FileExists(dst_path);
Status exist = backup_env_->FileExists(final_dest_path);
if (exist.ok()) {
file_exists = true;
} else if (exist.IsNotFound()) {
@@ -1351,7 +1353,7 @@ Status BackupEngineImpl::AddBackupFileWorkItem(
"overwrite the file.",
fname.c_str());
need_to_copy = true;
backup_env_->DeleteFile(dst_path);
backup_env_->DeleteFile(final_dest_path);
} else {
// the file is present and referenced by a backup
ROCKS_LOG_INFO(options_.info_log,
@@ -1360,25 +1362,25 @@ Status BackupEngineImpl::AddBackupFileWorkItem(
&checksum_value);
}
}
live_dst_paths.insert(dst_path);
live_dst_paths.insert(final_dest_path);
if (!contents.empty() || need_to_copy) {
ROCKS_LOG_INFO(options_.info_log, "Copying %s to %s", fname.c_str(),
dst_path_tmp.c_str());
copy_dest_path->c_str());
CopyOrCreateWorkItem copy_or_create_work_item(
src_dir.empty() ? "" : src_dir + fname, dst_path_tmp, contents, db_env_,
backup_env_, options_.sync, rate_limiter, size_limit,
src_dir.empty() ? "" : src_dir + fname, *copy_dest_path, contents,
db_env_, backup_env_, options_.sync, rate_limiter, size_limit,
progress_callback);
BackupAfterCopyOrCreateWorkItem after_copy_or_create_work_item(
copy_or_create_work_item.result.get_future(), shared, need_to_copy,
backup_env_, dst_path_tmp, dst_path, dst_relative);
backup_env_, temp_dest_path, final_dest_path, dst_relative);
files_to_copy_or_create_.write(std::move(copy_or_create_work_item));
backup_items_to_finish.push_back(std::move(after_copy_or_create_work_item));
} else {
std::promise<CopyOrCreateResult> promise_result;
BackupAfterCopyOrCreateWorkItem after_copy_or_create_work_item(
promise_result.get_future(), shared, need_to_copy, backup_env_,
dst_path_tmp, dst_path, dst_relative);
temp_dest_path, final_dest_path, dst_relative);
backup_items_to_finish.push_back(std::move(after_copy_or_create_work_item));
CopyOrCreateResult result;
result.status = s;
@@ -1533,7 +1535,7 @@ Status BackupEngineImpl::GarbageCollect() {
}
// here we have to delete the dir and all its children
std::string full_private_path =
GetAbsolutePath(GetPrivateFileRel(backup_id, tmp_dir));
GetAbsolutePath(GetPrivateFileRel(backup_id));
std::vector<std::string> subchildren;
backup_env_->GetChildren(full_private_path, &subchildren);
for (auto& subchild : subchildren) {
@@ -1733,51 +1735,25 @@ Status BackupEngineImpl::BackupMeta::StoreToFile(bool sync) {
if (!app_metadata_.empty()) {
std::string hex_encoded_metadata =
Slice(app_metadata_).ToString(/* hex */ true);
// +1 to accomodate newline character
size_t hex_meta_strlen = kMetaDataPrefix.ToString().length() + hex_encoded_metadata.length() + 1;
if (hex_meta_strlen >= buf_size) {
return Status::Corruption("Buffer too small to fit backup metadata");
}
else if (len + hex_meta_strlen >= buf_size) {
backup_meta_file->Append(Slice(buf.get(), len));
buf.reset();
unique_ptr<char[]> new_reset_buf(new char[max_backup_meta_file_size_]);
buf.swap(new_reset_buf);
len = 0;
}
len += snprintf(buf.get() + len, buf_size - len, "%s%s\n",
kMetaDataPrefix.ToString().c_str(),
hex_encoded_metadata.c_str());
if (len >= buf_size) {
return Status::Corruption("Buffer too small to fit backup metadata");
}
}
char writelen_temp[19];
if (len + sprintf(writelen_temp, "%" ROCKSDB_PRIszt "\n", files_.size()) >= buf_size) {
backup_meta_file->Append(Slice(buf.get(), len));
buf.reset();
unique_ptr<char[]> new_reset_buf(new char[max_backup_meta_file_size_]);
buf.swap(new_reset_buf);
len = 0;
len += snprintf(buf.get() + len, buf_size - len, "%" ROCKSDB_PRIszt "\n",
files_.size());
if (len >= buf_size) {
return Status::Corruption("Buffer too small to fit backup metadata");
}
{
const char *const_write = writelen_temp;
len += snprintf(buf.get() + len, buf_size - len, "%s", const_write);
}
for (const auto& file : files_) {
// use crc32 for now, switch to something else if needed
size_t newlen = len + file->filename.length() + sprintf(writelen_temp, " crc32 %u\n", file->checksum_value);
const char *const_write = writelen_temp;
if (newlen >= buf_size) {
backup_meta_file->Append(Slice(buf.get(), len));
buf.reset();
unique_ptr<char[]> new_reset_buf(new char[max_backup_meta_file_size_]);
buf.swap(new_reset_buf);
len = 0;
len += snprintf(buf.get() + len, buf_size - len, "%s crc32 %u\n",
file->filename.c_str(), file->checksum_value);
if (len >= buf_size) {
return Status::Corruption("Buffer too small to fit backup metadata");
}
len += snprintf(buf.get() + len, buf_size - len, "%s%s",
file->filename.c_str(), const_write);
}
s = backup_meta_file->Append(Slice(buf.get(), len));
+8 -9
View File
@@ -811,9 +811,8 @@ TEST_F(BackupableDBTest, NoDoubleCopy) {
test_db_env_->SetFilenamesForMockedAttrs(dummy_db_->live_files_);
ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), false));
std::vector<std::string> should_have_written = {
"/shared/.00010.sst.tmp", "/shared/.00011.sst.tmp",
"/private/1.tmp/CURRENT", "/private/1.tmp/MANIFEST-01",
"/private/1.tmp/00011.log", "/meta/.1.tmp"};
"/shared/.00010.sst.tmp", "/shared/.00011.sst.tmp", "/private/1/CURRENT",
"/private/1/MANIFEST-01", "/private/1/00011.log", "/meta/.1.tmp"};
AppendPath(backupdir_, should_have_written);
test_backup_env_->AssertWrittenFiles(should_have_written);
@@ -829,9 +828,9 @@ TEST_F(BackupableDBTest, NoDoubleCopy) {
ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), false));
// should not open 00010.sst - it's already there
should_have_written = {"/shared/.00015.sst.tmp", "/private/2.tmp/CURRENT",
"/private/2.tmp/MANIFEST-01",
"/private/2.tmp/00011.log", "/meta/.2.tmp"};
should_have_written = {"/shared/.00015.sst.tmp", "/private/2/CURRENT",
"/private/2/MANIFEST-01", "/private/2/00011.log",
"/meta/.2.tmp"};
AppendPath(backupdir_, should_have_written);
test_backup_env_->AssertWrittenFiles(should_have_written);
@@ -976,7 +975,7 @@ TEST_F(BackupableDBTest, InterruptCreationTest) {
backup_engine_->CreateNewBackup(db_.get(), !!(rnd.Next() % 2)).ok());
CloseDBAndBackupEngine();
// should also fail cleanup so the tmp directory stays behind
ASSERT_OK(backup_chroot_env_->FileExists(backupdir_ + "/private/1.tmp/"));
ASSERT_OK(backup_chroot_env_->FileExists(backupdir_ + "/private/1/"));
OpenDBAndBackupEngine(false /* destroy_old_data */);
test_backup_env_->SetLimitWrittenFiles(1000000);
@@ -1171,7 +1170,7 @@ TEST_F(BackupableDBTest, DeleteTmpFiles) {
shared_tmp += "/shared";
}
shared_tmp += "/.00006.sst.tmp";
std::string private_tmp_dir = backupdir_ + "/private/10.tmp";
std::string private_tmp_dir = backupdir_ + "/private/10";
std::string private_tmp_file = private_tmp_dir + "/00003.sst";
file_manager_->WriteToFile(shared_tmp, "tmp");
file_manager_->CreateDir(private_tmp_dir);
@@ -1589,7 +1588,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr, "SKIPPED as BackupableDB is not supported in ROCKSDB_LITE\n");
return 0;
}
+7 -17
View File
@@ -1218,7 +1218,9 @@ Status BlobDBImpl::CloseBlobFile(std::shared_ptr<BlobFile> bfile,
if (bfile->HasTTL()) {
size_t erased __attribute__((__unused__));
erased = open_ttl_files_.erase(bfile);
} else if (bfile == open_non_ttl_file_) {
assert(erased == 1);
} else {
assert(bfile == open_non_ttl_file_);
open_non_ttl_file_ = nullptr;
}
}
@@ -1299,9 +1301,7 @@ std::pair<bool, int64_t> BlobDBImpl::CheckSeqFiles(bool aborted) {
{
ReadLock lockbfile_r(&bfile->mutex_);
if (bfile->expiration_range_.second > epoch_now) {
continue;
}
if (bfile->expiration_range_.second > epoch_now) continue;
process_files.push_back(bfile);
}
}
@@ -1568,16 +1568,12 @@ Status BlobDBImpl::GCFileAndUpdateLSM(const std::shared_ptr<BlobFile>& bfptr,
newfile = NewBlobFile(reason);
new_writer = CheckOrCreateWriterLocked(newfile);
// Can't use header beyond this point
newfile->header_ = std::move(header);
// Can't use header beyond this point
newfile->header_valid_ = true;
newfile->file_size_ = BlobLogHeader::kSize;
newfile->SetColumnFamilyId(bfptr->column_family_id());
newfile->SetHasTTL(bfptr->HasTTL());
newfile->SetCompression(bfptr->compression());
newfile->expiration_range_ = bfptr->expiration_range_;
s = new_writer->WriteHeader(newfile->header_);
if (!s.ok()) {
ROCKS_LOG_ERROR(db_options_.info_log,
"File: %s - header writing failed",
@@ -1585,10 +1581,8 @@ Status BlobDBImpl::GCFileAndUpdateLSM(const std::shared_ptr<BlobFile>& bfptr,
break;
}
// We don't add the file to open_ttl_files_ or open_non_ttl_files_, to
// avoid user writes writing to the file, and avoid CheckSeqFiles close
// the file by mistake.
WriteLock wl(&mutex_);
blob_files_.insert(std::make_pair(newfile->BlobFileNumber(), newfile));
}
@@ -1656,10 +1650,6 @@ Status BlobDBImpl::GCFileAndUpdateLSM(const std::shared_ptr<BlobFile>& bfptr,
gc_stats->bytes_overwritten);
RecordTick(statistics_, BLOB_DB_GC_BYTES_EXPIRED, gc_stats->bytes_expired);
if (newfile != nullptr) {
{
MutexLock l(&write_mutex_);
CloseBlobFile(newfile);
}
total_blob_size_ += newfile->file_size_;
ROCKS_LOG_INFO(db_options_.info_log, "New blob file %" PRIu64 ".",
newfile->BlobFileNumber());
+1 -21
View File
@@ -763,26 +763,6 @@ TEST_F(BlobDBTest, GCExpiredKeyWhileOverwriting) {
VerifyDB({{"foo", "v2"}});
}
TEST_F(BlobDBTest, NewFileGeneratedFromGCShouldMarkAsImmutable) {
BlobDBOptions bdb_options;
bdb_options.min_blob_size = 0;
bdb_options.disable_background_tasks = true;
Open(bdb_options);
ASSERT_OK(Put("foo", "bar"));
auto blob_files = blob_db_impl()->TEST_GetBlobFiles();
auto blob_file1 = blob_files[0];
ASSERT_EQ(1, blob_files.size());
ASSERT_OK(blob_db_impl()->TEST_CloseBlobFile(blob_file1));
GCStats gc_stats;
ASSERT_OK(blob_db_impl()->TEST_GCFileAndUpdateLSM(blob_file1, &gc_stats));
ASSERT_EQ(1, gc_stats.blob_count);
ASSERT_EQ(1, gc_stats.num_keys_relocated);
blob_files = blob_db_impl()->TEST_GetBlobFiles();
ASSERT_EQ(2, blob_files.size());
ASSERT_EQ(blob_file1, blob_files[0]);
ASSERT_TRUE(blob_files[1]->Immutable());
}
// This test is no longer valid since we now return an error when we go
// over the configured max_db_size.
// The test needs to be re-written later in such a way that writes continue
@@ -1544,7 +1524,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr, "SKIPPED as BlobDB is not supported in ROCKSDB_LITE\n");
return 0;
}
+1 -13
View File
@@ -573,18 +573,6 @@ TEST_F(CheckpointTest, CheckpointInvalidDirectoryName) {
}
}
TEST_F(CheckpointTest, CheckpointWithParallelWrites) {
// When run with TSAN, this exposes the data race fixed in
// https://github.com/facebook/rocksdb/pull/3603
ASSERT_OK(Put("key1", "val1"));
port::Thread thread([this]() { ASSERT_OK(Put("key2", "val2")); });
Checkpoint* checkpoint;
ASSERT_OK(Checkpoint::Create(db_, &checkpoint));
ASSERT_OK(checkpoint->CreateCheckpoint(snapshot_name_));
delete checkpoint;
thread.join();
}
} // namespace rocksdb
int main(int argc, char** argv) {
@@ -596,7 +584,7 @@ int main(int argc, char** argv) {
#else
#include <stdio.h>
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr, "SKIPPED as Checkpoint is not supported in ROCKSDB_LITE\n");
return 0;
}
+1 -1
View File
@@ -169,7 +169,7 @@ int main() {
}
#endif // GFLAGS
#else
int main(int argc, char** argv) {
int main(int /*argc*/, char** /*argv*/) {
fprintf(stderr, "Not supported in lite mode.\n");
return 1;
}
Loaded 100 of 118 files, more files were not shown because too many files have changed in this diff. Show more