ipldfs 0.1.0

Virtual filesystem for ipld
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

//! Least Recently Used cache for ipld fs

use std::{
    future::Future,
    io::Result,
    pin::Pin,
    sync::{Arc, Mutex},
    task::Poll,
};

use crate::ipldfs::{IpldFS, Multipart};

/// LruCache Decorator for [`IpldFS`]
#[derive(Debug, Clone)]
pub struct LruCache<FS>
where
    FS: IpldFS + Clone,
{
    inner: FS,
    lru: Arc<Mutex<lru_mem::LruCache<String, Multipart>>>,
}

impl<FS> LruCache<FS>
where
    FS: IpldFS + Clone,
{
    /// Create new `LruCache Decorator` for input [`fs`](IpldFS)
    pub fn new(fs: FS, maxinum_memory_size: usize) -> Self {
        Self {
            inner: fs,
            lru: Arc::new(Mutex::new(lru_mem::LruCache::new(maxinum_memory_size))),
        }
    }
}

impl<FS> IpldFS for LruCache<FS>
where
    FS: IpldFS + Clone + Sync + Send + Unpin + 'static,
{
    type Read<'cx> = LruCacheRead<'cx, FS>;

    type ReadEntry<'cx> = FS::ReadEntry<'cx>;

    type Write<'cx> = FS::Write<'cx>;

    fn read_entry<'cx, 'a>(&'a mut self, cid: libipld::Cid) -> Self::ReadEntry<'cx>
    where
        'a: 'cx,
    {
        self.inner.read_entry(cid)
    }

    fn read<'cx, 'a>(&'a mut self, cid: &'cx libipld::Cid) -> Self::Read<'cx>
    where
        'a: 'cx,
    {
        let cid_string = cid.to_string();

        {
            let mut lru = self.lru.lock().unwrap();

            if let Some(multipart) = lru.get(&cid_string) {
                return LruCacheRead::Cached(Some(multipart.clone()));
            }
        }

        LruCacheRead::Invoke(cid.to_string(), self.inner.read(cid), self.lru.clone())
    }

    fn write<'cx, 'a>(
        &'a mut self,
        cid: libipld::Cid,
        multipart: Multipart,
        lease: std::time::Duration,
    ) -> Self::Write<'cx>
    where
        'a: 'cx,
    {
        self.inner.write(cid, multipart, lease)
    }
}

pub enum LruCacheRead<'cx, FS>
where
    FS: IpldFS + Send + Unpin + 'static,
{
    Cached(Option<Multipart>),
    Invoke(
        String,
        FS::Read<'cx>,
        Arc<Mutex<lru_mem::LruCache<String, Multipart>>>,
    ),
}

impl<'cx, FS> Future for LruCacheRead<'cx, FS>
where
    FS: IpldFS + Send + Unpin + 'static,
{
    type Output = Result<Multipart>;
    fn poll(self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
        match self.get_mut() {
            LruCacheRead::Cached(multpart) => return Poll::Ready(Ok(multpart.take().unwrap())),
            LruCacheRead::Invoke(key, read, lru) => match Pin::new(read).poll(cx) {
                Poll::Ready(Ok(multipart)) => {
                    {
                        lru.lock()
                            .unwrap()
                            .insert(key.clone(), multipart.clone())
                            .map_err(|err| {
                                std::io::Error::new(std::io::ErrorKind::OutOfMemory, err)
                            })?;
                    }

                    return Poll::Ready(Ok(multipart));
                }

                p => return p,
            },
        }
    }
}