shardmap 0.2.1

Sharded embedded in-memory map with optional cache, protocol, and server internals
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

use bytes::BytesMut;

#[cfg(feature = "server")]
use crate::commands::dump_restore::write_string_dump_resp;
use crate::commands::dump_restore::{
    DumpRestoreValue, encode_dump_value, encode_string_dump_value,
};
use crate::commands::redis::{
    bulk, define_redis_command, write_frame, write_resp_null, wrong_arity,
};
use crate::protocol::Frame;
#[cfg(feature = "server")]
use crate::server::wire::ServerWire;
#[cfg(feature = "server")]
use crate::storage::hash_key_tag_from_hash;
use crate::storage::{EmbeddedStore, RedisKeyStore};

define_redis_command!(Dump, "DUMP", false);

impl crate::commands::redis::RedisCommand for Dump {
    fn execute(store: &EmbeddedStore, args: &[&[u8]]) -> Frame {
        match args {
            [key] => dump_key(store, key).map_or(Frame::Null, bulk),
            _ => wrong_arity("DUMP"),
        }
    }

    #[cfg(feature = "server")]
    fn write_resp(store: &EmbeddedStore, args: &[&[u8]], out: &mut BytesMut) {
        match args {
            [key] => write_dump_resp(store, key, out),
            _ => write_frame(out, &wrong_arity("DUMP")),
        }
    }

    #[cfg(feature = "server")]
    fn write_resp_owned_shard(
        store: &EmbeddedStore,
        args: &[&[u8]],
        owned_shard_id: usize,
        out: &mut BytesMut,
    ) -> bool {
        match args {
            [key] => write_dump_resp_owned_shard(store, key, owned_shard_id, out),
            _ => false,
        }
    }
}

#[cfg(feature = "server")]
fn write_dump_resp(store: &EmbeddedStore, key: &[u8], out: &mut BytesMut) {
    if let Some(value) = store.get_ref(key) {
        write_string_dump_resp(out, value.value());
        return;
    }
    match store
        .clone_object_value(key)
        .map(|value| encode_dump_value(&DumpRestoreValue::Object(value)))
    {
        Some(payload) => ServerWire::write_resp_blob_string(out, &payload),
        None => write_resp_null(out),
    }
}

#[cfg(feature = "server")]
fn write_dump_resp_owned_shard(
    store: &EmbeddedStore,
    key: &[u8],
    owned_shard_id: usize,
    out: &mut BytesMut,
) -> bool {
    let route = store.route_key(key);
    if route.shard_id != owned_shard_id {
        return false;
    }

    let key_tag = hash_key_tag_from_hash(route.key_hash);
    #[cfg(feature = "unsafe")]
    {
        // SAFETY: direct-shard RESP requests are route-checked before command
        // dispatch, so this worker owns the shard for this key.
        let outcome = unsafe {
            store.with_shared_value_bytes_full_key_tagged_owned_shard_no_ttl(
                owned_shard_id,
                route.key_hash,
                key_tag,
                key.len(),
                |value| write_string_dump_resp(out, value),
            )
        };
        if matches!(outcome, Some(true)) {
            return true;
        }
    }

    #[cfg(not(feature = "unsafe"))]
    {
        let _ = key_tag;
        let outcome = store.with_shared_value_bytes_full_key_owned_shard_no_ttl(
            owned_shard_id,
            route.key_hash,
            key,
            |value| write_string_dump_resp(out, value),
        );
        if matches!(outcome, Some(true)) {
            return true;
        }
    }

    false
}

pub(crate) fn dump_key(store: &EmbeddedStore, key: &[u8]) -> Option<Vec<u8>> {
    if let Some(value) = store.get_value_bytes(key) {
        return Some(encode_string_dump_value(value.as_ref()));
    }
    store
        .clone_object_value(key)
        .map(|value| encode_dump_value(&DumpRestoreValue::Object(value)))
}