Helper functions to support direct IO

Summary:
This patch adds the helper functions and variables to allow a backend
implementing WritableFile to support direct IO when persisting a
memtable.

Test Plan:
Since there is no upstream implementation of WritableFile supporting
direct IO, the new behavior is disabled.

Tests should be provided by the backend implementing WritableFile.

include/rocksdb/env.h

@@ -457,6 +457,12 @@ class WritableFile {
     return false;
   }
 
+  // Indicates the upper layers if the current WritableFile implementation
+  // uses direct IO.
+  virtual bool UseDirectIO() const {
+    return false;
+  }
+
   /*
    * Change the priority in rate limiter if rate limiting is enabled.
    * If rate limiting is not enabled, this call has no effect.

util/file_reader_writer.cc

@@ -146,7 +146,7 @@ Status WritableFileWriter::Flush() {
   //     the page.
   // Xfs does neighbor page flushing outside of the specified ranges. We
   // need to make sure sync range is far from the write offset.
-  if (bytes_per_sync_) {
+  if (!direct_io_ && bytes_per_sync_) {
     uint64_t kBytesNotSyncRange = 1024 * 1024;  // recent 1MB is not synced.
     uint64_t kBytesAlignWhenSync = 4 * 1024;    // Align 4KB.
     if (filesize_ > kBytesNotSyncRange) {
@@ -170,7 +170,7 @@ Status WritableFileWriter::Sync(bool use_fsync) {
     return s;
   }
   TEST_KILL_RANDOM(rocksdb_kill_odds);
-  if (pending_sync_) {
+  if (!direct_io_ && pending_sync_) {
     s = SyncInternal(use_fsync);
     if (!s.ok()) {
       return s;

util/file_reader_writer.h

@@ -66,6 +66,7 @@ class WritableFileWriter {
   uint64_t filesize_;
   bool pending_sync_;
   bool pending_fsync_;
+  bool direct_io_;
   uint64_t last_sync_size_;
   uint64_t bytes_per_sync_;
   RateLimiter* rate_limiter_;
@@ -80,6 +81,7 @@ class WritableFileWriter {
         filesize_(0),
         pending_sync_(false),
         pending_fsync_(false),
+        direct_io_(writable_file_->UseDirectIO()),
         last_sync_size_(0),
         bytes_per_sync_(options.bytes_per_sync),
         rate_limiter_(options.rate_limiter) {}