sorock 0.12.0

A Multi-Raft implementation in Rust
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

use super::*;

mod communicator;

use communicator::{Communicator, RaftConnection};
use std::collections::HashMap;

/// `RaftNode` manages multiple `RaftProcess`es on shards.
pub struct RaftNode {
    self_node_id: NodeAddress,
    cache: moka::sync::Cache<NodeAddress, RaftConnection>,
    process_map: spin::RwLock<HashMap<ShardIndex, Arc<process::RaftProcess>>>,
}

impl RaftNode {
    /// Create a new Raft node with a given node ID.
    pub fn new(id: NodeAddress) -> Self {
        let builder = moka::sync::Cache::builder()
            .initial_capacity(3)
            .time_to_idle(Duration::from_secs(60));
        Self {
            self_node_id: id,
            cache: builder.build(),
            process_map: HashMap::new().into(),
        }
    }

    /// Get a Raft handle to give I/O capability to a Raft process on a shard.
    pub fn get_handle(&self, shard_index: ShardIndex) -> RaftHandle {
        RaftHandle {
            shard_index,
            self_node_id: self.self_node_id.clone(),
            connection_cache: self.cache.clone(),
        }
    }

    /// Attach a Raft process to a shard.
    pub fn attach_process(&self, shard_index: ShardIndex, p: process::RaftProcess) {
        self.process_map.write().insert(shard_index, Arc::new(p));
    }

    /// Detach a Raft process from a shard.
    pub fn detach_process(&self, shard_index: ShardIndex) {
        self.process_map.write().remove(&shard_index);
    }

    pub(super) fn get_process(&self, shard_index: ShardIndex) -> Option<Arc<process::RaftProcess>> {
        self.process_map.read().get(&shard_index).cloned()
    }
}

/// `RaftHandle` gives I/O capability to a Raft process on a shard.
#[derive(Clone)]
pub struct RaftHandle {
    pub self_node_id: NodeAddress,
    pub shard_index: ShardIndex,
    connection_cache: moka::sync::Cache<NodeAddress, RaftConnection>,
}
impl RaftHandle {
    pub(super) fn self_node_id(&self) -> NodeAddress {
        self.self_node_id.clone()
    }

    pub(super) fn connect(&self, dest_node_id: NodeAddress) -> Communicator {
        let conn: RaftConnection = self.connection_cache.get_with(dest_node_id.clone(), || {
            RaftConnection::new(self.self_node_id.clone(), dest_node_id.clone())
        });
        Communicator::new(conn, self.shard_index)
    }
}