spark-test commented on code in PR #59914:
URL: https://github.com/apache/spark/pull/59914#discussion_r4416538794
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSource.scala:
##########
@@ -468,6 +469,13 @@ object NativeWriterCommitMessage {
/**
* Reads a partition of a native data source: each array of the Arrow stream
exported by the
* library is checked, and imported into the Arrow vectors that back the
returned batch.
+ *
+ * A reader can be closed while another thread is in [[next]]: some operators
read their input on
+ * a thread of their own that the task does not wait for, such as the stdin
writer thread of
+ * `PipedRDD`, while `DataSourceRDD` closes the reader once the task
completes. The library must
+ * not get the release of the stream during get_next, and the task must stay
valid until then, so
+ * [[close]] then leaves the release to [[next]], once get_next returns,
rather than wait for it
+ * and keep the task from completing. [[next]] returns false once the reader
is closed.
Review Comment:
**Nit (P3):** This sentence follows the description of a `close()` that
lands while another thread is in `next()`, so it reads as covering that
in-flight call too. But `next()` wraps `readNext()` in
`Utils.tryWithSafeFinally(...)(stopReading())`, which rethrows the read's error
after the deferred release. So an in-flight `next()` whose `get_next`,
validation or import fails throws, as `closeWhileReading(failing = true)`
expects. Only a `next()` that starts after the close always returns false. A
maintainer who trusts the doc could assume that a concurrently closed reader
never throws from `next()`. Maybe:
```suggestion
* and keep the task from completing. That [[next]] then returns false, or
throws the error of its
* read. A [[next]] called once the reader is closed returns false without
calling the library.
```
**Verification:**
- **Inspection:** The class documentation of the native partition reader
describes both outcomes of a next() that is in flight when another thread
closes the reader, a false result and the rethrown error of a failed read,
consistently with the implementation of next() and with the failing-read case
of the cross-thread close test.
##########
sql/core/src/test/resources/native-datasource/test_native_datasource.cc:
##########
@@ -790,24 +812,52 @@ int streamGetNext(ArrowArrayStream* stream, ArrowArray*
out) {
state->last_error = "the task was killed";
return ECANCELED;
}
- if (state->next >= state->end) {
- out->release = nullptr; // The end of the stream.
- return 0;
+ if (!state->slow_read.empty()) {
+ // The flag covers the slow read, for streamRelease to tell whether it
comes during get_next.
+ state->in_get_next = true;
+ int result = EIO;
+ try {
+ std::string reading = state->slow_read + ".reading";
+ struct stat info;
+ if (stat(reading.c_str(), &info) != 0) {
+ writeFile(reading, "");
+ std::string go = state->slow_read + ".continue";
+ for (int i = 0; i < 6000 && stat(go.c_str(), &info) != 0; i++) {
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ }
+ }
+ if (state->fail_slow_read) {
Review Comment:
**Nit (P3):** After the wait, this branch never touches `state->task`, and
neither do `nextBatch` or `streamRelease`. So both new tests observe only when
the stream is released, not whether the `SparkDsTaskContext` is still valid.
Suppose `close()` freed the task while `get_next` runs but still deferred the
stream release, for example by calling `task.close()` before the `reading`
check. Both tests would still pass. A library that polls `task->is_interrupted`
here, as the header advises, would then call through freed memory: the crash
this PR describes.
A cheap guard: right after the wait, call
`state->task->is_interrupted(state->task)` and record the answer next to the
release marker, then assert in both tests that it was false. With this PR the
context is live and answers false: the `rdd.pipe` test's task completes
normally, and the direct test has no `TaskContext`. A context freed early is no
longer registered, so `NativeLibrary.isInterrupted` answers true.
**Verification:**
- **Regression:** Freeing the native task context in close() while a read is
in flight, with the stream release still deferred until get_next returns, makes
the cross-thread close tests fail on the recorded task-context probe.
- **Behavior:** With the deferred release as implemented, the cross-thread
close tests pass and observe a live task context after get_next resumes, both
for the read that succeeds and for the read that fails.
--
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]
