Port file system inspector (#575)

port related interface to rust, add `file_system_inspected_env` factory method.
Signed-off-by: tabokie <xy.tao@outlook.com>

librocksdb_sys/bindings/x86_64-unknown-linux-gnu-bindings.rs

@@ -3,14 +3,14 @@
 pub const __GNUC_VA_LIST: u32 = 1;
 pub const _STDINT_H: u32 = 1;
 pub const _FEATURES_H: u32 = 1;
-pub const _DEFAULT_SOURCE: u32 = 1;
-pub const __GLIBC_USE_ISOC2X: u32 = 0;
+pub const _BSD_SOURCE: u32 = 1;
+pub const _SVID_SOURCE: u32 = 1;
 pub const __USE_ISOC11: u32 = 1;
 pub const __USE_ISOC99: u32 = 1;
 pub const __USE_ISOC95: u32 = 1;
-pub const __USE_POSIX_IMPLICITLY: u32 = 1;
 pub const _POSIX_SOURCE: u32 = 1;
 pub const _POSIX_C_SOURCE: u32 = 200809;
+pub const __USE_POSIX_IMPLICITLY: u32 = 1;
 pub const __USE_POSIX: u32 = 1;
 pub const __USE_POSIX2: u32 = 1;
 pub const __USE_POSIX199309: u32 = 1;
@@ -19,42 +19,23 @@ pub const __USE_XOPEN2K: u32 = 1;
 pub const __USE_XOPEN2K8: u32 = 1;
 pub const _ATFILE_SOURCE: u32 = 1;
 pub const __USE_MISC: u32 = 1;
+pub const __USE_BSD: u32 = 1;
+pub const __USE_SVID: u32 = 1;
 pub const __USE_ATFILE: u32 = 1;
 pub const __USE_FORTIFY_LEVEL: u32 = 0;
-pub const __GLIBC_USE_DEPRECATED_GETS: u32 = 0;
-pub const __GLIBC_USE_DEPRECATED_SCANF: u32 = 0;
 pub const _STDC_PREDEF_H: u32 = 1;
 pub const __STDC_IEC_559__: u32 = 1;
 pub const __STDC_IEC_559_COMPLEX__: u32 = 1;
-pub const __STDC_ISO_10646__: u32 = 201706;
+pub const __STDC_ISO_10646__: u32 = 201103;
+pub const __STDC_NO_THREADS__: u32 = 1;
 pub const __GNU_LIBRARY__: u32 = 6;
 pub const __GLIBC__: u32 = 2;
-pub const __GLIBC_MINOR__: u32 = 31;
+pub const __GLIBC_MINOR__: u32 = 18;
 pub const _SYS_CDEFS_H: u32 = 1;
-pub const __glibc_c99_flexarr_available: u32 = 1;
 pub const __WORDSIZE: u32 = 64;
 pub const __WORDSIZE_TIME64_COMPAT32: u32 = 1;
 pub const __SYSCALL_WORDSIZE: u32 = 64;
-pub const __LONG_DOUBLE_USES_FLOAT128: u32 = 0;
-pub const __HAVE_GENERIC_SELECTION: u32 = 1;
-pub const __GLIBC_USE_LIB_EXT2: u32 = 0;
-pub const __GLIBC_USE_IEC_60559_BFP_EXT: u32 = 0;
-pub const __GLIBC_USE_IEC_60559_BFP_EXT_C2X: u32 = 0;
-pub const __GLIBC_USE_IEC_60559_FUNCS_EXT: u32 = 0;
-pub const __GLIBC_USE_IEC_60559_FUNCS_EXT_C2X: u32 = 0;
-pub const __GLIBC_USE_IEC_60559_TYPES_EXT: u32 = 0;
-pub const _BITS_TYPES_H: u32 = 1;
-pub const __TIMESIZE: u32 = 64;
-pub const _BITS_TYPESIZES_H: u32 = 1;
-pub const __OFF_T_MATCHES_OFF64_T: u32 = 1;
-pub const __INO_T_MATCHES_INO64_T: u32 = 1;
-pub const __RLIM_T_MATCHES_RLIM64_T: u32 = 1;
-pub const __STATFS_MATCHES_STATFS64: u32 = 1;
-pub const __FD_SETSIZE: u32 = 1024;
-pub const _BITS_TIME64_H: u32 = 1;
 pub const _BITS_WCHAR_H: u32 = 1;
-pub const _BITS_STDINT_INTN_H: u32 = 1;
-pub const _BITS_STDINT_UINTN_H: u32 = 1;
 pub const INT8_MIN: i32 = -128;
 pub const INT16_MIN: i32 = -32768;
 pub const INT32_MIN: i32 = -2147483648;
@@ -140,103 +121,14 @@ fn bindgen_test_layout_max_align_t() {
         )
     );
 }
