feat: fs add concurrent write #4817
Conversation
hoslo
force-pushed
the
fs-concurrent-write
branch
from
4047e39
to
29de2d8
Compare
|
@Xuanwo I don't understand why the behavior test is failing. Can you help me look at it? |
hoslo
force-pushed
the
fs-concurrent-write
branch
8 times, most recently
from
0c8a26a
to
acbc5e7
Compare
| @@ -49,6 +49,12 @@ pub trait PositionWrite: Send + Sync + Unpin + 'static { | |||
| offset: u64, | |||
| buf: Buffer, | |||
| ) -> impl Future<Output = Result<()>> + MaybeSend; | |||
|
|
|||
| /// close is used to close the underlying storage. | |||
There was a problem hiding this comment.
close will close the file instead of the storage.
| /// close is used to close the underlying storage. | ||
| fn close(&self, offset: u64, buf: Buffer) -> impl Future<Output = Result<()>> + MaybeSend; | ||
|
|
||
| /// abort is used to abort the underlying storage. |
| @@ -148,13 +154,16 @@ impl<W: PositionWrite> oio::Write for PositionWriter<W> { | |||
|
|
|||
| if let Some(buffer) = self.cache.clone() { | |||
| let offset = self.next_offset; | |||
| self.w.write_all_at(offset, buffer).await?; | |||
| self.w.close(offset, buffer).await?; | |||
core/src/services/fs/writer.rs
Outdated
|
|
||
| tokio::task::spawn_blocking(move || { | ||
| let mut offset = offset; | ||
| let mut buf: &[u8] = &buf.to_bytes(); |
There was a problem hiding this comment.
to_bytes will copy the entire buffer to bytes which is slow. Maybe we can use:
while !buf.is_empty() {
match f.seek_write(buf.chunk(), offset) {
Ok(n) => {
buf.advance(n);
offset += n as u64
}
Err(e) => return Err(e).map_err(new_std_io_error),
}
}
core/src/services/fs/writer.rs
Outdated
| .await; | ||
|
|
||
| tokio::task::spawn_blocking(move || { | ||
| f.write_all_at(&buf.to_bytes(), offset) |
There was a problem hiding this comment.
Please don't use to_bytes whenever possible.
There was a problem hiding this comment.
what method should i use?
There was a problem hiding this comment.
Use buf.chunk() and buf.advance(). I think we can implement a write_at under different target so we don't need to remove the write impl.
hoslo
force-pushed
the
fs-concurrent-write
branch
5 times, most recently
from
00964ed
to
5ce79e3
Compare
| @@ -149,12 +155,16 @@ impl<W: PositionWrite> oio::Write for PositionWriter<W> { | |||
| if let Some(buffer) = self.cache.clone() { | |||
| let offset = self.next_offset; | |||
| self.w.write_all_at(offset, buffer).await?; | |||
| self.w.as_ref().close().await?; | |||
There was a problem hiding this comment.
We should call close no matter there are cache or not.
core/src/services/fs/writer.rs
Outdated
| f.seek_write(buf, offset).map_err(new_std_io_error) | ||
| } | ||
|
|
||
| #[cfg(not(target_os = "windows"))] |
There was a problem hiding this comment.
Please use cfg(unix) and cfg(windows)
|
|
||
| #[cfg(target_os = "windows")] | ||
| fn write_at(f: &File, buf: &[u8], offset: u64) -> Result<usize> { | ||
| f.seek_write(buf, offset).map_err(new_std_io_error) |
There was a problem hiding this comment.
We can use use here to avoid leave them too away to find.
hoslo
force-pushed
the
fs-concurrent-write
branch
from
5ce79e3
to
93d5a6d
Compare
|
@Xuanwo any else problem? |
Part #4526