Calvin proactive proxy2

Cargo.toml

@@ -222,6 +222,8 @@ cmake = { git = "https://github.com/rust-lang/cmake-rs" }
 # remove this when https://github.com/rust-lang/backtrace-rs/pull/503 is merged.
 backtrace = { git = 'https://github.com/hehechen/backtrace-rs', branch = "v0.3.61" }
 
+sysinfo ={ git = "https://github.com/tikv/sysinfo", branch = "0.26-fix-cpu" }
+
 [target.'cfg(target_os = "linux")'.dependencies]
 procinfo = { git = "https://github.com/tikv/procinfo-rs", rev = "6599eb9dca74229b2c1fcc44118bef7eff127128" }
 # When you modify TiKV cooperatively with kvproto, this will be useful to submit the PR to TiKV and the PR to
@@ -424,7 +426,7 @@ opt-level = 1
 
 [profile.dev]
 opt-level = 0
-debug = 0
+debug = false
 codegen-units = 4
 lto = false
 incremental = true

components/raftstore/src/coprocessor/dispatcher.rs

@@ -554,13 +554,20 @@ impl<E: KvEngine> CoprocessorHost<E> {
     }
 
     // (index, term) is for the applying entry.
-    pub fn pre_exec(&self, region: &Region, cmd: &RaftCmdRequest, index: u64, term: u64) -> bool {
+    pub fn pre_exec(
+        &self,
+        region: &Region,
+        cmd: &RaftCmdRequest,
+        index: u64,
+        term: u64,
+        apply_state: &RaftApplyState,
+    ) -> bool {
         let mut ctx = ObserverContext::new(region);
         if !cmd.has_admin_request() {
             let query = cmd.get_requests();
             for observer in &self.registry.query_observers {
                 let observer = observer.observer.inner();
-                if observer.pre_exec_query(&mut ctx, query, index, term) {
+                if observer.pre_exec_query(&mut ctx, query, index, term, apply_state) {
                     return true;
                 }
             }
@@ -569,7 +576,7 @@ impl<E: KvEngine> CoprocessorHost<E> {
             let admin = cmd.get_admin_request();
             for observer in &self.registry.admin_observers {
                 let observer = observer.observer.inner();
-                if observer.pre_exec_admin(&mut ctx, admin, index, term) {
+                if observer.pre_exec_admin(&mut ctx, admin, index, term, apply_state) {
                     return true;
                 }
             }
@@ -846,6 +853,32 @@ impl<E: KvEngine> CoprocessorHost<E> {
         }
     }
 
+    pub fn post_compact_log_from_underlying_engine(
+        &self,
+        region_id: u64,
+        do_write: bool,
+        compact_index: u64,
+        compact_term: u64,
+        max_compact_index: u64,
+        max_compact_term: u64,
+        request_applied_index: u64,
+        raftstore_applied_index: u64,
+    ) {
+        for observer in &self.registry.region_change_observers {
+            let observer = observer.observer.inner();
+            observer.post_compact_log_from_underlying_engine(
+                region_id,
+                do_write,
+                compact_index,
+                compact_term,
+                max_compact_index,
+                max_compact_term,
+                request_applied_index,
+                raftstore_applied_index,
+            );
+        }
+    }
+
     pub fn shutdown(&self) {
         for entry in &self.registry.admin_observers {
             entry.observer.inner().stop();
@@ -947,6 +980,7 @@ mod tests {
             _: &AdminRequest,
             _: u64,
             _: u64,
+            _: &RaftApplyState,
         ) -> bool {
             self.called
                 .fetch_add(ObserverIndex::PreExecAdmin as usize, Ordering::SeqCst);
@@ -1002,6 +1036,7 @@ mod tests {
             _: &[Request],
             _: u64,
             _: u64,
+            _: &RaftApplyState,
         ) -> bool {
             self.called
                 .fetch_add(ObserverIndex::PreExecQuery as usize, Ordering::SeqCst);
@@ -1262,14 +1297,15 @@ mod tests {
         assert_all!([&ob.called], &[index]);
 
         let mut query_req = RaftCmdRequest::default();
+        let apply_state = RaftApplyState::default();
         query_req.set_requests(vec![Request::default()].into());
-        host.pre_exec(&region, &query_req, 0, 0);
+        host.pre_exec(&region, &query_req, 0, 0, &apply_state);
         index += ObserverIndex::PreExecQuery as usize;
         assert_all!([&ob.called], &[index]);
 
         let mut admin_req = RaftCmdRequest::default();
         admin_req.set_admin_request(AdminRequest::default());
-        host.pre_exec(&region, &admin_req, 0, 0);
+        host.pre_exec(&region, &admin_req, 0, 0, &apply_state);
         index += ObserverIndex::PreExecAdmin as usize;
         assert_all!([&ob.called], &[index]);
 

components/raftstore/src/coprocessor/mod.rs

@@ -113,6 +113,7 @@ pub trait AdminObserver: Coprocessor {
         _: &AdminRequest,
         _: u64,
         _: u64,
+        _: &RaftApplyState,
     ) -> bool {
         false
     }
@@ -153,7 +154,14 @@ pub trait QueryObserver: Coprocessor {
 
     /// Hook before exec write request, returns whether we should skip this
     /// write.
-    fn pre_exec_query(&self, _: &mut ObserverContext<'_>, _: &[Request], _: u64, _: u64) -> bool {
+    fn pre_exec_query(
+        &self,
+        _: &mut ObserverContext<'_>,
+        _: &[Request],
+        _: u64,
+        _: u64,
+        _: &RaftApplyState,
+    ) -> bool {
         false
     }
 
@@ -337,6 +345,19 @@ pub trait RegionChangeObserver: Coprocessor {
     fn pre_write_apply_state(&self, _: &mut ObserverContext<'_>) -> bool {
         true
     }
+
+    fn post_compact_log_from_underlying_engine(
+        &self,
+        _region_id: u64,
+        _do_write: bool,
+        _compact_index: u64,
+        _compact_term: u64,
+        _max_compact_index: u64,
+        _max_compact_term: u64,
+        _request_applied_index: u64,
+        _raftstore_applied_index: u64,
+    ) {
+    }
 }
 
 pub trait MessageObserver: Coprocessor {