Decouple engine_store_ffi and raftstore

components/raftstore/src/engine_store_ffi/lock_cf_reader.rs

@@ -9,10 +9,8 @@ use encryption::DataKeyManager;
 use file_system::File;
 use tikv_util::codec::bytes::CompactBytesFromFileDecoder;
 
-use crate::{
-    engine_store_ffi::interfaces::root::DB::{BaseBuffView, RawVoidPtr},
-    store::snap::snap_io::get_decrypter_reader,
-};
+use self::interfaces::root::DB::{BaseBuffView, RawVoidPtr};
+use crate::store::snap::snap_io::get_decrypter_reader;
 
 type LockCFDecoder = BufReader<Box<dyn Read + Send>>;
 

components/raftstore/src/engine_store_ffi/mod.rs

@@ -27,13 +27,13 @@ use kvproto::{kvrpcpb, metapb, raft_cmdpb};
 use protobuf::Message;
 pub use read_index_helper::ReadIndexClient;
 
-pub use crate::engine_store_ffi::interfaces::root::DB::{
+pub use self::interfaces::root::DB::{
     BaseBuffView, ColumnFamilyType, CppStrVecView, EngineStoreApplyRes, EngineStoreServerHelper,
     EngineStoreServerStatus, FileEncryptionRes, FsStats, HttpRequestRes, HttpRequestStatus,
     KVGetStatus, RaftCmdHeader, RaftProxyStatus, RaftStoreProxyFFIHelper, RawCppPtr,
     RawCppStringPtr, RawVoidPtr, SSTReaderPtr, StoreStats, WriteCmdType, WriteCmdsView,
 };
-use crate::engine_store_ffi::{
+use self::{
     interfaces::root::DB::{
         ConstRawVoidPtr, FileEncryptionInfoRaw, RaftStoreProxyPtr, RawCppPtrType, RawRustPtr,
         SSTReaderInterfaces, SSTView, SSTViewVec, RAFT_STORE_PROXY_MAGIC_NUMBER,
@@ -276,7 +276,7 @@ impl Into<u32> for RawRustPtrType {
 
 pub extern "C" fn ffi_gc_rust_ptr(
     data: RawVoidPtr,
-    type_: crate::engine_store_ffi::interfaces::root::DB::RawRustPtrType,
+    type_: self::interfaces::root::DB::RawRustPtrType,
 ) {
     if data.is_null() {
         return;
@@ -385,9 +385,7 @@ pub extern "C" fn ffi_poll_read_index_task(
             _ => {}
         }
     }
-    let task = unsafe {
-        &mut *(task_ptr as *mut crate::engine_store_ffi::read_index_helper::ReadIndexTask)
-    };
+    let task = unsafe { &mut *(task_ptr as *mut self::read_index_helper::ReadIndexTask) };
     let waker = if std::ptr::null_mut() == waker {
         None
     } else {

components/raftstore/src/engine_store_ffi/observer.rs

@@ -27,6 +27,12 @@ use yatp::{
     task::future::TaskCell,
 };
 
+use self::{
+    gen_engine_store_server_helper,
+    interfaces::root::{DB as ffi_interfaces, DB::EngineStoreApplyRes},
+    name_to_cf, ColumnFamilyType, EngineStoreServerHelper, RaftCmdHeader, RawCppPtr, TiFlashEngine,
+    WriteCmdType, WriteCmds, CF_DEFAULT, CF_LOCK, CF_WRITE,
+};
 use crate::{
     coprocessor,
     coprocessor::{
@@ -35,12 +41,6 @@ use crate::{
         Cmd, Coprocessor, CoprocessorHost, ObserverContext, PdTaskObserver, QueryObserver,
         RegionChangeEvent, RegionChangeObserver, RegionState, StoreSizeInfo,
     },
-    engine_store_ffi::{
-        gen_engine_store_server_helper,
-        interfaces::root::{DB as ffi_interfaces, DB::EngineStoreApplyRes},
-        name_to_cf, ColumnFamilyType, EngineStoreServerHelper, RaftCmdHeader, RawCppPtr,
-        TiFlashEngine, WriteCmdType, WriteCmds, CF_DEFAULT, CF_LOCK, CF_WRITE,
-    },
     store::{check_sst_for_ingestion, snap::plain_file_used, SnapKey},
     Error, Result,
 };
@@ -737,6 +737,7 @@ impl ApplySnapshotObserver for TiFlashObserver {
                         info!("background pre-handle snapshot get error";
                             "snap_key" => ?snap_key,
                             "region" => ?ob_ctx.region(),
+                            "pending" => self.engine.pending_applies_count.load(Ordering::SeqCst),
                         );
                         true
                     }
@@ -754,6 +755,11 @@ impl ApplySnapshotObserver for TiFlashObserver {
             None => {
                 // We can't find background pre-handle task,
                 // maybe we can't get snapshot at that time.
+                info!("pre-handled snapshot not found";
+                    "snap_key" => ?snap_key,
+                    "region" => ?ob_ctx.region(),
+                    "pending" => self.engine.pending_applies_count.load(Ordering::SeqCst),
+                );
                 true
             }
         };
@@ -772,12 +778,12 @@ impl ApplySnapshotObserver for TiFlashObserver {
             );
             self.engine_store_server_helper
                 .apply_pre_handled_snapshot(ptr.0);
+            info!("apply snapshot finished";
+                "peer_id" => ?snap_key,
+                "region" => ?ob_ctx.region(),
+                "pending" => self.engine.pending_applies_count.load(Ordering::SeqCst),
+            );
         }
-        info!("apply snapshot finished";
-            "peer_id" => ?peer_id,
-            "region" => ?ob_ctx.region(),
-            "pending" => self.engine.pending_applies_count.load(Ordering::SeqCst),
-        );
         Ok(())
     }
 

components/raftstore/src/store/fsm/apply.rs

@@ -75,9 +75,6 @@ use crate::{
         ApplyCtxInfo, Cmd, CmdBatch, CmdObserveInfo, CoprocessorHost, ObserveHandle, ObserveLevel,
         RegionState,
     },
-    engine_store_ffi::{
-        ColumnFamilyType, EngineStoreApplyRes, RaftCmdHeader, WriteCmdType, WriteCmds,
-    },
     store::{
         cmd_resp,
         fsm::RaftPollerBuilder,

components/raftstore/src/store/snap.rs

@@ -45,7 +45,6 @@ use tikv_util::{
 
 use crate::{
     coprocessor::CoprocessorHost,
-    engine_store_ffi,
     store::{
         metrics::{
             CfNames, INGEST_SST_DURATION_SECONDS, SNAPSHOT_BUILD_TIME_HISTOGRAM,
@@ -450,14 +449,6 @@ pub struct Snapshot {
     mgr: SnapManagerCore,
 }
 
-pub struct PreHandledSnapshot {
-    pub index: u64,
-    pub term: u64,
-    pub inner: engine_store_ffi::RawCppPtr,
-}
-
-unsafe impl Send for PreHandledSnapshot {}
-
 #[derive(PartialEq, Eq, Clone, Copy)]
 enum CheckPolicy {
     ErrAllowed,