Skip to content

Migrate translator to shared synchronous primitive #530

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Shourya742 wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Migrate translator to shared synchronous primitive #530

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
57e6cda
migrate translator to new shared primitives API
Shourya742 Jul 21, 2026
2ec31b8
Remove dashmap from translator
Shourya742 Jul 21, 2026
a8e04fe
open-channel success can orphan an upstream/SV2 channel if the SV1 do…
Shourya742 Jul 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion integration-tests/Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 0 additions & 1 deletion miner-apps/Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 0 additions & 1 deletion miner-apps/translator/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@ tracing = { version = "0.1" }
clap = { version = "4.5.39", features = ["derive"] }
hex = "0.4.3"
hotpath = "0.14.0"
dashmap = "6.1.0"

[features]
default = ["monitoring"]
Expand Down
102 changes: 52 additions & 50 deletions miner-apps/translator/src/lib/monitoring.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,53 +18,53 @@ impl ServerMonitoring for ChannelManager {
TproxyMode::Aggregated => {
// In Aggregated mode: one shared channel to the server
// stored under AGGREGATED_CHANNEL_ID
if let Some(aggregated_extended_channel) =
self.extended_channels.get(&AGGREGATED_CHANNEL_ID)
{
let channel_id = aggregated_extended_channel.get_channel_id();
let target = *aggregated_extended_channel.get_target();
let extranonce_prefix =
aggregated_extended_channel.get_extranonce_prefix().to_vec();
let user_identity = aggregated_extended_channel.get_user_identity().to_string();
let full_extranonce_size =
aggregated_extended_channel.get_full_extranonce_size();
let rollable_extranonce_size =
aggregated_extended_channel.get_rollable_extranonce_size();
let version_rolling = aggregated_extended_channel.is_version_rolling();
let nominal_hashrate = aggregated_extended_channel.get_nominal_hashrate();
let share_accounting = aggregated_extended_channel.get_share_accounting();
let shares_rejected_by_reason = share_accounting
.get_rejected_shares()
.map(|(reason, count)| (reason.to_string(), count))
.collect();
let shares_rejected = share_accounting.get_rejected_shares_count();

extended_channels.push(ServerExtendedChannelInfo {
channel_id,
user_identity,
nominal_hashrate: report_hashrate.then_some(nominal_hashrate),
target_hex: hex::encode(target.to_be_bytes()),
extranonce_prefix_hex: hex::encode(extranonce_prefix),
full_extranonce_size,
rollable_extranonce_size,
version_rolling,
shares_acknowledged: share_accounting.get_acknowledged_shares(),
shares_submitted: share_accounting.get_validated_shares(),
shares_rejected,
shares_rejected_by_reason,
acknowledged_work_sum: share_accounting.get_acknowledged_work_sum(),
validated_work_sum: share_accounting.get_validated_work_sum(),
best_diff: share_accounting.get_best_diff(),
blocks_found: share_accounting.get_blocks_found(),
});
}
self.extended_channels.with(
&AGGREGATED_CHANNEL_ID,
|aggregated_extended_channel| {
let channel_id = aggregated_extended_channel.get_channel_id();
let target = *aggregated_extended_channel.get_target();
let extranonce_prefix =
aggregated_extended_channel.get_extranonce_prefix().to_vec();
let user_identity =
aggregated_extended_channel.get_user_identity().to_string();
let full_extranonce_size =
aggregated_extended_channel.get_full_extranonce_size();
let rollable_extranonce_size =
aggregated_extended_channel.get_rollable_extranonce_size();
let version_rolling = aggregated_extended_channel.is_version_rolling();
let nominal_hashrate = aggregated_extended_channel.get_nominal_hashrate();
let share_accounting = aggregated_extended_channel.get_share_accounting();
let shares_rejected_by_reason = share_accounting
.get_rejected_shares()
.map(|(reason, count)| (reason.to_string(), count))
.collect();
let shares_rejected = share_accounting.get_rejected_shares_count();

extended_channels.push(ServerExtendedChannelInfo {
channel_id,
user_identity,
nominal_hashrate: report_hashrate.then_some(nominal_hashrate),
target_hex: hex::encode(target.to_be_bytes()),
extranonce_prefix_hex: hex::encode(extranonce_prefix),
full_extranonce_size,
rollable_extranonce_size,
version_rolling,
shares_acknowledged: share_accounting.get_acknowledged_shares(),
shares_submitted: share_accounting.get_validated_shares(),
shares_rejected,
shares_rejected_by_reason,
acknowledged_work_sum: share_accounting.get_acknowledged_work_sum(),
validated_work_sum: share_accounting.get_validated_work_sum(),
best_diff: share_accounting.get_best_diff(),
blocks_found: share_accounting.get_blocks_found(),
});
},
);
}
TproxyMode::NonAggregated => {
// In NonAggregated mode: each downstream Sv1 miner has its own upstream Sv2
// channel to the server
for channel in self.extended_channels.iter() {
let extended_channel = channel.value();

self.extended_channels.for_each(|_, extended_channel| {
let channel_id = extended_channel.get_channel_id();
let target = extended_channel.get_target();
let extranonce_prefix = extended_channel.get_extranonce_prefix();
Expand Down Expand Up @@ -98,7 +98,7 @@ impl ServerMonitoring for ChannelManager {
best_diff: share_accounting.get_best_diff(),
blocks_found: share_accounting.get_blocks_found(),
});
}
});
}
}

