foyer_memory/
pipe.rs

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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
// Copyright 2025 foyer Project Authors
//
// Licensed 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::{fmt::Debug, marker::PhantomData, sync::Arc};

use foyer_common::code::{Key, Value};

use crate::{record::Record, Eviction};

/// A piece of record that is irrelevant to the eviction algorithm.
///
/// With [`Piece`], the disk cache doesn't need to consider the eviction generic type.
pub struct Piece<K, V> {
    record: *const (),
    key: *const K,
    value: *const V,
    hash: u64,
    drop_fn: fn(*const ()),
}

impl<K, V> Debug for Piece<K, V> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Piece")
            .field("record", &self.record)
            .field("hash", &self.hash)
            .finish()
    }
}

unsafe impl<K, V> Send for Piece<K, V> {}
unsafe impl<K, V> Sync for Piece<K, V> {}

impl<K, V> Drop for Piece<K, V> {
    fn drop(&mut self) {
        (self.drop_fn)(self.record);
    }
}

impl<K, V> Piece<K, V> {
    /// Create a record piece from an record wrapped by [`Arc`].
    pub fn new<E>(record: Arc<Record<E>>) -> Self
    where
        E: Eviction<Key = K, Value = V>,
    {
        let raw = Arc::into_raw(record);
        let record = raw as *const ();
        let key = unsafe { (*raw).key() } as *const _;
        let value = unsafe { (*raw).value() } as *const _;
        let hash = unsafe { (*raw).hash() };
        let drop_fn = |ptr| unsafe {
            let _ = Arc::from_raw(ptr as *const Record<E>);
        };
        Self {
            record,
            key,
            value,
            hash,
            drop_fn,
        }
    }

    /// Get the key of the record.
    pub fn key(&self) -> &K {
        unsafe { &*self.key }
    }

    /// Get the value of the record.
    pub fn value(&self) -> &V {
        unsafe { &*self.value }
    }

    /// Get the hash of the record.
    pub fn hash(&self) -> u64 {
        self.hash
    }
}

/// Pipe is used to notify disk cache to cache entries from the in-memory cache.
pub trait Pipe: Send + Sync + 'static {
    /// Type of the key of the record.
    type Key;
    /// Type of the value of the record.
    type Value;

    /// Decide whether to send evicted entry to pipe.
    fn is_enabled(&self) -> bool;

    /// Send the piece to the disk cache.
    fn send(&self, piece: Piece<Self::Key, Self::Value>);
}

/// An no-op pipe that is never enabled.
#[derive(Debug)]
pub struct NoopPipe<K, V>(PhantomData<(K, V)>);

impl<K, V> Default for NoopPipe<K, V> {
    fn default() -> Self {
        Self(PhantomData)
    }
}

impl<K, V> Pipe for NoopPipe<K, V>
where
    K: Key,
    V: Value,
{
    type Key = K;
    type Value = V;

    fn is_enabled(&self) -> bool {
        false
    }

    fn send(&self, _: Piece<Self::Key, Self::Value>) {}
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        eviction::fifo::{Fifo, FifoHint},
        record::Data,
    };

    #[test]
    fn test_piece() {
        let r1 = Arc::new(Record::new(Data::<Fifo<Arc<Vec<u8>>, Arc<Vec<u8>>>> {
            key: Arc::new(vec![b'k'; 4096]),
            value: Arc::new(vec![b'k'; 16384]),
            hint: FifoHint,
            hash: 1,
            weight: 1,
        }));

        let p1 = Piece::new(r1.clone());
        let k1 = p1.key().clone();
        let r2 = r1.clone();
        let p2 = Piece::new(r1.clone());

        drop(p1);
        drop(r2);
        drop(p2);
        drop(r1);
        drop(k1);
    }
}