raftlog 0.6.0

An implementation of distributed replicated log based on the Raft algorithm
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
122
123
124
125
126
127
128

use super::Common;
use log::{LogPosition, LogPrefix, LogSuffix};
use message::{self, AppendEntriesReply, Message, MessageHeader, SequenceNumber};
use node::NodeId;
use Io;

/// RPC要求メッセージの送信を補助するためのビルダ.
pub struct RpcCaller<'a, IO: 'a + Io> {
    common: &'a mut Common<IO>,
}
impl<'a, IO: 'a + Io> RpcCaller<'a, IO> {
    pub fn new(common: &'a mut Common<IO>) -> Self {
        RpcCaller { common }
    }
    pub fn broadcast_request_vote(mut self) {
        let header = self.make_header(&NodeId::new(String::new())); // ブロードキャストノード時に空文字列を宛先に指定
        let log_tail = self.common.history.tail();
        let request = message::RequestVoteCall {
            header: header.clone(),
            log_tail,
        }
        .into();
        let self_reply = message::RequestVoteReply {
            header,
            voted: true,
        }
        .into();
        self.broadcast(request, self_reply);
    }
    pub fn broadcast_append_entries(mut self, suffix: LogSuffix) {
        let header = self.make_header(&NodeId::new(String::new())); // ブロードキャストノード時に空文字列を宛先に指定
        let request = message::AppendEntriesCall {
            header: header.clone(),
            committed_log_tail: self.common.history.committed_tail().index,
            suffix,
        }
        .into();
        let self_reply = AppendEntriesReply {
            header,
            log_tail: self.common.history.tail(),
            busy: false,
        }
        .into();
        self.broadcast(request, self_reply);
    }
    pub fn send_append_entries(mut self, peer: &NodeId, suffix: LogSuffix) {
        let message = message::AppendEntriesCall {
            header: self.make_header(peer),
            committed_log_tail: self.common.history.committed_tail().index,
            suffix,
        }
        .into();
        self.common.io.send_message(message);
    }
    pub fn send_install_snapshot(mut self, peer: &NodeId, prefix: LogPrefix) {
        let header = self.make_header(peer);
        let message = message::InstallSnapshotCast { header, prefix }.into();
        self.common.io.send_message(message);
    }

    fn make_header(&mut self, destination: &NodeId) -> MessageHeader {
        let seq_no = self.common.seq_no;
        self.common.seq_no = SequenceNumber::new(seq_no.as_u64() + 1);
        MessageHeader {
            sender: self.common.local_node.id.clone(),
            destination: destination.clone(),
            seq_no,
            term: self.common.local_node.ballot.term,
        }
    }
    fn broadcast(&mut self, mut message: Message, self_reply: Message) {
        let mut do_self_reply = false;
        for peer in self.common.history.config().members() {
            if *peer == self.common.local_node.id {
                do_self_reply = true;
            } else {
                message.set_destination(peer);
                self.common.io.send_message(message.clone());
            }
        }
        if do_self_reply {
            self.common.unread_message = Some(self_reply);
        }
    }
}

/// RPC応答メッセージの送信を補助するためのビルダ.
pub struct RpcCallee<'a, IO: 'a + Io> {
    common: &'a mut Common<IO>,
    caller: &'a MessageHeader,
}
impl<'a, IO: 'a + Io> RpcCallee<'a, IO> {
    pub fn new(common: &'a mut Common<IO>, caller: &'a MessageHeader) -> Self {
        RpcCallee { common, caller }
    }
    pub fn reply_request_vote(self, voted: bool) {
        let header = self.make_header();
        let message = message::RequestVoteReply { header, voted }.into();
        self.common.io.send_message(message);
    }
    pub fn reply_append_entries(self, log_tail: LogPosition) {
        let message = AppendEntriesReply {
            header: self.make_header(),
            log_tail,
            busy: false,
        }
        .into();
        self.common.io.send_message(message);
    }
    pub fn reply_busy(self) {
        let message = AppendEntriesReply {
            header: self.make_header(),
            log_tail: self.common.history.tail(),
            busy: true,
        }
        .into();
        self.common.io.send_message(message);
    }

    fn make_header(&self) -> MessageHeader {
        MessageHeader {
            sender: self.common.local_node.id.clone(),
            destination: self.caller.sender.clone(),
            seq_no: self.caller.seq_no,
            term: self.common.local_node.ballot.term,
        }
    }
}