You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2024/03/29 10:20:13 UTC
(opendal) 01/01: refactor(core/raw/oio): Mark oio::Write::write as an unsafe function
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch unsafe-oio-trait
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 6d6c53ea1552bfb4916b68f356f393d7b70f7a2a
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Fri Mar 29 18:19:38 2024 +0800
refactor(core/raw/oio): Mark oio::Write::write as an unsafe function
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
bindings/nodejs/src/lib.rs | 4 ++--
bindings/python/src/file.rs | 2 +-
core/benches/oio/utils.rs | 2 +-
core/benches/oio/write.rs | 2 +-
core/src/layers/blocking.rs | 2 +-
core/src/layers/complete.rs | 4 ++--
core/src/layers/concurrent_limit.rs | 4 ++--
core/src/layers/dtrace.rs | 4 ++--
core/src/layers/error_context.rs | 4 ++--
core/src/layers/logging.rs | 4 ++--
core/src/layers/madsim.rs | 2 +-
core/src/layers/metrics.rs | 4 ++--
core/src/layers/minitrace.rs | 4 ++--
core/src/layers/oteltrace.rs | 4 ++--
core/src/layers/prometheus.rs | 4 ++--
core/src/layers/prometheus_client.rs | 4 ++--
core/src/layers/retry.rs | 4 ++--
core/src/layers/throttle.rs | 4 ++--
core/src/layers/timeout.rs | 2 +-
core/src/layers/tracing.rs | 4 ++--
core/src/raw/adapters/kv/backend.rs | 4 ++--
core/src/raw/adapters/typed_kv/backend.rs | 4 ++--
core/src/raw/enum_utils.rs | 4 ++--
core/src/raw/oio/write/api.rs | 36 +++++++++++++++++++++-------
core/src/raw/oio/write/append_write.rs | 2 +-
core/src/raw/oio/write/block_write.rs | 4 ++--
core/src/raw/oio/write/exact_buf_write.rs | 17 ++++++-------
core/src/raw/oio/write/multipart_write.rs | 4 ++--
core/src/raw/oio/write/one_shot_write.rs | 2 +-
core/src/raw/oio/write/range_write.rs | 4 ++--
core/src/services/alluxio/writer.rs | 2 +-
core/src/services/fs/writer.rs | 4 ++--
core/src/services/ghac/writer.rs | 2 +-
core/src/services/hdfs/writer.rs | 4 ++--
core/src/services/hdfs_native/writer.rs | 2 +-
core/src/services/sftp/writer.rs | 2 +-
core/src/types/operator/blocking_operator.rs | 2 +-
core/src/types/writer.rs | 21 ++++++++++------
38 files changed, 108 insertions(+), 80 deletions(-)
diff --git a/bindings/nodejs/src/lib.rs b/bindings/nodejs/src/lib.rs
index b0bab1b4cc..5fac01b0ba 100644
--- a/bindings/nodejs/src/lib.rs
+++ b/bindings/nodejs/src/lib.rs
@@ -705,7 +705,7 @@ impl BlockingWriter {
/// await writer.close();
/// ```
#[napi]
- pub unsafe fn write(&mut self, content: Either<Buffer, String>) -> Result<()> {
+ pub unsafe unsafe fn write(&mut self, content: Either<Buffer, String>) -> Result<()> {
let c = match content {
Either::A(buf) => buf.as_ref().to_owned(),
Either::B(s) => s.into_bytes(),
@@ -752,7 +752,7 @@ impl Writer {
/// await writer.close();
/// ```
#[napi]
- pub async unsafe fn write(&mut self, content: Either<Buffer, String>) -> Result<()> {
+ pub async unsafe unsafe fn write(&mut self, content: Either<Buffer, String>) -> Result<()> {
let c = match content {
Either::A(buf) => buf.as_ref().to_owned(),
Either::B(s) => s.into_bytes(),
diff --git a/bindings/python/src/file.rs b/bindings/python/src/file.rs
index e407a8568e..66ffa6ca67 100644
--- a/bindings/python/src/file.rs
+++ b/bindings/python/src/file.rs
@@ -96,7 +96,7 @@ impl File {
}
/// Write bytes into the file.
- pub fn write(&mut self, bs: &[u8]) -> PyResult<usize> {
+ pub unsafe fn write(&mut self, bs: &[u8]) -> PyResult<usize> {
let writer = match &mut self.0 {
FileState::Reader(_) => {
return Err(PyIOError::new_err(
diff --git a/core/benches/oio/utils.rs b/core/benches/oio/utils.rs
index 18305f7fba..fcc2ba2f88 100644
--- a/core/benches/oio/utils.rs
+++ b/core/benches/oio/utils.rs
@@ -24,7 +24,7 @@ use rand::RngCore;
pub struct BlackHoleWriter;
impl oio::Write for BlackHoleWriter {
- async fn write(&mut self, bs: oio::ReadableBuf) -> opendal::Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> opendal::Result<usize> {
Ok(bs.len())
}
diff --git a/core/benches/oio/write.rs b/core/benches/oio/write.rs
index 633d604727..83e4b24795 100644
--- a/core/benches/oio/write.rs
+++ b/core/benches/oio/write.rs
@@ -48,7 +48,7 @@ pub fn bench_exact_buf_write(c: &mut Criterion) {
let mut bs = content.clone();
while !bs.is_empty() {
- let n = w.write(bs.clone().into()).await.unwrap();
+ let n = unsafe { w.write(bs.clone().into()).await.unwrap() };
bs.advance(n);
}
w.close().await.unwrap();
diff --git a/core/src/layers/blocking.rs b/core/src/layers/blocking.rs
index 278d5acca4..ff8ab5f61a 100644
--- a/core/src/layers/blocking.rs
+++ b/core/src/layers/blocking.rs
@@ -291,7 +291,7 @@ impl<I: oio::Read + 'static> oio::BlockingRead for BlockingWrapper<I> {
}
impl<I: oio::Write + 'static> oio::BlockingWrite for BlockingWrapper<I> {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
self.handle.block_on(self.inner.write(bs))
}
diff --git a/core/src/layers/complete.rs b/core/src/layers/complete.rs
index c7c232382b..bf25543f7e 100644
--- a/core/src/layers/complete.rs
+++ b/core/src/layers/complete.rs
@@ -638,7 +638,7 @@ impl<W> oio::Write for CompleteWriter<W>
where
W: oio::Write,
{
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let w = self.inner.as_mut().ok_or_else(|| {
Error::new(ErrorKind::Unexpected, "writer has been closed or aborted")
})?;
@@ -673,7 +673,7 @@ impl<W> oio::BlockingWrite for CompleteWriter<W>
where
W: oio::BlockingWrite,
{
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let w = self.inner.as_mut().ok_or_else(|| {
Error::new(ErrorKind::Unexpected, "writer has been closed or aborted")
})?;
diff --git a/core/src/layers/concurrent_limit.rs b/core/src/layers/concurrent_limit.rs
index e1af9ddefb..464c531b19 100644
--- a/core/src/layers/concurrent_limit.rs
+++ b/core/src/layers/concurrent_limit.rs
@@ -266,7 +266,7 @@ impl<R: oio::BlockingRead> oio::BlockingRead for ConcurrentLimitWrapper<R> {
}
impl<R: oio::Write> oio::Write for ConcurrentLimitWrapper<R> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
self.inner.write(bs).await
}
@@ -280,7 +280,7 @@ impl<R: oio::Write> oio::Write for ConcurrentLimitWrapper<R> {
}
impl<R: oio::BlockingWrite> oio::BlockingWrite for ConcurrentLimitWrapper<R> {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
self.inner.write(bs)
}
diff --git a/core/src/layers/dtrace.rs b/core/src/layers/dtrace.rs
index ca64b4e143..2fb23e768a 100644
--- a/core/src/layers/dtrace.rs
+++ b/core/src/layers/dtrace.rs
@@ -381,7 +381,7 @@ impl<R: oio::BlockingRead> oio::BlockingRead for DtraceLayerWrapper<R> {
}
impl<R: oio::Write> oio::Write for DtraceLayerWrapper<R> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let c_path = CString::new(self.path.clone()).unwrap();
probe_lazy!(opendal, writer_write_start, c_path.as_ptr());
self.inner
@@ -429,7 +429,7 @@ impl<R: oio::Write> oio::Write for DtraceLayerWrapper<R> {
}
impl<R: oio::BlockingWrite> oio::BlockingWrite for DtraceLayerWrapper<R> {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let c_path = CString::new(self.path.clone()).unwrap();
probe_lazy!(opendal, blocking_writer_write_start, c_path.as_ptr());
self.inner
diff --git a/core/src/layers/error_context.rs b/core/src/layers/error_context.rs
index aeed6b4adb..129ec52c2f 100644
--- a/core/src/layers/error_context.rs
+++ b/core/src/layers/error_context.rs
@@ -366,7 +366,7 @@ impl<T: oio::BlockingRead> oio::BlockingRead for ErrorContextWrapper<T> {
}
impl<T: oio::Write> oio::Write for ErrorContextWrapper<T> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
self.inner.write(bs.clone()).await.map_err(|err| {
err.with_operation(WriteOperation::Write)
.with_context("service", self.scheme)
@@ -393,7 +393,7 @@ impl<T: oio::Write> oio::Write for ErrorContextWrapper<T> {
}
impl<T: oio::BlockingWrite> oio::BlockingWrite for ErrorContextWrapper<T> {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
self.inner.write(bs.clone()).map_err(|err| {
err.with_operation(WriteOperation::BlockingWrite)
.with_context("service", self.scheme)
diff --git a/core/src/layers/logging.rs b/core/src/layers/logging.rs
index 0c29fa9f3c..4341325061 100644
--- a/core/src/layers/logging.rs
+++ b/core/src/layers/logging.rs
@@ -1075,7 +1075,7 @@ impl<W> LoggingWriter<W> {
}
impl<W: oio::Write> oio::Write for LoggingWriter<W> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
match self.inner.write(bs.clone()).await {
Ok(n) => {
self.written += n as u64;
@@ -1173,7 +1173,7 @@ impl<W: oio::Write> oio::Write for LoggingWriter<W> {
}
impl<W: oio::BlockingWrite> oio::BlockingWrite for LoggingWriter<W> {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
match self.inner.write(bs.clone()) {
Ok(n) => {
self.written += n as u64;
diff --git a/core/src/layers/madsim.rs b/core/src/layers/madsim.rs
index 54f0163c1a..ec2f2b1696 100644
--- a/core/src/layers/madsim.rs
+++ b/core/src/layers/madsim.rs
@@ -284,7 +284,7 @@ pub struct MadsimWriter {
}
impl oio::Write for MadsimWriter {
- async fn write(&mut self, bs: oio::ReadableBuf) -> crate::Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> crate::Result<usize> {
#[cfg(madsim)]
{
let req = Request::Write(self.path.to_string(), bs);
diff --git a/core/src/layers/metrics.rs b/core/src/layers/metrics.rs
index 349c8c7658..de380ea7e7 100644
--- a/core/src/layers/metrics.rs
+++ b/core/src/layers/metrics.rs
@@ -788,7 +788,7 @@ impl<R: oio::BlockingRead> oio::BlockingRead for MetricWrapper<R> {
}
impl<R: oio::Write> oio::Write for MetricWrapper<R> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let start = Instant::now();
self.inner
@@ -822,7 +822,7 @@ impl<R: oio::Write> oio::Write for MetricWrapper<R> {
}
impl<R: oio::BlockingWrite> oio::BlockingWrite for MetricWrapper<R> {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
self.inner
.write(bs)
.map(|n| {
diff --git a/core/src/layers/minitrace.rs b/core/src/layers/minitrace.rs
index b4e740b3cb..796398ce31 100644
--- a/core/src/layers/minitrace.rs
+++ b/core/src/layers/minitrace.rs
@@ -309,7 +309,7 @@ impl<R: oio::BlockingRead> oio::BlockingRead for MinitraceWrapper<R> {
}
impl<R: oio::Write> oio::Write for MinitraceWrapper<R> {
- fn write(&mut self, bs: oio::ReadableBuf) -> impl Future<Output = Result<usize>> + Send {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> impl Future<Output = Result<usize>> + Send {
let _g = self.span.set_local_parent();
let _span = LocalSpan::enter_with_local_parent(WriteOperation::Write.into_static());
self.inner.write(bs)
@@ -329,7 +329,7 @@ impl<R: oio::Write> oio::Write for MinitraceWrapper<R> {
}
impl<R: oio::BlockingWrite> oio::BlockingWrite for MinitraceWrapper<R> {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let _g = self.span.set_local_parent();
let _span = LocalSpan::enter_with_local_parent(WriteOperation::BlockingWrite.into_static());
self.inner.write(bs)
diff --git a/core/src/layers/oteltrace.rs b/core/src/layers/oteltrace.rs
index 3949cba765..d4c8bfecc3 100644
--- a/core/src/layers/oteltrace.rs
+++ b/core/src/layers/oteltrace.rs
@@ -287,7 +287,7 @@ impl<R: oio::BlockingRead> oio::BlockingRead for OtelTraceWrapper<R> {
}
impl<R: oio::Write> oio::Write for OtelTraceWrapper<R> {
- fn write(&mut self, bs: oio::ReadableBuf) -> impl Future<Output = Result<usize>> + Send {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> impl Future<Output = Result<usize>> + Send {
self.inner.write(bs)
}
@@ -301,7 +301,7 @@ impl<R: oio::Write> oio::Write for OtelTraceWrapper<R> {
}
impl<R: oio::BlockingWrite> oio::BlockingWrite for OtelTraceWrapper<R> {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
self.inner.write(bs)
}
diff --git a/core/src/layers/prometheus.rs b/core/src/layers/prometheus.rs
index b889cdb682..ef6f31bfe3 100644
--- a/core/src/layers/prometheus.rs
+++ b/core/src/layers/prometheus.rs
@@ -729,7 +729,7 @@ impl<R: oio::BlockingRead> oio::BlockingRead for PrometheusMetricWrapper<R> {
}
impl<R: oio::Write> oio::Write for PrometheusMetricWrapper<R> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let labels = self.stats.generate_metric_label(
self.scheme.into_static(),
Operation::Write.into_static(),
@@ -767,7 +767,7 @@ impl<R: oio::Write> oio::Write for PrometheusMetricWrapper<R> {
}
impl<R: oio::BlockingWrite> oio::BlockingWrite for PrometheusMetricWrapper<R> {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let labels = self.stats.generate_metric_label(
self.scheme.into_static(),
Operation::BlockingWrite.into_static(),
diff --git a/core/src/layers/prometheus_client.rs b/core/src/layers/prometheus_client.rs
index 78e81dd7fe..230429bd9f 100644
--- a/core/src/layers/prometheus_client.rs
+++ b/core/src/layers/prometheus_client.rs
@@ -577,7 +577,7 @@ impl<R: oio::BlockingRead> oio::BlockingRead for PrometheusMetricWrapper<R> {
}
impl<R: oio::Write> oio::Write for PrometheusMetricWrapper<R> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let start = Instant::now();
self.inner
@@ -614,7 +614,7 @@ impl<R: oio::Write> oio::Write for PrometheusMetricWrapper<R> {
}
impl<R: oio::BlockingWrite> oio::BlockingWrite for PrometheusMetricWrapper<R> {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
self.inner
.write(bs)
.map(|n| {
diff --git a/core/src/layers/retry.rs b/core/src/layers/retry.rs
index eca63d61cf..ae7149c40f 100644
--- a/core/src/layers/retry.rs
+++ b/core/src/layers/retry.rs
@@ -707,7 +707,7 @@ impl<R: oio::BlockingRead, I: RetryInterceptor> oio::BlockingRead for RetryWrapp
}
impl<R: oio::Write, I: RetryInterceptor> oio::Write for RetryWrapper<R, I> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
use backon::RetryableWithContext;
let inner = self.inner.take().expect("inner must be valid");
@@ -805,7 +805,7 @@ impl<R: oio::Write, I: RetryInterceptor> oio::Write for RetryWrapper<R, I> {
}
impl<R: oio::BlockingWrite, I: RetryInterceptor> oio::BlockingWrite for RetryWrapper<R, I> {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
{ || self.inner.as_mut().unwrap().write(bs.clone()) }
.retry(&self.builder)
.when(|e| e.is_temporary())
diff --git a/core/src/layers/throttle.rs b/core/src/layers/throttle.rs
index ff3111b74d..8cc9a48933 100644
--- a/core/src/layers/throttle.rs
+++ b/core/src/layers/throttle.rs
@@ -196,7 +196,7 @@ impl<R: oio::BlockingRead> oio::BlockingRead for ThrottleWrapper<R> {
}
impl<R: oio::Write> oio::Write for ThrottleWrapper<R> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let buf_length = NonZeroU32::new(bs.len() as u32).unwrap();
loop {
@@ -231,7 +231,7 @@ impl<R: oio::Write> oio::Write for ThrottleWrapper<R> {
}
impl<R: oio::BlockingWrite> oio::BlockingWrite for ThrottleWrapper<R> {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let buf_length = NonZeroU32::new(bs.len() as u32).unwrap();
loop {
diff --git a/core/src/layers/timeout.rs b/core/src/layers/timeout.rs
index cd42156695..d93bdbec90 100644
--- a/core/src/layers/timeout.rs
+++ b/core/src/layers/timeout.rs
@@ -298,7 +298,7 @@ impl<R: oio::Read> oio::Read for TimeoutWrapper<R> {
}
impl<R: oio::Write> oio::Write for TimeoutWrapper<R> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let fut = self.inner.write(bs);
Self::io_timeout(self.timeout, WriteOperation::Write.into_static(), fut).await
}
diff --git a/core/src/layers/tracing.rs b/core/src/layers/tracing.rs
index 1d7f538075..fbce9b591b 100644
--- a/core/src/layers/tracing.rs
+++ b/core/src/layers/tracing.rs
@@ -289,7 +289,7 @@ impl<R: oio::Write> oio::Write for TracingWrapper<R> {
parent = &self.span,
level = "trace",
skip_all)]
- fn write(&mut self, bs: oio::ReadableBuf) -> impl Future<Output = Result<usize>> + Send {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> impl Future<Output = Result<usize>> + Send {
self.inner.write(bs)
}
@@ -315,7 +315,7 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for TracingWrapper<R> {
parent = &self.span,
level = "trace",
skip_all)]
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
self.inner.write(bs)
}
diff --git a/core/src/raw/adapters/kv/backend.rs b/core/src/raw/adapters/kv/backend.rs
index af54f86d91..42438272d6 100644
--- a/core/src/raw/adapters/kv/backend.rs
+++ b/core/src/raw/adapters/kv/backend.rs
@@ -260,7 +260,7 @@ enum Buffer {
unsafe impl<S: Adapter> Sync for KvWriter<S> {}
impl<S: Adapter> oio::Write for KvWriter<S> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
match &mut self.buffer {
Buffer::Active(buf) => {
buf.extend_from_slice(&bs);
@@ -290,7 +290,7 @@ impl<S: Adapter> oio::Write for KvWriter<S> {
}
impl<S: Adapter> oio::BlockingWrite for KvWriter<S> {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
match &mut self.buffer {
Buffer::Active(buf) => {
buf.extend_from_slice(&bs);
diff --git a/core/src/raw/adapters/typed_kv/backend.rs b/core/src/raw/adapters/typed_kv/backend.rs
index 852c6f57ec..3336ef8a41 100644
--- a/core/src/raw/adapters/typed_kv/backend.rs
+++ b/core/src/raw/adapters/typed_kv/backend.rs
@@ -279,7 +279,7 @@ impl<S> KvWriter<S> {
}
impl<S: Adapter> oio::Write for KvWriter<S> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let size = bs.len();
let mut buf = self.buf.take().unwrap_or_else(|| Vec::with_capacity(size));
@@ -309,7 +309,7 @@ impl<S: Adapter> oio::Write for KvWriter<S> {
}
impl<S: Adapter> oio::BlockingWrite for KvWriter<S> {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let size = bs.len();
let mut buf = self.buf.take().unwrap_or_else(|| Vec::with_capacity(size));
diff --git a/core/src/raw/enum_utils.rs b/core/src/raw/enum_utils.rs
index 97b3c40162..e8c5a43ab3 100644
--- a/core/src/raw/enum_utils.rs
+++ b/core/src/raw/enum_utils.rs
@@ -71,7 +71,7 @@ impl<ONE: oio::BlockingRead, TWO: oio::BlockingRead> oio::BlockingRead for TwoWa
}
impl<ONE: oio::Write, TWO: oio::Write> oio::Write for TwoWays<ONE, TWO> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
match self {
Self::One(v) => v.write(bs).await,
Self::Two(v) => v.write(bs).await,
@@ -130,7 +130,7 @@ impl<ONE: oio::BlockingRead, TWO: oio::BlockingRead, THREE: oio::BlockingRead> o
impl<ONE: oio::Write, TWO: oio::Write, THREE: oio::Write> oio::Write
for ThreeWays<ONE, TWO, THREE>
{
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
match self {
Self::One(v) => v.write(bs).await,
Self::Two(v) => v.write(bs).await,
diff --git a/core/src/raw/oio/write/api.rs b/core/src/raw/oio/write/api.rs
index ea824e1b1f..9e3f295b6d 100644
--- a/core/src/raw/oio/write/api.rs
+++ b/core/src/raw/oio/write/api.rs
@@ -82,10 +82,16 @@ pub trait Write: Unpin + Send + Sync {
///
/// It's possible that `n < bs.len()`, caller should pass the remaining bytes
/// repeatedly until all bytes has been written.
+ ///
+ /// # Safety
+ ///
+ /// - The caller MUST ensure that the buffer is valid before `write` returns `Ready`.
+ /// - The implementor SHOULD NOT store [`oio::ReadableBuf`] in anyways. The buf MUST
+ /// be passed by or copied out to an owned buffer.
#[cfg(not(target_arch = "wasm32"))]
- fn write(&mut self, bs: oio::ReadableBuf) -> impl Future<Output = Result<usize>> + Send;
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> impl Future<Output = Result<usize>> + Send;
#[cfg(target_arch = "wasm32")]
- fn write(&mut self, bs: oio::ReadableBuf) -> impl Future<Output = Result<usize>>;
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> impl Future<Output = Result<usize>>;
/// Close the writer and make sure all data has been flushed.
#[cfg(not(target_arch = "wasm32"))]
@@ -101,7 +107,7 @@ pub trait Write: Unpin + Send + Sync {
}
impl Write for () {
- async fn write(&mut self, _: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, _: oio::ReadableBuf) -> Result<usize> {
unimplemented!("write is required to be implemented for oio::Write")
}
@@ -130,7 +136,7 @@ pub trait WriteDyn: Unpin + Send + Sync {
impl<T: Write + ?Sized> WriteDyn for T {
fn write_dyn(&mut self, bs: oio::ReadableBuf) -> BoxedFuture<Result<usize>> {
- Box::pin(self.write(bs))
+ unsafe { Box::pin(self.write(bs)) }
}
fn close_dyn(&mut self) -> BoxedFuture<Result<()>> {
@@ -143,7 +149,7 @@ impl<T: Write + ?Sized> WriteDyn for T {
}
impl<T: WriteDyn + ?Sized> Write for Box<T> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
self.deref_mut().write_dyn(bs).await
}
@@ -162,14 +168,28 @@ pub type BlockingWriter = Box<dyn BlockingWrite>;
/// BlockingWrite is the trait that OpenDAL returns to callers.
pub trait BlockingWrite: Send + Sync + 'static {
/// Write whole content at once.
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize>;
+ ///
+ /// # Behavior
+ ///
+ /// - `Ok(n)` means `n` bytes has been written successfully.
+ /// - `Err(err)` means error happens and no bytes has been written.
+ ///
+ /// It's possible that `n < bs.len()`, caller should pass the remaining bytes
+ /// repeatedly until all bytes has been written.
+ ///
+ /// # Safety
+ ///
+ /// - The caller MUST ensure that the buffer is valid before `write` returns `Ready`.
+ /// - The implementor SHOULD NOT store [`oio::ReadableBuf`] in anyways. The buf MUST
+ /// be passed by or copied out to an owned buffer.
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize>;
/// Close the writer and make sure all data has been flushed.
fn close(&mut self) -> Result<()>;
}
impl BlockingWrite for () {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let _ = bs;
unimplemented!("write is required to be implemented for oio::BlockingWrite")
@@ -187,7 +207,7 @@ impl BlockingWrite for () {
///
/// To make BlockingWriter work as expected, we must add this impl.
impl<T: BlockingWrite + ?Sized> BlockingWrite for Box<T> {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
(**self).write(bs)
}
diff --git a/core/src/raw/oio/write/append_write.rs b/core/src/raw/oio/write/append_write.rs
index 742173939a..8c2d70b07a 100644
--- a/core/src/raw/oio/write/append_write.rs
+++ b/core/src/raw/oio/write/append_write.rs
@@ -86,7 +86,7 @@ impl<W> oio::Write for AppendWriter<W>
where
W: AppendWrite,
{
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let offset = match self.offset {
Some(offset) => offset,
None => {
diff --git a/core/src/raw/oio/write/block_write.rs b/core/src/raw/oio/write/block_write.rs
index 329d8f3d06..fe364d4128 100644
--- a/core/src/raw/oio/write/block_write.rs
+++ b/core/src/raw/oio/write/block_write.rs
@@ -183,7 +183,7 @@ impl<W> oio::Write for BlockWriter<W>
where
W: BlockWrite,
{
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
loop {
if self.futures.has_remaining() {
// Fill cache with the first write.
@@ -363,7 +363,7 @@ mod tests {
expected_content.extend_from_slice(&bs);
loop {
- match w.write(bs.clone().into()).await {
+ match unsafe { w.write(bs.clone().into()).await } {
Ok(_) => break,
Err(_) => continue,
}
diff --git a/core/src/raw/oio/write/exact_buf_write.rs b/core/src/raw/oio/write/exact_buf_write.rs
index f041371249..b67a4c6928 100644
--- a/core/src/raw/oio/write/exact_buf_write.rs
+++ b/core/src/raw/oio/write/exact_buf_write.rs
@@ -49,7 +49,7 @@ impl<W: oio::Write> ExactBufWriter<W> {
}
impl<W: oio::Write> oio::Write for ExactBufWriter<W> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
// Quick Path
//
// if buffer is empty and bs is larger than buffer_size, we can directly
@@ -86,10 +86,11 @@ impl<W: oio::Write> oio::Write for ExactBufWriter<W> {
break;
}
- let written = self
- .inner
- .write(oio::ReadableBuf::from_slice(&self.buffer))
- .await?;
+ let written = unsafe {
+ self.inner
+ .write(oio::ReadableBuf::from_slice(&self.buffer))
+ .await?
+ };
self.buffer.advance(written);
}
@@ -121,7 +122,7 @@ mod tests {
}
impl Write for MockWriter {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
debug!("test_fuzz_exact_buf_writer: flush size: {}", &bs.len());
self.buf.extend_from_slice(&bs);
@@ -153,7 +154,7 @@ mod tests {
let mut bs = Bytes::from(expected.clone());
while !bs.is_empty() {
- let n = w.write(bs.clone().into()).await?;
+ let n = unsafe { w.write(bs.clone().into()).await? };
bs.advance(n);
}
@@ -192,7 +193,7 @@ mod tests {
let mut bs = Bytes::from(content.clone());
while !bs.is_empty() {
- let n = writer.write(bs.clone().into()).await?;
+ let n = unsafe { writer.write(bs.clone().into()).await? };
bs.advance(n);
}
}
diff --git a/core/src/raw/oio/write/multipart_write.rs b/core/src/raw/oio/write/multipart_write.rs
index 89cb992922..eed1c57f2b 100644
--- a/core/src/raw/oio/write/multipart_write.rs
+++ b/core/src/raw/oio/write/multipart_write.rs
@@ -229,7 +229,7 @@ impl<W> oio::Write for MultipartWriter<W>
where
W: MultipartWrite,
{
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let upload_id = match self.upload_id.clone() {
Some(v) => v,
None => {
@@ -446,7 +446,7 @@ mod tests {
rng.fill_bytes(&mut bs);
loop {
- match w.write(bs.clone().into()).await {
+ match unsafe { w.write(bs.clone().into()).await } {
Ok(_) => break,
Err(_) => continue,
}
diff --git a/core/src/raw/oio/write/one_shot_write.rs b/core/src/raw/oio/write/one_shot_write.rs
index b78b4e625c..aea92a11a0 100644
--- a/core/src/raw/oio/write/one_shot_write.rs
+++ b/core/src/raw/oio/write/one_shot_write.rs
@@ -55,7 +55,7 @@ impl<W: OneShotWrite> OneShotWriter<W> {
}
impl<W: OneShotWrite> oio::Write for OneShotWriter<W> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
match &self.buffer {
Some(_) => Err(Error::new(
ErrorKind::Unsupported,
diff --git a/core/src/raw/oio/write/range_write.rs b/core/src/raw/oio/write/range_write.rs
index dd3f223b8b..117003c7c7 100644
--- a/core/src/raw/oio/write/range_write.rs
+++ b/core/src/raw/oio/write/range_write.rs
@@ -163,7 +163,7 @@ impl<W: RangeWrite> RangeWriter<W> {
}
impl<W: RangeWrite> oio::Write for RangeWriter<W> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let location = match self.location.clone() {
Some(location) => location,
None => {
@@ -363,7 +363,7 @@ mod tests {
rng.fill_bytes(&mut bs);
loop {
- match w.write(bs.clone().into()).await {
+ match unsafe { w.write(bs.clone().into()).await } {
Ok(_) => break,
Err(_) => continue,
}
diff --git a/core/src/services/alluxio/writer.rs b/core/src/services/alluxio/writer.rs
index ccb5b6b980..de7fe2d760 100644
--- a/core/src/services/alluxio/writer.rs
+++ b/core/src/services/alluxio/writer.rs
@@ -43,7 +43,7 @@ impl AlluxioWriter {
}
impl oio::Write for AlluxioWriter {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let stream_id = match self.stream_id {
Some(stream_id) => stream_id,
None => {
diff --git a/core/src/services/fs/writer.rs b/core/src/services/fs/writer.rs
index f04a6ebe38..063d78814f 100644
--- a/core/src/services/fs/writer.rs
+++ b/core/src/services/fs/writer.rs
@@ -47,7 +47,7 @@ impl<F> FsWriter<F> {
unsafe impl<F> Sync for FsWriter<F> {}
impl oio::Write for FsWriter<tokio::fs::File> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let f = self.f.as_mut().expect("FsWriter must be initialized");
f.write(&bs).await.map_err(new_std_io_error)
@@ -81,7 +81,7 @@ impl oio::Write for FsWriter<tokio::fs::File> {
}
impl oio::BlockingWrite for FsWriter<std::fs::File> {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let f = self.f.as_mut().expect("FsWriter must be initialized");
f.write(&bs).map_err(new_std_io_error)
diff --git a/core/src/services/ghac/writer.rs b/core/src/services/ghac/writer.rs
index 142a7fa241..016440e394 100644
--- a/core/src/services/ghac/writer.rs
+++ b/core/src/services/ghac/writer.rs
@@ -38,7 +38,7 @@ impl GhacWriter {
}
impl oio::Write for GhacWriter {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let size = bs.len();
let offset = self.size;
diff --git a/core/src/services/hdfs/writer.rs b/core/src/services/hdfs/writer.rs
index ce6b2b0661..87b5c1066c 100644
--- a/core/src/services/hdfs/writer.rs
+++ b/core/src/services/hdfs/writer.rs
@@ -52,7 +52,7 @@ impl<F> HdfsWriter<F> {
}
impl oio::Write for HdfsWriter<hdrs::AsyncFile> {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let f = self.f.as_mut().expect("HdfsWriter must be initialized");
f.write(&bs).await.map_err(new_std_io_error)
@@ -81,7 +81,7 @@ impl oio::Write for HdfsWriter<hdrs::AsyncFile> {
}
impl oio::BlockingWrite for HdfsWriter<hdrs::File> {
- fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
let f = self.f.as_mut().expect("HdfsWriter must be initialized");
f.write(&bs).map_err(new_std_io_error)
}
diff --git a/core/src/services/hdfs_native/writer.rs b/core/src/services/hdfs_native/writer.rs
index 39b642b421..c8b428aa06 100644
--- a/core/src/services/hdfs_native/writer.rs
+++ b/core/src/services/hdfs_native/writer.rs
@@ -31,7 +31,7 @@ impl HdfsNativeWriter {
}
impl oio::Write for HdfsNativeWriter {
- async fn write(&mut self, _bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, _bs: oio::ReadableBuf) -> Result<usize> {
todo!()
}
diff --git a/core/src/services/sftp/writer.rs b/core/src/services/sftp/writer.rs
index 960a5e5ebd..12d0b7cf47 100644
--- a/core/src/services/sftp/writer.rs
+++ b/core/src/services/sftp/writer.rs
@@ -38,7 +38,7 @@ impl SftpWriter {
}
impl oio::Write for SftpWriter {
- async fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
+ async unsafe fn write(&mut self, bs: oio::ReadableBuf) -> Result<usize> {
self.file
.write(bs.as_slice())
.await
diff --git a/core/src/types/operator/blocking_operator.rs b/core/src/types/operator/blocking_operator.rs
index 03cedacf15..982c5f94f1 100644
--- a/core/src/types/operator/blocking_operator.rs
+++ b/core/src/types/operator/blocking_operator.rs
@@ -644,7 +644,7 @@ impl BlockingOperator {
let (_, mut w) = inner.blocking_write(&path, args)?;
while !bs.is_empty() {
- let n = w.write(bs.clone().into())?;
+ let n = unsafe { w.write(bs.clone().into())? };
bs.advance(n);
}
w.close()?;
diff --git a/core/src/types/writer.rs b/core/src/types/writer.rs
index 4627957787..5af89a65cc 100644
--- a/core/src/types/writer.rs
+++ b/core/src/types/writer.rs
@@ -94,7 +94,8 @@ impl Writer {
pub async fn write(&mut self, bs: impl Into<Bytes>) -> Result<()> {
let mut bs = bs.into();
while !bs.is_empty() {
- let n = self.inner.write(bs.clone().into()).await?;
+ // Safety: buf is valid before write return.
+ let n = unsafe { self.inner.write(bs.clone().into()).await? };
bs.advance(n);
}
@@ -138,7 +139,8 @@ impl Writer {
while let Some(bs) = sink_from.try_next().await? {
let mut bs = bs.into();
while !bs.is_empty() {
- let n = self.inner.write(bs.clone().into()).await?;
+ // Safety: bs must be valid before write return.
+ let n = unsafe { self.inner.write(bs.clone().into()).await? };
bs.advance(n);
written += n as u64;
}
@@ -230,7 +232,8 @@ pub mod into_futures_async_write {
let mut w = w.take().expect("writer must be valid");
let bs = oio::ReadableBuf::from_slice(buf);
let fut = async move {
- let res = w.write(bs).await;
+ // FIXME: we need to check if bs remains the same after future created.
+ let res = unsafe { w.write(bs).await };
(w, res)
};
self.state = State::Writing(Box::pin(fut));
@@ -302,7 +305,8 @@ impl BlockingWriter {
pub fn write(&mut self, bs: impl Into<Bytes>) -> Result<()> {
let mut bs = bs.into();
while !bs.is_empty() {
- let n = self.inner.write(bs.clone().into())?;
+ // Safety: bs must be valid before write return.
+ let n = unsafe { self.inner.write(bs.clone().into())? };
bs.advance(n);
}
@@ -317,9 +321,12 @@ impl BlockingWriter {
impl io::Write for BlockingWriter {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
- self.inner
- .write(buf.into())
- .map_err(|err| io::Error::new(io::ErrorKind::Other, err))
+ // Safety: buf is valid before write return.
+ unsafe {
+ self.inner
+ .write(buf.into())
+ .map_err(|err| io::Error::new(io::ErrorKind::Other, err))
+ }
}
fn flush(&mut self) -> io::Result<()> {