mirror of
https://github.com/facebook/rocksdb.git
synced 2026-07-07 22:55:23 +08:00
Compare commits
10 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c60df9d9e7 | |||
| 747c853203 | |||
| f5ee207c64 | |||
| f2fd21fa6f | |||
| 758527812c | |||
| 9d2e34ed4c | |||
| 1f5103d583 | |||
| 163dd4b81b | |||
| 1642582559 | |||
| 8d28083264 |
+11
-1
@@ -1,5 +1,10 @@
|
||||
# Rocksdb Change Log
|
||||
## Unreleased
|
||||
## 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
|
||||
* RocksDBOptionsParser::Parse()'s `ignore_unknown_options` argument will only be effective if the option file shows it is generated using a higher version of RocksDB than the current version.
|
||||
* Remove CompactionEventListener.
|
||||
@@ -13,6 +18,11 @@
|
||||
### 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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -510,6 +510,9 @@ Status ExternalSstFileIngestionJob::AssignGlobalSeqnoForIngestedFile(
|
||||
std::string seqno_val;
|
||||
PutFixed64(&seqno_val, seqno);
|
||||
status = rwfile->Write(file_to_ingest->global_seqno_offset, seqno_val);
|
||||
if (status.ok()) {
|
||||
status = rwfile->Fsync();
|
||||
}
|
||||
if (status.ok()) {
|
||||
file_to_ingest->assigned_seqno = seqno;
|
||||
}
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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
-1
@@ -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");
|
||||
|
||||
+1
-1
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -431,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;
|
||||
|
||||
Vendored
+9
@@ -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;
|
||||
|
||||
Vendored
+6
-2
@@ -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) {
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -5,8 +5,8 @@
|
||||
#pragma once
|
||||
|
||||
#define ROCKSDB_MAJOR 5
|
||||
#define ROCKSDB_MINOR 12
|
||||
#define ROCKSDB_PATCH 0
|
||||
#define ROCKSDB_MINOR 13
|
||||
#define ROCKSDB_PATCH 1
|
||||
|
||||
// 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
|
||||
|
||||
+20
-13
@@ -38,13 +38,14 @@ jlong Java_org_rocksdb_PlainTableConfig_newTableFactoryHandle(
|
||||
/*
|
||||
* Class: org_rocksdb_BlockBasedTableConfig
|
||||
* Method: newTableFactoryHandle
|
||||
* Signature: (ZJIJIIZIZZZJIBBI)J
|
||||
* Signature: (ZJIJJIIZIZZZJIBBI)J
|
||||
*/
|
||||
jlong Java_org_rocksdb_BlockBasedTableConfig_newTableFactoryHandle(
|
||||
JNIEnv* env, jobject jobj, jboolean no_block_cache, jlong block_cache_size,
|
||||
jint block_cache_num_shardbits, jlong block_size, jint block_size_deviation,
|
||||
jint block_restart_interval, jboolean whole_key_filtering,
|
||||
jlong jfilterPolicy, jboolean cache_index_and_filter_blocks,
|
||||
JNIEnv *env, jobject jobj, jboolean no_block_cache, jlong block_cache_size,
|
||||
jint block_cache_num_shardbits, jlong jblock_cache, jlong block_size,
|
||||
jint block_size_deviation, jint block_restart_interval,
|
||||
jboolean whole_key_filtering, jlong jfilter_policy,
|
||||
jboolean cache_index_and_filter_blocks,
|
||||
jboolean pin_l0_filter_and_index_blocks_in_cache,
|
||||
jboolean hash_index_allow_collision, jlong block_cache_compressed_size,
|
||||
jint block_cache_compressd_num_shard_bits, jbyte jchecksum_type,
|
||||
@@ -52,22 +53,28 @@ jlong Java_org_rocksdb_BlockBasedTableConfig_newTableFactoryHandle(
|
||||
rocksdb::BlockBasedTableOptions options;
|
||||
options.no_block_cache = no_block_cache;
|
||||
|
||||
if (!no_block_cache && block_cache_size > 0) {
|
||||
if (block_cache_num_shardbits > 0) {
|
||||
options.block_cache =
|
||||
rocksdb::NewLRUCache(block_cache_size, block_cache_num_shardbits);
|
||||
} else {
|
||||
options.block_cache = rocksdb::NewLRUCache(block_cache_size);
|
||||
if (!no_block_cache) {
|
||||
if (jblock_cache > 0) {
|
||||
std::shared_ptr<rocksdb::Cache> *pCache =
|
||||
reinterpret_cast<std::shared_ptr<rocksdb::Cache> *>(jblock_cache);
|
||||
options.block_cache = *pCache;
|
||||
} else if (block_cache_size > 0) {
|
||||
if (block_cache_num_shardbits > 0) {
|
||||
options.block_cache =
|
||||
rocksdb::NewLRUCache(block_cache_size, block_cache_num_shardbits);
|
||||
} else {
|
||||
options.block_cache = rocksdb::NewLRUCache(block_cache_size);
|
||||
}
|
||||
}
|
||||
}
|
||||
options.block_size = block_size;
|
||||
options.block_size_deviation = block_size_deviation;
|
||||
options.block_restart_interval = block_restart_interval;
|
||||
options.whole_key_filtering = whole_key_filtering;
|
||||
if (jfilterPolicy > 0) {
|
||||
if (jfilter_policy > 0) {
|
||||
std::shared_ptr<rocksdb::FilterPolicy> *pFilterPolicy =
|
||||
reinterpret_cast<std::shared_ptr<rocksdb::FilterPolicy> *>(
|
||||
jfilterPolicy);
|
||||
jfilter_policy);
|
||||
options.filter_policy = *pFilterPolicy;
|
||||
}
|
||||
options.cache_index_and_filter_blocks = cache_index_and_filter_blocks;
|
||||
|
||||
@@ -15,6 +15,7 @@ public class BlockBasedTableConfig extends TableFormatConfig {
|
||||
noBlockCache_ = false;
|
||||
blockCacheSize_ = 8 * 1024 * 1024;
|
||||
blockCacheNumShardBits_ = 0;
|
||||
blockCache_ = null;
|
||||
blockSize_ = 4 * 1024;
|
||||
blockSizeDeviation_ = 10;
|
||||
blockRestartInterval_ = 16;
|
||||
@@ -71,6 +72,24 @@ public class BlockBasedTableConfig extends TableFormatConfig {
|
||||
return blockCacheSize_;
|
||||
}
|
||||
|
||||
/**
|
||||
* Use the specified cache for blocks.
|
||||
* When not null this take precedence even if the user sets a block cache size.
|
||||
*
|
||||
* {@link org.rocksdb.Cache} should not be disposed before options instances
|
||||
* using this cache is disposed.
|
||||
*
|
||||
* {@link org.rocksdb.Cache} instance can be re-used in multiple options
|
||||
* instances.
|
||||
*
|
||||
* @param cache {@link org.rocksdb.Cache} Cache java instance (e.g. LRUCache).
|
||||
* @return the reference to the current config.
|
||||
*/
|
||||
public BlockBasedTableConfig setBlockCache(final Cache cache) {
|
||||
blockCache_ = cache;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Controls the number of shards for the block cache.
|
||||
* This is applied only if cacheSize is set to non-negative.
|
||||
@@ -413,25 +432,25 @@ public class BlockBasedTableConfig extends TableFormatConfig {
|
||||
filterHandle = filter_.nativeHandle_;
|
||||
}
|
||||
|
||||
return newTableFactoryHandle(noBlockCache_, blockCacheSize_,
|
||||
blockCacheNumShardBits_, blockSize_, blockSizeDeviation_,
|
||||
blockRestartInterval_, wholeKeyFiltering_,
|
||||
filterHandle, cacheIndexAndFilterBlocks_,
|
||||
pinL0FilterAndIndexBlocksInCache_,
|
||||
hashIndexAllowCollision_, blockCacheCompressedSize_,
|
||||
blockCacheCompressedNumShardBits_,
|
||||
checksumType_.getValue(), indexType_.getValue(),
|
||||
long blockCacheHandle = 0;
|
||||
if (blockCache_ != null) {
|
||||
blockCacheHandle = blockCache_.nativeHandle_;
|
||||
}
|
||||
|
||||
return newTableFactoryHandle(noBlockCache_, blockCacheSize_, blockCacheNumShardBits_,
|
||||
blockCacheHandle, blockSize_, blockSizeDeviation_, blockRestartInterval_,
|
||||
wholeKeyFiltering_, filterHandle, cacheIndexAndFilterBlocks_,
|
||||
pinL0FilterAndIndexBlocksInCache_, hashIndexAllowCollision_, blockCacheCompressedSize_,
|
||||
blockCacheCompressedNumShardBits_, checksumType_.getValue(), indexType_.getValue(),
|
||||
formatVersion_);
|
||||
}
|
||||
|
||||
private native long newTableFactoryHandle(
|
||||
boolean noBlockCache, long blockCacheSize, int blockCacheNumShardBits,
|
||||
long blockSize, int blockSizeDeviation, int blockRestartInterval,
|
||||
boolean wholeKeyFiltering, long filterPolicyHandle,
|
||||
private native long newTableFactoryHandle(boolean noBlockCache, long blockCacheSize,
|
||||
int blockCacheNumShardBits, long blockCacheHandle, long blockSize, int blockSizeDeviation,
|
||||
int blockRestartInterval, boolean wholeKeyFiltering, long filterPolicyHandle,
|
||||
boolean cacheIndexAndFilterBlocks, boolean pinL0FilterAndIndexBlocksInCache,
|
||||
boolean hashIndexAllowCollision, long blockCacheCompressedSize,
|
||||
int blockCacheCompressedNumShardBits, byte checkSumType,
|
||||
byte indexType, int formatVersion);
|
||||
int blockCacheCompressedNumShardBits, byte checkSumType, byte indexType, int formatVersion);
|
||||
|
||||
private boolean cacheIndexAndFilterBlocks_;
|
||||
private boolean pinL0FilterAndIndexBlocksInCache_;
|
||||
@@ -442,6 +461,7 @@ public class BlockBasedTableConfig extends TableFormatConfig {
|
||||
private long blockSize_;
|
||||
private long blockCacheSize_;
|
||||
private int blockCacheNumShardBits_;
|
||||
private Cache blockCache_;
|
||||
private long blockCacheCompressedSize_;
|
||||
private int blockCacheCompressedNumShardBits_;
|
||||
private int blockSizeDeviation_;
|
||||
|
||||
@@ -6,7 +6,11 @@
|
||||
package org.rocksdb;
|
||||
|
||||
import org.junit.ClassRule;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
import org.junit.rules.TemporaryFolder;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
@@ -16,6 +20,8 @@ public class BlockBasedTableConfigTest {
|
||||
public static final RocksMemoryResource rocksMemoryResource =
|
||||
new RocksMemoryResource();
|
||||
|
||||
@Rule public TemporaryFolder dbFolder = new TemporaryFolder();
|
||||
|
||||
@Test
|
||||
public void noBlockCache() {
|
||||
BlockBasedTableConfig blockBasedTableConfig = new BlockBasedTableConfig();
|
||||
@@ -31,6 +37,31 @@ public class BlockBasedTableConfigTest {
|
||||
isEqualTo(8 * 1024);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void sharedBlockCache() throws RocksDBException {
|
||||
try (final Cache cache = new LRUCache(8 * 1024 * 1024);
|
||||
final Statistics statistics = new Statistics()) {
|
||||
for (int shard = 0; shard < 8; shard++) {
|
||||
try (final Options options =
|
||||
new Options()
|
||||
.setCreateIfMissing(true)
|
||||
.setStatistics(statistics)
|
||||
.setTableFormatConfig(new BlockBasedTableConfig().setBlockCache(cache));
|
||||
final RocksDB db =
|
||||
RocksDB.open(options, dbFolder.getRoot().getAbsolutePath() + "/" + shard)) {
|
||||
final byte[] key = "some-key".getBytes(StandardCharsets.UTF_8);
|
||||
final byte[] value = "some-value".getBytes(StandardCharsets.UTF_8);
|
||||
|
||||
db.put(key, value);
|
||||
db.flush(new FlushOptions());
|
||||
db.get(key);
|
||||
|
||||
assertThat(statistics.getTickerCount(TickerType.BLOCK_CACHE_ADD)).isEqualTo(shard + 1);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void blockSizeDeviation() {
|
||||
BlockBasedTableConfig blockBasedTableConfig = new BlockBasedTableConfig();
|
||||
@@ -148,6 +179,14 @@ public class BlockBasedTableConfigTest {
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void blockBasedTableWithBlockCache() {
|
||||
try (final Options options = new Options().setTableFormatConfig(
|
||||
new BlockBasedTableConfig().setBlockCache(new LRUCache(17 * 1024 * 1024)))) {
|
||||
assertThat(options.tableFactoryName()).isEqualTo("BlockBasedTable");
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void blockBasedTableFormatVersion() {
|
||||
BlockBasedTableConfig config = new BlockBasedTableConfig();
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -244,6 +244,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
|
||||
@@ -302,6 +309,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);
|
||||
}
|
||||
|
||||
@@ -61,7 +61,8 @@ class PartitionedFilterBlockBuilder : public FullFilterBlockBuilder {
|
||||
uint32_t filters_in_partition_;
|
||||
};
|
||||
|
||||
class PartitionedFilterBlockReader : public FilterBlockReader {
|
||||
class PartitionedFilterBlockReader : public FilterBlockReader,
|
||||
public Cleanable {
|
||||
public:
|
||||
explicit PartitionedFilterBlockReader(const SliceTransform* prefix_extractor,
|
||||
bool whole_key_filtering,
|
||||
|
||||
+18
-11
@@ -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) {
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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);
|
||||
|
||||
|
||||
@@ -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
@@ -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);
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -613,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;
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -1524,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;
|
||||
}
|
||||
|
||||
@@ -584,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;
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -460,7 +460,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 DateTieredDB is not supported in ROCKSDB_LITE\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -328,7 +328,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 DocumentDB is not supported in ROCKSDB_LITE\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -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 JSONDocument is not supported in ROCKSDB_LITE\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -36,7 +36,7 @@ int main(int argc, char** argv) {
|
||||
#else // ROCKSDB_LITE
|
||||
#include <stdio.h>
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
int main(int /*argc*/, char** /*argv*/) {
|
||||
fprintf(stderr, "SKIPPED as TimedEnv is not supported in ROCKSDB_LITE\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -491,7 +491,7 @@ int main(int argc, char** argv) {
|
||||
|
||||
#else
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
int main(int /*argc*/, char** /*argv*/) {
|
||||
printf("Lua is not supported in RocksDBLite. Ignoring the test.\n");
|
||||
}
|
||||
|
||||
|
||||
@@ -269,7 +269,7 @@ int main(int argc, char** argv) {
|
||||
#else
|
||||
#include <cstdio>
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
int main(int /*argc*/, char** /*argv*/) {
|
||||
printf("Skipped in RocksDBLite as utilities are not supported.\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -65,7 +65,7 @@ int main(int argc, char** argv) {
|
||||
#else // ROCKSDB_LITE
|
||||
#include <stdio.h>
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
int main(int /*argc*/, char** /*argv*/) {
|
||||
fprintf(stderr, "SKIPPED as EnvRegistry is not supported in ROCKSDB_LITE\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -311,7 +311,7 @@ int main(int argc, char** argv) {
|
||||
#else
|
||||
#include <cstdio>
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
int main(int /*argc*/, char** /*argv*/) {
|
||||
printf("Skipped in RocksDBLite as utilities are not supported.\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -886,7 +886,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 redis is not supported in ROCKSDB_LITE\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -299,7 +299,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 SpatialDB is not supported in ROCKSDB_LITE\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -171,7 +171,7 @@ int main(int /*argc*/, char** /*argv*/) {
|
||||
fprintf(stderr, "PASSED\n");
|
||||
}
|
||||
#else
|
||||
int main(int argc, char** argv) {
|
||||
int main(int /*argc*/, char** /*argv*/) {
|
||||
fprintf(stderr, "SKIPPED as RocksDBLite does not include utilities.\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -1391,7 +1391,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 optimistic_transaction is not supported in ROCKSDB_LITE\n");
|
||||
|
||||
@@ -5542,7 +5542,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 Transactions are not supported in ROCKSDB_LITE\n");
|
||||
return 0;
|
||||
|
||||
@@ -2003,7 +2003,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 Transactions are not supported in ROCKSDB_LITE\n");
|
||||
return 0;
|
||||
|
||||
@@ -655,7 +655,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 DBWithTTL is not supported in ROCKSDB_LITE\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user