opendal/services/monoiofs/
backend.rs1use std::fmt::Debug;
19use std::io;
20use std::path::PathBuf;
21use std::sync::Arc;
22
23use chrono::DateTime;
24use monoio::fs::OpenOptions;
25
26use super::core::MonoiofsCore;
27use super::core::BUFFER_SIZE;
28use super::delete::MonoiofsDeleter;
29use super::reader::MonoiofsReader;
30use super::writer::MonoiofsWriter;
31use crate::raw::*;
32use crate::services::MonoiofsConfig;
33use crate::*;
34
35impl Configurator for MonoiofsConfig {
36 type Builder = MonoiofsBuilder;
37 fn into_builder(self) -> Self::Builder {
38 MonoiofsBuilder { config: self }
39 }
40}
41
42#[doc = include_str!("docs.md")]
44#[derive(Default, Debug)]
45pub struct MonoiofsBuilder {
46 config: MonoiofsConfig,
47}
48
49impl MonoiofsBuilder {
50 pub fn root(mut self, root: &str) -> Self {
54 self.config.root = if root.is_empty() {
55 None
56 } else {
57 Some(root.to_string())
58 };
59 self
60 }
61}
62
63impl Builder for MonoiofsBuilder {
64 const SCHEME: Scheme = Scheme::Monoiofs;
65 type Config = MonoiofsConfig;
66
67 fn build(self) -> Result<impl Access> {
68 let root = self.config.root.map(PathBuf::from).ok_or(
69 Error::new(ErrorKind::ConfigInvalid, "root is not specified")
70 .with_operation("Builder::build"),
71 )?;
72 if let Err(e) = std::fs::metadata(&root) {
73 if e.kind() == io::ErrorKind::NotFound {
74 std::fs::create_dir_all(&root).map_err(|e| {
75 Error::new(ErrorKind::Unexpected, "create root dir failed")
76 .with_operation("Builder::build")
77 .with_context("root", root.to_string_lossy())
78 .set_source(e)
79 })?;
80 }
81 }
82 let root = root.canonicalize().map_err(|e| {
83 Error::new(
84 ErrorKind::Unexpected,
85 "canonicalize of root directory failed",
86 )
87 .with_operation("Builder::build")
88 .with_context("root", root.to_string_lossy())
89 .set_source(e)
90 })?;
91 let worker_threads = 1; let io_uring_entries = 1024;
93 Ok(MonoiofsBackend {
94 core: Arc::new(MonoiofsCore::new(root, worker_threads, io_uring_entries)),
95 })
96 }
97}
98
99#[derive(Debug, Clone)]
100pub struct MonoiofsBackend {
101 core: Arc<MonoiofsCore>,
102}
103
104impl Access for MonoiofsBackend {
105 type Reader = MonoiofsReader;
106 type Writer = MonoiofsWriter;
107 type Lister = ();
108 type Deleter = oio::OneShotDeleter<MonoiofsDeleter>;
109
110 fn info(&self) -> Arc<AccessorInfo> {
111 self.core.info.clone()
112 }
113
114 async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
115 let path = self.core.prepare_path(path);
116 let meta = self
117 .core
118 .dispatch(move || monoio::fs::metadata(path))
119 .await
120 .map_err(new_std_io_error)?;
121 let mode = if meta.is_dir() {
122 EntryMode::DIR
123 } else if meta.is_file() {
124 EntryMode::FILE
125 } else {
126 EntryMode::Unknown
127 };
128 let m = Metadata::new(mode)
129 .with_content_length(meta.len())
130 .with_last_modified(
131 meta.modified()
132 .map(DateTime::from)
133 .map_err(new_std_io_error)?,
134 );
135 Ok(RpStat::new(m))
136 }
137
138 async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
139 let path = self.core.prepare_path(path);
140 let reader = MonoiofsReader::new(self.core.clone(), path, args.range()).await?;
141 Ok((RpRead::default(), reader))
142 }
143
144 async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
145 let path = self.core.prepare_write_path(path).await?;
146 let writer = MonoiofsWriter::new(self.core.clone(), path, args.append()).await?;
147 Ok((RpWrite::default(), writer))
148 }
149
150 async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
151 Ok((
152 RpDelete::default(),
153 oio::OneShotDeleter::new(MonoiofsDeleter::new(self.core.clone())),
154 ))
155 }
156
157 async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
158 let from = self.core.prepare_path(from);
159 self.core
161 .dispatch({
162 let from = from.clone();
163 move || monoio::fs::metadata(from)
164 })
165 .await
166 .map_err(new_std_io_error)?;
167 let to = self.core.prepare_write_path(to).await?;
168 self.core
169 .dispatch(move || monoio::fs::rename(from, to))
170 .await
171 .map_err(new_std_io_error)?;
172 Ok(RpRename::default())
173 }
174
175 async fn create_dir(&self, path: &str, _args: OpCreateDir) -> Result<RpCreateDir> {
176 let path = self.core.prepare_path(path);
177 self.core
178 .dispatch(move || monoio::fs::create_dir_all(path))
179 .await
180 .map_err(new_std_io_error)?;
181 Ok(RpCreateDir::default())
182 }
183
184 async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
185 let from = self.core.prepare_path(from);
186 self.core
188 .dispatch({
189 let from = from.clone();
190 move || monoio::fs::metadata(from)
191 })
192 .await
193 .map_err(new_std_io_error)?;
194 let to = self.core.prepare_write_path(to).await?;
195 self.core
196 .dispatch({
197 let core = self.core.clone();
198 move || async move {
199 let from = OpenOptions::new().read(true).open(from).await?;
200 let to = OpenOptions::new()
201 .write(true)
202 .create(true)
203 .truncate(true)
204 .open(to)
205 .await?;
206
207 let mut pos = 0;
212 let mut buf = core.buf_pool.get();
214 buf.reserve(BUFFER_SIZE);
216 let _ = buf.split_off(BUFFER_SIZE);
217
218 loop {
219 let result;
220 (result, buf) = from.read_at(buf, pos).await;
221 if result? == 0 {
222 break;
224 }
225 let result;
226 (result, buf) = to.write_all_at(buf, pos).await;
227 result?;
228 pos += buf.len() as u64;
229 buf.clear();
230 }
231 core.buf_pool.put(buf);
232 Ok(())
233 }
234 })
235 .await
236 .map_err(new_std_io_error)?;
237 Ok(RpCopy::default())
238 }
239}