opendal_service_monoiofs/
backend.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::fmt::Debug;
19use std::io;
20use std::path::PathBuf;
21use std::sync::Arc;
22
23use monoio::fs::OpenOptions;
24use opendal_core::raw::*;
25use opendal_core::*;
26
27use super::config::MonoiofsConfig;
28use super::core::BUFFER_SIZE;
29use super::core::MonoiofsCore;
30use super::deleter::MonoiofsDeleter;
31use super::reader::*;
32
33/// File system support via [`monoio`].
34#[doc = include_str!("docs.md")]
35#[derive(Debug, Default)]
36pub struct MonoiofsBuilder {
37    pub(super) config: MonoiofsConfig,
38}
39
40impl MonoiofsBuilder {
41    /// Set root of this backend.
42    ///
43    /// All operations will happen under this root.
44    pub fn root(mut self, root: &str) -> Self {
45        self.config.root = if root.is_empty() {
46            None
47        } else {
48            Some(root.to_string())
49        };
50        self
51    }
52}
53
54impl Builder for MonoiofsBuilder {
55    type Config = MonoiofsConfig;
56
57    fn build(self) -> Result<impl Service> {
58        let root = self.config.root.map(PathBuf::from).ok_or(
59            Error::new(ErrorKind::ConfigInvalid, "root is not specified")
60                .with_operation("Builder::build"),
61        )?;
62        if let Err(e) = std::fs::metadata(&root)
63            && e.kind() == io::ErrorKind::NotFound
64        {
65            std::fs::create_dir_all(&root).map_err(|e| {
66                Error::new(ErrorKind::Unexpected, "create root dir failed")
67                    .with_operation("Builder::build")
68                    .with_context("root", root.to_string_lossy())
69                    .set_source(e)
70            })?;
71        }
72        let root = root.canonicalize().map_err(|e| {
73            Error::new(
74                ErrorKind::Unexpected,
75                "canonicalize of root directory failed",
76            )
77            .with_operation("Builder::build")
78            .with_context("root", root.to_string_lossy())
79            .set_source(e)
80        })?;
81        let worker_threads = 1; // TODO: test concurrency and default to available_parallelism and bind cpu
82        let io_uring_entries = 1024;
83        Ok(MonoiofsBackend {
84            core: Arc::new(MonoiofsCore::new(root, worker_threads, io_uring_entries)),
85        })
86    }
87}
88
89#[derive(Debug, Clone)]
90pub struct MonoiofsBackend {
91    pub(crate) core: Arc<MonoiofsCore>,
92}
93
94impl Service for MonoiofsBackend {
95    type Reader = oio::PositionReader<MonoiofsPositionReader>;
96    type Writer = MonoiofsLazyWriter;
97    type Lister = ();
98    type Deleter = oio::OneShotDeleter<MonoiofsDeleter>;
99    type Copier = oio::OneShotCopier;
100    type Composer = ();
101
102    fn info(&self) -> ServiceInfo {
103        self.core.info.clone()
104    }
105
106    fn capability(&self) -> Capability {
107        self.core.capability
108    }
109
110    async fn stat(&self, _ctx: &OperationContext, path: &str, _args: OpStat) -> Result<RpStat> {
111        let path = self.core.prepare_path(path)?;
112        let meta = self
113            .core
114            .dispatch(move || monoio::fs::metadata(path))
115            .await
116            .map_err(new_std_io_error)?;
117        let mode = if meta.is_dir() {
118            EntryMode::DIR
119        } else if meta.is_file() {
120            EntryMode::FILE
121        } else {
122            EntryMode::Unknown
123        };
124        let mut m = match mode {
125            EntryMode::FILE => MetadataBuilder::file(meta.len()),
126            EntryMode::DIR => MetadataBuilder::dir(),
127            EntryMode::Unknown => MetadataBuilder::unknown(),
128        };
129        m.last_modified(Timestamp::try_from(
130            meta.modified().map_err(new_std_io_error)?,
131        )?);
132        Ok(RpStat::new(m.build()))
133    }
134    fn read(&self, _ctx: &OperationContext, path: &str, _args: OpRead) -> Result<Self::Reader> {
135        let path = self.core.prepare_path(path)?;
136        Ok(oio::PositionReader::new(MonoiofsPositionReader::new(
137            self.core.clone(),
138            path,
139        )))
140    }
141
142    fn write(&self, _ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
143        Ok(MonoiofsLazyWriter::new(self.core.clone(), path, args))
144    }
145
146    fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
147        let output: oio::OneShotDeleter<MonoiofsDeleter> = {
148            Ok(oio::OneShotDeleter::new(MonoiofsDeleter::new(
149                self.core.clone(),
150            )))
151        }?;
152
153        Ok(output)
154    }
155
156    fn list(&self, _ctx: &OperationContext, _path: &str, _args: OpList) -> Result<Self::Lister> {
157        Err(Error::new(
158            ErrorKind::Unsupported,
159            "operation is not supported",
160        ))
161    }
162
163    async fn rename(
164        &self,
165        _ctx: &OperationContext,
166        from: &str,
167        to: &str,
168        _args: OpRename,
169    ) -> Result<RpRename> {
170        let from = self.core.prepare_path(from)?;
171        // ensure file exists
172        self.core
173            .dispatch({
174                let from = from.clone();
175                move || monoio::fs::metadata(from)
176            })
177            .await
178            .map_err(new_std_io_error)?;
179        let to = self.core.prepare_write_path(to).await?;
180        self.core
181            .dispatch(move || monoio::fs::rename(from, to))
182            .await
183            .map_err(new_std_io_error)?;
184        Ok(RpRename::default())
185    }
186
187    async fn create_dir(
188        &self,
189        _ctx: &OperationContext,
190        path: &str,
191        _args: OpCreateDir,
192    ) -> Result<RpCreateDir> {
193        let path = self.core.prepare_path(path)?;
194        self.core
195            .dispatch(move || monoio::fs::create_dir_all(path))
196            .await
197            .map_err(new_std_io_error)?;
198        Ok(RpCreateDir::default())
199    }
200
201    fn copy(
202        &self,
203        _ctx: &OperationContext,
204        from: &str,
205        to: &str,
206        _args: OpCopy,
207    ) -> Result<Self::Copier> {
208        let core = self.core.clone();
209        let from = self.core.prepare_path(from)?;
210        let to = to.to_string();
211
212        let copier = oio::OneShotCopier::new(async move {
213            // ensure file exists
214            core.dispatch({
215                let from = from.clone();
216                move || monoio::fs::metadata(from)
217            })
218            .await
219            .map_err(new_std_io_error)?;
220            let to = core.prepare_write_path(&to).await?;
221            core.dispatch({
222                let core = core.clone();
223                move || async move {
224                    let from = OpenOptions::new().read(true).open(from).await?;
225                    let to = OpenOptions::new()
226                        .write(true)
227                        .create(true)
228                        .truncate(true)
229                        .open(to)
230                        .await?;
231
232                    // AsyncReadRent and AsyncWriteRent is not implemented
233                    // for File, so we can't write this:
234                    // monoio::io::copy(&mut from, &mut to).await?;
235
236                    let mut pos = 0;
237                    // allocate and resize buffer
238                    let mut buf = core.buf_pool.get();
239                    // set capacity of buf to exact size to avoid excessive read
240                    buf.reserve(BUFFER_SIZE);
241                    let _ = buf.split_off(BUFFER_SIZE);
242
243                    loop {
244                        let result;
245                        (result, buf) = from.read_at(buf, pos).await;
246                        if result? == 0 {
247                            // EOF
248                            break;
249                        }
250                        let result;
251                        (result, buf) = to.write_all_at(buf, pos).await;
252                        result?;
253                        pos += buf.len() as u64;
254                        buf.clear();
255                    }
256                    core.buf_pool.put(buf);
257                    Ok(pos)
258                }
259            })
260            .await
261            .map_err(new_std_io_error)
262            .map(|size| {
263                let metadata = MetadataBuilder::file(size);
264                metadata.build()
265            })
266        });
267
268        Ok(copier)
269    }
270
271    async fn presign(
272        &self,
273        _ctx: &OperationContext,
274        _path: &str,
275        _args: OpPresign,
276    ) -> Result<RpPresign> {
277        Err(Error::new(
278            ErrorKind::Unsupported,
279            "operation is not supported",
280        ))
281    }
282}
opendal_service_monoiofs/backend.rs

opendal_service_monoiofs/
backend.rs
