Ecosystem/gupaxx
1
0
Fork 0
mirror of https://github.com/Cyrix126/gupaxx.git synced 2024-12-22 14:49:21 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

helper: p2pool - stdout payouts/xmr parser, priv -> pub functions

Browse source
This commit is contained in:
hinto-janaiyo 2022-12-05 14:55:50 -05:00
parent 5d293054cf
commit 1f3a472869
No known key found for this signature in database
GPG key ID: B1C5A64B80691E45
7 changed files with 592 additions and 302 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

96
Cargo.lock generated
View file

@ -23,6 +23,7 @@ dependencies = [
"log",
"num-format",
"num_cpus",
"portable-pty",
"rand 0.8.5",
"regex",
"rfd",
@ -1355,6 +1356,17 @@ dependencies = [
"version_check",
]
[[package]]
name = "filedescriptor"
version = "0.8.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7199d965852c3bac31f779ef99cbb4537f80e952e2d6aa0ffeb30cce00f4f46e"
dependencies = [
"libc",
"thiserror",
"winapi",
]
[[package]]
name = "filetime"
version = "0.2.18"
@ -2049,6 +2061,15 @@ dependencies = [
"web-sys",
]
[[package]]
name = "ioctl-rs"
version = "0.1.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f7970510895cee30b3e9128319f2cefd4bde883a39f38baa279567ba3a7eb97d"
dependencies = [
"libc",
]
[[package]]
name = "itertools"
version = "0.10.5"
@ -2916,6 +2937,24 @@ dependencies = [
"miniz_oxide 0.6.2",
]
[[package]]
name = "portable-pty"
version = "0.7.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4e4d17ec050a6b7ea4b15c430183772bce8384072d3f328e0967e72b7eec46b5"
dependencies = [
"anyhow",
"bitflags",
"filedescriptor",
"lazy_static",
"libc",
"log",
"serial",
"shared_library",
"shell-words",
"winapi",
]
[[package]]
name = "postage"
version = "0.5.0"
@ -3419,6 +3458,48 @@ dependencies = [
"syn",
]
[[package]]
name = "serial"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a1237a96570fc377c13baa1b88c7589ab66edced652e43ffb17088f003db3e86"
dependencies = [
"serial-core",
"serial-unix",
"serial-windows",
]
[[package]]
name = "serial-core"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3f46209b345401737ae2125fe5b19a77acce90cd53e1658cda928e4fe9a64581"
dependencies = [
"libc",
]
[[package]]
name = "serial-unix"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f03fbca4c9d866e24a459cbca71283f545a37f8e3e002ad8c70593871453cab7"
dependencies = [
"ioctl-rs",
"libc",
"serial-core",
"termios",
]
[[package]]
name = "serial-windows"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "15c6d3b776267a75d31bbdfd5d36c0ca051251caafc285827052bc53bcdc8162"
dependencies = [
"libc",
"serial-core",
]
[[package]]
name = "servo-fontconfig"
version = "0.5.1"
@ -3506,6 +3587,12 @@ dependencies = [
"libc",
]
[[package]]
name = "shell-words"
version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "24188a676b6ae68c3b2cb3a01be17fbf7240ce009799bb56d5b1409051e78fde"
[[package]]
name = "shellexpand"
version = "2.1.2"
@ -3752,6 +3839,15 @@ dependencies = [
"winapi-util",
]
[[package]]
name = "termios"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d5d9cf598a6d7ce700a4e6a9199da127e6819a61e64b68609683cc9a01b5683a"
dependencies = [
"libc",
]
[[package]]
name = "test-cert-gen"
version = "0.9.0"

1
Cargo.toml
View file

@ -38,6 +38,7 @@ image = { version = "0.24.4", features = ["png"] }
log = "0.4.17"
num_cpus = "1.13.1"
num-format = { version = "0.4.3", default-features = false }
portable-pty = "0.7.0"
rand = "0.8.5"
regex = { version = "1.6.0", default-features = false, features = ["perf"] }
rfd = "0.10.0"

17
src/README.md
View file

@ -24,6 +24,21 @@
## Thread Model
![thread_model.png](https://github.com/hinto-janaiyo/gupax/blob/main/images/thread_model.png)
Note: The process I/O model depends on if the `[Simple]` or `[Advanced]` version is used.
`[Simple]` has:
- 1 OS thread for the watchdog (API fetching, watching signals, etc)
- 1 OS thread (with 2 tokio tasks) for STDOUT/STDERR
- No pseudo terminal allocated
- No STDIN pipe
`[Advanced]` has:
- 1 OS thread for the watchdog (API fetching, watching signals, relaying STDIN)
- 1 OS thread for a PTY-Child combo (combines STDOUT/STDERR for me, nice!)
- A PTY (pseudo terminal) whose underlying type is abstracted with the [`portable_pty`](https://docs.rs/portable-pty/) library
The reason `[Advanced]` is non-async is because P2Pool requires a `TTY` to take STDIN. The PTY library used, [`portable_pty`](https://docs.rs/portable-pty/), doesn't implement async traits. There seem to be tokio PTY libraries, but they are Unix-specific. Having separate PTY code for Windows/Unix is also a big pain. Since the threads will be sleeping most of the time (the pipes are lazily read and buffered), it's fine. Ideally, any I/O should be a tokio task, though.
## Bootstrap
This is how Gupax works internally when starting up:
@ -33,7 +48,7 @@ This is how Gupax works internally when starting up:
- Start initializing main `App` struct
- Parse command arguments
- Attempt to read disk files
- If errors were found, pop-up window
- If errors were found, set the `panic` error screen
2. **AUTO**
- If `auto_update` == `true`, spawn auto-updating thread

1
src/constants.rs
View file

@ -38,6 +38,7 @@ pub const BYTES_BANNER: &[u8] = include_bytes!("../images/banner.png");
pub const P2POOL_BASE_ARGS: &str = "";
pub const XMRIG_BASE_ARGS: &str = "--http-host=127.0.0.1 --http-port=18088 --algo=rx/0 --coin=Monero";
pub const HORIZONTAL: &str = "--------------------------------------------";
pub const HORI_DOUBLE: &str = "----------------------------------------------------------------------------------------";
// This is the typical space added when using
// [ui.separator()] or [ui.group()]

692
src/helper.rs
View file

@ -64,8 +64,6 @@ pub struct Helper {
priv_api_xmrig: Arc<Mutex<PrivXmrigApi>>, // For "watchdog" thread
}
// Impl found at the very bottom of this file.
//---------------------------------------------------------------------------------------------------- [Process] Struct
// This holds all the state of a (child) process.
// The main GUI thread will use this to display console text, online state, etc.
@ -73,9 +71,8 @@ pub struct Process {
pub name: ProcessName, // P2Pool or XMRig?
pub state: ProcessState, // The state of the process (alive, dead, etc)
pub signal: ProcessSignal, // Did the user click [Start/Stop/Restart]?
start: Instant, // Start time of process
pub start: Instant, // Start time of process
pub uptime: HumanTime, // Human readable process uptime
pub output: String, // This is the process's PUBLIC stdout + stderr
// STDIN Problem:
// - User can input many many commands in 1 second
// - The process loop only processes every 1 second
@ -86,11 +83,26 @@ pub struct Process {
// - When the user inputs something, push it to a [Vec]
// - In the process loop, loop over every [Vec] element and
// send each one individually to the process stdin
pub child: Option<Arc<Mutex<tokio::process::Child>>>, // A handle to the actual child process
stdout: Option<tokio::process::ChildStdout>, // A handle to the process's STDOUT
stderr: Option<tokio::process::ChildStderr>, // A handle to the process's STDERR
stdin: Option<tokio::process::ChildStdin>, // A handle to the process's STDIN
//
pub input: Vec<String>,
// The below are the handles to the actual child process.
// [Simple] has no STDIN, but [Advanced] does. A PTY (pseudo-terminal) is
// required for P2Pool/XMRig to open their STDIN pipe, so whether [child]
// or [child_pty] actually has a [Some] depends on the users setting.
// [Simple]
child: Option<Arc<Mutex<tokio::process::Child>>>,
stdout: Option<tokio::process::ChildStdout>, // Handle to STDOUT pipe
stderr: Option<tokio::process::ChildStderr>, // Handle to STDERR pipe
// [Advanced] (PTY)
child_pty: Option<Arc<Mutex<Box<dyn portable_pty::Child + Send + std::marker::Sync>>>>, // STDOUT/STDERR is combined automatically thanks to this PTY, nice
stdin: Option<Box<dyn portable_pty::MasterPty + Send>>, // A handle to the process's MasterPTY/STDIN
// This is the process's private output [String], used by both [Simple] and [Advanced].
// The "watchdog" threads mutate this, the "helper" thread synchronizes the [Pub*Api] structs
// so that the data in here is cloned there roughly once a second. GUI thread never touches this.
output: String,
}
//---------------------------------------------------------------------------------------------------- [Process] Impl
@ -107,6 +119,7 @@ impl Process {
stderr: Option::None,
stdin: Option::None,
child: Option::None,
child_pty: Option::None,
// P2Pool log level 1 produces a bit less than 100,000 lines a day.
// Assuming each line averages 80 UTF-8 scalars (80 bytes), then this
// initial buffer should last around a week (56MB) before resetting.
@ -124,13 +137,10 @@ impl Process {
//---------------------------------------------------------------------------------------------------- [Process*] Enum
#[derive(Copy,Clone,Eq,PartialEq,Debug)]
pub enum ProcessState {
Alive, // Process is online, GREEN!
Dead, // Process is dead, BLACK!
Failed, // Process is dead AND exited with a bad code, RED!
// Process is starting up, YELLOW!
// Really, processes start instantly, this just accounts for the delay
// between the main thread and this threads 1 second event loop.
Starting,
Alive, // Process is online, GREEN!
Dead, // Process is dead, BLACK!
Failed, // Process is dead AND exited with a bad code, RED!
Middle, // Process is in the middle of something (starting, stopping, etc), YELLOW!
}
#[derive(Copy,Clone,Eq,PartialEq,Debug)]
@ -151,6 +161,331 @@ impl std::fmt::Display for ProcessState { fn fmt(&self, f: &mut std::fmt::Forma
impl std::fmt::Display for ProcessSignal { fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { write!(f, "{:#?}", self) } }
impl std::fmt::Display for ProcessName { fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { write!(f, "{:#?}", self) } }
//---------------------------------------------------------------------------------------------------- [Helper]
use tokio::io::{BufReader,AsyncBufReadExt};
impl Helper {
//---------------------------------------------------------------------------------------------------- General Functions
pub fn new(instant: std::time::Instant, pub_api_p2pool: Arc<Mutex<PubP2poolApi>>, pub_api_xmrig: Arc<Mutex<PubXmrigApi>>) -> Self {
Self {
instant,
human_time: HumanTime::into_human(instant.elapsed()),
p2pool: Arc::new(Mutex::new(Process::new(ProcessName::P2pool, String::new(), PathBuf::new()))),
xmrig: Arc::new(Mutex::new(Process::new(ProcessName::Xmrig, String::new(), PathBuf::new()))),
priv_api_p2pool: Arc::new(Mutex::new(PrivP2poolApi::new())),
priv_api_xmrig: Arc::new(Mutex::new(PrivXmrigApi::new())),
// These are created when initializing [App], since it needs a handle to it as well
pub_api_p2pool,
pub_api_xmrig,
}
}
// The tokio runtime that blocks while async reading both STDOUT/STDERR
// Cheaper than spawning 2 OS threads just to read 2 pipes (...right? :D)
#[tokio::main]
async fn async_read_stdout_stderr(process: Arc<Mutex<Process>>) {
let process_stdout = Arc::clone(&process);
let process_stderr = Arc::clone(&process);
let stdout = process.lock().unwrap().child.as_ref().unwrap().lock().unwrap().stdout.take().unwrap();
let stderr = process.lock().unwrap().child.as_ref().unwrap().lock().unwrap().stderr.take().unwrap();
// Create STDOUT pipe job
let stdout_job = tokio::spawn(async move {
let mut reader = BufReader::new(stdout).lines();
while let Ok(Some(line)) = reader.next_line().await {
println!("{}", line); // For debugging.
writeln!(process_stdout.lock().unwrap().output, "{}", line);
}
});
// Create STDERR pipe job
let stderr_job = tokio::spawn(async move {
let mut reader = BufReader::new(stderr).lines();
while let Ok(Some(line)) = reader.next_line().await {
println!("{}", line); // For debugging.
writeln!(process_stderr.lock().unwrap().output, "{}", line);
}
});
// Block and read both until they are closed (automatic when process dies)
// The ordering of STDOUT/STDERR should be automatic thanks to the locks.
tokio::join![stdout_job, stderr_job];
}
// Reads a PTY which combines STDOUT/STDERR for me, yay
fn read_pty(process: Arc<Mutex<Process>>, reader: Box<dyn std::io::Read + Send>) {
use std::io::BufRead;
let mut stdout = std::io::BufReader::new(reader).lines();
while let Some(Ok(line)) = stdout.next() {
println!("{}", line); // For debugging.
writeln!(process.lock().unwrap().output, "{}", line);
}
}
//---------------------------------------------------------------------------------------------------- P2Pool specific
// Intermediate function that parses the arguments, and spawns the P2Pool watchdog thread.
pub fn spawn_p2pool(helper: &Arc<Mutex<Self>>, state: &crate::disk::P2pool, path: std::path::PathBuf) {
let mut args = Vec::with_capacity(500);
let path = path.clone();
let mut api_path = path.clone();
api_path.pop();
// [Simple]
if state.simple {
// Build the p2pool argument
let (ip, rpc, zmq) = crate::node::enum_to_ip_rpc_zmq_tuple(state.node); // Get: (IP, RPC, ZMQ)
args.push("--wallet".to_string()); args.push(state.address.clone()); // Wallet address
args.push("--host".to_string()); args.push(ip.to_string()); // IP Address
args.push("--rpc-port".to_string()); args.push(rpc.to_string()); // RPC Port
args.push("--zmq-port".to_string()); args.push(zmq.to_string()); // ZMQ Port
args.push("--data-api".to_string()); args.push(api_path.display().to_string()); // API Path
args.push("--local-api".to_string()); // Enable API
args.push("--no-color".to_string()); // Remove color escape sequences, Gupax terminal can't parse it :(
args.push("--mini".to_string()); // P2Pool Mini
// [Advanced]
} else {
// Overriding command arguments
if !state.arguments.is_empty() {
for arg in state.arguments.split_whitespace() {
args.push(arg.to_string());
}
// Else, build the argument
} else {
args.push(state.address.clone()); // Wallet
args.push(state.selected_ip.clone()); // IP
args.push(state.selected_rpc.clone()); // RPC
args.push(state.selected_zmq.clone()); // ZMQ
args.push("--local-api".to_string()); // Enable API
args.push("--no-color".to_string()); // Remove color escape sequences
if state.mini { args.push("--mini".to_string()); }; // Mini
args.push(format!("--loglevel {}", state.log_level)); // Log Level
args.push(format!("--out-peers {}", state.out_peers)); // Out Peers
args.push(format!("--in-peers {}", state.in_peers)); // In Peers
args.push(format!("--data-api {}", api_path.display())); // API Path
}
}
// Print arguments to console
crate::disk::print_dash(&format!("P2Pool | Launch arguments ... {:#?}", args));
// Spawn watchdog thread
let simple = !state.simple; // Will this process need a PTY (STDIN)?
let process = Arc::clone(&helper.lock().unwrap().p2pool);
let pub_api = Arc::clone(&helper.lock().unwrap().pub_api_p2pool);
let priv_api = Arc::clone(&helper.lock().unwrap().priv_api_p2pool);
thread::spawn(move || {
if simple {
Self::spawn_simple_p2pool_watchdog(process, pub_api, priv_api, args, path);
} else {
Self::spawn_pty_p2pool_watchdog(process, pub_api, priv_api, args, path);
}
});
}
// The [Simple] P2Pool watchdog tokio runtime, using async features with no PTY (STDIN).
#[tokio::main]
async fn spawn_simple_p2pool_watchdog(process: Arc<Mutex<Process>>, pub_api: Arc<Mutex<PubP2poolApi>>, priv_api: Arc<Mutex<PrivP2poolApi>>, args: Vec<String>, path: std::path::PathBuf) {
// 1a. Create command
let child = Arc::new(Mutex::new(tokio::process::Command::new(path)
.args(args)
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.stdin(Stdio::piped())
.spawn().unwrap()));
// 2. Set process state
let mut lock = process.lock().unwrap();
lock.state = ProcessState::Alive;
lock.signal = ProcessSignal::None;
lock.start = Instant::now();
lock.child = Some(Arc::clone(&child));
drop(lock);
// 3. Spawn STDOUT/STDERR thread
let process_clone = Arc::clone(&process);
thread::spawn(move || {
Self::async_read_stdout_stderr(process_clone);
});
// 4. Loop forever as watchdog until process dies
loop {
// a. Watch user SIGNAL
if process.lock().unwrap().signal == ProcessSignal::Stop {
process.lock().unwrap().child.as_mut().unwrap().lock().unwrap().kill().await;
process.lock().unwrap().signal = ProcessSignal::None;
}
// let signal = match process.lock().unwrap().signal {
// ProcessSignal::Stop => { crate::disk::print_dash("KILLING P2POOL"); process.lock().unwrap().child.as_mut().unwrap().lock().unwrap().kill().await.unwrap() },
// ProcessSignal::Restart => process.lock().unwrap().child.as_mut().unwrap().lock().unwrap().kill().await,
// _ => Ok(()),
// };
// b. Create STDIN task
if !process.lock().unwrap().input.is_empty() { /* process it */ }
// c. Create API task
let async_file_read = { /* tokio async file read job */ };
// d. Execute async tasks
// tokio::join![signal];
// f. Sleep (900ms)
std::thread::sleep(MILLI_900);
}
}
// The [Advanced] P2Pool watchdog. Spawns 1 OS thread for reading a PTY (STDOUT+STDERR), and combines the [Child] with a PTY so STDIN actually works.
#[tokio::main]
async fn spawn_pty_p2pool_watchdog(process: Arc<Mutex<Process>>, pub_api: Arc<Mutex<PubP2poolApi>>, priv_api: Arc<Mutex<PrivP2poolApi>>, args: Vec<String>, path: std::path::PathBuf) {
// 1a. Create PTY
let pty = portable_pty::native_pty_system();
let mut pair = pty.openpty(portable_pty::PtySize {
rows: 24,
cols: 80,
pixel_width: 0,
pixel_height: 0,
}).unwrap();
// 1b. Create command
let mut cmd = portable_pty::CommandBuilder::new(path);
cmd.args(args);
// 1c. Create child
let child_pty = Arc::new(Mutex::new(pair.slave.spawn_command(cmd).unwrap()));
// 2. Set process state
let mut lock = process.lock().unwrap();
lock.state = ProcessState::Alive;
lock.signal = ProcessSignal::None;
lock.start = Instant::now();
lock.child_pty = Some(Arc::clone(&child_pty));
let reader = pair.master.try_clone_reader().unwrap(); // Get STDOUT/STDERR before moving the PTY
lock.stdin = Some(pair.master);
drop(lock);
// 3. Spawn PTY read thread
let process_clone = Arc::clone(&process);
thread::spawn(move || {
Self::read_pty(process_clone, reader);
});
// 4. Loop as watchdog
loop {
// Set timer
let now = Instant::now();
// Check SIGNAL
if process.lock().unwrap().signal == ProcessSignal::Stop {
child_pty.lock().unwrap().kill(); // This actually sends a SIGHUP to p2pool (closes the PTY, hangs up on p2pool)
// Wait to get the exit status
let exit_status = match child_pty.lock().unwrap().wait() {
Ok(e) => if e.success() { "Successful" } else { "Failed" },
_ => "Unknown Error",
};
let mut lock = process.lock().unwrap();
let uptime = lock.uptime.clone();
info!("P2Pool | Stopped ... Uptime was: [{}], Exit status: [{}]", uptime, exit_status);
writeln!(lock.output, "{}\nP2Pool stopped | Uptime: [{}] | Exit status: [{}]\n{}\n\n", HORI_DOUBLE, uptime, exit_status, HORI_DOUBLE);
lock.signal = ProcessSignal::None;
break
}
// Check vector of user input
let mut lock = process.lock().unwrap();
if !lock.input.is_empty() {
let input = std::mem::take(&mut lock.input);
for line in input {
writeln!(lock.stdin.as_mut().unwrap(), "{}", line);
}
}
drop(lock);
// Sleep (only if 900ms hasn't passed)
let elapsed = now.elapsed().as_millis();
// Since logic goes off if less than 1000, casting should be safe
if elapsed < 900 { std::thread::sleep(std::time::Duration::from_millis((900-elapsed) as u64)); }
}
// 5. If loop broke, we must be done here.
info!("P2Pool | Watchdog thread exiting... Goodbye!");
}
//---------------------------------------------------------------------------------------------------- XMRig specific
// Intermediate function that parses the arguments, and spawns the XMRig watchdog thread.
pub fn spawn_xmrig(state: &crate::disk::Xmrig, api_path: &std::path::Path) {
let mut args = Vec::with_capacity(500);
if state.simple {
let rig_name = if state.simple_rig.is_empty() { GUPAX_VERSION.to_string() } else { state.simple_rig.clone() }; // Rig name
args.push(format!("--threads {}", state.current_threads)); // Threads
args.push(format!("--user {}", state.simple_rig)); // Rig name
args.push(format!("--url 127.0.0.1:3333")); // Local P2Pool (the default)
args.push("--no-color".to_string()); // No color escape codes
if state.pause != 0 { args.push(format!("--pause-on-active {}", state.pause)); } // Pause on active
} else {
if !state.arguments.is_empty() {
for arg in state.arguments.split_whitespace() {
args.push(arg.to_string());
}
} else {
args.push(format!("--user {}", state.address.clone())); // Wallet
args.push(format!("--threads {}", state.current_threads)); // Threads
args.push(format!("--rig-id {}", state.selected_rig)); // Rig ID
args.push(format!("--url {}:{}", state.selected_ip.clone(), state.selected_port.clone())); // IP/Port
args.push(format!("--http-host {}", state.api_ip).to_string()); // HTTP API IP
args.push(format!("--http-port {}", state.api_port).to_string()); // HTTP API Port
args.push("--no-color".to_string()); // No color escape codes
if state.tls { args.push("--tls".to_string()); } // TLS
if state.keepalive { args.push("--keepalive".to_string()); } // Keepalive
if state.pause != 0 { args.push(format!("--pause-on-active {}", state.pause)); } // Pause on active
}
}
// Print arguments to console
crate::disk::print_dash(&format!("XMRig | Launch arguments ... {:#?}", args));
// Spawn watchdog thread
thread::spawn(move || {
Self::spawn_xmrig_watchdog(args);
});
}
// The actual XMRig watchdog tokio runtime.
#[tokio::main]
pub async fn spawn_xmrig_watchdog(args: Vec<String>) {
}
//---------------------------------------------------------------------------------------------------- The "helper"
// Intermediate function that spawns the helper thread.
pub fn spawn_helper(helper: &Arc<Mutex<Self>>) {
let helper = Arc::clone(helper);
thread::spawn(move || { Self::helper(helper); });
}
// [helper] = Actual Arc
// [h] = Temporary lock that gets dropped
// [jobs] = Vector of async jobs ready to go
// #[tokio::main]
pub fn helper(helper: Arc<Mutex<Self>>) {
// Begin loop
loop {
// 1. Create "jobs" vector holding async tasks
// let jobs: Vec<tokio::task::JoinHandle<Result<(), anyhow::Error>>> = vec![];
// 2. Loop init timestamp
let start = Instant::now();
// 7. Set Gupax/P2Pool/XMRig uptime
let mut h = helper.lock().unwrap();
h.human_time = HumanTime::into_human(h.instant.elapsed());
drop(h);
// 8. Calculate if we should sleep or not.
// If we should sleep, how long?
let elapsed = start.elapsed().as_millis();
if elapsed < 1000 {
// Casting from u128 to u64 should be safe here, because [elapsed]
// is less than 1000, meaning it can fit into a u64 easy.
std::thread::sleep(std::time::Duration::from_millis((1000-elapsed) as u64));
}
// 9. End loop
}
}
}
//---------------------------------------------------------------------------------------------------- [HumanTime]
// This converts a [std::time::Duration] into something more readable.
// Used for uptime display purposes: [7 years, 8 months, 15 days, 23 hours, 35 minutes, 1 second]
@ -316,13 +651,43 @@ impl HumanNumber {
}
}
//---------------------------------------------------------------------------------------------------- Regexes
// Not to be confused with the [Regexes] struct in [main.rs], this one is meant
// for parsing the output of P2Pool and finding payouts and total XMR found.
// Why Regex instead of the standard library?
// 1. I'm already using Regex
// 2. It's insanely faster
//
// The following STDLIB implementation takes [0.003~] seconds to find all matches given a [String] with 30k lines:
// let mut n = 0;
// for line in P2POOL_OUTPUT.lines() {
// if line.contains("[0-9].[0-9]+ XMR") { n += 1; }
// }
//
// This regex function takes [0.0003~] seconds (10x faster):
// let regex = Regex::new("[0-9].[0-9]+ XMR").unwrap();
// let n = regex.find_iter(P2POOL_OUTPUT).count();
//
// Both are nominally fast enough where it doesn't matter too much but meh, why not use regex.
struct P2poolRegex {
xmr: regex::Regex,
}
impl P2poolRegex {
fn new() -> Self {
Self { xmr: regex::Regex::new("[0-9].[0-9]+ XMR").unwrap(), }
}
}
//---------------------------------------------------------------------------------------------------- Public P2Pool API
// GUI thread interfaces with this.
pub struct PubP2poolApi {
// One off
pub mini: bool,
// Output
pub output: String,
// These are manually parsed from the STDOUT.
pub payouts: f64,
pub payouts: u128,
pub payouts_hour: f64,
pub payouts_day: f64,
pub payouts_month: f64,
@ -344,7 +709,8 @@ impl PubP2poolApi {
pub fn new() -> Self {
Self {
mini: true,
payouts: 0.0,
output: String::with_capacity(56_000_000),
payouts: 0,
payouts_hour: 0.0,
payouts_day: 0.0,
payouts_month: 0.0,
@ -362,10 +728,53 @@ impl PubP2poolApi {
}
}
// Mutate [PubP2poolApi] with data from a [PrivP2poolApi].
fn update_from_priv(self, output: String, regex: P2poolRegex, private: PrivP2poolApi, uptime: f64) -> Self {
// 1. Parse STDOUT
let (payouts, xmr) = Self::calc_payouts_and_xmr(&output, &regex);
let stdout_parse = Self {
output: output.clone(),
payouts,
xmr,
..self // <- So useful
};
// 2. Time calculations
let hour_day_month = Self::update_hour_day_month(stdout_parse, uptime);
// 3. Final priv -> pub conversion
Self {
hashrate_15m: HumanNumber::from_u128(private.hashrate_15m),
hashrate_1h: HumanNumber::from_u128(private.hashrate_1h),
hashrate_24h: HumanNumber::from_u128(private.hashrate_24h),
shares_found: HumanNumber::from_u128(private.shares_found),
average_effort: HumanNumber::to_percent(private.average_effort),
current_effort: HumanNumber::to_percent(private.current_effort),
connections: HumanNumber::from_u16(private.connections),
..hour_day_month // <- Holy cow this is so good
}
}
// Essentially greps the output for [x.xxxxxxxxxxxx XMR] where x = a number.
// It sums each match and counts along the way, handling an error by not adding and printing to console.
fn calc_payouts_and_xmr(output: &str, regex: &P2poolRegex) -> (u128 /* payout count */, f64 /* total xmr */) {
let mut iter = regex.xmr.find_iter(output);
let mut result: f64 = 0.0;
let mut count: u128 = 0;
for i in iter {
if let Some(text) = i.as_str().split_whitespace().next() {
match text.parse::<f64>() {
Ok(num) => result += num,
Err(e) => error!("P2Pool | Total XMR sum calculation error: [{}]", e),
}
count += 1;
}
}
(count, result)
}
// Updates the struct with hour/day/month calculations given an uptime in f64 seconds.
pub fn update_hour_day_month(self, elapsed: f64) -> Self {
fn update_hour_day_month(self, elapsed: f64) -> Self {
// Payouts
let per_sec = self.payouts / elapsed;
let per_sec = (self.payouts as f64) / elapsed;
let payouts_hour = (per_sec * 60.0) * 60.0;
let payouts_day = payouts_hour * 24.0;
let payouts_month = payouts_day * 30.0;
@ -381,7 +790,7 @@ impl PubP2poolApi {
xmr_hour,
xmr_day,
xmr_month,
..self // <- wow this is so useful
..self
}
}
}
@ -413,16 +822,11 @@ impl PrivP2poolApi {
connections: 0,
}
}
// Formats raw private data into ready-to-print human readable version.
// pub fn from_priv(private: PrivP2poolApi) -> Self {
// Self {
// }
// }
}
//---------------------------------------------------------------------------------------------------- Public XMRig API
pub struct PubXmrigApi {
output: String,
worker_id: String,
resources: HumanNumber,
hashrate: HumanNumber,
@ -435,6 +839,7 @@ pub struct PubXmrigApi {
impl PubXmrigApi {
pub fn new() -> Self {
Self {
output: String::with_capacity(56_000_000),
worker_id: "???".to_string(),
resources: HumanNumber::unknown(),
hashrate: HumanNumber::unknown(),
@ -446,8 +851,9 @@ impl PubXmrigApi {
}
// Formats raw private data into ready-to-print human readable version.
fn from_priv(private: PrivXmrigApi) -> Self {
fn from_priv(private: PrivXmrigApi, output: String) -> Self {
Self {
output: output.clone(),
worker_id: private.worker_id,
resources: HumanNumber::from_load(private.resources.load_average),
hashrate: HumanNumber::from_hashrate(private.hashrate.total),
@ -524,235 +930,3 @@ impl Hashrate {
}
}
}
//---------------------------------------------------------------------------------------------------- [Helper]
use tokio::io::{BufReader,AsyncBufReadExt};
impl Helper {
//---------------------------------------------------------------------------------------------------- General Functions
pub fn new(instant: std::time::Instant) -> Self {
Self {
instant,
human_time: HumanTime::into_human(instant.elapsed()),
p2pool: Arc::new(Mutex::new(Process::new(ProcessName::P2pool, String::new(), PathBuf::new()))),
xmrig: Arc::new(Mutex::new(Process::new(ProcessName::Xmrig, String::new(), PathBuf::new()))),
pub_api_p2pool: Arc::new(Mutex::new(PubP2poolApi::new())),
pub_api_xmrig: Arc::new(Mutex::new(PubXmrigApi::new())),
priv_api_p2pool: Arc::new(Mutex::new(PrivP2poolApi::new())),
priv_api_xmrig: Arc::new(Mutex::new(PrivXmrigApi::new())),
}
}
// The tokio runtime that blocks while async reading both STDOUT/STDERR
// Cheaper than spawning 2 OS threads just to read 2 pipes (...right? :D)
#[tokio::main]
async fn read_stdout_stderr(process: Arc<Mutex<Process>>) {
let process_stdout = Arc::clone(&process);
let process_stderr = Arc::clone(&process);
let stdout = process.lock().unwrap().stdout.take().unwrap();
let stderr = process.lock().unwrap().stderr.take().unwrap();
// Create STDOUT pipe job
let stdout_job = tokio::spawn(async move {
let mut stdout_reader = BufReader::new(stdout).lines();
while let Ok(Some(line)) = stdout_reader.next_line().await {
// println!("{}", line); // For debugging.
writeln!(process_stdout.lock().unwrap().output, "{}", line);
}
});
// Create STDERR pipe job
let stderr_job = tokio::spawn(async move {
let mut stderr_reader = BufReader::new(stderr).lines();
while let Ok(Some(line)) = stderr_reader.next_line().await {
// println!("{}", line); // For debugging.
writeln!(process_stderr.lock().unwrap().output, "{}", line);
}
});
// Block and read both until they are closed (automatic when process dies)
// The ordering of STDOUT/STDERR should be automatic thanks to the locks.
tokio::join![stdout_job, stderr_job];
}
//---------------------------------------------------------------------------------------------------- P2Pool specific
// Intermediate function that parses the arguments, and spawns the P2Pool watchdog thread.
pub fn spawn_p2pool(helper: &Arc<Mutex<Self>>, state: &crate::disk::P2pool, path: std::path::PathBuf) {
let mut args = Vec::with_capacity(500);
let path = path.clone();
let mut api_path = path.clone();
api_path.pop();
// [Simple]
if state.simple {
// Build the p2pool argument
let (ip, rpc, zmq) = crate::node::enum_to_ip_rpc_zmq_tuple(state.node); // Get: (IP, RPC, ZMQ)
args.push("--wallet".to_string()); args.push(state.address.clone()); // Wallet address
args.push("--host".to_string()); args.push(ip.to_string()); // IP Address
args.push("--rpc-port".to_string()); args.push(rpc.to_string()); // RPC Port
args.push("--zmq-port".to_string()); args.push(zmq.to_string()); // ZMQ Port
args.push("--data-api".to_string()); args.push(api_path.display().to_string()); // API Path
args.push("--local-api".to_string()); // Enable API
args.push("--no-color".to_string()); // Remove color escape sequences, Gupax terminal can't parse it :(
args.push("--mini".to_string()); // P2Pool Mini
// [Advanced]
} else {
// Overriding command arguments
if !state.arguments.is_empty() {
for arg in state.arguments.split_whitespace() {
args.push(arg.to_string());
}
// Else, build the argument
} else {
args.push(state.address.clone()); // Wallet
args.push(state.selected_ip.clone()); // IP
args.push(state.selected_rpc.clone()); // RPC
args.push(state.selected_zmq.clone()); // ZMQ
args.push("--local-api".to_string()); // Enable API
args.push("--no-color".to_string()); // Remove color escape sequences
if state.mini { args.push("--mini".to_string()); }; // Mini
args.push(format!("--loglevel {}", state.log_level)); // Log Level
args.push(format!("--out-peers {}", state.out_peers)); // Out Peers
args.push(format!("--in-peers {}", state.in_peers)); // In Peers
args.push(format!("--data-api {}", api_path.display())); // API Path
}
}
// Print arguments to console
crate::disk::print_dash(&format!("P2Pool | Launch arguments ... {:#?}", args));
// Spawn watchdog thread
let process = Arc::clone(&helper.lock().unwrap().p2pool);
let pub_api = Arc::clone(&helper.lock().unwrap().pub_api_p2pool);
let priv_api = Arc::clone(&helper.lock().unwrap().priv_api_p2pool);
thread::spawn(move || {
Self::spawn_p2pool_watchdog(process, pub_api, priv_api, args, path);
});
}
// The actual P2Pool watchdog tokio runtime.
#[tokio::main]
async fn spawn_p2pool_watchdog(process: Arc<Mutex<Process>>, pub_api: Arc<Mutex<PubP2poolApi>>, priv_api: Arc<Mutex<PrivP2poolApi>>, args: Vec<String>, path: std::path::PathBuf) {
// 1. Create command
let child = Arc::new(Mutex::new(tokio::process::Command::new(path)
.args(args)
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.stdin(Stdio::piped())
.spawn().unwrap()));
// 2. Set process state
let mut lock = process.lock().unwrap();
lock.state = ProcessState::Alive;
lock.signal = ProcessSignal::None;
lock.start = Instant::now();
lock.child = Some(Arc::clone(&child));
lock.stdin = Some(child.lock().unwrap().stdin.take().unwrap());
drop(lock);
// 3. Spawn STDOUT/STDERR thread
let process_clone = Arc::clone(&process);
thread::spawn(move || {
Self::read_stdout_stderr(process_clone);
});
// 4. Loop forever as watchdog until process dies
loop {
// a. Watch user SIGNAL
match process.lock().unwrap().signal {
ProcessSignal::Stop => {},
ProcessSignal::Restart => {},
_ => {},
}
// b. Create STDIN task
if !process.lock().unwrap().input.is_empty() { /* process it */ }
// c. Create API task
let async_file_read = { /* tokio async file read job */ };
// d. Execute async tasks
tokio::join![/* jobs */];
// f. Sleep (900ms)
std::thread::sleep(MILLI_900);
}
}
//---------------------------------------------------------------------------------------------------- XMRig specific
// Intermediate function that parses the arguments, and spawns the XMRig watchdog thread.
pub fn spawn_xmrig(state: &crate::disk::Xmrig, api_path: &std::path::Path) {
let mut args = Vec::with_capacity(500);
if state.simple {
let rig_name = if state.simple_rig.is_empty() { GUPAX_VERSION.to_string() } else { state.simple_rig.clone() }; // Rig name
args.push(format!("--threads {}", state.current_threads)); // Threads
args.push(format!("--user {}", state.simple_rig)); // Rig name
args.push(format!("--url 127.0.0.1:3333")); // Local P2Pool (the default)
args.push("--no-color".to_string()); // No color escape codes
if state.pause != 0 { args.push(format!("--pause-on-active {}", state.pause)); } // Pause on active
} else {
if !state.arguments.is_empty() {
for arg in state.arguments.split_whitespace() {
args.push(arg.to_string());
}
} else {
args.push(format!("--user {}", state.address.clone())); // Wallet
args.push(format!("--threads {}", state.current_threads)); // Threads
args.push(format!("--rig-id {}", state.selected_rig)); // Rig ID
args.push(format!("--url {}:{}", state.selected_ip.clone(), state.selected_port.clone())); // IP/Port
args.push(format!("--http-host {}", state.api_ip).to_string()); // HTTP API IP
args.push(format!("--http-port {}", state.api_port).to_string()); // HTTP API Port
args.push("--no-color".to_string()); // No color escape codes
if state.tls { args.push("--tls".to_string()); } // TLS
if state.keepalive { args.push("--keepalive".to_string()); } // Keepalive
if state.pause != 0 { args.push(format!("--pause-on-active {}", state.pause)); } // Pause on active
}
}
// Print arguments to console
crate::disk::print_dash(&format!("XMRig | Launch arguments ... {:#?}", args));
// Spawn watchdog thread
thread::spawn(move || {
Self::spawn_xmrig_watchdog(args);
});
}
// The actual XMRig watchdog tokio runtime.
#[tokio::main]
pub async fn spawn_xmrig_watchdog(args: Vec<String>) {
}
//---------------------------------------------------------------------------------------------------- The "helper"
// Intermediate function that spawns the helper thread.
pub fn spawn_helper(helper: &Arc<Mutex<Self>>) {
let helper = Arc::clone(helper);
thread::spawn(move || { Self::helper(helper); });
}
// [helper] = Actual Arc
// [h] = Temporary lock that gets dropped
// [jobs] = Vector of async jobs ready to go
// #[tokio::main]
pub fn helper(helper: Arc<Mutex<Self>>) {
// Begin loop
loop {
// 1. Create "jobs" vector holding async tasks
// let jobs: Vec<tokio::task::JoinHandle<Result<(), anyhow::Error>>> = vec![];
// 2. Loop init timestamp
let start = Instant::now();
// 7. Set Gupax/P2Pool/XMRig uptime
let mut h = helper.lock().unwrap();
h.human_time = HumanTime::into_human(h.instant.elapsed());
drop(h);
// 8. Calculate if we should sleep or not.
// If we should sleep, how long?
let elapsed = start.elapsed().as_millis();
if elapsed < 1000 {
// Casting from u128 to u64 should be safe here, because [elapsed]
// is less than 1000, meaning it can fit into a u64 easy.
std::thread::sleep(std::time::Duration::from_millis((1000-elapsed) as u64));
}
// 9. End loop
}
}
}

56
src/main.rs
View file

@ -103,10 +103,12 @@ pub struct App {
// indicate if an error message needs to be displayed
// (it takes up the whole screen with [error_msg] and buttons for ok/quit/etc)
error_state: ErrorState,
// Helper State:
// This holds everything related to the data
// processed by the "helper thread", including
// Helper/API State:
// This holds everything related to the data processed by the "helper thread".
// This includes the "helper" threads public P2Pool/XMRig's API.
helper: Arc<Mutex<Helper>>,
p2pool_api: Arc<Mutex<PubP2poolApi>>,
xmrig_api: Arc<Mutex<PubXmrigApi>>,
// Fix-me.
// These shouldn't exist
@ -144,6 +146,8 @@ impl App {
fn new(now: Instant) -> Self {
info!("Initializing App Struct...");
let p2pool_api = Arc::new(Mutex::new(PubP2poolApi::new()));
let xmrig_api = Arc::new(Mutex::new(PubXmrigApi::new()));
let mut app = Self {
tab: Tab::default(),
ping: Arc::new(Mutex::new(Ping::new())),
@ -161,8 +165,9 @@ impl App {
restart: Arc::new(Mutex::new(Restart::No)),
diff: false,
error_state: ErrorState::new(),
helper: Arc::new(Mutex::new(Helper::new(now))),
helper: Arc::new(Mutex::new(Helper::new(now, p2pool_api.clone(), xmrig_api.clone()))),
p2pool_api,
xmrig_api,
// TODO
// these p2pool/xmrig bools are here for debugging purposes
// they represent the online/offline status.
@ -1014,22 +1019,29 @@ impl eframe::App for App {
});
ui.group(|ui| {
let width = (ui.available_width()/3.0)-5.0;
if self.p2pool {
if ui.add_sized([width, height], Button::new("⟲")).on_hover_text("Restart P2Pool").clicked() { self.p2pool = false; }
if ui.add_sized([width, height], Button::new("⏹")).on_hover_text("Stop P2Pool").clicked() { self.p2pool = false; }
ui.add_enabled_ui(false, |ui| {
if ui.add_sized([width, height], Button::new("⏺")).on_hover_text("Start P2Pool").clicked() {
Helper::spawn_p2pool(&self.helper, &self.state.p2pool, self.state.gupax.absolute_p2pool_path.clone());
}
});
} else {
ui.add_enabled_ui(false, |ui| {
ui.add_sized([width, height], Button::new("⟲")).on_hover_text("Restart P2Pool");
ui.add_sized([width, height], Button::new("⏹")).on_hover_text("Stop P2Pool");
});
if ui.add_sized([width, height], Button::new("⏺")).on_hover_text("Start P2Pool").clicked() {
Helper::spawn_p2pool(&self.helper, &self.state.p2pool, self.state.gupax.absolute_p2pool_path.clone());
}
// if self.p2pool {
// if ui.add_sized([width, height], Button::new("⟲")).on_hover_text("Restart P2Pool").clicked() { self.p2pool = false; }
// if ui.add_sized([width, height], Button::new("⏹")).on_hover_text("Stop P2Pool").clicked() { self.p2pool = false; }
// ui.add_enabled_ui(false, |ui| {
// if ui.add_sized([width, height], Button::new("⏺")).on_hover_text("Start P2Pool").clicked() {
// Helper::spawn_p2pool(&self.helper, &self.state.p2pool, self.state.gupax.absolute_p2pool_path.clone());
// }
// });
// } else {
// ui.add_enabled_ui(false, |ui| {
// ui.add_sized([width, height], Button::new("⟲")).on_hover_text("Restart P2Pool");
// ui.add_sized([width, height], Button::new("⏹")).on_hover_text("Stop P2Pool");
// });
// if ui.add_sized([width, height], Button::new("⏺")).on_hover_text("Start P2Pool").clicked() {
// Helper::spawn_p2pool(&self.helper, &self.state.p2pool, self.state.gupax.absolute_p2pool_path.clone());
// }
// }
ui.add_sized([width, height], Button::new("⟲")).on_hover_text("Restart P2Pool");
if ui.add_sized([width, height], Button::new("⏹")).on_hover_text("Stop P2Pool").clicked() {
self.helper.lock().unwrap().p2pool.lock().unwrap().signal = ProcessSignal::Stop;
}
if ui.add_sized([width, height], Button::new("⏺")).on_hover_text("Start P2Pool").clicked() {
Helper::spawn_p2pool(&self.helper, &self.state.p2pool, self.state.gupax.absolute_p2pool_path.clone());
}
});
},
@ -1101,7 +1113,7 @@ impl eframe::App for App {
Gupax::show(&mut self.state.gupax, &self.og, &self.state_path, &self.update, &self.file_window, &mut self.error_state, &self.restart, self.width, self.height, frame, ctx, ui);
}
Tab::P2pool => {
P2pool::show(&mut self.state.p2pool, &mut self.node_vec, &self.og, self.p2pool, &self.ping, &self.regex, self.width, self.height, ctx, ui);
P2pool::show(&mut self.state.p2pool, &mut self.node_vec, &self.og, self.p2pool, &self.ping, &self.regex, &self.helper, &self.p2pool_api, self.width, self.height, ctx, ui);
}
Tab::Xmrig => {
Xmrig::show(&mut self.state.xmrig, &mut self.pool_vec, &self.regex, self.width, self.height, ctx, ui);

31
src/p2pool.rs
View file

@ -32,32 +32,23 @@ use regex::Regex;
use log::*;
impl P2pool {
pub fn show(&mut self, node_vec: &mut Vec<(String, Node)>, og: &Arc<Mutex<State>>, _online: bool, ping: &Arc<Mutex<Ping>>, regex: &Regexes, width: f32, height: f32, ctx: &egui::Context, ui: &mut egui::Ui) {
pub fn show(&mut self, node_vec: &mut Vec<(String, Node)>, og: &Arc<Mutex<State>>, _online: bool, ping: &Arc<Mutex<Ping>>, regex: &Regexes, helper: &Arc<Mutex<Helper>>, api: &Arc<Mutex<PubP2poolApi>>, width: f32, height: f32, ctx: &egui::Context, ui: &mut egui::Ui) {
let text_edit = height / 22.0;
//---------------------------------------------------------------------------------------------------- Console
ui.group(|ui| {
let height = height / 10.0;
let height = height / 2.5;
let width = width - SPACE;
ui.style_mut().override_text_style = Some(Monospace);
//ui.add_sized([width, height*3.5], TextEdit::multiline(&mut "asdf"));
egui::Frame::none()
.fill(Color32::from_rgb(18, 18, 18))
.show(ui, |ui| {
let text_style = egui::TextStyle::Monospace;
let row_height = ui.text_style_height(&text_style);
let total_rows = 700_000;
let width = width-(SPACE*2.0);
egui::ScrollArea::vertical().max_width(width).max_height(height*3.5).auto_shrink([false; 2]).show_rows(ui, row_height, total_rows, |ui, row_range| {
let mut text = "".to_string();
for row in row_range {
text = format!("{}Row {}/{}\n", text, row + 1, total_rows);
// ui.label(text);
}
ui.add_sized([width, height*3.5], TextEdit::multiline(&mut text.as_str()));
egui::Frame::none().fill(Color32::from_rgb(18, 18, 18)).show(ui, |ui| {
ui.style_mut().override_text_style = Some(Name("MonospaceSmall".into()));
egui::ScrollArea::vertical().stick_to_bottom(true).max_width(width).max_height(height).auto_shrink([false; 2]).show_viewport(ui, |ui, _| {
let lock = api.lock().unwrap();
ui.add_sized([width, height], TextEdit::multiline(&mut lock.output.as_str()));
// if lock.p2pool.lock().unwrap().state == ProcessState::Alive { ctx.request_repaint(); }
});
});
});
ui.separator();
ui.add_sized([width, text_edit], TextEdit::hint_text(TextEdit::singleline(&mut "".to_string()), r#"Type a command (e.g "help" or "status") and press Enter"#));
// ui.separator();
// ui.add_sized([width, text_edit], TextEdit::hint_text(TextEdit::singleline(&mut "".to_string()), r#"Type a command (e.g "help" or "status") and press Enter"#));
});
//---------------------------------------------------------------------------------------------------- Args