Repository: tikv/tikv (language: rust)

Issue:
gc_worker: deadlock in gc_a_round
## Bug Report

when compaction-filter is disable, we will perform GC in a traditional way, and need to GC each region one by one
https://github.com/tikv/tikv/blob/25897b73e1f0e93f616c764dbe65ed7db02edd07/src/server/gc_worker/gc_manager.rs#L516-L528

and here we should notice that we hold the lock for `current_tasks` when we call `async_gc_next_region`, while in the `cb` function, we need the same lock too, which means, we should release the lock before run `cb`

however, when we gc a region with the function `async_gc_next_region` 

https://github.com/tikv/tikv/blob/caf82d2864d1501af733bf60c99a7732e105144b/src/server/gc_worker/gc_manager.rs#L581-L609 

 if we meet any error we will goto `hand_gc_task_schedule_error`
https://github.com/tikv/tikv/blob/25897b73e1f0e93f616c764dbe65ed7db02edd07/src/server/gc_worker/gc_worker.rs#L1139-L1153 

if it is a busy error,we will run the func `callback`

https://github.com/tikv/tikv/blob/caf82d2864d1501af733bf60c99a7732e105144b/src/server/gc_worker/gc_worker.rs#L1139-L1153

### What version of TiKV are you using?
<!-- You can run `tikv-server --version` -->

### What operating system and CPU are you using?
<!-- If you're using Linux, you can run `cat /proc/cpuinfo` -->

### Steps to reproduce
<!-- If possible, provide a recipe for reproducing the error. A complete runnable program is good. -->

### What did you expect?

no deadlock
### What did happened?
GC blocks when meet error like `"failed to schedule gc task"] [err="channel is full"] `



Discussion / hints:

Fix patch:
diff --git a/src/server/gc_worker/gc_manager.rs b/src/server/gc_worker/gc_manager.rs
index d9c5287b67d..1c92142e274 100644
--- a/src/server/gc_worker/gc_manager.rs
+++ b/src/server/gc_worker/gc_manager.rs
@@ -21,7 +21,7 @@ use super::{
     compaction_filter::is_compaction_filter_allowed,
     config::GcWorkerConfigManager,
     gc_worker::{schedule_gc, GcSafePointProvider, GcTask},
-    Result,
+    Error, ErrorInner, Result,
 };
 use crate::{server::metrics::*, storage::Callback, tikv_util::sys::thread::StdThreadBuildWrapper};
 
@@ -514,14 +514,18 @@ impl<S: GcSafePointProvider, R: RegionInfoProvider + 'static, E: KvEngine> GcMan
             self.check_if_need_rewind(&progress, &mut need_rewind, &mut end);
 
             let controller: Arc<(Mutex<usize>, Condvar)> = Arc::clone(&task_controller);
-            let cb = Box::new(move |_res| {
-                let (lock, cvar) = &*controller;
-                let mut current_tasks = lock.lock().unwrap();
-                *current_tasks -= 1;
-                cvar.notify_one();
-                AUTO_GC_PROCESSED_REGIONS_GAUGE_VEC
-                    .with_label_values(&[PROCESS_TYPE_GC])
-                    .inc();
+            let cb = Box::new(move |res| {
+                if let Err(Error(box ErrorInner::GcWorkerTooBusy)) = res {
+                    // the task is actually not scheduled, so no action here.
+                } else {
+                    let (lock, cvar) = &*controller;
+                    let mut current_tasks = lock.lock().unwrap();
+                    *current_tasks -= 1;
+                    cvar.notify_one();
+                    AUTO_GC_PROCESSED_REGIONS_GAUGE_VEC
+                        .with_label_values(&[PROCESS_TYPE_GC])
+                        .inc();
+                }
             });
             maybe_wait(self.max_concurrent_tasks - 1);
             let mut current_tasks = lock.lock().unwrap();
@@ -1010,4 +1014,35 @@ mod tests {
             );
         }
     }
+
+    #[test]
+    fn test_gc_worker_full() {
+        let safe_points = vec![233];
+        let regions: BTreeMap<_, _> = vec![
+            (b"".to_vec(), b"1".to_vec(), 1),
+            (b"1".to_vec(), b"".to_vec(), 2),
+        ]
+        .into_iter()
+        .map(|(start_key, end_key, id)| {
+            let mut r = metapb::Region::default();
+            r.set_id(id);
+            r.set_start_key(start_key.clone());
+            r.set_end_key(end_key);
+            r.mut_peers().push(new_peer(1, 1));
+            let info = RegionInfo::new(r, StateRole::Leader);
+            (start_key, info)
+        })
+        .collect();
+
+        let mut test_util = GcManagerTestUtil::new(regions);
+
+        for safe_point in &safe_points {
+            test_util.add_next_safe_point(*safe_point);
+        }
+        test_util.gc_manager.as_mut().unwrap().initialize();
+
+        fail::cfg("schedule_gc_full", "return").unwrap();
+        test_util.gc_manager.as_mut().unwrap().gc_a_round().unwrap();
+        test_util.stop();
+    }
 }
diff --git a/src/server/gc_worker/gc_worker.rs b/src/server/gc_worker/gc_worker.rs
index a0537a478d0..bd26182ed6c 100644
--- a/src/server/gc_worker/gc_worker.rs
+++ b/src/server/gc_worker/gc_worker.rs
@@ -1130,13 +1130,18 @@ pub fn schedule_gc(
     safe_point: TimeStamp,
     callback: Callback<()>,
 ) -> Result<()> {
-    scheduler
-        .schedule(GcTask::Gc {
-            region,
-            safe_point,
-            callback,
-        })
-        .or_else(handle_gc_task_schedule_error)
+    let task = GcTask::Gc {
+        region,
+        safe_point,
+        callback,
+    };
+    if fail::eval("schedule_gc_full", |_| true).is_some() {
+        handle_gc_task_schedule_error(ScheduleError::Full(task))
+    } else {
+        scheduler
+            .schedule(task)
+            .or_else(handle_gc_task_schedule_error)
+    }
 }
 
 /// Does GC synchronously.


Test patch: