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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
use crate::{
local_client::LocalClient,
persistent_safety_storage::PersistentSafetyStorage,
process::ProcessService,
remote_service::RemoteService,
serializer::{SerializerClient, SerializerService},
thread::ThreadService,
SafetyRules, TSafetyRules,
};
use diem_config::config::{SafetyRulesConfig, SafetyRulesService};
use diem_infallible::RwLock;
use diem_secure_storage::{KVStorage, Storage};
use std::{convert::TryInto, net::SocketAddr, sync::Arc};
pub fn storage(config: &SafetyRulesConfig) -> PersistentSafetyStorage {
let backend = &config.backend;
let internal_storage: Storage = backend.try_into().expect("Unable to initialize storage");
if let Err(error) = internal_storage.available() {
panic!("Storage is not available: {:?}", error);
}
if let Some(test_config) = &config.test {
let author = test_config.author;
let consensus_private_key = test_config
.consensus_key
.as_ref()
.expect("Missing consensus key in test config")
.private_key();
let execution_private_key = test_config
.execution_key
.as_ref()
.expect("Missing execution key in test config")
.private_key();
let waypoint = test_config.waypoint.expect("No waypoint in config");
PersistentSafetyStorage::initialize(
internal_storage,
author,
consensus_private_key,
execution_private_key,
waypoint,
config.enable_cached_safety_data,
)
} else {
PersistentSafetyStorage::new(internal_storage, config.enable_cached_safety_data)
}
}
enum SafetyRulesWrapper {
Local(Arc<RwLock<SafetyRules>>),
Process(ProcessService),
Serializer(Arc<RwLock<SerializerService>>),
Thread(ThreadService),
}
pub struct SafetyRulesManager {
internal_safety_rules: SafetyRulesWrapper,
}
impl SafetyRulesManager {
pub fn new(config: &SafetyRulesConfig) -> Self {
if let SafetyRulesService::Process(conf) = &config.service {
return Self::new_process(conf.server_address(), config.network_timeout_ms);
}
let storage = storage(config);
let verify_vote_proposal_signature = config.verify_vote_proposal_signature;
let export_consensus_key = config.export_consensus_key;
match config.service {
SafetyRulesService::Local => Self::new_local(
storage,
verify_vote_proposal_signature,
export_consensus_key,
config.decoupled_execution,
),
SafetyRulesService::Serializer => Self::new_serializer(
storage,
verify_vote_proposal_signature,
export_consensus_key,
config.decoupled_execution,
),
SafetyRulesService::Thread => Self::new_thread(
storage,
verify_vote_proposal_signature,
export_consensus_key,
config.network_timeout_ms,
config.decoupled_execution,
),
_ => panic!("Unimplemented SafetyRulesService: {:?}", config.service),
}
}
pub fn new_local(
storage: PersistentSafetyStorage,
verify_vote_proposal_signature: bool,
export_consensus_key: bool,
decoupled_execution: bool,
) -> Self {
let safety_rules = SafetyRules::new(
storage,
verify_vote_proposal_signature,
export_consensus_key,
decoupled_execution,
);
Self {
internal_safety_rules: SafetyRulesWrapper::Local(Arc::new(RwLock::new(safety_rules))),
}
}
pub fn new_process(server_addr: SocketAddr, timeout_ms: u64) -> Self {
let process_service = ProcessService::new(server_addr, timeout_ms);
Self {
internal_safety_rules: SafetyRulesWrapper::Process(process_service),
}
}
pub fn new_serializer(
storage: PersistentSafetyStorage,
verify_vote_proposal_signature: bool,
export_consensus_key: bool,
decoupled_execution: bool,
) -> Self {
let safety_rules = SafetyRules::new(
storage,
verify_vote_proposal_signature,
export_consensus_key,
decoupled_execution,
);
let serializer_service = SerializerService::new(safety_rules);
Self {
internal_safety_rules: SafetyRulesWrapper::Serializer(Arc::new(RwLock::new(
serializer_service,
))),
}
}
pub fn new_thread(
storage: PersistentSafetyStorage,
verify_vote_proposal_signature: bool,
export_consensus_key: bool,
timeout_ms: u64,
decoupled_execution: bool,
) -> Self {
let thread = ThreadService::new(
storage,
verify_vote_proposal_signature,
export_consensus_key,
timeout_ms,
decoupled_execution,
);
Self {
internal_safety_rules: SafetyRulesWrapper::Thread(thread),
}
}
pub fn client(&self) -> Box<dyn TSafetyRules + Send + Sync> {
match &self.internal_safety_rules {
SafetyRulesWrapper::Local(safety_rules) => {
Box::new(LocalClient::new(safety_rules.clone()))
}
SafetyRulesWrapper::Process(process) => Box::new(process.client()),
SafetyRulesWrapper::Serializer(serializer_service) => {
Box::new(SerializerClient::new(serializer_service.clone()))
}
SafetyRulesWrapper::Thread(thread) => Box::new(thread.client()),
}
}
}