Skip to content

fix: batch save concurrent limit #2044

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

Merged
benjamin-747 merged 1 commit into from
Apr 7, 2026
Merged

fix: batch save concurrent limit #2044

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
102 changes: 100 additions & 2 deletions ceres/src/pack/import_repo.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ use std::{
Arc, Mutex,
atomic::{AtomicUsize, Ordering},
},
time::Instant,
};

use async_recursion::async_recursion;
Expand Down Expand Up @@ -47,6 +48,7 @@ pub struct ImportRepo {
pub command_list: Mutex<Vec<RefCommand>>,
pub unpack_redlock: Arc<RedLock>,
pub git_object_cache: Arc<GitObjectCache>,
pub receive_pack_extra_timings_ms: Mutex<Vec<(String, u128)>>,
}

#[async_trait]
Expand All @@ -55,6 +57,19 @@ impl RepoHandler for ImportRepo {
false
}

fn save_entry_concurrency(&self) -> usize {
self.storage.config().pack.save_entry_concurrency
}

fn receive_pack_extra_timings_ms(&self) -> Vec<(String, u128)> {
std::mem::take(
&mut self
.receive_pack_extra_timings_ms
.lock()
.expect("receive_pack_extra_timings_ms lock poisoned"),
)
}

fn sync_commands_after_unpack(&self, commands: &[RefCommand]) {
*self
.command_list
Expand All @@ -75,10 +90,32 @@ impl RepoHandler for ImportRepo {
}

async fn finalize_receive_pack(&self) -> Result<(), MegaError> {
let guard = self.unpack_redlock.clone().lock().await?;
let t0 = Instant::now();
let t_fp = Instant::now();
self.traverses_tree_and_update_filepath().await?;
self.receive_pack_extra_timings_ms
.lock()
.expect("receive_pack_extra_timings_ms lock poisoned")
.push((
"import_filepath_update_ms".to_string(),
t_fp.elapsed().as_millis(),
));

let t_attach = Instant::now();
self.attach_to_monorepo_parent().await?;
guard.unlock().await?;
self.receive_pack_extra_timings_ms
.lock()
.expect("receive_pack_extra_timings_ms lock poisoned")
.extend([
(
"import_attach_to_monorepo_parent_ms".to_string(),
t_attach.elapsed().as_millis(),
),
(
"import_finalize_total_ms".to_string(),
t0.elapsed().as_millis(),
),
]);
Ok(())
}

Expand Down Expand Up @@ -424,8 +461,18 @@ impl ImportRepo {

// Concurrent attaches need CAS on root mega_refs; retry when head moved.
const MAX_ATTACH_ATTEMPTS: u32 = 64;
let mut root_lock_wait_max_ms: u128 = 0;
let mut root_lock_wait_sum_ms: u128 = 0;

for attempt in 0..MAX_ATTACH_ATTEMPTS {
// Only the root mega_refs update needs cross-repo serialization.
// Keep the lock scope as small as possible: just the root ref read + attach transaction.
let t_lock = Instant::now();
let guard = self.unpack_redlock.clone().lock().await?;
let lock_wait_ms = t_lock.elapsed().as_millis();
root_lock_wait_max_ms = root_lock_wait_max_ms.max(lock_wait_ms);
root_lock_wait_sum_ms += lock_wait_ms;

let root_ref = storage
.get_main_ref("/")
.await?
Expand Down Expand Up @@ -479,6 +526,7 @@ impl ImportRepo {
}
}

let t_attach_txn = Instant::now();
match storage
.attach_to_monorepo_parent_in_txn(
&txn,
Expand All @@ -492,10 +540,38 @@ impl ImportRepo {
{
Ok(()) => {
txn.commit().await.map_err(MegaError::Db)?;
let t_unlock = Instant::now();
guard.unlock().await?;
self.receive_pack_extra_timings_ms
.lock()
.expect("receive_pack_extra_timings_ms lock poisoned")
.extend([
(
"import_attach_attempts_count".to_string(),
(attempt + 1) as u128,
),
(
"import_root_lock_wait_sum_ms".to_string(),
root_lock_wait_sum_ms,
),
(
"import_root_lock_wait_max_ms".to_string(),
root_lock_wait_max_ms,
),
(
"import_attach_txn_ms".to_string(),
t_attach_txn.elapsed().as_millis(),
),
(
"import_root_lock_unlock_ms".to_string(),
t_unlock.elapsed().as_millis(),
),
]);
return Ok(());
}
Err(MegaError::StaleMonorepoRootRef) if attempt + 1 < MAX_ATTACH_ATTEMPTS => {
let _ = txn.rollback().await;
let _ = guard.unlock().await;
tracing::warn!(
attempt = attempt,
repo_path = %self.repo.repo_path,
Expand All @@ -505,6 +581,28 @@ impl ImportRepo {
}
Err(e) => {
let _ = txn.rollback().await;
let _ = guard.unlock().await;
self.receive_pack_extra_timings_ms
.lock()
.expect("receive_pack_extra_timings_ms lock poisoned")
.extend([
(
"import_attach_attempts_count".to_string(),
(attempt + 1) as u128,
),
(
"import_root_lock_wait_sum_ms".to_string(),
root_lock_wait_sum_ms,
),
(
"import_root_lock_wait_max_ms".to_string(),
root_lock_wait_max_ms,
),
(
"import_attach_txn_ms".to_string(),
t_attach_txn.elapsed().as_millis(),
),
]);
return Err(e);
}
}
Expand Down
72 changes: 67 additions & 5 deletions ceres/src/pack/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ use std::{
Arc,
atomic::{AtomicUsize, Ordering},
},
time::Instant,
};

use async_trait::async_trait;
Expand Down Expand Up @@ -41,6 +42,20 @@ pub mod monorepo;
pub trait RepoHandler: Send + Sync + 'static {
fn is_monorepo(&self) -> bool;

/// Concurrency limit for `save_entry` batches during receive-pack.
/// `0` means unbounded.
fn save_entry_concurrency(&self) -> usize {
1
}

/// Optional extra timing key-values to be included in the final receive-pack report.
/// Implementations may return an empty vec.
///
/// Intended for repo-type-specific breakdown (e.g. import attach lock waits).
fn receive_pack_extra_timings_ms(&self) -> Vec<(String, u128)> {
Vec::new()
}

/// Replace the handler's ref-command snapshot after unpack and the protocol status loop
/// (`default_branch`, `status`, etc.). The handler is built from an early `commands.clone()`,
/// so without this, metadata updated in `git_receive_pack_stream` would be stale at finalize.
Expand All @@ -53,33 +68,72 @@ pub trait RepoHandler: Send + Sync + 'static {
mut rx: UnboundedReceiver<MetaAttached<Entry, EntryMeta>>,
_rx_pack_id: UnboundedReceiver<ObjectHash>,
) -> Result<(), MegaError> {
let t0 = Instant::now();
let mut entry_list = vec![];
let semaphore = Arc::new(Semaphore::new(1)); //这里暂时改动
let limit = self.save_entry_concurrency();
let semaphore = if limit == 0 {
None
} else {
Some(Arc::new(Semaphore::new(limit)))
};
let mut join_tasks = vec![];

//let temp_pack_id = Uuid::new_v4().to_string();
let temp_pack_id = String::new();

let mut total_entries: usize = 0;
let mut flush_batches: usize = 0;
while let Some(mut entry) = rx.recv().await {
self.check_entry(&entry.inner).await?;
entry.meta.set_pack_id(temp_pack_id.clone());
entry_list.push(entry);
total_entries += 1;
if entry_list.len() >= 1000 {
let acquired = semaphore.clone().acquire_owned().await.unwrap();
flush_batches += 1;
let entries = std::mem::take(&mut entry_list);
let shared = self.clone();
let sem = semaphore.clone();
let handle = tokio::spawn(async move {
let _acquired = acquired;
shared.save_entry(entries).await
let _permit = match sem {
Some(s) => Some(s.acquire_owned().await.expect("semaphore closed")),
None => None,
Comment on lines 96 to +99

@chatgpt-codex-connector chatgpt-codex-connector Bot Apr 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

P1 Badge Acquire save-entry permit before spawning batch task

Moving semaphore acquisition inside the spawned task removes backpressure from receiver_handler: for large receive-pack uploads with slow save_entry writes and save_entry_concurrency set low, the loop now keeps draining rx and spawning one task per 1000 entries, each retaining its batch while waiting for a permit. That creates unbounded queued tasks and memory growth (potential OOM) instead of bounded in-flight work; the permit should be acquired before tokio::spawn (or via a bounded worker queue) to preserve memory safety under load.

Useful? React with 👍 / 👎.

};
let t = Instant::now();
let len = entries.len();
shared.save_entry(entries).await.map(|_| {
tracing::debug!(
service = "git-receive-pack",
batch_entries = len,
elapsed_ms = t.elapsed().as_millis(),
"save_entry batch completed"
);
})
});
join_tasks.push(handle);
}
}
// process left entries
if !entry_list.is_empty() {
flush_batches += 1;
let handler = self.clone();
let entries = std::mem::take(&mut entry_list);
let handle = tokio::spawn(async move { handler.save_entry(entries).await });
let sem = semaphore.clone();
let handle = tokio::spawn(async move {
let _permit = match sem {
Some(s) => Some(s.acquire_owned().await.expect("semaphore closed")),
None => None,
};
let t = Instant::now();
let len = entries.len();
handler.save_entry(entries).await.map(|_| {
tracing::debug!(
service = "git-receive-pack",
batch_entries = len,
elapsed_ms = t.elapsed().as_millis(),
"save_entry batch completed"
);
})
});
join_tasks.push(handle);
}

Expand Down Expand Up @@ -120,6 +174,14 @@ pub trait RepoHandler: Send + Sync + 'static {
// }
// }

tracing::debug!(
service = "git-receive-pack",
total_entries = total_entries,
flush_batches = flush_batches,
save_entry_concurrency = limit,
elapsed_ms = t0.elapsed().as_millis(),
"receiver_handler done"
);
Ok(())
}

Expand Down
4 changes: 4 additions & 0 deletions ceres/src/pack/monorepo.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,10 @@ impl RepoHandler for MonoRepo {
true
}

fn save_entry_concurrency(&self) -> usize {
self.storage.config().pack.save_entry_concurrency
}

fn sync_commands_after_unpack(&self, commands: &[RefCommand]) {
*self
.command_list
Expand Down
5 changes: 4 additions & 1 deletion ceres/src/protocol/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,7 +187,9 @@ impl SmartSession {

let unpack_redlock = Arc::new(RedLock::new(
state.git_object_cache.connection.clone(),
"git:receive-pack:lock",
// Serialize monorepo root mega_refs update across concurrent import attaches.
// Filepath updates and per-repo work should not be blocked by this lock.
"git:receive-pack:lock:monorepo-root".to_string(),
30_000, // 30s TTL
));
Ok(Arc::new(ImportRepo {
Expand All @@ -196,6 +198,7 @@ impl SmartSession {
repo,
command_list: Mutex::new(commands),
unpack_redlock,
receive_pack_extra_timings_ms: Mutex::new(Vec::new()),
}) as Arc<dyn RepoHandler>)
} else {
let mut res = MonoRepo {
Expand Down
Loading
Loading