andygrove commented on code in PR #6092:
URL: https://github.com/apache/datafusion-comet/pull/6092#discussion_r4067214575
##########
native/core/src/execution/jni_api.rs:
##########
@@ -1113,30 +1133,26 @@ pub unsafe extern "system" fn
Java_org_apache_comet_Native_executePlan(
}
}
- // ScanExec path: busy-poll to interleave JVM batch pulls with
stream polling
+ // ScanExec path: JVM-fed scans return Pending without a waker and
are refilled here.
+ // Nothing pulled means the stream waits on native I/O, so park
instead of spinning.
get_runtime().block_on(async {
loop {
let next_item =
exec_context.stream.as_mut().unwrap().next();
let poll_output = poll!(next_item);
- // Only check time/tracing every 100 polls to reduce
overhead
- exec_context.poll_count_since_metrics_check += 1;
- if exec_context.poll_count_since_metrics_check >= 100 {
- exec_context.poll_count_since_metrics_check = 0;
- if let Some(interval) =
exec_context.metrics_update_interval {
- let now = Instant::now();
- if now - exec_context.metrics_last_update_time >=
interval {
- update_metrics(env, exec_context)?;
- exec_context.metrics_last_update_time = now;
- }
- }
- if exec_context.tracing_enabled {
- log_memory_usage(
- &exec_context.tracing_memory_metric_name,
-
total_reserved_for_thread(exec_context.rust_thread_id) as u64,
- );
+ if let Some(interval) =
exec_context.metrics_update_interval {
+ let now = Instant::now();
+ if now - exec_context.metrics_last_update_time >=
interval {
+ update_metrics(env, exec_context)?;
+ exec_context.metrics_last_update_time = now;
}
}
+ if exec_context.tracing_enabled {
Review Comment:
The tracing block came out of the 100-poll gate along with the metrics
check, and unlike the metrics check there is no interval behind it now. On the
path where the loop still turns quickly, a shuffle read or a broadcast build
side where every iteration does a real JVM pull, that is one `log_memory_usage`
per pulled batch instead of one per hundred. Each one is a `format!` plus the
process-wide `RECORDER` writer mutex in `native/common/src/tracing.rs`, and
`total_reserved_for_thread` takes the global pool registry lock on top of that.
Every task on the executor shares both of those locks.
During an I/O wait the new code is far quieter than the old spin, so this is
only about the pulling path. That is also the path someone opens a trace to
look at, though, so the denser sampling lands right where it perturbs the thing
being measured. Could the tracing emission sit behind the same interval check
as `update_metrics`? That would make trace density independent of how many
times the loop went round, which seems closer to what a counter track wants
anyway.
##########
native/core/src/execution/operators/scan.rs:
##########
@@ -112,23 +112,26 @@ impl ScanExec {
*self.batch.try_lock().unwrap() = Some(input);
}
- /// Pull next input batch from the upstream `ArrowArrayStreamReader`.
- pub fn get_next_batch(&mut self) -> Result<(), CometError> {
+ /// Pulls the next input batch from the upstream `ArrowArrayStreamReader`
unless one is
+ /// already buffered; returns whether it did.
+ pub fn get_next_batch(&mut self) -> Result<bool, CometError> {
if self.input_source.is_none() {
// This is a unit test. Input batches are seeded via
`set_input_batch`.
- return Ok(());
+ return Ok(false);
}
let mut current_batch = self.batch.try_lock().unwrap();
- if current_batch.is_none() {
- let mut timer = self.baseline_metrics.elapsed_compute().timer();
- let next_batch =
- ScanExec::pull_next(self.exec_context_id,
self.input_source.as_ref().unwrap())?;
- *current_batch = Some(next_batch);
- timer.stop();
+ if current_batch.is_some() {
+ return Ok(false);
}
- Ok(())
+ let mut timer = self.baseline_metrics.elapsed_compute().timer();
+ let next_batch =
+ ScanExec::pull_next(self.exec_context_id,
self.input_source.as_ref().unwrap())?;
+ *current_batch = Some(next_batch);
+ timer.stop();
+
+ Ok(true)
Review Comment:
This is the return value I want to poke at. Down in `ScanStream::poll_next`,
the `*scan_batch = None` at line 350 runs for the `InputBatch::EOF` arm too, so
EOF is not sticky in the stream. Once a JVM-fed scan finishes, the next
`get_next_batch` finds an empty buffer, makes another JNI round trip into an
exhausted reader, gets `InputBatch::EOF` back, and returns `true` here. #6091
describes the resting state as every JVM-fed scan holding a batch or having
reached EOF making the pull a no-op, and the EOF half of that isn't quite what
the code does today.
In the broadcast shape it costs one extra non-parking iteration and then
settles, because the buffer holds `Some(EOF)` afterwards and nothing polls the
exhausted stream again. What I am less sure about is an operator that does
re-poll it. An exhausted `ScanStream` returns `Pending` rather than
`Ready(None)` on a re-poll, the loop refills it with EOF, gets `true` back,
skips the park, and the cycle turns at full speed. Would leaving the EOF in
place, so `poll_next` returns `Ready(None)` without clearing, close that off?
`get_next_batch` becomes a real no-op after EOF, the wasted JNI call goes away,
and the invariant the issue leans on becomes true. Same thing in
`shuffle_scan.rs` at line 397.
--
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]
