andreacavalli/rocksdb
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

change use_direct_writes to use_direct_io_for_flush_and_compaction

Browse Source 
Summary:
Replace Options::use_direct_writes with Options::use_direct_io_for_flush_and_compaction
Now if Options::use_direct_io_for_flush_and_compaction = true, we will enable direct io for both reads and writes for flush and compaction job. Whereas Options::use_direct_reads controls user reads like iterator and Get().
Closes https://github.com/facebook/rocksdb/pull/2117

Differential Revision: D4860912

Pulled By: lightmark

fbshipit-source-id: d93575a8a5e780cf7e40797287edc425ee648c19
This commit is contained in:
Aaron Gao 2017-04-13 13:07:33 -07:00
parent b6f6b73a9c
commit 8d7edd5908
34 changed files with 253 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
HISTORY.md
View File

@ -8,6 +8,7 @@
* DB::Get in place of std::string accepts PinnableSlice, which avoids the extra memcpy of value to std::string in most of cases.
* PinnableSlice releases the pinned resources that contain the value when it is destructed or when ::Reset() is called on it.
* The old API that accepts std::string, although discouraged, is still supported.
* Replace Options::use_direct_writes with Options::use_direct_io_for_flush_and_compaction. Read Direct IO wiki for details.
### New Features
* Memtable flush can be avoided during checkpoint creation if total log file size is smaller than a threshold specified by the user.

10
db/builder.cc
View File

