github-jwinterm/grim
1
0
Fork 0
mirror of https://github.com/jwinterm/grim.git synced 2026-10-19 14:58:33 +02:00
Code Issues Packages Projects Releases Wiki Activity

node: handle errors, random p2p and api ports, optimize rwlock

Browse Source
This commit is contained in:
ardocrat
2024-05-14 17:36:49 +03:00
parent ad3bcd2e0a
commit ae5631a717
14 changed files with 301 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/config.rs
+29 -18
View File
@@ -17,6 +17,8 @@ use std::io::{BufRead, BufReader, Write};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4, TcpListener, ToSocketAddrs};
use std::path::PathBuf;
use std::str::FromStr;
use local_ip_address::list_afinet_netifas;
use serde::{Deserialize, Serialize};
use grin_config::{config, ConfigError, ConfigMembers, GlobalConfig};
use grin_config::config::{API_SECRET_FILE_NAME, FOREIGN_API_SECRET_FILE_NAME, SERVER_CONFIG_FILE_NAME};
@@ -24,8 +26,7 @@ use grin_core::global::ChainTypes;
use grin_p2p::{PeerAddr, Seeding};
use grin_p2p::msg::PeerAddrs;
use grin_servers::common::types::ChainValidationMode;
use local_ip_address::list_afinet_netifas;
use serde::{Deserialize, Serialize};
use rand::Rng;
use crate::{AppConfig, Settings};
use crate::node::Node;
@@ -68,7 +69,7 @@ impl PeersConfig {
}
/// Load saved peers to node server [`ConfigMembers`] config.
fn load_to_server_config() {
pub(crate) fn load_to_server_config() {
let mut w_node_config = Settings::node_config_to_update();
// Load seeds.
for seed in w_node_config.peers.seeds.clone() {
@@ -111,7 +112,6 @@ impl PeersConfig {
denied.peers.insert(denied.peers.len(), p);
w_node_config.node.server.p2p_config.peers_deny = Some(denied);
}
}
// Load preferred peers.
for peer in &w_node_config.peers.preferred.clone() {
@@ -175,13 +175,35 @@ impl NodeConfig {
fn save_default_node_server_config(chain_type: &ChainTypes) -> ConfigMembers {
let sub_dir = Some(chain_type.shortname());
let path = Settings::get_config_path(SERVER_CONFIG_FILE_NAME, sub_dir.clone());
let mut default_config = GlobalConfig::for_chain(chain_type);
default_config.update_paths(&Settings::get_base_path(sub_dir));
let config = default_config.members.unwrap();
let mut config = default_config.members.unwrap();
Self::setup_default_ports(&mut config);
Settings::write_to_file(&config, path);
config
}
/// Generate random p2p and api ports in ranges based on [`ChainTypes`].
fn setup_default_ports(config: &mut ConfigMembers) {
let (api, p2p) = match config.server.chain_type {
ChainTypes::Mainnet => {
let api = rand::thread_rng().gen_range(30000..33000);
let p2p = rand::thread_rng().gen_range(33000..37000);
(api, p2p)
},
_ => {
let api = rand::thread_rng().gen_range(40000..43000);
let p2p = rand::thread_rng().gen_range(43000..47000);
(api, p2p)
}
};
let api_addr = config.server.api_http_addr.split_once(":").unwrap().0;
config.server.api_http_addr = format!("{}:{}", api_addr, api);
config.server.p2p_config.port = p2p;
}
/// Save default peers config for specified [`ChainTypes`].
fn save_default_peers_config(chain_type: &ChainTypes) -> PeersConfig {
let sub_dir = Some(chain_type.shortname());
@@ -200,7 +222,6 @@ impl NodeConfig {
/// Get server config to use for node server before start.
pub fn node_server_config() -> ConfigMembers {
PeersConfig::load_to_server_config();
let r_config = Settings::node_config_to_read();
r_config.node.clone()
}
@@ -444,11 +465,7 @@ impl NodeConfig {
pub fn is_api_port_available(ip: &String, port: &String) -> bool {
if Node::is_running() {
// Check if API server with same address is running.
let same_running = if let Some(running_addr) = Node::get_api_addr() {
running_addr == format!("{}:{}", ip, port)
} else {
false
};
let same_running = NodeConfig::get_api_address() == format!("{}:{}", ip, port);
if same_running || Self::is_host_port_available(ip, port) {
return &Self::get_p2p_port() != port;
}
@@ -598,8 +615,6 @@ impl NodeConfig {
w_node_config.save();
}
// P2P settings
/// Get P2P server port.
pub fn get_p2p_port() -> String {
Settings::node_config_to_read().node.server.p2p_config.port.to_string()
@@ -613,11 +628,7 @@ impl NodeConfig {
let (_, api_port) = Self::get_api_ip_port();
if Node::is_running() {
// Check if P2P server with same port is running.
let same_running = if let Some(running_port) = Node::get_p2p_port() {
&running_port.to_string() == port
} else {
false
};
let same_running = &NodeConfig::get_p2p_port() == port;
if same_running || Self::is_port_available(port) {
return &api_port != port;
}
src/node/mod.rs
+4 -1
View File
@@ -19,4 +19,7 @@ mod node;
pub use node::Node;
mod config;
pub use config::*;
pub use config::*;
mod types;
pub use types::*;
src/node/node.rs
+111 -139
View File
@@ -14,18 +14,20 @@
use std::{fs, thread};
use std::path::PathBuf;
use std::sync::{Arc, RwLock, RwLockReadGuard};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use lazy_static::lazy_static;
use parking_lot::RwLock;
use futures::channel::oneshot;
use grin_chain::SyncStatus;
use grin_core::global;
use grin_core::global::ChainTypes;
use grin_servers::{Server, ServerStats, StratumServerConfig, StratumStats};
use grin_servers::common::types::Error;
use lazy_static::lazy_static;
use crate::node::NodeConfig;
use crate::node::{NodeConfig, NodeError};
use crate::node::stratum::{StratumStopState, StratumServer};
lazy_static! {
@@ -35,28 +37,27 @@ lazy_static! {
/// Provides [`Server`] control, holds current status and statistics.
pub struct Node {
/// Node [`Server`] statistics for UI.
/// Node [`Server`] statistics information.
stats: Arc<RwLock<Option<ServerStats>>>,
/// [`StratumServer`] statistics.
/// [`StratumServer`] statistics information.
stratum_stats: Arc<grin_util::RwLock<StratumStats>>,
/// Flag to start [`StratumServer`].
start_stratum_needed: AtomicBool,
/// State to stop [`StratumServer`] from outside.
stratum_stop_state: Arc<StratumStopState>,
/// Running API [`Server`] address.
api_addr: Arc<RwLock<Option<String>>>,
/// Running P2P [`grin_p2p::Server`] port.
p2p_port: Arc<RwLock<Option<u16>>>,
/// Indicator if node [`Server`] is starting.
starting: AtomicBool,
/// Thread flag to stop the [`Server`] and start it again.
/// Flag to stop the [`Server`] and start it again.
restart_needed: AtomicBool,
/// Thread flag to stop the [`Server`].
/// Flag to stop the [`Server`].
stop_needed: AtomicBool,
/// Flag to check if app exit is needed after [`Server`] stop.
exit_after_stop: AtomicBool,
/// Thread flag to start [`StratumServer`].
start_stratum_needed: AtomicBool,
/// Error on [`Server`] start.
init_error: Option<Error>
/// An error occurred on [`Server`] start.
error: Arc<RwLock<Option<Error>>>
}
impl Default for Node {
@@ -65,14 +66,12 @@ impl Default for Node {
stats: Arc::new(RwLock::new(None)),
stratum_stats: Arc::new(grin_util::RwLock::new(StratumStats::default())),
stratum_stop_state: Arc::new(StratumStopState::default()),
api_addr: Arc::new(RwLock::new(None)),
p2p_port: Arc::new(RwLock::new(None)),
starting: AtomicBool::new(false),
restart_needed: AtomicBool::new(false),
stop_needed: AtomicBool::new(false),
exit_after_stop: AtomicBool::new(false),
start_stratum_needed: AtomicBool::new(false),
init_error: None
error: Arc::new(RwLock::new(None))
}
}
}
@@ -103,26 +102,6 @@ impl Node {
}
}
/// Get API [`Server`] address if [`Node`] is running.
pub fn get_api_addr() -> Option<String> {
let r_api_addr = NODE_STATE.api_addr.read().unwrap();
if r_api_addr.is_some() {
Some(r_api_addr.as_ref().unwrap().clone())
} else {
None
}
}
/// Get P2P [`grin_p2p::Server`] port if node is running.
pub fn get_p2p_port() -> Option<u16> {
let r_p2p_port = NODE_STATE.p2p_port.read().unwrap();
if r_p2p_port.is_some() {
Some(r_p2p_port.unwrap())
} else {
None
}
}
/// Request to start [`StratumServer`].
pub fn start_stratum() {
NODE_STATE.start_stratum_needed.store(true, Ordering::Relaxed);
@@ -170,7 +149,7 @@ impl Node {
/// Get node [`Server`] statistics.
pub fn get_stats() -> Option<ServerStats> {
NODE_STATE.stats.read().unwrap().clone()
NODE_STATE.stats.read().clone()
}
/// Check if [`Server`] is not syncing (disabled or just running after synchronization).
@@ -201,6 +180,45 @@ impl Node {
None
}
/// Get [`Server`] error.
pub fn get_error() -> Option<NodeError> {
let r_err = NODE_STATE.error.read();
if r_err.is_some() {
let e = r_err.as_ref().unwrap();
// Setup a flag to show an error to clean up data.
let store_err = match e {
Error::Store(_) => true,
Error::Chain(_) => true,
_ => false
};
if store_err {
return Some(NodeError::Storage);
}
// Setup a flag to show P2P or API server error.
let p2p_api_err = match e {
Error::P2P(_) => Some(NodeError::P2P),
Error::API(_) => Some(NodeError::API),
_ => None
};
if p2p_api_err.is_some() {
return p2p_api_err;
}
// Setup a flag to show configuration error.
let config_err = match e {
Error::Configuration(_) => true,
_ => false
};
return if config_err {
Some(NodeError::Configuration)
} else {
Some(NodeError::Unknown)
}
}
None
}
/// Start the [`Server`] at separate thread to update state with stats and handle statuses.
fn start_server_thread() {
thread::spawn(move || {
@@ -226,7 +244,13 @@ impl Node {
NODE_STATE.restart_needed.store(false, Ordering::Relaxed);
}
Err(e) => {
Self::on_start_error(&e);
// Setup an error.
{
let mut w_err = NODE_STATE.error.write();
*w_err = Some(e);
}
// Reset server state.
Self::reset_server_state(true);
break;
}
}
@@ -234,8 +258,7 @@ impl Node {
// Stop the server.
server.stop();
// Clean stats and statuses.
Self::on_thread_stop();
// Exit thread loop.
Self::reset_server_state(false);
break;
}
@@ -256,7 +279,7 @@ impl Node {
// Update server stats.
if let Ok(stats) = server.get_server_stats() {
{
let mut w_stats = NODE_STATE.stats.write().unwrap();
let mut w_stats = NODE_STATE.stats.write();
*w_stats = Some(stats.clone());
}
@@ -274,14 +297,20 @@ impl Node {
}
}
Err(e) => {
Self::on_start_error(&e);
// Setup an error.
{
let mut w_err = NODE_STATE.error.write();
*w_err = Some(e);
}
// Reset server state.
Self::reset_server_state(true);
}
}
});
}
/// Reset stats and statuses on [`Server`] thread stop.
fn on_thread_stop() {
/// Clean up [`Server`] stats and statuses.
fn reset_server_state(has_error: bool) {
NODE_STATE.starting.store(false, Ordering::Relaxed);
NODE_STATE.restart_needed.store(false, Ordering::Relaxed);
NODE_STATE.start_stratum_needed.store(false, Ordering::Relaxed);
@@ -292,91 +321,40 @@ impl Node {
let mut w_stratum_stats = NODE_STATE.stratum_stats.write();
*w_stratum_stats = StratumStats::default();
}
// Clean server stats.
// Reset server stats.
{
let mut w_stats = NODE_STATE.stats.write().unwrap();
let mut w_stats = NODE_STATE.stats.write();
*w_stats = None;
}
// Clean launched API server address.
{
let mut w_api_addr = NODE_STATE.api_addr.write().unwrap();
*w_api_addr = None;
}
// Clean launched P2P server port.
{
let mut w_p2p_port = NODE_STATE.p2p_port.write().unwrap();
*w_p2p_port = None;
// Reset an error if needed.
if !has_error {
let mut w_err = NODE_STATE.error.write();
*w_err = None;
}
}
/// Handle node [`Server`] error on start.
fn on_start_error(e: &Error) {
Self::on_thread_stop();
/// Clean-up [`Server`] data if server is not running.
pub fn clean_up_data() {
if Self::is_running() {
return;
}
let config = NodeConfig::node_server_config();
let server_config = config.server.clone();
//TODO: Create error
// NODE_STATE.init_error = Some(e);
// Remove lock file if exists.
let mut lock_path = PathBuf::from(&server_config.db_root);
lock_path.push("grin.lock");
if lock_path.exists() {
fs::remove_file(lock_path).unwrap();
}
// // Clean-up server data on data init error.
// // TODO: Ask user to clean-up data
// let clean_server_and_recreate = || -> Server {
// let mut db_path = PathBuf::from(&server_config.db_root);
// db_path.push("grin.lock");
// fs::remove_file(db_path).unwrap();
//
// // Remove chain data on server start error
// let dirs_to_remove: Vec<&str> = vec!["header", "lmdb", "txhashset"];
// for dir in dirs_to_remove {
// let mut path = PathBuf::from(&server_config.db_root);
// path.push(dir);
// fs::remove_dir_all(path).unwrap();
// }
//
// // Recreate server
// let api_chan: &'static mut (oneshot::Sender<()>, oneshot::Receiver<()>) =
// Box::leak(Box::new(oneshot::channel::<()>()));
// server_result = Server::new(server_config.clone(), None, api_chan);
// server_result.unwrap()
// };
// Show err on server init error.
// TODO: Ask user to clean-up data
let show_error = |err: String| {
println!("Node server creation error:\n{}", err);
//TODO don't panic maybe
panic!("{}", err);
};
//TODO: Better error handling
match e {
Error::Store(e) => {
//TODO: Set err to ask user to clean data
panic!("{}", e);
//(clean_server_and_recreate)()
}
Error::Chain(e) => {
//TODO: Set err to ask user to clean data
panic!("{}", e);
//(clean_server_and_recreate)()
}
//TODO: Handle P2P error (Show config error msg)
Error::P2P(ref e) => {
(show_error)("P2P error".to_string());
}
//TODO: Handle API error (Show config error msg)
Error::API(ref e) => {
(show_error)(e.to_string());
}
//TODO: Seems like another node instance running?
Error::IOError(ref e) => {
(show_error)(e.to_string());
}
//TODO: Show config error msg
Error::Configuration(ref e) => {
(show_error)(e.to_string());
}
//TODO: Unknown error
_ => {
(show_error)("Unknown error".to_string());
// Remove chain data.
let dirs_to_remove: Vec<&str> = vec!["header", "lmdb", "txhashset"];
for dir in dirs_to_remove {
let mut path = PathBuf::from(&server_config.db_root);
path.push(dir);
if path.exists() {
fs::remove_dir_all(path).unwrap();
}
}
}
@@ -564,26 +542,20 @@ fn start_node_server() -> Result<Server, Error> {
global::set_global_future_time_limit(future_time_limit);
}
let api_chan: &'static mut (oneshot::Sender<()>, oneshot::Receiver<()>) =
Box::leak(Box::new(oneshot::channel::<()>()));
// Set launching API server address from config to state.
{
let mut w_api_addr = NODE_STATE.api_addr.write().unwrap();
*w_api_addr = Some(config.server.api_http_addr);
}
// Set launching P2P server port from config to state.
{
let mut w_p2p_port = NODE_STATE.p2p_port.write().unwrap();
*w_p2p_port = Some(config.server.p2p_config.port);
}
// Put flag to start stratum server if autorun is available.
if NodeConfig::is_stratum_autorun_enabled() {
NODE_STATE.start_stratum_needed.store(true, Ordering::Relaxed);
}
// Reset an error.
{
let mut w_err = NODE_STATE.error.write();
*w_err = None;
}
// Start integrated node server.
let api_chan: &'static mut (oneshot::Sender<()>, oneshot::Receiver<()>) =
Box::leak(Box::new(oneshot::channel::<()>()));
let server_result = Server::new(server_config, None, api_chan);
// Delay after server start.
src/node/types.rs
+28
View File
@@ -0,0 +1,28 @@
// Copyright 2024 The Grim Developers
//
// 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.
/// Integrated node error type.
#[derive(Clone)]
pub enum NodeError {
/// Storage issue.
Storage,
/// P2P server issue.
P2P,
/// API server issue.
API,
/// Configuration issue.
Configuration,
/// Unknown error.
Unknown
}