feat(core/types): Implement concurrent read for blocking read #4545

hoslo · 2024-04-28T03:27:14Z

No description provided.

hoslo · 2024-04-28T03:40:32Z

@Xuanwo Plz take a review for this design.

Xuanwo · 2024-04-28T04:45:18Z

core/src/types/blocking_read/blocking_reader.rs

+        let (interval_size, mut intervals) = end
+            .map(|end| {
+                // let interval_size = (end - start + concurrent - 1) / concurrent;
+                let interval_size = (end - start) / concurrent;


interval_size should be decided by chunk.

core/src/types/blocking_read/blocking_reader.rs

Xuanwo · 2024-04-29T11:07:03Z

core/src/types/blocking_read/blocking_reader.rs

 /// BlockingReader is designed to read data from given path in an blocking
 /// manner.
 pub struct BlockingReader {
-    pub(crate) inner: oio::BlockingReader,
+    pub(crate) inner: Arc<dyn oio::BlockingRead>,


Please change oio::BlockingReader to Arc<dyn oio::BlockingRead>, maybe worth a seperate PR.

Would you like to submit this PR first?

Xuanwo · 2024-04-29T11:09:48Z

core/src/types/blocking_read/buffer_iterator.rs

+        let mut bufs = Vec::with_capacity(self.concurrent);
+        let interval_size = self.chunk.unwrap_or(4 * 1024 * 1024) as u64;
+
+        let intervals: Vec<(u64, u64)> = (0..self.concurrent as u64)


Please don't calcluate ranges at once.

We can build a RangeIterator that only generate the next range to read. BufferIterator stores RangeIterator, every time users call next, BufferIterator fetchs a new range and than read as a new buffer.

Please don't calcluate ranges at once.

How can this be paralleled?

How can this be paralleled?

After re-visit the rayon API docs, I found that we can't call next on a ParallelIterator directly. So we can change the RangeIterator to return a ParallelIterator instead.

Every time users call next on BufferIterator, we build a new ParallelIterator, extend into buffer and yield from it.

The whole workflow looks like:

RangeIterator::next() -> Range<u64> struct BufferIterator { it: RangeIterator, buf: Vec<Buffer> } impl Iterator for BufferIterator { type Item = Result<Buffer>; fn next(&mut self) -> Option<Self::Item> { // fecth from buffer first. // Build a new ParallelIterator and collect it. // Return None if ranges have been all consumed. } }

@Xuanwo Plz review.

Xuanwo · 2024-05-06T08:02:25Z

core/src/types/blocking_read/blocking_reader.rs

 /// BlockingReader is designed to read data from given path in an blocking
 /// manner.
 pub struct BlockingReader {
-    pub(crate) inner: oio::BlockingReader,
+    pub(crate) inner: Arc<dyn oio::BlockingRead>,


Would you like to submit this PR first?

Xuanwo · 2024-05-06T08:02:41Z

core/src/types/blocking_read/blocking_reader.rs

@@ -74,24 +84,26 @@ impl BlockingReader {
            }
        }

+        let iter = BufferIterator::new(
+            self.inner.clone(),
+            self.options.chunk(),


We can accpet an OpReader directly.

Would you like to submit this PR first?

Do you mean submit a new PR to change oio::BlockingReader to Arc first?

Do you mean submit a new PR to change oio::BlockingReader to Arc first?

Yes

Xuanwo · 2024-05-06T08:05:52Z

core/src/types/blocking_read/blocking_reader.rs

-            offset += n as u64;
-            if Some(offset) == end {
-                return Ok(bufs.into_iter().flatten().collect());
+        for buffer in iter {


Maybe we can use collect?

How to return err early when using collect?

How to return err early when using collect?

collect will exit once error happened. See impl<A, E, V> FromIterator<Result<A, E>> for Result<V, E>:

Takes each element in the Iterator: if it is an Err, no further elements are taken, and the Err is returned. Should no Err occur, a container with the values of each Result is returned.

Xuanwo · 2024-05-06T08:06:52Z

core/src/types/blocking_read/blocking_reader.rs

-                return Ok(read as _);
+        let mut total_len = 0;
+        for buffer in iter {
+            match buffer {


Please try use if-let-else:

let mut read = 0; loop { let Some(bs) = iter.try_next()? else { return Ok(read); }; read += bs.len(); buf.put(bs); }

Xuanwo · 2024-05-06T08:07:12Z

core/src/types/blocking_read/blocking_reader.rs

    }

    /// Convert reader into [`StdReader`] which implements [`futures::AsyncRead`],
    /// [`futures::AsyncSeek`] and [`futures::AsyncBufRead`].
    #[inline]
    pub fn into_std_read(self, range: Range<u64>) -> StdReader {
        // TODO: the capacity should be decided by services.
-        StdReader::new(self.inner, range)
+        StdReader::new(self.inner.clone(), range)


Maybe we should pass OpReader?

Xuanwo · 2024-05-06T08:09:31Z

core/src/types/blocking_read/buffer_iterator.rs

+
+        let mut bufs = Vec::with_capacity(self.concurrent);
+
+        let intervals: Vec<Range<u64>> = (0..self.concurrent)


Maybe we should put self.finished inside the RangeIterator should it can know when to stop returning new ranges.

Maybe we should put self.finished inside the RangeIterator should it can know when to stop returning new ranges.

When end is None, We only know end when after read.

When end is None, We only know end when after read.

That's why we need to store finished in RangeIterator.

Do you mean read also put into RangeIterator?

@Xuanwo I submit a new one, is this what you want?

Close but not what I want. Can I push changes to this PR directly?

Close but not what I want. Can I push changes to this PR directly?

Of course, thank for your patient.

@Xuanwo Do you still remember this?

github-actions bot added the releases-note/feat label Apr 28, 2024

github-actions bot requested review from ClSlaid and xyjixyjixyji April 28, 2024 03:27

Xuanwo reviewed Apr 28, 2024

View reviewed changes

hoslo force-pushed the feat-concurrent-blocking-read branch from 274fa3d to 5a597d3 Compare April 28, 2024 08:35

Xuanwo reviewed Apr 28, 2024

View reviewed changes

core/src/types/blocking_read/blocking_reader.rs Outdated Show resolved Hide resolved

hoslo force-pushed the feat-concurrent-blocking-read branch 2 times, most recently from d7640e3 to 9720e2b Compare April 28, 2024 11:02

Xuanwo reviewed Apr 29, 2024

View reviewed changes

hoslo force-pushed the feat-concurrent-blocking-read branch from 9720e2b to d500961 Compare May 6, 2024 07:30

Xuanwo reviewed May 6, 2024

View reviewed changes

hoslo mentioned this pull request May 7, 2024

feat(core/types): change oio::BlockingReader to Arc<dyn oio::BlockingReader> #4577

Merged

hoslo force-pushed the feat-concurrent-blocking-read branch 3 times, most recently from 661ba73 to 40322ce Compare May 7, 2024 06:56

feat(core/types): Implement concurrent read for blocking read

cb882aa

hoslo force-pushed the feat-concurrent-blocking-read branch from 40322ce to cb882aa Compare May 7, 2024 07:20

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(core/types): Implement concurrent read for blocking read #4545

feat(core/types): Implement concurrent read for blocking read #4545

hoslo commented Apr 28, 2024

hoslo commented Apr 28, 2024

Xuanwo Apr 28, 2024

Xuanwo Apr 29, 2024

Xuanwo May 6, 2024

Xuanwo Apr 29, 2024

hoslo Apr 30, 2024 •

edited

Loading

Xuanwo Apr 30, 2024

hoslo May 6, 2024

Xuanwo May 6, 2024

Xuanwo May 6, 2024

hoslo May 6, 2024

Xuanwo May 6, 2024

Xuanwo May 6, 2024

hoslo May 6, 2024

Xuanwo May 6, 2024

Xuanwo May 6, 2024

Xuanwo May 6, 2024

Xuanwo May 6, 2024

hoslo May 6, 2024

Xuanwo May 6, 2024

hoslo May 7, 2024

hoslo May 7, 2024

Xuanwo May 7, 2024

hoslo May 7, 2024

hoslo Jun 7, 2024 •

edited

Loading


		let mut bufs = Vec::with_capacity(self.concurrent);

		let intervals: Vec<Range<u64>> = (0..self.concurrent)

feat(core/types): Implement concurrent read for blocking read #4545

Are you sure you want to change the base?

feat(core/types): Implement concurrent read for blocking read #4545

Conversation

hoslo commented Apr 28, 2024

hoslo commented Apr 28, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hoslo Apr 30, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hoslo Jun 7, 2024 • edited Loading

Choose a reason for hiding this comment

hoslo Apr 30, 2024 •

edited

Loading

hoslo Jun 7, 2024 •

edited

Loading