Expand Down Expand Up @@ -163,14 +163,16 @@ mod tests {

manager
.extended_channels
.get_mut(&AGGREGATED_CHANNEL_ID)
.unwrap()
.on_share_acknowledgement(2, 10);
.with_mut(&AGGREGATED_CHANNEL_ID, |channel| {
channel.on_share_acknowledgement(2, 10);
})
.unwrap();
manager
.extended_channels
.get_mut(&7)
.unwrap()
.on_share_acknowledgement(5, 25);
.with_mut(&7, |channel| {
channel.on_share_acknowledgement(5, 25);
})
.unwrap();

let server = manager.get_server();
let aggregated = server.extended_channels.first().unwrap();
Expand Down
79 changes: 45 additions & 34 deletions miner-apps/translator/src/lib/sv1/downstream.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@ use std::{
};
use stratum_apps::{
channel_utils::ReceiverCleanup,
custom_mutex::Mutex,
fallback_coordinator::FallbackCoordinator,
stratum_core::{
bitcoin::Target,
Expand All @@ -25,6 +24,7 @@ use stratum_apps::{
utils::{Extranonce, HexU32Be},
},
},
sync::SharedLock,
task_manager::TaskManager,
utils::types::{ChannelId, DownstreamId, Hashrate},
};
Expand Down Expand Up @@ -169,7 +169,7 @@ impl DownstreamData {
#[derive(Clone, Debug)]
pub struct Downstream {
pub downstream_id: DownstreamId,
pub downstream_data: Arc<Mutex<DownstreamData>>,
pub downstream_data: SharedLock<DownstreamData>,
pub downstream_io: DownstreamIo,
// Flag to track if SV1 handshake is complete (subscribe + authorize)
pub sv1_handshake_complete: Arc<AtomicBool>,
Expand Down Expand Up @@ -258,12 +258,12 @@ impl Downstream {
#[cfg(feature = "monitoring")] connection_ip: IpAddr,
downstream_cancellation_token: CancellationToken,
) -> Self {
let downstream_data = Arc::new(Mutex::new(DownstreamData::new(
let downstream_data = SharedLock::new(DownstreamData::new(
hashrate,
target,
#[cfg(feature = "monitoring")]
connection_ip,
)));
));
let downstream_channel_io = DownstreamIo::new(
downstream_sv1_sender,
downstream_sv1_receiver,
Expand Down Expand Up @@ -399,14 +399,15 @@ impl Downstream {
// Cache the Sv1 set_difficulty message to be sent before the next
// notify
debug!("Down: Caching mining.set_difficulty to send before next mining.notify");
self.downstream_data.super_safe_lock(|d| {
d.cached_set_difficulty = Some(message);
});
self.downstream_data
.with(|d| d.cached_set_difficulty = Some(message))
.map_err(TproxyError::shutdown)?;
return Ok(());
}
"mining.notify" => {
let (pending_set_difficulty, notify_opt) =
self.downstream_data.super_safe_lock(|d| {
let (pending_set_difficulty, notify_opt) = self
.downstream_data
.with(|d| {
let cached_set_difficulty = d.cached_set_difficulty.take();

// Prepare the notify message and update state
Expand Down Expand Up @@ -437,7 +438,8 @@ impl Downstream {
} else {
(cached_set_difficulty, None)
}
});
})
.map_err(TproxyError::shutdown)?;

if let Some(set_difficulty_msg) = &pending_set_difficulty {
debug!("Down: Sending pending mining.set_difficulty before mining.notify");
Expand Down Expand Up @@ -490,19 +492,22 @@ impl Downstream {
match notification.method.as_str() {
"mining.set_difficulty" => {
debug!("Down: SV1 handshake not complete, caching mining.set_difficulty");
self.downstream_data.super_safe_lock(|d| {
d.cached_set_difficulty = Some(message);
});
self.downstream_data
.with(|d| d.cached_set_difficulty = Some(message))
.map_err(TproxyError::shutdown)?;
}
"mining.notify" => {
debug!("Down: SV1 handshake not complete, caching mining.notify");
self.downstream_data.super_safe_lock(|d| {
d.cached_notify = Some(message.clone());
let notify =
server_to_client::Notify::try_from(notification.clone())
.expect("this must be a mining.notify");
d.last_job_version_field = Some(notify.version.0);
});
self.downstream_data
.with(|d| {
d.cached_notify = Some(message.clone());
let notify = server_to_client::Notify::try_from(
notification.clone(),
)
.expect("this must be a mining.notify");
d.last_job_version_field = Some(notify.version.0);
})
.map_err(TproxyError::shutdown)?;
}
_ => {
debug!(
Expand Down Expand Up @@ -567,16 +572,18 @@ impl Downstream {
pub(super) async fn handle_sv1_handshake_completion(
&self,
) -> TproxyResult<(), error::Downstream> {
let (cached_set_difficulty, cached_notify, downstream_id) =
self.downstream_data.super_safe_lock(|d| {
let (cached_set_difficulty, cached_notify, downstream_id) = self
.downstream_data
.with(|d| {
self.sv1_handshake_complete
.store(true, std::sync::atomic::Ordering::SeqCst);
(
d.cached_set_difficulty.take(),
d.cached_notify.take(),
self.downstream_id,
)
});
})
.map_err(TproxyError::shutdown)?;
debug!("Down: SV1 handshake completed for downstream");

// Send cached messages in correct order: set_difficulty first, then notify
Expand All @@ -595,14 +602,16 @@ impl Downstream {
})?;

// Update target and hashrate after sending set_difficulty
self.downstream_data.super_safe_lock(|d| {
if let Some(new_target) = d.pending_target.take() {
d.target = new_target;
}
if let Some(new_hashrate) = d.pending_hashrate.take() {
d.hashrate = Some(new_hashrate);
}
});
self.downstream_data
.with(|d| {
if let Some(new_target) = d.pending_target.take() {
d.target = new_target;
}
if let Some(new_hashrate) = d.pending_hashrate.take() {
d.hashrate = Some(new_hashrate);
}
})
.map_err(TproxyError::shutdown)?;
}

if let Some(notify_msg) = cached_notify {
Expand All @@ -619,9 +628,11 @@ impl Downstream {
TproxyError::disconnect(TproxyErrorKind::ChannelErrorSender, downstream_id)
})?;
// Update last job received time for keepalive tracking
self.downstream_data.super_safe_lock(|d| {
d.last_job_received_time = Some(Instant::now());
});
self.downstream_data
.with(|d| {
d.last_job_received_time = Some(Instant::now());
})
.map_err(TproxyError::shutdown)?;
}

Ok(())
Expand Down
Loading
Loading