refactor: spawn to read avro ObjectStore stream - #1
Conversation
Use the runtime handle provided to the AvroObjectReader to spawn tasks that perform I/O operations on the ObjectStore.
| use std::sync::Arc; | ||
| use tokio::runtime::Handle; | ||
|
|
||
| const STREAM_BUFFER_SIZE: usize = 8; |
There was a problem hiding this comment.
How is this magic number chosen? It shapes the backpressure profile of the streams, so some context on how this number is picked will help reason about the change.
(Or, can it be calculated with some heuristics based on the number of cores and/or available memory?)
There was a problem hiding this comment.
I did not have any better idea than to not make the backpressure buffer too large, since the chunk size can be configured in object store implementations. For local filesystem this buffer of 8 chunks can grow up to 64 MiB, and the cloud backends typically chunk at 8-64 KiB. I don't think the number of cores should figure because it's typically one producer task and one consumer that drives the decoder per file. The channel buffer size could be 1 and still provide optimization, since the main idea is to parallelize I/O and decoding.
There was a problem hiding this comment.
This is great context, could you please add that as code comment? I'm a big fan of documenting magic numbers
There was a problem hiding this comment.
Will do. Another reason for keeping it small is the potential cancellation, as discussed in the other comment.
| handle.spawn(async move { | ||
| while let Some(item) = stream.next().await { | ||
| let send_res = sender.send(item).await; | ||
| if send_res.is_err() { |
There was a problem hiding this comment.
Doesn't this mean we lose data in this scenario? What are the implications of a channel that's closed prematurely?
There was a problem hiding this comment.
Encountering an error here means the receiver part has been dropped without consuming all the items, so it's just propagating the cancellation.
There was a problem hiding this comment.
Yeah, that I know. But what's the implication of a closed consumer before the producer is finished here? Does it mean we lose data? Do files that should have been decoded are now thrown away silently?
If so, I believe this error should be louder, not a silent break. That's why I'm asking
There was a problem hiding this comment.
Since this is a low-level reader library, the possibilities for reporting are limited by design. The crate does not use tracing or any other logger API.
I think it's OK to follow the general async Rust principle of silently canceling an async operation if the consumer task has dropped the future/stream. There is no way to use that data any more because the application is not interested in decoding any more record batches, and object GET requests are supposed to be idempotent, so nothing is persistently lost.
|
Wait dont we already have this mechanism via the object store itself? |
Not explicitly in the implementations provided in the This is done to resolve comments in apache#9668 (comment) |
|
Note that the rationale in the doc of The issue with the original implementation in apache#9632 is that we didn't use the spawn mechanism to read the byte stream, mainly just due to expediency. |
Finally, this behavior is conditional on whether |
|
LGTM, though I can't say whether this is what Andrew Lamb was objecting, I guess we'll find out |
| .boxed(), | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
nit: I would add a comment here explaining that we need to spawn (via the runtime handle) both the initial retrieval of the handle to the stream and all subsequent item retrievals via the .next calls; and this is the reason why the mpsc::channel is needed; and this is the main difference between spawn and spawn_stream. It may be obvious, but it would be nice to explicitly state
4 participants
Use the runtime handle provided to the
AvroObjectReaderto spawn tasks that perform I/O operations on the ObjectStore.