Use a ThreadPool to clean up API throttling in Fixer

Also fix mutex unwrap() in executor.

Addresses #1519

Author: lsegal

qlty-check/Cargo.toml

@@ -32,6 +32,7 @@ globset.workspace = true
 ignore.workspace = true
 indicatif.workspace = true
 itertools.workspace = true
+lazy_static.workspace = true
 lzma-rs.workspace = true
 num_cpus.workspace = true
 once_cell.workspace = true

qlty-check/src/executor.rs

@@ -15,7 +15,7 @@ use crate::{
     ui::{Progress, ProgressBar},
 };
 use crate::{cache::IssuesCacheHit, planner::Plan, Results};
-use anyhow::{bail, Context, Result};
+use anyhow::{anyhow, bail, Context, Result};
 use chrono::Utc;
 pub use driver::Driver;
 use ignore::{DirEntry, WalkBuilder, WalkState};
@@ -640,7 +640,10 @@ fn run_invocation(
                     file_result.issues.len(),
                     MAX_ISSUES_PER_FILE
                 );
-                issue_limit_reached.lock().unwrap().insert(PathBuf::from(&file_result.path));
+                match issue_limit_reached.lock() {
+                    Ok(mut limit) => limit.insert(file_result.path.clone().into()),
+                    Err(_) => { debug!("Poison error in thread"); false },
+                };
                 file_result.issues.truncate(MAX_ISSUES_PER_FILE);
                 file_result.issues.shrink_to_fit();
                 return;
@@ -671,7 +674,9 @@ fn run_invocation(
     progress.increment(plan.workspace_entries.len() as u64);
     task.clear();
 
-    let issue_limit_reached = issue_limit_reached.lock().unwrap();
+    let issue_limit_reached = issue_limit_reached
+        .lock()
+        .map_err(|_| anyhow!("Posion error in thread"))?;
     if !issue_limit_reached.is_empty() {
         result.push_message(
             MessageLevel::Error,

qlty-check/src/llm/fixer.rs

@@ -3,31 +3,35 @@ use crate::source_reader::SourceReader;
 use crate::ui::ProgressBar as _;
 use crate::{executor::staging_area::StagingArea, Progress};
 use anyhow::{bail, Result};
+use lazy_static::lazy_static;
 use qlty_cloud::Client;
 use qlty_config::issue_transformer::IssueTransformer;
 use qlty_types::analysis::v1::{Issue, Suggestion};
+use rayon::{ThreadPool, ThreadPoolBuilder};
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
-use std::thread::sleep;
-use std::time::Duration;
-use tracing::{debug, warn};
-use tracing::{info, trace};
+use tracing::{debug, info, warn};
 use ureq::json;
 
 const MAX_FIXES: usize = 500;
 const MAX_FIXES_PER_FILE: usize = 30;
 const MAX_CONCURRENT_FIXES: usize = 10;
 
+lazy_static! {
+    static ref API_THREAD_POOL: ThreadPool = ThreadPoolBuilder::new()
+        .num_threads(MAX_CONCURRENT_FIXES)
+        .build()
+        .unwrap();
+}
+
 #[derive(Clone, Debug)]
 pub struct Fixer {
     progress: Progress,
     staging_area: StagingArea,
     r#unsafe: bool,
     attempts_per_file: Arc<Mutex<HashMap<Option<String>, AtomicUsize>>>,
     total_attempts: Arc<AtomicUsize>,
-    api_concurrency_lock: Arc<AtomicUsize>,
-    api_concurrency_guard: Arc<Mutex<()>>,
 }
 
 impl IssueTransformer for Fixer {
@@ -55,8 +59,6 @@ impl Fixer {
             r#unsafe: plan.settings.r#unsafe,
             attempts_per_file: Arc::new(Mutex::new(HashMap::new())),
             total_attempts: Arc::new(AtomicUsize::new(0)),
-            api_concurrency_lock: Arc::new(AtomicUsize::new(0)),
-            api_concurrency_guard: Arc::new(Mutex::new(())),
         }
     }
 
@@ -129,18 +131,18 @@ impl Fixer {
         if let Some(path) = issue.path() {
             let client = Client::authenticated()?;
             let content = self.staging_area.read(issue.path().unwrap().into())?;
-            self.try_fix_barrier();
-            let response = client.post("/fixes").send_json(json!({
-                "issue": issue.clone(),
-                "files": [{ "content": content, "path": path }],
-                "options": {
-                    "unsafe": self.r#unsafe
-                },
-            }));
-            self.api_concurrency_lock.fetch_sub(1, Ordering::SeqCst);
+            let response = API_THREAD_POOL.scope(|_| {
+                client.post("/fixes").send_json(json!({
+                    "issue": issue.clone(),
+                    "files": [{ "content": content, "path": path }],
+                    "options": {
+                        "unsafe": self.r#unsafe
+                    },
+                }))
+            })?;
 
             let response_debug = format!("{:?}", &response);
-            let suggestions: Vec<Suggestion> = response?.into_json()?;
+            let suggestions: Vec<Suggestion> = response.into_json()?;
             debug!("{} with {} suggestions", response_debug, suggestions.len());
 
             let mut issue = issue.clone();
@@ -151,14 +153,4 @@ impl Fixer {
             bail!("Issue {} has no path", issue.id);
         }
     }
-
-    fn try_fix_barrier(&self) {
-        let guard = self.api_concurrency_guard.lock().unwrap();
-        while self.api_concurrency_lock.load(Ordering::SeqCst) >= MAX_CONCURRENT_FIXES {
-            sleep(Duration::from_millis(100));
-        }
-        let value = self.api_concurrency_lock.fetch_add(1, Ordering::SeqCst);
-        trace!("API request made with {} concurrent fixes", value);
-        drop(guard);
-    }
 }