-pub type __u_char = libc::c_uchar;
-pub type __u_short = libc::c_ushort;
-pub type __u_int = libc::c_uint;
-pub type __u_long = libc::c_ulong;
-pub type __int8_t = libc::c_schar;
-pub type __uint8_t = libc::c_uchar;
-pub type __int16_t = libc::c_short;
-pub type __uint16_t = libc::c_ushort;
-pub type __int32_t = libc::c_int;
-pub type __uint32_t = libc::c_uint;
-pub type __int64_t = libc::c_long;
-pub type __uint64_t = libc::c_ulong;
-pub type __int_least8_t = __int8_t;
-pub type __uint_least8_t = __uint8_t;
-pub type __int_least16_t = __int16_t;
-pub type __uint_least16_t = __uint16_t;
-pub type __int_least32_t = __int32_t;
-pub type __uint_least32_t = __uint32_t;
-pub type __int_least64_t = __int64_t;
-pub type __uint_least64_t = __uint64_t;
-pub type __quad_t = libc::c_long;
-pub type __u_quad_t = libc::c_ulong;
-pub type __intmax_t = libc::c_long;
-pub type __uintmax_t = libc::c_ulong;
-pub type __dev_t = libc::c_ulong;
-pub type __uid_t = libc::c_uint;
-pub type __gid_t = libc::c_uint;
-pub type __ino_t = libc::c_ulong;
-pub type __ino64_t = libc::c_ulong;
-pub type __mode_t = libc::c_uint;
-pub type __nlink_t = libc::c_ulong;
-pub type __off_t = libc::c_long;
-pub type __off64_t = libc::c_long;
-pub type __pid_t = libc::c_int;
-#[repr(C)]
-#[derive(Debug, Copy, Clone)]
-pub struct __fsid_t {
-    pub __val: [libc::c_int; 2usize],
-}
-#[test]
-fn bindgen_test_layout___fsid_t() {
-    assert_eq!(
-        ::std::mem::size_of::<__fsid_t>(),
-        8usize,
-        concat!("Size of: ", stringify!(__fsid_t))
-    );
-    assert_eq!(
-        ::std::mem::align_of::<__fsid_t>(),
-        4usize,
-        concat!("Alignment of ", stringify!(__fsid_t))
-    );
-    assert_eq!(
-        unsafe { &(*(::std::ptr::null::<__fsid_t>())).__val as *const _ as usize },
-        0usize,
-        concat!(
-            "Offset of field: ",
-            stringify!(__fsid_t),
-            "::",
-            stringify!(__val)
-        )
-    );
-}
-pub type __clock_t = libc::c_long;
-pub type __rlim_t = libc::c_ulong;
-pub type __rlim64_t = libc::c_ulong;
-pub type __id_t = libc::c_uint;
-pub type __time_t = libc::c_long;
-pub type __useconds_t = libc::c_uint;
-pub type __suseconds_t = libc::c_long;
-pub type __daddr_t = libc::c_int;
-pub type __key_t = libc::c_int;
-pub type __clockid_t = libc::c_int;
-pub type __timer_t = *mut libc::c_void;
-pub type __blksize_t = libc::c_long;
-pub type __blkcnt_t = libc::c_long;
-pub type __blkcnt64_t = libc::c_long;
-pub type __fsblkcnt_t = libc::c_ulong;
-pub type __fsblkcnt64_t = libc::c_ulong;
-pub type __fsfilcnt_t = libc::c_ulong;
-pub type __fsfilcnt64_t = libc::c_ulong;
-pub type __fsword_t = libc::c_long;
-pub type __ssize_t = libc::c_long;
-pub type __syscall_slong_t = libc::c_long;
-pub type __syscall_ulong_t = libc::c_ulong;
-pub type __loff_t = __off64_t;
-pub type __caddr_t = *mut libc::c_char;
-pub type __intptr_t = libc::c_long;
-pub type __socklen_t = libc::c_uint;
-pub type __sig_atomic_t = libc::c_int;
-pub type int_least8_t = __int_least8_t;
-pub type int_least16_t = __int_least16_t;
-pub type int_least32_t = __int_least32_t;
-pub type int_least64_t = __int_least64_t;
-pub type uint_least8_t = __uint_least8_t;
-pub type uint_least16_t = __uint_least16_t;
-pub type uint_least32_t = __uint_least32_t;
-pub type uint_least64_t = __uint_least64_t;
+pub type int_least8_t = libc::c_schar;
+pub type int_least16_t = libc::c_short;
+pub type int_least32_t = libc::c_int;
+pub type int_least64_t = libc::c_long;
+pub type uint_least8_t = libc::c_uchar;
+pub type uint_least16_t = libc::c_ushort;
+pub type uint_least32_t = libc::c_uint;
+pub type uint_least64_t = libc::c_ulong;
 pub type int_fast8_t = libc::c_schar;
 pub type int_fast16_t = libc::c_long;
 pub type int_fast32_t = libc::c_long;
