Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Decouple engine_store_ffi and raftstore #157

Merged
merged 5 commits into from
Sep 5, 2022
Merged

Decouple engine_store_ffi and raftstore #157

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
4a23a54
add log
CalvinNeo Sep 2, 2022
5519c0e
remove engine_store_ffi out of other modules of raftstore
CalvinNeo Sep 2, 2022
f1eb69d
f
CalvinNeo Sep 4, 2022
9eca8df
rename thread pool
CalvinNeo Sep 5, 2022
eb3d5e4
f
CalvinNeo Sep 5, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 4 additions & 6 deletions components/raftstore/src/engine_store_ffi/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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 {
Expand Down
16 changes: 11 additions & 5 deletions components/raftstore/src/engine_store_ffi/observer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Expand All @@ -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
}
};
Expand All @@ -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(())
}

Expand Down
3 changes: 0 additions & 3 deletions components/raftstore/src/store/fsm/apply.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
9 changes: 0 additions & 9 deletions components/raftstore/src/store/snap.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
8 changes: 3 additions & 5 deletions components/raftstore/src/store/worker/region.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ use crate::{
JOB_STATUS_CANCELLED, JOB_STATUS_CANCELLING, JOB_STATUS_FAILED, JOB_STATUS_FINISHED,
JOB_STATUS_PENDING, JOB_STATUS_RUNNING,
},
snap::{plain_file_used, Error, PreHandledSnapshot, Result, SNAPSHOT_CFS},
snap::{plain_file_used, Error, Result, SNAPSHOT_CFS},
transport::CasualRouter,
ApplyOptions, CasualMessage, SnapEntry, SnapKey, SnapManager,
},
Expand Down Expand Up @@ -119,8 +119,6 @@ struct EngineStoreApplySnapTask {
region_id: u64,
peer_id: u64,
status: Arc<AtomicUsize>,
recv: mpsc::Receiver<Option<PreHandledSnapshot>>,
pre_handled_snap: Option<PreHandledSnapshot>,
}

#[derive(Clone)]
Expand Down Expand Up @@ -737,8 +735,8 @@ where
pool_size,
pre_handle_snap,
},
pool: Builder::new(thd_name!("region-task"))
.max_thread_count(pool_size)
pool: Builder::new(thd_name!("snap-generator"))
.max_thread_count(snap_generator_pool_size)
Comment on lines +738 to +739
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is also a "snap_handle_pool_size". Not sure whether this should be used or removed.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

snap_generator_pool_size and snap_handle_pool_size are different configs:

  1. snap_generator_pool_size is used by TiKV to generate snapshot
  2. snap_handle_pool_size is used by Proxy to perform row2column work, and is to be decoupled

.build_future_pool(),
ctx: SnapContext {
engine,
Expand Down