gengliangwang commented on code in PR #59961:
URL: https://github.com/apache/spark/pull/59961#discussion_r4429956745
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSource.scala:
##########
@@ -858,7 +896,9 @@ class NativePartitionReader(
override def close(): Unit = synchronized {
if (!closed) {
closed = true
- if (!reading) {
+ // Another thread may be in next, or still read the batch that next
returned to it.
+ if (!reading && consumerDone) {
Review Comment:
**[P1] A deferred release never happens if the reading thread ends without
calling next**
When close() runs on one thread while the thread that last called next() is
still alive, close() hands the release to that thread and does nothing else.
The hand-off only takes effect if that thread calls next() again: consumerDone
is checked only in close() and startReading(). If the reading thread instead
exits (interrupted, IOException, or simply stops iterating), nothing ever
releases the child allocator, the library's ArrowArrayStream, the dictionaries
or the NativeTask. Before this PR, close() always released them.
**Failure scenario:** The class doc names PipedRDD as the multi-threaded
case. Its stdin writer thread reads the native source while the task thread's
completion listener closes the reader. The writer is still alive, so close()
defers the release. PipedRDD's own completion listener (PipedRDD.scala:173-183)
then destroys the process and interrupts the writer, which exits through an
IOException or the interrupt without calling next() again. The native stream,
the NativeTask and the Arrow child allocator leak permanently, once for every
such task, for example under a limit or when the piped process exits early.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSource.scala:
##########
@@ -858,7 +896,9 @@ class NativePartitionReader(
override def close(): Unit = synchronized {
if (!closed) {
Review Comment:
**[P2] close() on the thread the release was deferred to does nothing, so
resources leak**
close() returns immediately once closed is set, even when the caller is the
very thread the release was deferred to. In that state !reading && consumerDone
already holds, so it should release. next() returning false also leaves
consumer set to that thread. The result is that a thread which sees next() ==
false and then calls close(), as DataSourceRDD.PartitionIterator does, never
releases the reader.
**Failure scenario:** Thread A, for example a PipedRDD writer reaching the
end of the stream, finishes stopReading() while closed is still false. Thread
B, the task completion listener, then calls close(): reading is false and A is
alive, so the release is deferred to A. A's final synchronized(!closed) returns
false, so next() returns false. PartitionIterator.hasNext then calls
reader.close() on A (DataSourceRDD.scala:276-279), which skips everything
because closed is already true. The allocator, the native stream and the
NativeTask are never released.
--
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]
