JeonDaehong commented on PR #3275:
URL: https://github.com/apache/iceberg-rust/pull/3275#issuecomment-6109012174
Thanks, handling the dropped sender looks right to me. While checking
whether a waiter can still hang, I found two more scenarios on `main` (not
introduced in this PR) where a waiter can get stuck:
1. **Overwritten Notifier in `try_start_eq_del_load` &
`insert_equality_delete`**
* **Issue:** `try_start_eq_del_load` stores `Loading(notify_a)`, and then
`insert_equality_delete` replaces it with `Loading(notify_b)`. A waiter that
reads the entry in between ends up waiting on `notify_a`, which is never
notified (the returned `notify` at `caching_delete_file_loader.rs:297` is
currently unused).
* **Fix:** Reuse the existing notifier from `try_start_eq_del_load`
inside `insert_equality_delete`.
2. **Race Condition in Waiter Wakeup (`notified()` vs `notified_owned()`)**
* **Issue:** The waiter clones the `Notify`, drops the read lock, and
only then calls `notified()`. Since `notify_waiters()` doesn't store a permit,
if the load finishes in that gap, the wakeup is lost.
* **Fix:** Call `notifier.clone().notified_owned()` while still holding
the lock (as `try_start_pos_del_load` does).
I reproduced both on this branch (for #2, I added a small sleep between the
read lock and `notified()` to make the gap deterministic). Both waiters hang as
expected, and with the fix applied (reusing the notifier and using
`notified_owned()` inside the lock), both wake up correctly.
Happy to send this as a follow-up PR if you'd rather keep this PR focused!
<details><summary>Repro for #1 (multi_thread runtime)</summary>
```rust
let filter = DeleteFilter::new(Runtime::current());
let _claimed = filter.try_start_eq_del_load("eq.parquet").unwrap();
let f2 = filter.clone();
let waiter = tokio::spawn(async move {
f2.get_equality_delete_predicate_for_delete_file_path("eq.parquet").await
});
tokio::time::sleep(Duration::from_millis(200)).await;
let (tx, rx) = tokio::sync::oneshot::channel::<Predicate>();
filter.insert_equality_delete("eq.parquet", rx);
tx.send(AlwaysTrue).unwrap();
// times out: the waiter is parked on the first notifier
assert!(tokio::time::timeout(Duration::from_secs(3), waiter).await.is_err());
</details>
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
