You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by "yah01 (via GitHub)" <gi...@apache.org> on 2023/08/04 10:30:31 UTC
[GitHub] [incubator-opendal] yah01 opened a new pull request, #2780: Add blocking layer
yah01 opened a new pull request, #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780
- Auto add blocking layer for backends which don't support blocking APIs
fix #2678
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo merged pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo merged PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#discussion_r1285194751
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ Self {
+ runtime: Arc::new(runtime),
+ }
+ }
+}
+
+impl<A: Accessor> Layer<A> for BlockingLayer {
+ type LayeredAccessor = BlockingAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccessor {
+ BlockingAccessor {
+ inner: inner,
+ runtime: self.runtime.clone(),
+ }
+ }
+}
+
+#[derive(Clone, Debug)]
+pub struct BlockingAccessor<A: Accessor> {
+ inner: A,
+
+ runtime: Arc<runtime::Runtime>,
+}
+
+#[async_trait]
+impl<A: Accessor> LayeredAccessor for BlockingAccessor<A> {
+ type Inner = A;
+ type Reader = A::Reader;
+ type BlockingReader = BlockingReader<A::Reader>;
+ type Writer = A::Writer;
+ type BlockingWriter = BlockingWriter<A::Writer>;
+ type Appender = A::Appender;
+ type Pager = A::Pager;
+ type BlockingPager = BlockingPager<A::Pager>;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ fn metadata(&self) -> AccessorInfo {
+ let mut info = self.inner.info();
+ let capability = info.capability_mut();
+ capability.blocking = true;
+ capability.read_can_next = true;
Review Comment:
> What about `read_can_seek`? I found `BlockingReader` requires this to create
No, it's handled by internally. Layer should not change them.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] yah01 commented on a diff in pull request #2780: feat(layers/blocking): add blocking layer
Posted by "yah01 (via GitHub)" <gi...@apache.org>.
yah01 commented on code in PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#discussion_r1285193537
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ Self {
+ runtime: Arc::new(runtime),
+ }
+ }
+}
+
+impl<A: Accessor> Layer<A> for BlockingLayer {
+ type LayeredAccessor = BlockingAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccessor {
+ BlockingAccessor {
+ inner: inner,
+ runtime: self.runtime.clone(),
+ }
+ }
+}
+
+#[derive(Clone, Debug)]
+pub struct BlockingAccessor<A: Accessor> {
+ inner: A,
+
+ runtime: Arc<runtime::Runtime>,
+}
+
+#[async_trait]
+impl<A: Accessor> LayeredAccessor for BlockingAccessor<A> {
+ type Inner = A;
+ type Reader = A::Reader;
+ type BlockingReader = BlockingReader<A::Reader>;
+ type Writer = A::Writer;
+ type BlockingWriter = BlockingWriter<A::Writer>;
+ type Appender = A::Appender;
+ type Pager = A::Pager;
+ type BlockingPager = BlockingPager<A::Pager>;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ fn metadata(&self) -> AccessorInfo {
+ let mut info = self.inner.info();
+ let capability = info.capability_mut();
+ capability.blocking = true;
+ capability.read_can_next = true;
Review Comment:
> > Should the blocking capability be `false` even with the blocking layer?
>
> It's a bit tricky. We need a new issue to address this. But at least, we should not change `read_can_next`.
What about `read_can_seek`? I found `BlockingReader` requires this to create
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] NobodyXu commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "NobodyXu (via GitHub)" <gi...@apache.org>.
NobodyXu commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1667384924
> Hi @NobodyXu , could you release a new version of openssh-sftp-client? Then we can use it for this, I'm going to bump this directly in this PR to make sure it works
@silver-ymz has released openssh-sftp-client v0.13.8
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] yah01 commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "yah01 (via GitHub)" <gi...@apache.org>.
yah01 commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1667449416
Thanks for you guys @Xuanwo @silver-ymz @NobodyXu, it's impossible to get this work done well without your help!
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] yah01 commented on a diff in pull request #2780: feat(layers/blocking): add blocking layer
Posted by "yah01 (via GitHub)" <gi...@apache.org>.
yah01 commented on code in PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#discussion_r1285193410
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
Review Comment:
> > Or maybe we could use the Handle to the current runtime?
>
> It's possible that we don't have an async runtime. Maybe we need to check the result of [`try_current`](https://docs.rs/tokio/latest/tokio/runtime/struct.Handle.html#method.try_current)?
As the `default()` can't return a `Result`, which means the user can't handle this error, I prefer to not implement the `Default` trait for `BlockingLayer`, just provide a `with_handle(Handle) -> Self` method
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1666806933
> what's the diff between sftp and the other backends, could you help on this?
cc @silver-ymz, do you have any ideas?
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] yah01 commented on a diff in pull request #2780: feat(layers/blocking): add blocking layer
Posted by "yah01 (via GitHub)" <gi...@apache.org>.
yah01 commented on code in PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#discussion_r1285179491
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ Self {
+ runtime: Arc::new(runtime),
+ }
+ }
+}
+
+impl<A: Accessor> Layer<A> for BlockingLayer {
+ type LayeredAccessor = BlockingAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccessor {
+ BlockingAccessor {
+ inner: inner,
+ runtime: self.runtime.clone(),
+ }
+ }
+}
+
+#[derive(Clone, Debug)]
+pub struct BlockingAccessor<A: Accessor> {
+ inner: A,
+
+ runtime: Arc<runtime::Runtime>,
+}
+
+#[async_trait]
+impl<A: Accessor> LayeredAccessor for BlockingAccessor<A> {
+ type Inner = A;
+ type Reader = A::Reader;
+ type BlockingReader = BlockingReader<A::Reader>;
+ type Writer = A::Writer;
+ type BlockingWriter = BlockingWriter<A::Writer>;
+ type Appender = A::Appender;
+ type Pager = A::Pager;
+ type BlockingPager = BlockingPager<A::Pager>;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ fn metadata(&self) -> AccessorInfo {
+ let mut info = self.inner.info();
+ let capability = info.capability_mut();
+ capability.blocking = true;
+ capability.read_can_next = true;
+ capability.read_can_seek = true;
+ info
+ }
+
+ async fn create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
+ self.inner.create_dir(path, args).await
+ }
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ self.inner.read(path, args).await
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ self.inner.write(path, args).await
+ }
+
+ async fn append(&self, path: &str, args: OpAppend) -> Result<(RpAppend, Self::Appender)> {
+ self.inner.append(path, args).await
+ }
+
+ async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ self.inner.copy(from, to, args).await
+ }
+
+ async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
+ self.inner.rename(from, to, args).await
+ }
+
+ async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ self.inner.stat(path, args).await
+ }
+
+ async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+ self.inner.delete(path, args).await
+ }
+
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Pager)> {
+ self.inner.list(path, args).await
+ }
+
+ async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
+ self.inner.presign(path, args).await
+ }
+
+ async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
+ self.inner.batch(args).await
+ }
+
+ fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
+ self.runtime.block_on(self.inner.create_dir(path, args))
+ }
+
+ fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+ self.runtime.block_on(async {
+ let (rp, reader) = self.inner.read(path, args).await?;
+ let blocking_reader = BlockingReader::new(self.runtime.clone(), reader);
+
+ Ok((rp, blocking_reader))
+ })
+ }
+
+ fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
+ self.runtime.block_on(async {
+ let (rp, writer) = self.inner.write(path, args).await?;
+ let blocking_writer = Self::BlockingWriter::new(self.runtime.clone(), writer);
+ Ok((rp, blocking_writer))
+ })
+ }
+
+ fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ self.runtime.block_on(self.inner.copy(from, to, args))
+ }
+
+ fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
+ self.runtime.block_on(self.inner.rename(from, to, args))
+ }
+
+ fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ self.runtime.block_on(self.inner.stat(path, args))
+ }
+
+ fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+ self.runtime.block_on(self.inner.delete(path, args))
+ }
+
+ fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingPager)> {
+ self.runtime.block_on(async {
+ let (rp, pager) = self.inner.list(path, args).await?;
+ let blocking_pager = Self::BlockingPager::new(self.runtime.clone(), pager);
+ Ok((rp, blocking_pager))
+ })
+ }
+}
+
+pub struct BlockingReader<R> {
+ runtime: Arc<runtime::Runtime>,
+ inner: R,
+}
+
+impl<R> BlockingReader<R> {
+ fn new(runtime: Arc<runtime::Runtime>, inner: R) -> Self {
+ Self { runtime, inner }
+ }
+}
+
+// impl<R: oio::Read> oio::Read for BlockingReader<R> {
+// fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
+// self.inner.poll_read(cx, buf)
+// }
+
+// fn poll_seek(&mut self, cx: &mut Context<'_>, pos: io::SeekFrom) -> Poll<Result<u64>> {
+// self.inner.poll_seek(cx, pos)
+// }
+
+// fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
+// self.inner.poll_next(cx)
+// }
+// }
+
+impl<R: oio::Read + 'static> oio::BlockingRead for BlockingReader<R> {
+ fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
+ self.runtime
+ .block_on(oio::ReadExt::read(&mut self.inner, buf))
+ }
+
+ fn seek(&mut self, pos: std::io::SeekFrom) -> Result<u64> {
+ self.runtime.block_on(self.inner.seek(pos))
+ }
+
+ fn next(&mut self) -> Option<Result<Bytes>> {
+ self.runtime.block_on(self.inner.next())
+ }
+}
+
+pub struct BlockingWriter<W> {
Review Comment:
Great idea!
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1667170661
I have fixed the seek support for blocking reader at: https://github.com/apache/incubator-opendal/pull/2799
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1667422007
> Thank you, but @silver-ymz is the one who fixed the `<OwnedHandle as Drop>::drop`.
Fantastic! I'm thrilled to have @NobodyXu and @silver-ymz as part of our community!
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] silver-ymz commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "silver-ymz (via GitHub)" <gi...@apache.org>.
silver-ymz commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1666814978
It seems that the dropping of sftp client does some magical things.
```
thread '<unnamed>' panicked at 'there is no reactor running, must be called from the context of a Tokio 1.x runtime', /home/runner/.cargo/registry/src/index.crates.io-6f17d22bba15001f/openssh-sftp-client-0.13.6/src/file/tokio_compat_file.rs:763:9
```
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1667221495
> I'm guessing we can add following unit tests in blocking layer:
>
> * init in blocking context, drop in blocking context
> * init in blocking context, drop in async context
> * init in async context, drop in blocking context
> * init in async context, drop in async context
This seems not a blocker here, we can improve this in later PRs.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#discussion_r1285191612
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ Self {
+ runtime: Arc::new(runtime),
+ }
+ }
+}
+
+impl<A: Accessor> Layer<A> for BlockingLayer {
+ type LayeredAccessor = BlockingAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccessor {
+ BlockingAccessor {
+ inner: inner,
+ runtime: self.runtime.clone(),
+ }
+ }
+}
+
+#[derive(Clone, Debug)]
+pub struct BlockingAccessor<A: Accessor> {
+ inner: A,
+
+ runtime: Arc<runtime::Runtime>,
+}
+
+#[async_trait]
+impl<A: Accessor> LayeredAccessor for BlockingAccessor<A> {
+ type Inner = A;
+ type Reader = A::Reader;
+ type BlockingReader = BlockingReader<A::Reader>;
+ type Writer = A::Writer;
+ type BlockingWriter = BlockingWriter<A::Writer>;
+ type Appender = A::Appender;
+ type Pager = A::Pager;
+ type BlockingPager = BlockingPager<A::Pager>;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ fn metadata(&self) -> AccessorInfo {
+ let mut info = self.inner.info();
+ let capability = info.capability_mut();
+ capability.blocking = true;
+ capability.read_can_next = true;
Review Comment:
> Should the blocking capability be `false` even with the blocking layer?
It's a bit tricky. We need a new issue to address this. But at least, we should not change `read_can_next`.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#discussion_r1285197008
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ Self {
+ runtime: Arc::new(runtime),
+ }
+ }
+}
+
+impl<A: Accessor> Layer<A> for BlockingLayer {
+ type LayeredAccessor = BlockingAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccessor {
+ BlockingAccessor {
+ inner: inner,
+ runtime: self.runtime.clone(),
+ }
+ }
+}
+
+#[derive(Clone, Debug)]
+pub struct BlockingAccessor<A: Accessor> {
+ inner: A,
+
+ runtime: Arc<runtime::Runtime>,
+}
+
+#[async_trait]
+impl<A: Accessor> LayeredAccessor for BlockingAccessor<A> {
+ type Inner = A;
+ type Reader = A::Reader;
+ type BlockingReader = BlockingReader<A::Reader>;
+ type Writer = A::Writer;
+ type BlockingWriter = BlockingWriter<A::Writer>;
+ type Appender = A::Appender;
+ type Pager = A::Pager;
+ type BlockingPager = BlockingPager<A::Pager>;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ fn metadata(&self) -> AccessorInfo {
+ let mut info = self.inner.info();
+ let capability = info.capability_mut();
+ capability.blocking = true;
+ capability.read_can_next = true;
Review Comment:
`CompleteLayer` implements the seek operation internall, but we didn't do the same thing for blocking seek. We need to implement this. Change `read_can_seek` is not helpful here.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] yah01 commented on a diff in pull request #2780: feat(layers/blocking): add blocking layer
Posted by "yah01 (via GitHub)" <gi...@apache.org>.
yah01 commented on code in PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#discussion_r1285196043
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ Self {
+ runtime: Arc::new(runtime),
+ }
+ }
+}
+
+impl<A: Accessor> Layer<A> for BlockingLayer {
+ type LayeredAccessor = BlockingAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccessor {
+ BlockingAccessor {
+ inner: inner,
+ runtime: self.runtime.clone(),
+ }
+ }
+}
+
+#[derive(Clone, Debug)]
+pub struct BlockingAccessor<A: Accessor> {
+ inner: A,
+
+ runtime: Arc<runtime::Runtime>,
+}
+
+#[async_trait]
+impl<A: Accessor> LayeredAccessor for BlockingAccessor<A> {
+ type Inner = A;
+ type Reader = A::Reader;
+ type BlockingReader = BlockingReader<A::Reader>;
+ type Writer = A::Writer;
+ type BlockingWriter = BlockingWriter<A::Writer>;
+ type Appender = A::Appender;
+ type Pager = A::Pager;
+ type BlockingPager = BlockingPager<A::Pager>;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ fn metadata(&self) -> AccessorInfo {
+ let mut info = self.inner.info();
+ let capability = info.capability_mut();
+ capability.blocking = true;
+ capability.read_can_next = true;
Review Comment:
https://github.com/yah01/incubator-opendal/blob/2fbe8b59f5f2bdc1d5d2f224eface5b9a57194c4/core/src/types/operator/blocking_operator.rs#L413-L422
Is it possible to remove the requirement for `read_can_seek`? As the async `Reader` doesn't require this
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1667204669
Hi, @yah01, it's better not to do force push..
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] yah01 commented on a diff in pull request #2780: feat(layers/blocking): add blocking layer
Posted by "yah01 (via GitHub)" <gi...@apache.org>.
yah01 commented on code in PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#discussion_r1285187406
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ Self {
+ runtime: Arc::new(runtime),
+ }
+ }
+}
+
+impl<A: Accessor> Layer<A> for BlockingLayer {
+ type LayeredAccessor = BlockingAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccessor {
+ BlockingAccessor {
+ inner: inner,
+ runtime: self.runtime.clone(),
+ }
+ }
+}
+
+#[derive(Clone, Debug)]
+pub struct BlockingAccessor<A: Accessor> {
+ inner: A,
+
+ runtime: Arc<runtime::Runtime>,
+}
+
+#[async_trait]
+impl<A: Accessor> LayeredAccessor for BlockingAccessor<A> {
+ type Inner = A;
+ type Reader = A::Reader;
+ type BlockingReader = BlockingReader<A::Reader>;
+ type Writer = A::Writer;
+ type BlockingWriter = BlockingWriter<A::Writer>;
+ type Appender = A::Appender;
+ type Pager = A::Pager;
+ type BlockingPager = BlockingPager<A::Pager>;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ fn metadata(&self) -> AccessorInfo {
+ let mut info = self.inner.info();
+ let capability = info.capability_mut();
+ capability.blocking = true;
+ capability.read_can_next = true;
Review Comment:
Should the blocking capability be `false` even with the blocking layer?
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] yah01 commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "yah01 (via GitHub)" <gi...@apache.org>.
yah01 commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1667364177
Hi @NobodyXu , could you release a new version of openssh-sftp-client? Then we can use it for this, I'm going to bump this directly in this PR to make sure it works
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#discussion_r1284283153
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
Review Comment:
Start a runtime in async context chould be a problem. Maybe we should spawn a new thread to do that?
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ Self {
+ runtime: Arc::new(runtime),
+ }
+ }
+}
+
+impl<A: Accessor> Layer<A> for BlockingLayer {
+ type LayeredAccessor = BlockingAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccessor {
+ BlockingAccessor {
+ inner: inner,
+ runtime: self.runtime.clone(),
+ }
+ }
+}
+
+#[derive(Clone, Debug)]
+pub struct BlockingAccessor<A: Accessor> {
+ inner: A,
+
+ runtime: Arc<runtime::Runtime>,
+}
+
+#[async_trait]
+impl<A: Accessor> LayeredAccessor for BlockingAccessor<A> {
+ type Inner = A;
+ type Reader = A::Reader;
+ type BlockingReader = BlockingReader<A::Reader>;
+ type Writer = A::Writer;
+ type BlockingWriter = BlockingWriter<A::Writer>;
+ type Appender = A::Appender;
+ type Pager = A::Pager;
+ type BlockingPager = BlockingPager<A::Pager>;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ fn metadata(&self) -> AccessorInfo {
+ let mut info = self.inner.info();
+ let capability = info.capability_mut();
+ capability.blocking = true;
+ capability.read_can_next = true;
Review Comment:
It's better not to change them.
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ Self {
+ runtime: Arc::new(runtime),
+ }
+ }
+}
+
+impl<A: Accessor> Layer<A> for BlockingLayer {
+ type LayeredAccessor = BlockingAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccessor {
+ BlockingAccessor {
+ inner: inner,
+ runtime: self.runtime.clone(),
+ }
+ }
+}
+
+#[derive(Clone, Debug)]
+pub struct BlockingAccessor<A: Accessor> {
+ inner: A,
+
+ runtime: Arc<runtime::Runtime>,
+}
+
+#[async_trait]
+impl<A: Accessor> LayeredAccessor for BlockingAccessor<A> {
+ type Inner = A;
+ type Reader = A::Reader;
+ type BlockingReader = BlockingReader<A::Reader>;
+ type Writer = A::Writer;
+ type BlockingWriter = BlockingWriter<A::Writer>;
+ type Appender = A::Appender;
+ type Pager = A::Pager;
+ type BlockingPager = BlockingPager<A::Pager>;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ fn metadata(&self) -> AccessorInfo {
+ let mut info = self.inner.info();
+ let capability = info.capability_mut();
+ capability.blocking = true;
+ capability.read_can_next = true;
+ capability.read_can_seek = true;
+ info
+ }
+
+ async fn create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
+ self.inner.create_dir(path, args).await
+ }
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ self.inner.read(path, args).await
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ self.inner.write(path, args).await
+ }
+
+ async fn append(&self, path: &str, args: OpAppend) -> Result<(RpAppend, Self::Appender)> {
+ self.inner.append(path, args).await
+ }
+
+ async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ self.inner.copy(from, to, args).await
+ }
+
+ async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
+ self.inner.rename(from, to, args).await
+ }
+
+ async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ self.inner.stat(path, args).await
+ }
+
+ async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+ self.inner.delete(path, args).await
+ }
+
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Pager)> {
+ self.inner.list(path, args).await
+ }
+
+ async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
+ self.inner.presign(path, args).await
+ }
+
+ async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
+ self.inner.batch(args).await
+ }
+
+ fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
+ self.runtime.block_on(self.inner.create_dir(path, args))
+ }
+
+ fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+ self.runtime.block_on(async {
+ let (rp, reader) = self.inner.read(path, args).await?;
+ let blocking_reader = BlockingReader::new(self.runtime.clone(), reader);
+
+ Ok((rp, blocking_reader))
+ })
+ }
+
+ fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
+ self.runtime.block_on(async {
+ let (rp, writer) = self.inner.write(path, args).await?;
+ let blocking_writer = Self::BlockingWriter::new(self.runtime.clone(), writer);
+ Ok((rp, blocking_writer))
+ })
+ }
+
+ fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ self.runtime.block_on(self.inner.copy(from, to, args))
+ }
+
+ fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
+ self.runtime.block_on(self.inner.rename(from, to, args))
+ }
+
+ fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ self.runtime.block_on(self.inner.stat(path, args))
+ }
+
+ fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+ self.runtime.block_on(self.inner.delete(path, args))
+ }
+
+ fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingPager)> {
+ self.runtime.block_on(async {
+ let (rp, pager) = self.inner.list(path, args).await?;
+ let blocking_pager = Self::BlockingPager::new(self.runtime.clone(), pager);
+ Ok((rp, blocking_pager))
+ })
+ }
+}
+
+pub struct BlockingReader<R> {
+ runtime: Arc<runtime::Runtime>,
+ inner: R,
+}
+
+impl<R> BlockingReader<R> {
+ fn new(runtime: Arc<runtime::Runtime>, inner: R) -> Self {
+ Self { runtime, inner }
+ }
+}
+
+// impl<R: oio::Read> oio::Read for BlockingReader<R> {
Review Comment:
Let's remove not needed code.
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ Self {
+ runtime: Arc::new(runtime),
+ }
+ }
+}
+
+impl<A: Accessor> Layer<A> for BlockingLayer {
+ type LayeredAccessor = BlockingAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccessor {
+ BlockingAccessor {
+ inner: inner,
+ runtime: self.runtime.clone(),
+ }
+ }
+}
+
+#[derive(Clone, Debug)]
+pub struct BlockingAccessor<A: Accessor> {
+ inner: A,
+
+ runtime: Arc<runtime::Runtime>,
Review Comment:
Maybe we should implement drop by ourselves.
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ Self {
+ runtime: Arc::new(runtime),
+ }
+ }
+}
+
+impl<A: Accessor> Layer<A> for BlockingLayer {
+ type LayeredAccessor = BlockingAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccessor {
+ BlockingAccessor {
+ inner: inner,
+ runtime: self.runtime.clone(),
+ }
+ }
+}
+
+#[derive(Clone, Debug)]
+pub struct BlockingAccessor<A: Accessor> {
+ inner: A,
+
+ runtime: Arc<runtime::Runtime>,
+}
+
+#[async_trait]
+impl<A: Accessor> LayeredAccessor for BlockingAccessor<A> {
+ type Inner = A;
+ type Reader = A::Reader;
+ type BlockingReader = BlockingReader<A::Reader>;
+ type Writer = A::Writer;
+ type BlockingWriter = BlockingWriter<A::Writer>;
+ type Appender = A::Appender;
+ type Pager = A::Pager;
+ type BlockingPager = BlockingPager<A::Pager>;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ fn metadata(&self) -> AccessorInfo {
+ let mut info = self.inner.info();
+ let capability = info.capability_mut();
+ capability.blocking = true;
+ capability.read_can_next = true;
+ capability.read_can_seek = true;
+ info
+ }
+
+ async fn create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
+ self.inner.create_dir(path, args).await
+ }
+
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ self.inner.read(path, args).await
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ self.inner.write(path, args).await
+ }
+
+ async fn append(&self, path: &str, args: OpAppend) -> Result<(RpAppend, Self::Appender)> {
+ self.inner.append(path, args).await
+ }
+
+ async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ self.inner.copy(from, to, args).await
+ }
+
+ async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
+ self.inner.rename(from, to, args).await
+ }
+
+ async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ self.inner.stat(path, args).await
+ }
+
+ async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+ self.inner.delete(path, args).await
+ }
+
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Pager)> {
+ self.inner.list(path, args).await
+ }
+
+ async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
+ self.inner.presign(path, args).await
+ }
+
+ async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
+ self.inner.batch(args).await
+ }
+
+ fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
+ self.runtime.block_on(self.inner.create_dir(path, args))
+ }
+
+ fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+ self.runtime.block_on(async {
+ let (rp, reader) = self.inner.read(path, args).await?;
+ let blocking_reader = BlockingReader::new(self.runtime.clone(), reader);
+
+ Ok((rp, blocking_reader))
+ })
+ }
+
+ fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
+ self.runtime.block_on(async {
+ let (rp, writer) = self.inner.write(path, args).await?;
+ let blocking_writer = Self::BlockingWriter::new(self.runtime.clone(), writer);
+ Ok((rp, blocking_writer))
+ })
+ }
+
+ fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+ self.runtime.block_on(self.inner.copy(from, to, args))
+ }
+
+ fn blocking_rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
+ self.runtime.block_on(self.inner.rename(from, to, args))
+ }
+
+ fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ self.runtime.block_on(self.inner.stat(path, args))
+ }
+
+ fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+ self.runtime.block_on(self.inner.delete(path, args))
+ }
+
+ fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingPager)> {
+ self.runtime.block_on(async {
+ let (rp, pager) = self.inner.list(path, args).await?;
+ let blocking_pager = Self::BlockingPager::new(self.runtime.clone(), pager);
+ Ok((rp, blocking_pager))
+ })
+ }
+}
+
+pub struct BlockingReader<R> {
+ runtime: Arc<runtime::Runtime>,
+ inner: R,
+}
+
+impl<R> BlockingReader<R> {
+ fn new(runtime: Arc<runtime::Runtime>, inner: R) -> Self {
+ Self { runtime, inner }
+ }
+}
+
+// impl<R: oio::Read> oio::Read for BlockingReader<R> {
+// fn poll_read(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
+// self.inner.poll_read(cx, buf)
+// }
+
+// fn poll_seek(&mut self, cx: &mut Context<'_>, pos: io::SeekFrom) -> Poll<Result<u64>> {
+// self.inner.poll_seek(cx, pos)
+// }
+
+// fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
+// self.inner.poll_next(cx)
+// }
+// }
+
+impl<R: oio::Read + 'static> oio::BlockingRead for BlockingReader<R> {
+ fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
+ self.runtime
+ .block_on(oio::ReadExt::read(&mut self.inner, buf))
+ }
+
+ fn seek(&mut self, pos: std::io::SeekFrom) -> Result<u64> {
+ self.runtime.block_on(self.inner.seek(pos))
+ }
+
+ fn next(&mut self) -> Option<Result<Bytes>> {
+ self.runtime.block_on(self.inner.next())
+ }
+}
+
+pub struct BlockingWriter<W> {
Review Comment:
We can implement a `BlockingWrapper` and implement different traits for it.
##########
core/src/types/operator/builder.rs:
##########
@@ -371,6 +371,11 @@ impl<A: Accessor> OperatorBuilder<A> {
pub fn finish(self) -> Operator {
let ob = self.layer(TypeEraseLayer);
- Operator::from_inner(Arc::new(ob.accessor) as FusedAccessor)
+ let op = Operator::from_inner(Arc::new(ob.accessor) as FusedAccessor);
+ if !op.info().can_blocking() {
+ op.layer(BlockingLayer::default())
Review Comment:
blocking layer should not be enabled by default since it will introduce cost for users don't want this. We can enable blocking layer in behavior tests directly for test.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1667442651
nextcloud failed for file locked, we should address it by add redis cache for it. Not related with this PR.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] NobodyXu commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "NobodyXu (via GitHub)" <gi...@apache.org>.
NobodyXu commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1666855533
openssh-sftp-client v0.13.7 has released, which should fix this.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] chitralverma commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "chitralverma (via GitHub)" <gi...@apache.org>.
chitralverma commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1667669790
@yah01 amazing, couple of questions regarding this,
- Are there any user-facing examples to use the blocking API?
- Which services now support/ don't support blocking API?
- Is there a separate PR required to update the docs.rs with service-level capabilities? Since this will be part of the next release
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#discussion_r1285195434
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ Self {
+ runtime: Arc::new(runtime),
+ }
+ }
+}
+
+impl<A: Accessor> Layer<A> for BlockingLayer {
+ type LayeredAccessor = BlockingAccessor<A>;
+
+ fn layer(&self, inner: A) -> Self::LayeredAccessor {
+ BlockingAccessor {
+ inner: inner,
+ runtime: self.runtime.clone(),
+ }
+ }
+}
+
+#[derive(Clone, Debug)]
+pub struct BlockingAccessor<A: Accessor> {
+ inner: A,
+
+ runtime: Arc<runtime::Runtime>,
+}
+
+#[async_trait]
+impl<A: Accessor> LayeredAccessor for BlockingAccessor<A> {
+ type Inner = A;
+ type Reader = A::Reader;
+ type BlockingReader = BlockingReader<A::Reader>;
+ type Writer = A::Writer;
+ type BlockingWriter = BlockingWriter<A::Writer>;
+ type Appender = A::Appender;
+ type Pager = A::Pager;
+ type BlockingPager = BlockingPager<A::Pager>;
+
+ fn inner(&self) -> &Self::Inner {
+ &self.inner
+ }
+
+ fn metadata(&self) -> AccessorInfo {
+ let mut info = self.inner.info();
+ let capability = info.capability_mut();
+ capability.blocking = true;
+ capability.read_can_next = true;
Review Comment:
Tracked at https://github.com/apache/incubator-opendal/issues/2790
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1666815272
> It seems that the dropping of sftp client does some magical things.
Also cc @NobodyXu for some ideas.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#discussion_r1285195590
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
Review Comment:
BlockingLayer's method is also OpenDAL's public API. How about do those thing at `BlockingLayer::create() -> Result<Self>`?
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] yah01 commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "yah01 (via GitHub)" <gi...@apache.org>.
yah01 commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1666806617
Hi @Xuanwo The sftp related tests failed, but I don't get it how this occurs, I did clone the `Handle` from `RUNTIME`, what's the diff between sftp and the other backends, could you help on this?
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] NobodyXu commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "NobodyXu (via GitHub)" <gi...@apache.org>.
NobodyXu commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1667107074
> Hi @Xuanwo, the sftp problem and the `BlockingReader` problem are blocking this PR, just wonder whether you are working on them? I could resolve them if you haven't started (should I create two new PRs?)
For the `openssh-sftp-client` bug, a simple bump is required, IMO you can open a new PR for that.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1667215471
I'm guessing we can add following unit tests in blocking layer:
- init in blocking context, drop in blocking context
- init in blocking context, drop in async context
- init in async context, drop in blocking context
- init in async context, drop in async context
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] NobodyXu commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "NobodyXu (via GitHub)" <gi...@apache.org>.
NobodyXu commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1667418751
Thank you, but @silver-ymz is the one who fixed the `<OwnedHandle as Drop>::drop`.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1667226428
> @Xuanwo @NobodyXu It seems the sftp client is still in trouble, any idea?
https://github.com/openssh-rust/openssh-sftp-client/blob/main/src/handle.rs#L22C15-L59
Seems `OwnedHandle` should also need some work.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1667679088
> * Are there any user-facing examples to use the blocking API?
Yes, docs provided at `BlockingOperator`. This PR doesn't change our public API.
> Which services now support/ don't support blocking API?
By adding blocking layers, services that only provide async API can have blocking API now.
> Is there a separate PR required to update the docs.rs with service-level capabilities? Since this will be part of the next release
This question is related to https://github.com/apache/incubator-opendal/issues/2790. I expect to address them before next release.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#discussion_r1285195845
##########
core/tests/behavior/blocking_list.rs:
##########
@@ -26,10 +26,17 @@ use crate::*;
pub fn behavior_blocking_list_tests(op: &Operator) -> Vec<Trial> {
let cap = op.info().capability();
- if !(cap.read && cap.write && cap.copy && cap.blocking && cap.list) {
+ if !(cap.read && cap.write && cap.copy && cap.list) {
return vec![];
}
+ let mut op = op.clone();
Review Comment:
We should avoid modifying the blocking test suites. Instead, we can make changes at https://github.com/apache/incubator-opendal/blob/main/core/tests/behavior/utils.rs#L46.
This approach is more aligned with how OpenDAL users will interact with the software.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1666871561
> openssh-sftp-client v0.13.7 has released, which should fix this.
Great! Thanks a lot!
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1667413825
> has released openssh-sftp-client v0.13.8
@NobodyXu, you are so awesome!
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] yah01 commented on a diff in pull request #2780: feat(layers/blocking): add blocking layer
Posted by "yah01 (via GitHub)" <gi...@apache.org>.
yah01 commented on code in PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#discussion_r1285179195
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
Review Comment:
Or maybe we could use the Handle to the current runtime? Tried this and it works
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] yah01 commented on a diff in pull request #2780: feat(layers/blocking): add blocking layer
Posted by "yah01 (via GitHub)" <gi...@apache.org>.
yah01 commented on code in PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#discussion_r1285194941
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
Review Comment:
> > I prefer to not implement the `Default` trait for `BlockingLayer`, just provide a `with_handle(Handle) -> Self` method
>
> Removing `default()` is ok, since it's not a requirement for `Layer`.
>
> But adding a new `with_handle()` API for `Layer` since a bit tricky. We don't want to expose this as our public API.
I added it as `BlockingLayer`'s method, but not `Layer`, just like the `LoggingLayer`'s `with_error_level`, is this OK?
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#discussion_r1285193716
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
Review Comment:
> I prefer to not implement the `Default` trait for `BlockingLayer`, just provide a `with_handle(Handle) -> Self` method
Removing `default()` is ok, since it's not a requirement for `Layer`.
But adding a new `with_handle()` API for `Layer` since a bit tricky. We don't want to expose this as our public API.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] NobodyXu commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "NobodyXu (via GitHub)" <gi...@apache.org>.
NobodyXu commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1666847092
> It seems that the dropping of sftp client does some magical things.
>
> ```
> thread '<unnamed>' panicked at 'there is no reactor running, must be called from the context of a Tokio 1.x runtime', /home/runner/.cargo/registry/src/index.crates.io-6f17d22bba15001f/openssh-sftp-client-0.13.6/src/file/tokio_compat_file.rs:763:9
> ```
[`TokioCompatFile::drop`](https://docs.rs/openssh-sftp-client/latest/src/openssh_sftp_client/file/tokio_compat_file.rs.html#763) calls `tokio::spawn` to wait on all `write_futures` and log the errors if any.
We can change it to only do that if a `tokio::Runtime` is available.
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] yah01 commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "yah01 (via GitHub)" <gi...@apache.org>.
yah01 commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1667104118
Hi @Xuanwo, the sftp problem and the `BlockingReader` problem are blocking this PR, just wonder whether you are working on them? I could resolve them if you haven't started (should I create two new PRs?)
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] yah01 commented on pull request #2780: feat(layers/blocking): add blocking layer
Posted by "yah01 (via GitHub)" <gi...@apache.org>.
yah01 commented on PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#issuecomment-1667225732
@Xuanwo @NobodyXu It seems the sftp client is still in trouble, any idea?
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-opendal] Xuanwo commented on a diff in pull request #2780: feat(layers/blocking): add blocking layer
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #2780:
URL: https://github.com/apache/incubator-opendal/pull/2780#discussion_r1285191518
##########
core/src/layers/blocking.rs:
##########
@@ -0,0 +1,262 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes;
+use bytes::Bytes;
+use tokio::runtime;
+
+use crate::raw::oio::ReadExt;
+use crate::raw::*;
+use crate::*;
+
+/// Add blocking API support for every operations.
+///
+/// # Blocking API
+///
+/// - This layer is auto-added to the operator if it's accessor doesn't support blocking APIs.
+///
+/// Tracking issue: #2678
+#[derive(Debug, Clone)]
+pub struct BlockingLayer {
+ runtime: Arc<runtime::Runtime>,
+}
+
+impl Default for BlockingLayer {
+ fn default() -> Self {
+ let runtime = runtime::Builder::new_current_thread()
Review Comment:
> Or maybe we could use the Handle to the current runtime?
It's possible that we don't have an async runtime. Maybe we need to check the result of [`try_current`](https://docs.rs/tokio/latest/tokio/runtime/struct.Handle.html#method.try_current)?
--
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: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org