@ -32,6 +32,7 @@
#include "util/file_reader_writer.h"
#include "util/filename.h"
#include "util/stop_watch.h"
#include "util/sync_point.h"
namespace rocksdb {
@ -103,6 +104,10 @@ Status BuildTable(
unique_ptr<WritableFileWriter> file_writer;
{
unique_ptr<WritableFile> file;
#ifndef NDEBUG
bool use_direct_writes = env_options.use_direct_writes;
TEST_SYNC_POINT_CALLBACK("BuildTable:create_file", &use_direct_writes);
#endif // !NDEBUG
s = NewWritableFile(env, fname, &file, env_options);
if (!s.ok()) {
EventHelpers::LogAndNotifyTableFileCreationFinished(
@ -180,6 +185,11 @@ Status BuildTable(
if (s.ok() && !empty) {
// Verify that the table is usable
// We set for_compaction to false and don't OptimizeForCompactionTableRead
// here because this is a special case after we finish the table building
// No matter whether use_direct_io_for_flush_and_compaction is true,
// we will regrad this verification as user reads since the goal is
// to cache it here for further user reads
std::unique_ptr<InternalIterator> it(table_cache->NewIterator(
ReadOptions(), env_options, internal_comparator, meta->fd,
nullptr /* range_del_agg */, nullptr,

6
db/c.cc
View File

@ -2121,9 +2121,9 @@ void rocksdb_options_set_use_direct_reads(rocksdb_options_t* opt,
opt->rep.use_direct_reads = v;
}
void rocksdb_options_set_use_direct_writes(rocksdb_options_t* opt,
unsigned char v) {
opt->rep.use_direct_writes = v;
void rocksdb_options_set_use_direct_io_for_flush_and_compaction(
rocksdb_options_t* opt, unsigned char v) {
opt->rep.use_direct_io_for_flush_and_compaction = v;
}
void rocksdb_options_set_allow_mmap_reads(

11
db/compaction_job.cc
View File

@ -1072,6 +1072,11 @@ Status CompactionJob::FinishCompactionOutputFile(
TableProperties tp;
if (s.ok() && current_entries > 0) {
// Verify that the table is usable
// We set for_compaction to false and don't OptimizeForCompactionTableRead
// here because this is a special case after we finish the table building
// No matter whether use_direct_io_for_flush_and_compaction is true,
// we will regrad this verification as user reads since the goal is
// to cache it here for further user reads
InternalIterator* iter = cfd->table_cache()->NewIterator(
ReadOptions(), env_options_, cfd->internal_comparator(), meta->fd,
nullptr /* range_del_agg */, nullptr,
@ -1198,7 +1203,11 @@ Status CompactionJob::OpenCompactionOutputFile(
#endif // !ROCKSDB_LITE
// Make the output file
unique_ptr<WritableFile> writable_file;
Status s = NewWritableFile(env_, fname, &writable_file, env_options_);
EnvOptions opt_env_opts =
env_->OptimizeForCompactionTableWrite(env_options_, db_options_);
TEST_SYNC_POINT_CALLBACK("CompactionJob::OpenCompactionOutputFile",
&opt_env_opts.use_direct_writes);
Status s = NewWritableFile(env_, fname, &writable_file, opt_env_opts);
if (!s.ok()) {
ROCKS_LOG_ERROR(
db_options_.info_log,

39
db/db_compaction_test.cc
View File

@ -40,6 +40,12 @@ class DBCompactionTestWithParam
bool exclusive_manual_compaction_;
};
class DBCompactionDirectIOTest : public DBCompactionTest,
public ::testing::WithParamInterface<bool> {
public:
DBCompactionDirectIOTest() : DBCompactionTest() {}
};
namespace {
class FlushedFileCollector : public EventListener {
@ -2552,6 +2558,39 @@ INSTANTIATE_TEST_CASE_P(DBCompactionTestWithParam, DBCompactionTestWithParam,
std::make_tuple(4, true),
std::make_tuple(4, false)));
TEST_P(DBCompactionDirectIOTest, DirectIO) {
Options options = CurrentOptions();
Destroy(options);
options.create_if_missing = true;
options.disable_auto_compactions = true;
options.use_direct_io_for_flush_and_compaction = GetParam();
options.env = new MockEnv(Env::Default());
Reopen(options);
SyncPoint::GetInstance()->SetCallBack(
"TableCache::NewIterator:for_compaction", [&](void* arg) {
bool* use_direct_reads = static_cast<bool*>(arg);
ASSERT_EQ(*use_direct_reads,
options.use_direct_io_for_flush_and_compaction);
});
SyncPoint::GetInstance()->SetCallBack(
"CompactionJob::OpenCompactionOutputFile", [&](void* arg) {
bool* use_direct_writes = static_cast<bool*>(arg);
ASSERT_EQ(*use_direct_writes,
options.use_direct_io_for_flush_and_compaction);
});
SyncPoint::GetInstance()->EnableProcessing();
CreateAndReopenWithCF({"pikachu"}, options);
MakeTables(3, "p", "q", 1);
ASSERT_EQ("1,1,1", FilesPerLevel(1));
Compact(1, "p1", "p9");
ASSERT_EQ("0,0,1", FilesPerLevel(1));
Destroy(options);
delete options.env;
}
INSTANTIATE_TEST_CASE_P(DBCompactionDirectIOTest, DBCompactionDirectIOTest,
testing::Bool());
class CompactionPriTest : public DBTestBase,
public testing::WithParamInterface<uint32_t> {
public:

33
db/db_flush_test.cc
View File

@ -19,6 +19,12 @@ class DBFlushTest : public DBTestBase {
DBFlushTest() : DBTestBase("/db_flush_test") {}
};
class DBFlushDirectIOTest : public DBFlushTest,
public ::testing::WithParamInterface<bool> {
public:
DBFlushDirectIOTest() : DBFlushTest() {}
};
// We had issue when two background threads trying to flush at the same time,
// only one of them get committed. The test verifies the issue is fixed.
TEST_F(DBFlushTest, FlushWhileWritingManifest) {
@ -83,6 +89,33 @@ TEST_F(DBFlushTest, SyncFail) {
Destroy(options);
}
TEST_P(DBFlushDirectIOTest, DirectIO) {
Options options;
options.create_if_missing = true;
options.disable_auto_compactions = true;
options.max_background_flushes = 2;
options.use_direct_io_for_flush_and_compaction = GetParam();
options.env = new MockEnv(Env::Default());
SyncPoint::GetInstance()->SetCallBack(
"BuildTable:create_file", [&](void* arg) {
bool* use_direct_writes = static_cast<bool*>(arg);
ASSERT_EQ(*use_direct_writes,
options.use_direct_io_for_flush_and_compaction);
});
SyncPoint::GetInstance()->EnableProcessing();
Reopen(options);
ASSERT_OK(Put("foo", "v"));
FlushOptions flush_options;
flush_options.wait = true;
ASSERT_OK(dbfull()->Flush(flush_options));
Destroy(options);
delete options.env;
}
INSTANTIATE_TEST_CASE_P(DBFlushDirectIOTest, DBFlushDirectIOTest,
testing::Bool());
} // namespace rocksdb
int main(int argc, char** argv) {

15
db/db_impl_open.cc
View File

@ -101,7 +101,8 @@ DBOptions SanitizeOptions(const std::string& dbname, const DBOptions& src) {
result.compaction_readahead_size = 1024 * 1024 * 2;
}
if (result.compaction_readahead_size > 0) {
if (result.compaction_readahead_size > 0 ||
result.use_direct_io_for_flush_and_compaction) {
result.new_table_reader_for_compaction_inputs = true;
}
@ -165,10 +166,12 @@ static Status ValidateOptions(
"then direct I/O reads (use_direct_reads) must be disabled. ");
}
if (db_options.allow_mmap_writes && db_options.use_direct_writes) {
if (db_options.allow_mmap_writes &&
db_options.use_direct_io_for_flush_and_compaction) {
return Status::NotSupported(
"If memory mapped writes (allow_mmap_writes) are enabled "
"then direct I/O writes (use_direct_writes) must be disabled. ");
"then direct I/O writes (use_direct_io_for_flush_and_compaction) must "
"be disabled. ");
}
if (db_options.keep_log_file_num == 0) {
@ -823,9 +826,11 @@ Status DBImpl::WriteLevel0TableForRecovery(int job_id, ColumnFamilyData* cfd,
std::vector<SequenceNumber> snapshot_seqs =
snapshots_.GetAll(&earliest_write_conflict_snapshot);
EnvOptions optimized_env_options =
env_->OptimizeForCompactionTableWrite(env_options_, immutable_db_options_);
s = BuildTable(
dbname_, env_, *cfd->ioptions(), mutable_cf_options, env_options_,
cfd->table_cache(), iter.get(),
dbname_, env_, *cfd->ioptions(), mutable_cf_options,
optimized_env_options, cfd->table_cache(), iter.get(),
std::unique_ptr<InternalIterator>(mem->NewRangeTombstoneIterator(ro)),
&meta, cfd->internal_comparator(),
cfd->int_tbl_prop_collector_factories(), cfd->GetID(), cfd->GetName(),

3
db/db_test2.cc
View File

@ -1995,7 +1995,8 @@ TEST_F(DBTest2, DirectIO) {
return;
}
Options options = CurrentOptions();
options.use_direct_reads = options.use_direct_writes = true;
options.use_direct_reads = options.use_direct_io_for_flush_and_compaction =
true;
options.allow_mmap_reads = options.allow_mmap_writes = false;
DestroyAndReopen(options);

4
db/flush_job.cc
View File

@ -294,9 +294,11 @@ Status FlushJob::WriteLevel0Table() {
TEST_SYNC_POINT_CALLBACK("FlushJob::WriteLevel0Table:output_compression",
&output_compression_);
EnvOptions optimized_env_options =
db_options_.env->OptimizeForCompactionTableWrite(env_options_, db_options_);
s = BuildTable(
dbname_, db_options_.env, *cfd_->ioptions(), mutable_cf_options_,
env_options_, cfd_->table_cache(), iter.get(),
optimized_env_options, cfd_->table_cache(), iter.get(),
std::move(range_del_iter), &meta_, cfd_->internal_comparator(),
cfd_->int_tbl_prop_collector_factories(), cfd_->GetID(),
cfd_->GetName(), existing_snapshots_,

2
db/forward_iterator_bench.cc
View File

@ -331,7 +331,7 @@ int main(int argc, char** argv) {
options.compaction_style = rocksdb::CompactionStyle::kCompactionStyleNone;
options.level0_slowdown_writes_trigger = 99999;
options.level0_stop_writes_trigger = 99999;
options.use_direct_writes = true;
options.use_direct_io_for_flush_and_compaction = true;
options.write_buffer_size = FLAGS_memtable_size;
rocksdb::BlockBasedTableOptions table_options;
table_options.block_cache = rocksdb::NewLRUCache(FLAGS_block_cache_size);

4
db/repair.cc
View File

@ -377,9 +377,11 @@ class Repairer {
ro.total_order_seek = true;
Arena arena;
ScopedArenaIterator iter(mem->NewIterator(ro, &arena));
EnvOptions optimized_env_options =
env_->OptimizeForCompactionTableWrite(env_options_, immutable_db_options_);
status = BuildTable(
dbname_, env_, *cfd->ioptions(), *cfd->GetLatestMutableCFOptions(),
env_options_, table_cache_, iter.get(),
optimized_env_options, table_cache_, iter.get(),
std::unique_ptr<InternalIterator>(mem->NewRangeTombstoneIterator(ro)),
&meta, cfd->internal_comparator(),
cfd->int_tbl_prop_collector_factories(), cfd->GetID(), cfd->GetName(),

5
db/table_cache.cc
View File

@ -184,6 +184,11 @@ InternalIterator* TableCache::NewIterator(
}
size_t readahead = 0;
if (for_compaction) {
#ifndef NDEBUG
bool use_direct_reads_for_compaction = env_options.use_direct_reads;
TEST_SYNC_POINT_CALLBACK("TableCache::NewIterator:for_compaction",
&use_direct_reads_for_compaction);
#endif // !NDEBUG
if (ioptions_.new_table_reader_for_compaction_inputs) {
readahead = ioptions_.compaction_readahead_size;
create_new_table_reader = true;

5
db/version_set.cc
View File

@ -2218,7 +2218,8 @@ VersionSet::VersionSet(const std::string& dbname,
current_version_number_(0),
manifest_file_size_(0),
env_options_(storage_options),
env_options_compactions_(env_options_) {}
env_options_compactions_(
env_->OptimizeForCompactionTableRead(env_options_, *db_options_)) {}
void CloseTables(void* ptr, size_t) {
TableReader* table_reader = reinterpret_cast<TableReader*>(ptr);
@ -3477,7 +3478,7 @@ InternalIterator* VersionSet::MakeInputIterator(
// Create concatenating iterator for the files from this level
list[num++] = NewTwoLevelIterator(
new LevelFileIteratorState(
cfd->table_cache(), read_options, env_options_,
cfd->table_cache(), read_options, env_options_compactions_,
cfd->internal_comparator(),
nullptr /* no per level latency histogram */,
true /* for_compaction */, false /* prefix enabled */,

19
env/env.cc vendored
View File

@ -10,9 +10,9 @@
#include "rocksdb/env.h"
#include <thread>
#include "options/db_options.h"
#include "port/port.h"
#include "port/sys_time.h"
#include "rocksdb/options.h"
#include "util/arena.h"
#include "util/autovector.h"
@ -316,7 +316,6 @@ void AssignEnvOptions(EnvOptions* env_options, const DBOptions& options) {
env_options->use_mmap_reads = options.allow_mmap_reads;
env_options->use_mmap_writes = options.allow_mmap_writes;
env_options->use_direct_reads = options.use_direct_reads;
env_options->use_direct_writes = options.use_direct_writes;
env_options->set_fd_cloexec = options.is_fd_close_on_exec;
env_options->bytes_per_sync = options.bytes_per_sync;
env_options->compaction_readahead_size = options.compaction_readahead_size;
@ -341,6 +340,22 @@ EnvOptions Env::OptimizeForManifestWrite(const EnvOptions& env_options) const {
return env_options;
}
EnvOptions Env::OptimizeForCompactionTableWrite(
const EnvOptions& env_options, const ImmutableDBOptions& db_options) const {
EnvOptions optimized_env_options(env_options);
optimized_env_options.use_direct_writes =
db_options.use_direct_io_for_flush_and_compaction;
return optimized_env_options;
}
EnvOptions Env::OptimizeForCompactionTableRead(
const EnvOptions& env_options, const ImmutableDBOptions& db_options) const {
EnvOptions optimized_env_options(env_options);
optimized_env_options.use_direct_reads =
db_options.use_direct_io_for_flush_and_compaction;
return optimized_env_options;
}
EnvOptions::EnvOptions(const DBOptions& options) {
AssignEnvOptions(this, options);
}

5
include/rocksdb/c.h
View File

@ -804,8 +804,9 @@ extern ROCKSDB_LIBRARY_API void rocksdb_options_set_allow_mmap_writes(
rocksdb_options_t*, unsigned char);
extern ROCKSDB_LIBRARY_API void rocksdb_options_set_use_direct_reads(
rocksdb_options_t*, unsigned char);
extern ROCKSDB_LIBRARY_API void rocksdb_options_set_use_direct_writes(
rocksdb_options_t*, unsigned char);
extern ROCKSDB_LIBRARY_API void
rocksdb_options_set_use_direct_io_for_flush_and_compaction(rocksdb_options_t*,
unsigned char);
extern ROCKSDB_LIBRARY_API void rocksdb_options_set_is_fd_close_on_exec(
rocksdb_options_t*, unsigned char);
extern ROCKSDB_LIBRARY_API void rocksdb_options_set_skip_log_error_on_recovery(

15
include/rocksdb/env.h
View File

@ -44,6 +44,7 @@ class WritableFile;
class RandomRWFile;
class Directory;
struct DBOptions;
struct ImmutableDBOptions;
class RateLimiter;
class ThreadStatusUpdater;
struct ThreadStatus;
@ -375,6 +376,20 @@ class Env {
virtual EnvOptions OptimizeForManifestWrite(
const EnvOptions& env_options) const;
// OptimizeForCompactionTableWrite will create a new EnvOptions object that is a copy
// of the EnvOptions in the parameters, but is optimized for writing table
// files. Default implementation returns the copy of the same object.
virtual EnvOptions OptimizeForCompactionTableWrite(
const EnvOptions& env_options,
const ImmutableDBOptions& db_options) const;
// OptimizeForCompactionTableWrite will create a new EnvOptions object that is a copy
// of the EnvOptions in the parameters, but is optimized for reading table
// files. Default implementation returns the copy of the same object.
virtual EnvOptions OptimizeForCompactionTableRead(
const EnvOptions& env_options,
const ImmutableDBOptions& db_options) const;
// Returns the status of all threads that belong to the current Env.
virtual Status GetThreadList(std::vector<ThreadStatus>* thread_list) {
return Status::NotSupported("Not supported.");

7
include/rocksdb/options.h
View File

@ -581,15 +581,16 @@ struct DBOptions {
// bufferized. The hardware buffer of the devices may however still
// be used. Memory mapped files are not impacted by these parameters.
// Use O_DIRECT for reading file
// Use O_DIRECT for user reads
// Default: false
// Not supported in ROCKSDB_LITE mode!
bool use_direct_reads = false;
// Use O_DIRECT for writing file
// Use O_DIRECT for both reads and writes in background flush and compactions
// When true, we also force new_table_reader_for_compaction_inputs to true.
// Default: false
// Not supported in ROCKSDB_LITE mode!
bool use_direct_writes = false;
bool use_direct_io_for_flush_and_compaction = false;
// If false, fallocate() calls are bypassed
bool allow_fallocate = true;

42
java/rocksjni/options.cc
View File

@ -1052,24 +1052,26 @@ void Java_org_rocksdb_Options_setUseDirectReads(JNIEnv* env, jobject jobj,
/*
* Class: org_rocksdb_Options
* Method: useDirectWrites
* Method: useDirectIoForFlushAndCompaction
* Signature: (J)Z
*/
jboolean Java_org_rocksdb_Options_useDirectWrites(JNIEnv* env, jobject jobj,
jlong jhandle) {
return reinterpret_cast<rocksdb::Options*>(jhandle)->use_direct_writes;
jboolean Java_org_rocksdb_Options_useDirectIoForFlushAndCompaction(
JNIEnv* env, jobject jobj, jlong jhandle) {
return reinterpret_cast<rocksdb::Options*>(jhandle)
->use_direct_io_for_flush_and_compaction;
}
/*
* Class: org_rocksdb_Options
* Method: setUseDirectReads
* Method: setUseDirectIoForFlushAndCompaction
* Signature: (JZ)V
*/
void Java_org_rocksdb_Options_setUseDirectWrites(JNIEnv* env, jobject jobj,
jlong jhandle,
jboolean use_direct_writes) {
reinterpret_cast<rocksdb::Options*>(jhandle)->use_direct_writes =
static_cast<bool>(use_direct_writes);
void Java_org_rocksdb_Options_setUseDirectIoForFlushAndCompaction(
JNIEnv* env, jobject jobj, jlong jhandle,
jboolean use_direct_io_for_flush_and_compaction) {
reinterpret_cast<rocksdb::Options*>(jhandle)
->use_direct_io_for_flush_and_compaction =
static_cast<bool>(use_direct_io_for_flush_and_compaction);
}
/*
@ -4920,12 +4922,13 @@ void Java_org_rocksdb_DBOptions_setUseDirectReads(JNIEnv* env, jobject jobj,
/*
* Class: org_rocksdb_DBOptions
* Method: useDirectWrites
* Method: useDirectIoForFlushAndCompaction
* Signature: (J)Z
*/
jboolean Java_org_rocksdb_DBOptions_useDirectWrites(JNIEnv* env, jobject jobj,
jlong jhandle) {
return reinterpret_cast<rocksdb::DBOptions*>(jhandle)->use_direct_writes;
jboolean Java_org_rocksdb_DBOptions_useDirectIoForFlushAndCompaction(
JNIEnv* env, jobject jobj, jlong jhandle) {
return reinterpret_cast<rocksdb::DBOptions*>(jhandle)
->use_direct_io_for_flush_and_compaction;
}
/*
@ -4933,11 +4936,12 @@ jboolean Java_org_rocksdb_DBOptions_useDirectWrites(JNIEnv* env, jobject jobj,
* Method: setUseDirectReads
* Signature: (JZ)V
*/
void Java_org_rocksdb_DBOptions_setUseDirectWrites(JNIEnv* env, jobject jobj,
jlong jhandle,
jboolean use_direct_writes) {
reinterpret_cast<rocksdb::DBOptions*>(jhandle)->use_direct_writes =
static_cast<bool>(use_direct_writes);
void Java_org_rocksdb_DBOptions_setUseDirectIoForFlushAndCompaction(
JNIEnv* env, jobject jobj, jlong jhandle,
jboolean use_direct_io_for_flush_and_compaction) {
reinterpret_cast<rocksdb::DBOptions*>(jhandle)
->use_direct_io_for_flush_and_compaction =
static_cast<bool>(use_direct_io_for_flush_and_compaction);
}
/*

16
java/src/main/java/org/rocksdb/DBOptions.java
View File

@ -531,17 +531,18 @@ public class DBOptions
}
@Override
public DBOptions setUseDirectWrites(
final boolean useDirectWrites) {
public DBOptions setUseDirectIoForFlushAndCompaction(
final boolean useDirectIoForFlushAndCompaction) {
assert(isOwningHandle());
setUseDirectWrites(nativeHandle_, useDirectWrites);
setUseDirectIoForFlushAndCompaction(nativeHandle_,
useDirectIoForFlushAndCompaction);
return this;
}
@Override
public boolean useDirectWrites() {
public boolean useDirectIoForFlushAndCompaction() {
assert(isOwningHandle());
return useDirectWrites(nativeHandle_);
return useDirectIoForFlushAndCompaction(nativeHandle_);
}
@Override
@ -1025,8 +1026,9 @@ public class DBOptions
private native long manifestPreallocationSize(long handle);
private native void setUseDirectReads(long handle, boolean useDirectReads);
private native boolean useDirectReads(long handle);
private native void setUseDirectWrites(long handle, boolean useDirectWrites);
private native boolean useDirectWrites(long handle);
private native void setUseDirectIoForFlushAndCompaction(
long handle, boolean useDirectIoForFlushAndCompaction);
private native boolean useDirectIoForFlushAndCompaction(long handle);
private native void setAllowFAllocate(final long handle,
final boolean allowFAllocate);
private native boolean allowFAllocate(final long handle);

17
java/src/main/java/org/rocksdb/DBOptionsInterface.java
View File

@ -804,21 +804,24 @@ public interface DBOptionsInterface<T extends DBOptionsInterface> {
boolean useDirectReads();
/**
* Enable the OS to use direct I/O for writing sst tables.
* Enable the OS to use direct reads and writes in flush and
* compaction
* Default: false
*
* @param useDirectWrites if true, then direct write is enabled
* @param useDirectIoForFlushAndCompaction if true, then direct
* I/O will be enabled for background flush and compactions
* @return the instance of the current object.
*/
T setUseDirectWrites(boolean useDirectWrites);
T setUseDirectIoForFlushAndCompaction(boolean useDirectIoForFlushAndCompaction);
/**
* Enable the OS to use direct I/O for writing sst tables.
* Default: false
* Enable the OS to use direct reads and writes in flush and
* compaction
*
* @return if true, then direct writes are enabled
* @return if true, then direct I/O is enabled for flush and
* compaction
*/
boolean useDirectWrites();
boolean useDirectIoForFlushAndCompaction();
/**
* Whether fallocate calls are allowed

14
java/src/main/java/org/rocksdb/Options.java
View File

@ -593,16 +593,17 @@ public class Options extends RocksObject
}
@Override
public Options setUseDirectWrites(final boolean useDirectWrites) {
public Options setUseDirectIoForFlushAndCompaction(
final boolean useDirectIoForFlushAndCompaction) {
assert(isOwningHandle());
setUseDirectWrites(nativeHandle_, useDirectWrites);
setUseDirectIoForFlushAndCompaction(nativeHandle_, useDirectIoForFlushAndCompaction);
return this;
}
@Override
public boolean useDirectWrites() {
public boolean useDirectIoForFlushAndCompaction() {
assert(isOwningHandle());
return useDirectWrites(nativeHandle_);
return useDirectIoForFlushAndCompaction(nativeHandle_);
}
@Override
@ -1621,8 +1622,9 @@ public class Options extends RocksObject
private native long manifestPreallocationSize(long handle);
private native void setUseDirectReads(long handle, boolean useDirectReads);
private native boolean useDirectReads(long handle);
private native void setUseDirectWrites(long handle, boolean useDirectWrites);
private native boolean useDirectWrites(long handle);
private native void setUseDirectIoForFlushAndCompaction(
long handle, boolean useDirectIoForFlushAndCompaction);
private native boolean useDirectIoForFlushAndCompaction(long handle);
private native void setAllowFAllocate(final long handle,
final boolean allowFAllocate);
private native boolean allowFAllocate(final long handle);

6
java/src/test/java/org/rocksdb/DBOptionsTest.java
View File

@ -331,11 +331,11 @@ public class DBOptionsTest {
}
@Test
public void useDirectWrites() {
public void useDirectIoForFlushAndCompaction() {
try(final DBOptions opt = new DBOptions()) {
final boolean boolValue = rand.nextBoolean();
opt.setUseDirectWrites(boolValue);
assertThat(opt.useDirectWrites()).isEqualTo(boolValue);
opt.setUseDirectIoForFlushAndCompaction(boolValue);
assertThat(opt.useDirectIoForFlushAndCompaction()).isEqualTo(boolValue);
}
}

6
java/src/test/java/org/rocksdb/OptionsTest.java
View File

@ -553,11 +553,11 @@ public class OptionsTest {
}
@Test
public void useDirectWrites() {
public void useDirectIoForFlushAndCompaction() {
try(final Options opt = new Options()) {
final boolean boolValue = rand.nextBoolean();
opt.setUseDirectWrites(boolValue);
assertThat(opt.useDirectWrites()).isEqualTo(boolValue);
opt.setUseDirectIoForFlushAndCompaction(boolValue);
assertThat(opt.useDirectIoForFlushAndCompaction()).isEqualTo(boolValue);
}
}

9
options/db_options.cc
View File

@ -53,7 +53,8 @@ ImmutableDBOptions::ImmutableDBOptions(const DBOptions& options)
allow_mmap_reads(options.allow_mmap_reads),
allow_mmap_writes(options.allow_mmap_writes),
use_direct_reads(options.use_direct_reads),
use_direct_writes(options.use_direct_writes),
use_direct_io_for_flush_and_compaction(
options.use_direct_io_for_flush_and_compaction),
allow_fallocate(options.allow_fallocate),
is_fd_close_on_exec(options.is_fd_close_on_exec),
advise_random_on_open(options.advise_random_on_open),
@ -127,8 +128,10 @@ void ImmutableDBOptions::Dump(Logger* log) const {
allow_mmap_writes);
ROCKS_LOG_HEADER(log, " Options.use_direct_reads: %d",
use_direct_reads);
ROCKS_LOG_HEADER(log, " Options.use_direct_writes: %d",
use_direct_writes);
ROCKS_LOG_HEADER(log,
" "
"Options.use_direct_io_for_flush_and_compaction: %d",
use_direct_io_for_flush_and_compaction);
ROCKS_LOG_HEADER(log, " Options.create_missing_column_families: %d",
create_missing_column_families);
ROCKS_LOG_HEADER(log, " Options.db_log_dir: %s",

2
options/db_options.h
View File

@ -48,7 +48,7 @@ struct ImmutableDBOptions {
bool allow_mmap_reads;
bool allow_mmap_writes;
bool use_direct_reads;
bool use_direct_writes;
bool use_direct_io_for_flush_and_compaction;
bool allow_fallocate;
bool is_fd_close_on_exec;
bool advise_random_on_open;

3
options/options.cc
View File

@ -153,7 +153,8 @@ DBOptions::DBOptions(const Options& options)
allow_mmap_reads(options.allow_mmap_reads),
allow_mmap_writes(options.allow_mmap_writes),
use_direct_reads(options.use_direct_reads),
use_direct_writes(options.use_direct_writes),
use_direct_io_for_flush_and_compaction(
options.use_direct_io_for_flush_and_compaction),
allow_fallocate(options.allow_fallocate),
is_fd_close_on_exec(options.is_fd_close_on_exec),
skip_log_error_on_recovery(options.skip_log_error_on_recovery),

3
options/options_helper.cc
View File

@ -70,7 +70,8 @@ DBOptions BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
options.allow_mmap_reads = immutable_db_options.allow_mmap_reads;
options.allow_mmap_writes = immutable_db_options.allow_mmap_writes;
options.use_direct_reads = immutable_db_options.use_direct_reads;
options.use_direct_writes = immutable_db_options.use_direct_writes;
options.use_direct_io_for_flush_and_compaction =
immutable_db_options.use_direct_io_for_flush_and_compaction;
options.allow_fallocate = immutable_db_options.allow_fallocate;
options.is_fd_close_on_exec = immutable_db_options.is_fd_close_on_exec;
options.stats_dump_period_sec = mutable_db_options.stats_dump_period_sec;

6
options/options_helper.h
View File

@ -163,8 +163,10 @@ static std::unordered_map<std::string, OptionTypeInfo> db_options_type_info = {
{offsetof(struct DBOptions, use_direct_reads), OptionType::kBoolean,
OptionVerificationType::kNormal, false, 0}},
{"use_direct_writes",
{offsetof(struct DBOptions, use_direct_writes), OptionType::kBoolean,
OptionVerificationType::kNormal, false, 0}},
{0, OptionType::kBoolean, OptionVerificationType::kDeprecated, false, 0}},
{"use_direct_io_for_flush_and_compaction",
{offsetof(struct DBOptions, use_direct_io_for_flush_and_compaction),
OptionType::kBoolean, OptionVerificationType::kNormal, false, 0}},
{"allow_2pc",
{offsetof(struct DBOptions, allow_2pc), OptionType::kBoolean,
OptionVerificationType::kNormal, false, 0}},

2
options/options_settable_test.cc
View File

@ -274,7 +274,7 @@ TEST_F(OptionsSettableTest, DBOptionsAllFieldsSettable) {
"allow_fallocate=true;"
"allow_mmap_reads=false;"
"use_direct_reads=false;"
"use_direct_writes=false;"
"use_direct_io_for_flush_and_compaction=false;"
"max_log_file_size=4607;"
"random_access_max_buffer_size=1048576;"
"advise_random_on_open=true;"

4
options/options_test.cc
View File

@ -120,7 +120,7 @@ TEST_F(OptionsTest, GetOptionsFromMapTest) {
{"allow_mmap_reads", "true"},
{"allow_mmap_writes", "false"},
{"use_direct_reads", "false"},
{"use_direct_writes", "false"},
{"use_direct_io_for_flush_and_compaction", "false"},
{"is_fd_close_on_exec", "true"},
{"skip_log_error_on_recovery", "false"},
{"stats_dump_period_sec", "46"},
@ -234,7 +234,7 @@ TEST_F(OptionsTest, GetOptionsFromMapTest) {
ASSERT_EQ(new_db_opt.allow_mmap_reads, true);
ASSERT_EQ(new_db_opt.allow_mmap_writes, false);
ASSERT_EQ(new_db_opt.use_direct_reads, false);
ASSERT_EQ(new_db_opt.use_direct_writes, false);
ASSERT_EQ(new_db_opt.use_direct_io_for_flush_and_compaction, false);
ASSERT_EQ(new_db_opt.is_fd_close_on_exec, true);
ASSERT_EQ(new_db_opt.skip_log_error_on_recovery, false);
ASSERT_EQ(new_db_opt.stats_dump_period_sec, 46U);

14
tools/db_bench_tool.cc
View File

@ -800,17 +800,18 @@ DEFINE_uint64(wal_size_limit_MB, 0, "Set the size limit for the WAL Files"
" in MB.");
DEFINE_uint64(max_total_wal_size, 0, "Set total max WAL size");
DEFINE_bool(mmap_read, rocksdb::EnvOptions().use_mmap_reads,
DEFINE_bool(mmap_read, rocksdb::Options().allow_mmap_reads,
"Allow reads to occur via mmap-ing files");
DEFINE_bool(mmap_write, rocksdb::EnvOptions().use_mmap_writes,
DEFINE_bool(mmap_write, rocksdb::Options().allow_mmap_writes,
"Allow writes to occur via mmap-ing files");
DEFINE_bool(use_direct_reads, rocksdb::EnvOptions().use_direct_reads,
DEFINE_bool(use_direct_reads, rocksdb::Options().use_direct_reads,
"Use O_DIRECT for reading data");
DEFINE_bool(use_direct_writes, rocksdb::EnvOptions().use_direct_writes,
"Use O_DIRECT for writing data");
DEFINE_bool(use_direct_io_for_flush_and_compaction,
rocksdb::Options().use_direct_io_for_flush_and_compaction,
"Use O_DIRECT for background flush and compaction I/O");
DEFINE_bool(advise_random_on_open, rocksdb::Options().advise_random_on_open,
"Advise random access on table file open");
@ -2813,7 +2814,8 @@ void VerifyDBFromDB(std::string& truth_db_name) {
options.allow_mmap_reads = FLAGS_mmap_read;
options.allow_mmap_writes = FLAGS_mmap_write;
options.use_direct_reads = FLAGS_use_direct_reads;
options.use_direct_writes = FLAGS_use_direct_writes;
options.use_direct_io_for_flush_and_compaction =
FLAGS_use_direct_io_for_flush_and_compaction;
#ifndef ROCKSDB_LITE
options.compaction_options_fifo = CompactionOptionsFIFO(
FLAGS_fifo_compaction_max_table_files_size_mb * 1024 * 1024);

2
tools/db_bench_tool_test.cc
View File

@ -211,7 +211,7 @@ const std::string options_file_content = R"OPTIONS_FILE(
allow_mmap_reads=false
allow_mmap_writes=false
use_direct_reads=false
use_direct_writes=false
use_direct_io_for_flush_and_compaction=false
stats_dump_period_sec=600
allow_fallocate=true
max_log_file_size=83886080

12
tools/db_stress.cc
View File

@ -269,16 +269,17 @@ DEFINE_string(db, "", "Use the db with the following name.");
DEFINE_bool(verify_checksum, false,
"Verify checksum for every block read from storage");
DEFINE_bool(mmap_read, rocksdb::EnvOptions().use_mmap_reads,
DEFINE_bool(mmap_read, rocksdb::Options().allow_mmap_reads,
"Allow reads to occur via mmap-ing files");
DEFINE_bool(mmap_write, rocksdb::EnvOptions().use_mmap_writes,
DEFINE_bool(mmap_write, rocksdb::Options().allow_mmap_writes,
"Allow writes to occur via mmap-ing files");
DEFINE_bool(use_direct_reads, rocksdb::EnvOptions().use_direct_reads,
DEFINE_bool(use_direct_reads, rocksdb::Options().use_direct_reads,
"Use O_DIRECT for reading data");
DEFINE_bool(use_direct_writes, rocksdb::EnvOptions().use_direct_writes,
DEFINE_bool(use_direct_io_for_flush_and_compaction,
rocksdb::Options().use_direct_io_for_flush_and_compaction,
"Use O_DIRECT for writing data");
// Database statistics
@ -2156,7 +2157,8 @@ class StressTest {
options_.allow_mmap_reads = FLAGS_mmap_read;
options_.allow_mmap_writes = FLAGS_mmap_write;
options_.use_direct_reads = FLAGS_use_direct_reads;
options_.use_direct_writes = FLAGS_use_direct_writes;
options_.use_direct_io_for_flush_and_compaction =
FLAGS_use_direct_io_for_flush_and_compaction;
options_.target_file_size_base = FLAGS_target_file_size_base;
options_.target_file_size_multiplier = FLAGS_target_file_size_multiplier;
options_.max_bytes_for_level_base = FLAGS_max_bytes_for_level_base;

2
util/testutil.cc
View File

@ -242,7 +242,7 @@ void RandomInitDBOptions(DBOptions* db_opt, Random* rnd) {
db_opt->allow_mmap_reads = rnd->Uniform(2);
db_opt->allow_mmap_writes = rnd->Uniform(2);
db_opt->use_direct_reads = rnd->Uniform(2);
db_opt->use_direct_writes = rnd->Uniform(2);
db_opt->use_direct_io_for_flush_and_compaction = rnd->Uniform(2);
db_opt->create_if_missing = rnd->Uniform(2);
db_opt->create_missing_column_families = rnd->Uniform(2);
db_opt->enable_thread_tracking = rnd->Uniform(2);