@@ -245,8 +137,8 @@ pub type uint_fast8_t = libc::c_uchar;
 pub type uint_fast16_t = libc::c_ulong;
 pub type uint_fast32_t = libc::c_ulong;
 pub type uint_fast64_t = libc::c_ulong;
-pub type intmax_t = __intmax_t;
-pub type uintmax_t = __uintmax_t;
+pub type intmax_t = libc::c_long;
+pub type uintmax_t = libc::c_ulong;
 #[repr(C)]
 #[derive(Debug, Copy, Clone)]
 pub struct crocksdb_cloud_envoptions_t {
@@ -637,6 +529,11 @@ pub const crocksdb_backgrounderrorreason_t_kCompaction: crocksdb_backgrounderror
 pub const crocksdb_backgrounderrorreason_t_kWriteCallback: crocksdb_backgrounderrorreason_t = 3;
 pub const crocksdb_backgrounderrorreason_t_kMemTable: crocksdb_backgrounderrorreason_t = 4;
 pub type crocksdb_backgrounderrorreason_t = u32;
+#[repr(C)]
+#[derive(Debug, Copy, Clone)]
+pub struct crocksdb_file_system_inspector_t {
+    _unused: [u8; 0],
+}
 extern "C" {
     pub fn crocksdb_open(
         options: *const crocksdb_options_t,
@@ -2874,6 +2771,19 @@ extern "C" {
         context: *mut crocksdb_compactionfiltercontext_t,
     ) -> libc::c_uchar;
 }
+extern "C" {
+    pub fn crocksdb_compactionfiltercontext_file_numbers(
+        context: *mut crocksdb_compactionfiltercontext_t,
+        buffer: *mut *const u64,
+        len: *mut usize,
+    );
+}
+extern "C" {
+    pub fn crocksdb_compactionfiltercontext_table_properties(
+        context: *mut crocksdb_compactionfiltercontext_t,
+        offset: usize,
+    ) -> *mut crocksdb_table_properties_t;
+}
 extern "C" {
     pub fn crocksdb_compactionfilterfactory_create(
         state: *mut libc::c_void,
@@ -3323,6 +3233,51 @@ extern "C" {
 extern "C" {
     pub fn crocksdb_sequential_file_destroy(arg1: *mut crocksdb_sequential_file_t);
 }
+pub type crocksdb_file_system_inspector_read_cb = ::std::option::Option<
+    unsafe extern "C" fn(
+        state: *mut libc::c_void,
+        len: usize,
+        errptr: *mut *mut libc::c_char,
+    ) -> usize,
+>;
+pub type crocksdb_file_system_inspector_write_cb = ::std::option::Option<
+    unsafe extern "C" fn(
+        state: *mut libc::c_void,
+        len: usize,
+        errptr: *mut *mut libc::c_char,
+    ) -> usize,
+>;
+extern "C" {
+    pub fn crocksdb_file_system_inspector_create(
+        state: *mut libc::c_void,
+        destructor: ::std::option::Option<unsafe extern "C" fn(arg1: *mut libc::c_void)>,
+        read: crocksdb_file_system_inspector_read_cb,
+        write: crocksdb_file_system_inspector_write_cb,
+    ) -> *mut crocksdb_file_system_inspector_t;
+}
+extern "C" {
+    pub fn crocksdb_file_system_inspector_destroy(arg1: *mut crocksdb_file_system_inspector_t);
+}
+extern "C" {
+    pub fn crocksdb_file_system_inspector_read(
+        inspector: *mut crocksdb_file_system_inspector_t,
+        len: usize,
+        errptr: *mut *mut libc::c_char,
+    ) -> usize;
+}
+extern "C" {
+    pub fn crocksdb_file_system_inspector_write(
+        inspector: *mut crocksdb_file_system_inspector_t,
+        len: usize,
+        errptr: *mut *mut libc::c_char,
+    ) -> usize;
+}
+extern "C" {
+    pub fn crocksdb_file_system_inspected_env_create(
+        arg1: *mut crocksdb_env_t,
+        arg2: *mut crocksdb_file_system_inspector_t,
+    ) -> *mut crocksdb_env_t;
+}
 extern "C" {
     pub fn crocksdb_sstfilereader_create(
         io_options: *const crocksdb_options_t,

librocksdb_sys/crocksdb/c.cc

@@ -23,6 +23,7 @@
 #include "rocksdb/encryption.h"
 #include "rocksdb/env.h"
 #include "rocksdb/env_encryption.h"
+#include "rocksdb/env_inspected.h"
 #include "rocksdb/filter_policy.h"
 #include "rocksdb/iostats_context.h"
 #include "rocksdb/iterator.h"
@@ -198,6 +199,8 @@ using rocksdb::encryption::KeyManager;
 using rocksdb::encryption::NewKeyManagedEncryptedEnv;
 #endif
 
+using rocksdb::FileSystemInspector;
+using rocksdb::NewFileSystemInspectedEnv;
 using std::shared_ptr;
 
 extern "C" {
@@ -665,6 +668,10 @@ struct crocksdb_sst_partitioner_factory_t {
   std::shared_ptr<SstPartitionerFactory> rep;
 };
 
+struct crocksdb_file_system_inspector_t {
+  std::shared_ptr<FileSystemInspector> rep;
+};
+
 static bool SaveError(char** errptr, const Status& s) {
   assert(errptr != nullptr);
   if (s.ok()) {
@@ -4093,6 +4100,89 @@ crocksdb_env_t* crocksdb_key_managed_encrypted_env_create(
 }
 #endif
 
+struct crocksdb_file_system_inspector_impl_t : public FileSystemInspector {
+  void* state;
+  void (*destructor)(void*);
+  crocksdb_file_system_inspector_read_cb read;
+  crocksdb_file_system_inspector_write_cb write;
+
+  virtual ~crocksdb_file_system_inspector_impl_t() { destructor(state); }
+
+  Status Read(size_t len, size_t* allowed) {
+    assert(allowed);
+    char* err = nullptr;
+    *allowed = read(state, len, &err);
+    if (err) {
+      Status s = Status::IOError(err);
+      // malloc-ed by strdup
+      free(err);
+      return s;
+    } else {
+      return Status::OK();
+    }
+  }
+
+  Status Write(size_t len, size_t* allowed) {
+    assert(allowed);
+    char* err = nullptr;
+    *allowed = write(state, len, &err);
+    if (err) {
+      Status s = Status::IOError(err);
+      // malloc-ed by strdup
+      free(err);
+      return s;
+    } else {
+      return Status::OK();
+    }
+  }
+};
+
+crocksdb_file_system_inspector_t* crocksdb_file_system_inspector_create(
+    void* state, void (*destructor)(void*),
+    crocksdb_file_system_inspector_read_cb read,
+    crocksdb_file_system_inspector_write_cb write) {
+  std::shared_ptr<crocksdb_file_system_inspector_impl_t> inspector_impl =
+      std::make_shared<crocksdb_file_system_inspector_impl_t>();
+  inspector_impl->state = state;
+  inspector_impl->destructor = destructor;
+  inspector_impl->read = read;
+  inspector_impl->write = write;
+  crocksdb_file_system_inspector_t* inspector =
+      new crocksdb_file_system_inspector_t;
+  inspector->rep = inspector_impl;
+  return inspector;
+}
+
+void crocksdb_file_system_inspector_destroy(
+    crocksdb_file_system_inspector_t* inspector) {
+  delete inspector;
+}
+
+size_t crocksdb_file_system_inspector_read(
+    crocksdb_file_system_inspector_t* inspector, size_t len, char** errptr) {
+  assert(inspector != nullptr && inspector->rep != nullptr);
+  size_t allowed = 0;
+  SaveError(errptr, inspector->rep->Read(len, &allowed));
+  return allowed;
+}
+
+size_t crocksdb_file_system_inspector_write(
+    crocksdb_file_system_inspector_t* inspector, size_t len, char** errptr) {
+  assert(inspector != nullptr && inspector->rep != nullptr);
+  size_t allowed = 0;
+  SaveError(errptr, inspector->rep->Write(len, &allowed));
+  return allowed;
+}
+
+crocksdb_env_t* crocksdb_file_system_inspected_env_create(
+    crocksdb_env_t* base_env, crocksdb_file_system_inspector_t* inspector) {
+  assert(base_env != nullptr);
+  assert(inspector != nullptr);
+  crocksdb_env_t* result = new crocksdb_env_t;
+  result->rep = NewFileSystemInspectedEnv(base_env->rep, inspector->rep);
+  return result;
+}
+
 crocksdb_sstfilereader_t* crocksdb_sstfilereader_create(
     const crocksdb_options_t* io_options) {
   auto reader = new crocksdb_sstfilereader_t;

librocksdb_sys/crocksdb/crocksdb/c.h

@@ -214,6 +214,9 @@ typedef struct crocksdb_encryption_key_manager_t
     crocksdb_encryption_key_manager_t;
 #endif
 
+typedef struct crocksdb_file_system_inspector_t
+    crocksdb_file_system_inspector_t;
+
 /* DB operations */
 
 extern C_ROCKSDB_LIBRARY_API crocksdb_t* crocksdb_open(
@@ -1621,6 +1624,31 @@ crocksdb_key_managed_encrypted_env_create(crocksdb_env_t*,
                                           crocksdb_encryption_key_manager_t*);
 #endif
 
+/* FileSystemInspectedEnv */
+
+typedef size_t (*crocksdb_file_system_inspector_read_cb)(void* state,
+                                                         size_t len,
+                                                         char** errptr);
+typedef size_t (*crocksdb_file_system_inspector_write_cb)(void* state,
+                                                          size_t len,
+                                                          char** errptr);
+
+extern C_ROCKSDB_LIBRARY_API crocksdb_file_system_inspector_t*
+crocksdb_file_system_inspector_create(
+    void* state, void (*destructor)(void*),
+    crocksdb_file_system_inspector_read_cb read,
+    crocksdb_file_system_inspector_write_cb write);
+extern C_ROCKSDB_LIBRARY_API void crocksdb_file_system_inspector_destroy(
+    crocksdb_file_system_inspector_t*);
+extern C_ROCKSDB_LIBRARY_API size_t crocksdb_file_system_inspector_read(
+    crocksdb_file_system_inspector_t* inspector, size_t len, char** errptr);
+extern C_ROCKSDB_LIBRARY_API size_t crocksdb_file_system_inspector_write(
+    crocksdb_file_system_inspector_t* inspector, size_t len, char** errptr);
+
+extern C_ROCKSDB_LIBRARY_API crocksdb_env_t*
+crocksdb_file_system_inspected_env_create(crocksdb_env_t*,
+                                          crocksdb_file_system_inspector_t*);
+
 /* SstFile */
 
 extern C_ROCKSDB_LIBRARY_API crocksdb_sstfilereader_t*

rocksdb @ a487d256

-Subproject commit 93e89a58edb820f2daa15362120617e49ef6ddd6
+Subproject commit a487d256ee927afc17ffe2dd863e4769c4ae8f3b

librocksdb_sys/src/lib.rs

@@ -173,6 +173,8 @@ pub struct DBSstPartitionerContext(c_void);
 pub struct DBSstPartitionerFactory(c_void);
 #[repr(C)]
 pub struct DBWriteBatchIterator(c_void);
+#[repr(C)]
+pub struct DBFileSystemInspectorInstance(c_void);
 
 #[derive(Copy, Clone, Debug, Eq, PartialEq)]
 #[repr(C)]
@@ -1717,6 +1719,30 @@ extern "C" {
         key_manager: *mut DBEncryptionKeyManagerInstance,
     ) -> *mut DBEnv;
 
+    // FileSystemInspectedEnv
+    pub fn crocksdb_file_system_inspector_create(
+        state: *mut c_void,
+        destructor: extern "C" fn(*mut c_void),
+        read: extern "C" fn(*mut c_void, size_t, *mut *mut c_char) -> size_t,
+        write: extern "C" fn(*mut c_void, size_t, *mut *mut c_char) -> size_t,
+    ) -> *mut DBFileSystemInspectorInstance;
+    pub fn crocksdb_file_system_inspector_destroy(inspector: *mut DBFileSystemInspectorInstance);
+    pub fn crocksdb_file_system_inspector_read(
+        inspector: *mut DBFileSystemInspectorInstance,
+        len: size_t,
+        errptr: *mut *mut c_char,
+    ) -> size_t;
+    pub fn crocksdb_file_system_inspector_write(
+        inspector: *mut DBFileSystemInspectorInstance,
+        len: size_t,
+        errptr: *mut *mut c_char,
+    ) -> size_t;
+
+    pub fn crocksdb_file_system_inspected_env_create(
+        base_env: *mut DBEnv,
+        inspector: *mut DBFileSystemInspectorInstance,
+    ) -> *mut DBEnv;
+
     // SstFileReader
     pub fn crocksdb_sstfilereader_create(io_options: *const Options) -> *mut SstFileReader;
 

src/encryption.rs

@@ -194,7 +194,7 @@ unsafe impl Sync for DBEncryptionKeyManager {}
 impl DBEncryptionKeyManager {
     pub fn new(key_manager: Arc<dyn EncryptionKeyManager>) -> DBEncryptionKeyManager {
         // Size of Arc<dyn T>::into_raw is of 128-bits, which couldn't be used as C-style pointer.
-        // Bixing it to make a 64-bits pointer.
+        // Boxing it to make a 64-bits pointer.
         let ctx = Box::into_raw(Box::new(key_manager)) as *mut c_void;
         let instance = unsafe {
             crocksdb_ffi::crocksdb_encryption_key_manager_create(

src/file_system.rs

+// Copyright 2020 TiKV Project Authors. Licensed under Apache-2.0.
+
+pub use crocksdb_ffi::{self, DBFileSystemInspectorInstance};
+
+use libc::{c_char, c_void, size_t, strdup};
+use std::sync::Arc;
+
+// Inspect global IO flow. No per-file inspection for now.
+pub trait FileSystemInspector: Sync + Send {
+    fn read(&self, len: usize) -> Result<usize, String>;
+    fn write(&self, len: usize) -> Result<usize, String>;
+}
+
+extern "C" fn file_system_inspector_destructor(ctx: *mut c_void) {
+    unsafe {
+        // Recover from raw pointer and implicitly drop.
+        Box::from_raw(ctx as *mut Arc<dyn FileSystemInspector>);
+    }
+}
+
+extern "C" fn file_system_inspector_read(
+    ctx: *mut c_void,
+    len: size_t,
+    errptr: *mut *mut c_char,
+) -> size_t {
+    let file_system_inspector = unsafe { &*(ctx as *mut Arc<dyn FileSystemInspector>) };
+    match file_system_inspector.read(len) {
+        Ok(ret) => ret,
+        Err(e) => {
+            unsafe {
+                *errptr = strdup(e.as_ptr() as *const c_char);
+            }
+            0
+        }
+    }
+}
+
+extern "C" fn file_system_inspector_write(
+    ctx: *mut c_void,
+    len: size_t,
+    errptr: *mut *mut c_char,
+) -> size_t {
+    let file_system_inspector = unsafe { &*(ctx as *mut Arc<dyn FileSystemInspector>) };
+    match file_system_inspector.write(len) {
+        Ok(ret) => ret,
+        Err(e) => {
+            unsafe {
+                *errptr = strdup(e.as_ptr() as *const c_char);
+            }
+            0
+        }
+    }
+}
+
+pub struct DBFileSystemInspector {
+    pub inner: *mut DBFileSystemInspectorInstance,
+}
+
+unsafe impl Send for DBFileSystemInspector {}
+unsafe impl Sync for DBFileSystemInspector {}
+
+impl DBFileSystemInspector {
+    pub fn new(file_system_inspector: Arc<dyn FileSystemInspector>) -> DBFileSystemInspector {
+        // Size of Arc<dyn T>::into_raw is of 128-bits, which couldn't be used as C-style pointer.
+        // Boxing it to make a 64-bits pointer.
+        let ctx = Box::into_raw(Box::new(file_system_inspector)) as *mut c_void;
+        let instance = unsafe {
+            crocksdb_ffi::crocksdb_file_system_inspector_create(
+                ctx,
+                file_system_inspector_destructor,
+                file_system_inspector_read,
+                file_system_inspector_write,
+            )
+        };
+        DBFileSystemInspector { inner: instance }
+    }
+}
+
+impl Drop for DBFileSystemInspector {
+    fn drop(&mut self) {
+        unsafe {
+            crocksdb_ffi::crocksdb_file_system_inspector_destroy(self.inner);
+        }
+    }
+}
+
+#[cfg(test)]
+impl FileSystemInspector for DBFileSystemInspector {
+    fn read(&self, len: usize) -> Result<usize, String> {
+        let ret = unsafe { ffi_try!(crocksdb_file_system_inspector_read(self.inner, len)) };
+        Ok(ret)
+    }
+    fn write(&self, len: usize) -> Result<usize, String> {
+        let ret = unsafe { ffi_try!(crocksdb_file_system_inspector_write(self.inner, len)) };
+        Ok(ret)
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::sync::{Arc, Mutex};
+
+    use super::*;
+
+    struct TestDrop {
+        called: Arc<AtomicUsize>,
+    }
+
+    impl Drop for TestDrop {
+        fn drop(&mut self) {
+            self.called.fetch_add(1, Ordering::SeqCst);
+        }
+    }
+
+    struct TestFileSystemInspector {
+        pub refill_bytes: usize,
+        pub read_called: usize,
+        pub write_called: usize,
+        pub drop: Option<TestDrop>,
+    }
+
+    impl Default for TestFileSystemInspector {
+        fn default() -> Self {
+            TestFileSystemInspector {
+                refill_bytes: 0,
+                read_called: 0,
+                write_called: 0,
+                drop: None,
+            }
+        }
+    }
+
+    impl FileSystemInspector for Mutex<TestFileSystemInspector> {
+        fn read(&self, len: usize) -> Result<usize, String> {
+            let mut inner = self.lock().unwrap();
+            inner.read_called += 1;
+            if len <= inner.refill_bytes {
+                Ok(len)
+            } else {
+                Err("request exceeds refill bytes".into())
+            }
+        }
+        fn write(&self, len: usize) -> Result<usize, String> {
+            let mut inner = self.lock().unwrap();
+            inner.write_called += 1;
+            if len <= inner.refill_bytes {
+                Ok(len)
+            } else {
+                Err("request exceeds refill bytes".into())
+            }
+        }
+    }
+
+    #[test]
+    fn test_create_and_destroy_inspector() {
+        let drop_called = Arc::new(AtomicUsize::new(0));
+        let fs_inspector = Arc::new(Mutex::new(TestFileSystemInspector {
+            drop: Some(TestDrop {
+                called: drop_called.clone(),
+            }),
+            ..Default::default()
+        }));
+        let db_fs_inspector = DBFileSystemInspector::new(fs_inspector.clone());
+        drop(fs_inspector);
+        assert_eq!(0, drop_called.load(Ordering::SeqCst));
+        drop(db_fs_inspector);
+        assert_eq!(1, drop_called.load(Ordering::SeqCst));
+    }
+
+    #[test]
+    fn test_inspected_operation() {
+        let fs_inspector = Arc::new(Mutex::new(TestFileSystemInspector {
+            refill_bytes: 4,
+            ..Default::default()
+        }));
+        let db_fs_inspector = DBFileSystemInspector::new(fs_inspector.clone());
+        assert_eq!(2, db_fs_inspector.read(2).unwrap());
+        assert!(db_fs_inspector.read(8).is_err());
+        assert_eq!(2, db_fs_inspector.write(2).unwrap());
+        assert!(db_fs_inspector.write(8).is_err());
+        let record = fs_inspector.lock().unwrap();
+        assert_eq!(2, record.read_called);
+        assert_eq!(2, record.write_called);
+    }
+}

src/lib.rs

@@ -86,6 +86,7 @@ pub mod comparator;
 #[cfg(feature = "encryption")]
 mod encryption;
 mod event_listener;
+mod file_system;
 pub mod logger;
 pub mod merge_operator;
 mod metadata;

src/rocksdb.rs

@@ -42,6 +42,7 @@ use std::{fs, ptr, slice};
 use cloud::CloudEnvOptions;
 #[cfg(feature = "encryption")]
 use encryption::{DBEncryptionKeyManager, EncryptionKeyManager};
+use file_system::{DBFileSystemInspector, FileSystemInspector};
 use table_properties::{TableProperties, TablePropertiesCollection};
 use table_properties_rc::TablePropertiesCollection as RcTablePropertiesCollection;
 use titan::TitanDBOptions;
@@ -2670,6 +2671,23 @@ impl Env {
         })
     }
 
+    pub fn new_file_system_inspected_env(
+        base_env: Arc<Env>,
+        file_system_inspector: Arc<dyn FileSystemInspector>,
+    ) -> Result<Env, String> {
+        let db_file_system_inspector = DBFileSystemInspector::new(file_system_inspector);
+        let env = unsafe {
+            crocksdb_ffi::crocksdb_file_system_inspected_env_create(
+                base_env.inner,
+                db_file_system_inspector.inner,
+            )
+        };
+        Ok(Env {
+            inner: env,
+            base: Some(base_env),
+        })
+    }
+
     pub fn new_sequential_file(
         &self,
         path: &str,