HyukjinKwon commented on code in PR #59282: URL: https://github.com/apache/spark/pull/59282#discussion_r4228283943
########## sql/core/src/main/scala/org/apache/spark/sql/execution/python/RowQueue.scala: ########## @@ -181,8 +181,13 @@ private[python] case class DiskRowQueue( * * HybridRowQueue could be safely appended in one thread, and pulled in another thread in the same * time. + * + * This is intentionally not a case class. As a memory consumer it must use identity equality: + * otherwise two queues created with the same arguments in one task are equal, so only one of + * them is tracked (and offered for spilling) by the [[TaskMemoryManager]], and each refuses to + * spill when the other one triggers it. */ -case class HybridRowQueue( +class HybridRowQueue( Review Comment: **Non-blocking (P2):** With identity equality, two queues that share the same `TaskMemoryManager`, local dir, width, `SerializerManager` and `lockFree` flag now become spill candidates for each other. That newly exposes them to a lock-order inversion that already exists in `HybridQueue`: - `TaskMemoryManager.acquireExecutionMemory` holds the TMM monitor while it calls `consumerToSpill.spill(...)`, and `HybridQueue.spill` then takes the queue's monitor. - `HybridQueue.close()` holds the queue's monitor while it calls `queues.remove().close()`. For an in-memory page that ends in `TaskMemoryManager.freePage`, which takes `synchronized(this)` on the TMM. With `spark.python.udf.pipelined.enabled=true`, `add()` runs on a pooled writer thread. Take two Python eval operators in one task, for example both sides of a sort-merge join or stacked eval nodes with the same input width. Completion listeners run LIFO: writer2 join, `queue2.close()`, writer1 join, `queue1.close()`. So `queue2.close()` can run on the task thread while writer1 is still adding to `queue1`, for example after the other join side runs out, a limit, or a task failure. Suppose writer1 allocates a page under memory pressure at that moment and picks `queue2` as the spill victim. Writer1 then holds the TMM monitor and waits on `queue2`'s monitor, while the task thread holds `queue2`'s monitor and waits on the TMM monitor in `freePage`. Monitor waits are not interruptible, so the task hangs in cleanup. Before this change an equal-argument pair could not get here: the second queue was never added to `consumers`, and `spill` returned 0 at `trigger == this` before locking. Pairs that differ in width can already deadlock this way, but this PR extends it to the equal-width case that the description calls out. A small fix: in `HybridQueue.close()`, swap `queues` for an empty list while holding the monitor, then close the detached queues (freeing their pages) after releasing it. Close then never holds the queue monitor while calling into `TaskMemoryManager`, and a spill that runs after the swap just sees an empty list. **Verification:** - **Regression:** Two equal-argument hybrid row queues on one TaskMemoryManager, with the second holding several in-memory pages: one thread spills the second queue for the first while holding the TaskMemoryManager monitor, and another thread concurrently closes the second queue. Both threads complete within a bounded timeout instead of deadlocking. - **Behavior:** After a close that overlaps a spill, the queue's memory usage is zero and the TaskMemoryManager reports no unreleased memory. - **Compatibility:** The existing hybrid queue tests and the two new identity-equality tests keep their results: equal-argument queues spill for each other, self-triggered spill returns 0, and rows drain in FIFO order. -- 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]
