You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "fresh-borzoni (via GitHub)" <gi...@apache.org> on 2026/01/11 13:07:53 UTC
[PR] [ISSUE-135] Fix potential deadlock with holding lock during await [fluss-rust]
fresh-borzoni opened a new pull request, #145:
URL: https://github.com/apache/fluss-rust/pull/145
### Purpose
Linked issue: close #135
Fix critical bug where DashMap shard lock was held across `.await` points in the record accumulator, which could cause deadlocks via lock inversion and make futures `!Send`.
### Brief change log
- Changed `batches` field type in `BucketAndWriteBatches` from `HashMap<BucketId, Mutex<VecDeque<WriteBatch>>>` to `HashMap<BucketId, Arc<Mutex<VecDeque<WriteBatch>>>>>`
- Updated `append()`, `re_enqueue()`, `drain_batches_for_one_node()`, and `bucket_ready()` methods to clone the `Arc` before awaiting, ensuring DashMap locks are dropped before async suspension points
- Fixed unrelated test compilation error in `datum.rs`
### Tests
- Existing test `re_enqueue_increments_attempts` passes
- All accumulator unit tests pass
- Full build and clippy pass with no warnings
### API and Format
No API or storage format changes.
### Documentation
No new features introduced - this is a bug fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE-135] Fix potential deadlock with holding lock during await [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #145:
URL: https://github.com/apache/fluss-rust/pull/145#discussion_r2680679261
##########
crates/fluss/src/client/write/accumulator.rs:
##########
@@ -281,60 +301,76 @@ impl RecordAccumulator {
return Ok(ready);
}
- let mut nodes_drain_index_guard = self.nodes_drain_index.lock().await;
- let drain_index = nodes_drain_index_guard.entry(node.id()).or_insert(0);
- let start = *drain_index % buckets.len();
+ // Get the start index without holding the lock across awaits
+ let start = {
+ let mut nodes_drain_index_guard = self.nodes_drain_index.lock().await;
+ let drain_index = nodes_drain_index_guard.entry(node.id()).or_insert(0);
+ *drain_index % buckets.len()
+ };
+
let mut current_index = start;
+ let mut last_processed_index;
Review Comment:
It would require #[allow(unused_assignments)] with this suggestion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE-135] Fix potential deadlock with holding lock during await [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #145:
URL: https://github.com/apache/fluss-rust/pull/145#issuecomment-3735044414
Made more thorough audit:
### Fix 1: `ready()` (lines 173-212) - DashMap iterator across await
The DashMap iterator is held while calling `bucket_ready().await`. Since `bucket_ready()` internally awaits on Mutex locks, the DashMap shard guard stays alive across async suspension points, blocking concurrent operations on the same shard.
**Fix:** Snapshot the data into `Vec<(TablePath, Vec<(BucketId, Arc<Mutex<...>>)>)>` first, then iterate and await on the snapshot after all DashMap guards are dropped.
### Fix 2: `bucket_ready()` (lines 214-244) - signature change to work with snapshot
Changed from taking `&BucketAndWriteBatches` (which borrows from DashMap entry) to taking an owned `Vec<(BucketId, Arc<Mutex<...>>)>` snapshot. This ensures no DashMap references survive to the await points inside the method.
### Fix 3: `drain_batches_for_one_node()` (lines 290-365) - Mutex held across loop
The `nodes_drain_index` Mutex is locked once at the start and held throughout the entire loop, including across `deque.lock().await` calls. This unnecessarily serializes all drain operations (even for different nodes) and creates lock ordering issues.
**Fix:** Split into three short critical sections: (1) lock and read start index, (2) do drain loop with no lock, (3) lock and write back final index.
### Fix 4: `await_flush_completion()` (lines 437-444) - RwLock guard across await
The RwLock read guard on `incomplete_batches` is held while awaiting on result handles. This blocks any concurrent `remove_incomplete_batches()` writes for the entire flush duration, causing writer starvation or potential deadlock.
**Fix:** Clone handles into a Vec before awaiting - same pattern as fixes above.
All four fixes follow the "snapshot/clone Arc, drop guard, then await" pattern established in this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE-135] Fix potential deadlock with holding lock during await [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #145:
URL: https://github.com/apache/fluss-rust/pull/145
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE-135] Fix potential deadlock with holding lock during await [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #145:
URL: https://github.com/apache/fluss-rust/pull/145#discussion_r2679567895
##########
crates/fluss/src/client/write/accumulator.rs:
##########
@@ -203,6 +208,7 @@ impl RecordAccumulator {
let mut next_delay = next_ready_check_delay_ms;
for (bucket_id, batch) in batches.batches.iter() {
+ let batch = batch.clone();
Review Comment:
batch is an Arc<Mutex<_>> now, so the clone is just cloning the Arc.
We do it to avoid holding a reference into the HashMap/entry guard across .lock().await. Cloning gives us an owned handle we can await safely.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE-135] Fix potential deadlock with holding lock during await [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #145:
URL: https://github.com/apache/fluss-rust/pull/145#issuecomment-3736648781
@luoyuxia thanks for the review, pushed fix for CI
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE-135] Fix potential deadlock with holding lock during await [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #145:
URL: https://github.com/apache/fluss-rust/pull/145#issuecomment-3734537619
@zhaohaidao @luoyuxia PTAL 🙏
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE-135] Fix potential deadlock with holding lock during await [fluss-rust]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #145:
URL: https://github.com/apache/fluss-rust/pull/145#discussion_r2680658423
##########
crates/fluss/src/client/write/accumulator.rs:
##########
@@ -281,60 +301,76 @@ impl RecordAccumulator {
return Ok(ready);
}
- let mut nodes_drain_index_guard = self.nodes_drain_index.lock().await;
- let drain_index = nodes_drain_index_guard.entry(node.id()).or_insert(0);
- let start = *drain_index % buckets.len();
+ // Get the start index without holding the lock across awaits
+ let start = {
+ let mut nodes_drain_index_guard = self.nodes_drain_index.lock().await;
+ let drain_index = nodes_drain_index_guard.entry(node.id()).or_insert(0);
+ *drain_index % buckets.len()
+ };
+
let mut current_index = start;
+ let mut last_processed_index;
Review Comment:
The variable `last_processed_index` is declared without initialization. While it's always assigned at line 318 before being used at line 371, initializing it (e.g., to `start`) would make the code clearer and more maintainable.
```suggestion
let mut last_processed_index = start;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE-135] Fix potential deadlock with holding lock during await [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #145:
URL: https://github.com/apache/fluss-rust/pull/145#discussion_r2679567895
##########
crates/fluss/src/client/write/accumulator.rs:
##########
@@ -203,6 +208,7 @@ impl RecordAccumulator {
let mut next_delay = next_ready_check_delay_ms;
for (bucket_id, batch) in batches.batches.iter() {
+ let batch = batch.clone();
Review Comment:
batch is an Arc<Mutex<_>> now, so the clone is just cloning the Arc.
We do it to avoid holding a reference into the HashMap/entry guard across .lock().await. Cloning gives us an owned handle we can await safely.
every other .await already uses an owned Arc()(.cloned() or scoped entry())
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE-135] Fix potential deadlock with holding lock during await [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #145:
URL: https://github.com/apache/fluss-rust/pull/145#discussion_r2679564265
##########
crates/fluss/src/client/write/accumulator.rs:
##########
@@ -203,6 +208,7 @@ impl RecordAccumulator {
let mut next_delay = next_ready_check_delay_ms;
for (bucket_id, batch) in batches.batches.iter() {
+ let batch = batch.clone();
Review Comment:
just curious about why clone in here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org