opendal_service_foyer/
lib.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
18#![doc = include_str!("../README.md")]
19#![cfg_attr(docsrs, feature(doc_cfg))]
20#![cfg_attr(docsrs, doc(auto_cfg))]
21#![deny(missing_docs)]
22
23mod backend;
24mod config;
25mod core;
26mod deleter;
27mod reader;
28mod writer;
29
30use std::ops::Deref;
31
32use foyer::Code;
33use foyer::Result as FoyerResult;
34
35use opendal_core::Buffer;
36
37pub use backend::FoyerBuilder as Foyer;
38pub use config::FoyerConfig;
39
40/// URI scheme used for service registration and scheme-driven construction.
41pub const FOYER_SCHEME: &str = "foyer";
42
43/// [`FoyerKey`] is a key for the foyer cache.
44///
45/// It implements foyer's [`Code`] trait directly, so the service does not depend on
46/// foyer's `serde` feature (and its bincode dependency).
47#[derive(Debug, Clone, PartialEq, Eq, Hash)]
48pub struct FoyerKey {
49    /// The path of the key.
50    pub path: String,
51}
52
53impl Code for FoyerKey {
54    fn encode(&self, writer: &mut impl std::io::Write) -> FoyerResult<()> {
55        let path = self.path.as_bytes();
56        writer.write_all(&(path.len() as u64).to_le_bytes())?;
57        writer.write_all(path)?;
58        Ok(())
59    }
60
61    fn decode(reader: &mut impl std::io::Read) -> FoyerResult<Self>
62    where
63        Self: Sized,
64    {
65        let mut len_bytes = [0u8; 8];
66        reader.read_exact(&mut len_bytes)?;
67        let len = u64::from_le_bytes(len_bytes) as usize;
68        let mut buf = vec![0u8; len];
69        reader.read_exact(&mut buf)?;
70        let path = String::from_utf8(buf)
71            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
72        Ok(FoyerKey { path })
73    }
74
75    fn estimated_size(&self) -> usize {
76        8 + self.path.len()
77    }
78}
79
80/// [`FoyerValue`] is a wrapper around `Buffer` that implements the `Code` trait.
81#[derive(Debug)]
82pub struct FoyerValue(pub Buffer);
83
84impl Deref for FoyerValue {
85    type Target = Buffer;
86
87    fn deref(&self) -> &Self::Target {
88        &self.0
89    }
90}
91
92impl Code for FoyerValue {
93    fn encode(&self, writer: &mut impl std::io::Write) -> FoyerResult<()> {
94        let len = self.0.len() as u64;
95        writer.write_all(&len.to_le_bytes())?;
96        std::io::copy(&mut self.0.clone(), writer)?;
97        Ok(())
98    }
99
100    fn decode(reader: &mut impl std::io::Read) -> FoyerResult<Self>
101    where
102        Self: Sized,
103    {
104        let mut len_bytes = [0u8; 8];
105        reader.read_exact(&mut len_bytes)?;
106        let len = u64::from_le_bytes(len_bytes) as usize;
107        let mut buffer = vec![0u8; len];
108        reader.read_exact(&mut buffer[..len])?;
109        Ok(FoyerValue(buffer.into()))
110    }
111
112    fn estimated_size(&self) -> usize {
113        8 + self.0.len()
114    }
115}
116
117/// Register this service's URI scheme or schemes with an operator registry.
118///
119/// Registration enables scheme-driven construction through
120/// [`opendal_core::Operator::from_uri`] and
121/// [`opendal_core::Operator::via_iter`]. Direct construction through
122/// [`opendal_core::Operator::new`] does not require registration.
123pub fn register_foyer_service(registry: &opendal_core::OperatorRegistry) {
124    registry.register::<Foyer>(FOYER_SCHEME);
125}
126
127#[cfg(test)]
128mod tests {
129    use foyer::{
130        BlockEngineConfig, DeviceBuilder, FsDeviceBuilder, HybridCacheBuilder, RecoverMode,
131    };
132    use opendal_core::ErrorKind;
133    use opendal_core::Operator;
134    use size::consts::MiB;
135
136    use super::*;
137
138    fn key(i: u8) -> String {
139        format!("obj-{i}")
140    }
141
142    fn value(i: u8) -> Vec<u8> {
143        vec![i; 1024]
144    }
145
146    #[tokio::test]
147    async fn test_basic_operations() {
148        let dir = tempfile::tempdir().unwrap();
149
150        let cache = HybridCacheBuilder::new()
151            .memory(10)
152            .with_shards(1)
153            .storage()
154            .with_engine_config(
155                BlockEngineConfig::new(
156                    FsDeviceBuilder::new(dir.path())
157                        .with_capacity(16 * MiB as usize)
158                        .build()
159                        .unwrap(),
160                )
161                .with_block_size(MiB as usize),
162            )
163            .with_recover_mode(RecoverMode::None)
164            .build()
165            .await
166            .unwrap();
167
168        let op = Operator::new(Foyer::new().cache(cache)).unwrap();
169
170        // Write some data
171        for i in 0..10 {
172            op.write(&key(i), value(i)).await.unwrap();
173        }
174
175        // Read back
176        for i in 0..10 {
177            let buf = op.read(&key(i)).await.unwrap();
178            assert_eq!(buf.to_vec(), value(i));
179        }
180
181        // Stat
182        for i in 0..10 {
183            let meta = op.stat(&key(i)).await.unwrap();
184            assert_eq!(meta.content_length(), 1024);
185        }
186
187        // Delete
188        for i in 0..10 {
189            op.delete(&key(i)).await.unwrap();
190        }
191
192        // Verify deleted
193        for i in 0..10 {
194            let res = op.read(&key(i)).await;
195            assert!(res.is_err(), "should fail to read deleted file");
196        }
197    }
198
199    #[tokio::test]
200    async fn test_range_read() {
201        let dir = tempfile::tempdir().unwrap();
202
203        let cache = HybridCacheBuilder::new()
204            .memory(1024 * 1024)
205            .with_shards(1)
206            .storage()
207            .with_engine_config(
208                BlockEngineConfig::new(
209                    FsDeviceBuilder::new(dir.path())
210                        .with_capacity(16 * MiB as usize)
211                        .build()
212                        .unwrap(),
213                )
214                .with_block_size(MiB as usize),
215            )
216            .with_recover_mode(RecoverMode::None)
217            .build()
218            .await
219            .unwrap();
220
221        let op = Operator::new(Foyer::new().cache(cache)).unwrap();
222
223        let data: Vec<u8> = (0..100).collect();
224        op.write("test", data.clone()).await.unwrap();
225
226        // Range read
227        let buf = op.read_with("test").range(10..20).await.unwrap();
228        assert_eq!(buf.to_vec(), data[10..20]);
229
230        let err = op.read_with("test").range(95..105).await.unwrap_err();
231        assert_eq!(err.kind(), ErrorKind::RangeNotSatisfied);
232
233        let buf = op.read_with("test").range(100..).await.unwrap();
234        assert!(buf.is_empty());
235
236        let buf = op.read_with("test").range(200..).await.unwrap();
237        assert!(buf.is_empty());
238
239        let buf = op.read_with("test").range(200..200).await.unwrap();
240        assert!(buf.is_empty());
241    }
242
243    #[tokio::test]
244    async fn test_hybrid_cache_via_config() {
245        let dir = tempfile::tempdir().unwrap();
246
247        // Test using the builder API with disk configuration
248        let op = Operator::new(
249            Foyer::new()
250                .memory(1024 * 1024) // 1MB memory
251                .disk_path(dir.path().to_str().unwrap())
252                .disk_capacity(16 * 1024 * 1024) // 16MB disk
253                .disk_file_size(1024 * 1024) // 1MB per file
254                .recover_mode("none")
255                .shards(1),
256        )
257        .unwrap();
258
259        // Write some data
260        for i in 0..5 {
261            op.write(&key(i), value(i)).await.unwrap();
262        }
263
264        // Read back
265        for i in 0..5 {
266            let buf = op.read(&key(i)).await.unwrap();
267            assert_eq!(buf.to_vec(), value(i));
268        }
269
270        // Delete
271        for i in 0..5 {
272            op.delete(&key(i)).await.unwrap();
273        }
274    }
275}
opendal_service_foyer/lib.rs

opendal_service_foyer/
lib.rs
