Merge pull request #228 from Shourya742/2026-02-02-migrate-tproxy-to-dashmap

Migrate tproxy to dashmap

Author: GitGab19

miner-apps/translator/src/lib/mod.rs

@@ -105,12 +105,12 @@ impl TranslatorSv2 {
             self.config.downstream_port,
         );
 
-        let sv1_server = Sv1Server::new(
+        let sv1_server = Arc::new(Sv1Server::new(
             downstream_addr,
             channel_manager_to_sv1_server_receiver,
             sv1_server_to_channel_manager_sender,
             self.config.clone(),
-        );
+        ));
 
         info!("Initializing upstream connection...");
 
@@ -143,14 +143,15 @@ impl TranslatorSv2 {
         ));
 
         info!("Launching ChannelManager tasks...");
-        ChannelManager::run_channel_manager_tasks(
-            channel_manager.clone(),
-            notify_shutdown.clone(),
-            shutdown_complete_tx.clone(),
-            status_sender.clone(),
-            task_manager.clone(),
-        )
-        .await;
+        channel_manager
+            .clone()
+            .run_channel_manager_tasks(
+                notify_shutdown.clone(),
+                shutdown_complete_tx.clone(),
+                status_sender.clone(),
+                task_manager.clone(),
+            )
+            .await;
 
         // Start monitoring server if configured
         if let Some(monitoring_addr) = self.config.monitoring_address() {
@@ -166,7 +167,7 @@ impl TranslatorSv2 {
                                                 * handled separately) */
             )
             .expect("Failed to initialize monitoring server")
-            .with_sv1_monitoring(Arc::new(sv1_server.clone())) // SV1 client connections
+            .with_sv1_monitoring(sv1_server.clone()) // SV1 client connections
             .expect("Failed to add SV1 monitoring");
 
             // Create shutdown signal that waits for ShutdownAll
@@ -281,7 +282,7 @@ impl TranslatorSv2 {
         status_sender: Sender<Status>,
         shutdown_complete_tx: mpsc::Sender<()>,
         task_manager: Arc<TaskManager>,
-        sv1_server_instance: Sv1Server,
+        sv1_server_instance: Arc<Sv1Server>,
         required_extensions: Vec<u16>,
     ) -> Result<(), TproxyErrorKind> {
         const MAX_RETRIES: usize = 3;

miner-apps/translator/src/lib/monitoring.rs

@@ -13,71 +13,67 @@ impl ServerMonitoring for ChannelManager {
         let mut extended_channels = Vec::new();
         let standard_channels = Vec::new(); // tProxy only uses extended channels
 
-        self.channel_manager_data
-            .safe_lock(|d| {
-                match tproxy_mode() {
-                    TproxyMode::Aggregated => {
+        match tproxy_mode() {
+            TproxyMode::Aggregated => {
+                self.upstream_extended_channel
+                    .super_safe_lock(|upstream_channel| {
                         // In Aggregated mode: one shared upstream channel to the server
-                        if let Some(upstream_channel) = &d.upstream_extended_channel {
-                            if let Ok(channel_lock) = upstream_channel.read() {
-                                let channel_id = channel_lock.get_channel_id();
-                                let target = channel_lock.get_target();
-                                let extranonce_prefix = channel_lock.get_extranonce_prefix();
-                                let user_identity = channel_lock.get_user_identity();
-                                let share_accounting = channel_lock.get_share_accounting();
+                        if let Some(upstream_channel) = upstream_channel.as_mut() {
+                            let channel_id = upstream_channel.get_channel_id();
+                            let target = upstream_channel.get_target();
+                            let extranonce_prefix = upstream_channel.get_extranonce_prefix();
+                            let user_identity = upstream_channel.get_user_identity();
+                            let share_accounting = upstream_channel.get_share_accounting();
 
-                                extended_channels.push(ServerExtendedChannelInfo {
-                                    channel_id,
-                                    user_identity: user_identity.clone(),
-                                    nominal_hashrate: channel_lock.get_nominal_hashrate(),
-                                    target_hex: hex::encode(target.to_be_bytes()),
-                                    extranonce_prefix_hex: hex::encode(extranonce_prefix),
-                                    full_extranonce_size: channel_lock.get_full_extranonce_size(),
-                                    rollable_extranonce_size: channel_lock
-                                        .get_rollable_extranonce_size(),
-                                    version_rolling: channel_lock.is_version_rolling(),
-                                    shares_accepted: share_accounting.get_shares_accepted(),
-                                    share_work_sum: share_accounting.get_share_work_sum(),
-                                    last_share_sequence_number: share_accounting
-                                        .get_last_share_sequence_number(),
-                                    best_diff: share_accounting.get_best_diff(),
-                                });
-                            }
+                            extended_channels.push(ServerExtendedChannelInfo {
+                                channel_id,
+                                user_identity: user_identity.clone(),
+                                nominal_hashrate: upstream_channel.get_nominal_hashrate(),
+                                target_hex: hex::encode(target.to_be_bytes()),
+                                extranonce_prefix_hex: hex::encode(extranonce_prefix),
+                                full_extranonce_size: upstream_channel.get_full_extranonce_size(),
+                                rollable_extranonce_size: upstream_channel
+                                    .get_rollable_extranonce_size(),
+                                version_rolling: upstream_channel.is_version_rolling(),
+                                shares_accepted: share_accounting.get_shares_accepted(),
+                                share_work_sum: share_accounting.get_share_work_sum(),
+                                last_share_sequence_number: share_accounting
+                                    .get_last_share_sequence_number(),
+                                best_diff: share_accounting.get_best_diff(),
+                            });
                         }
-                    }
-                    TproxyMode::NonAggregated => {
-                        // In NonAggregated mode: each downstream Sv1 miner has its own upstream Sv2
-                        // channel to the server
-                        for (_channel_id, extended_channel) in d.extended_channels.iter() {
-                            if let Ok(channel_lock) = extended_channel.read() {
-                                let channel_id = channel_lock.get_channel_id();
-                                let target = channel_lock.get_target();
-                                let extranonce_prefix = channel_lock.get_extranonce_prefix();
-                                let user_identity = channel_lock.get_user_identity();
-                                let share_accounting = channel_lock.get_share_accounting();
+                    });
+            }
+            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();
 
-                                extended_channels.push(ServerExtendedChannelInfo {
-                                    channel_id,
-                                    user_identity: user_identity.clone(),
-                                    nominal_hashrate: channel_lock.get_nominal_hashrate(),
-                                    target_hex: hex::encode(target.to_be_bytes()),
-                                    extranonce_prefix_hex: hex::encode(extranonce_prefix),
-                                    full_extranonce_size: channel_lock.get_full_extranonce_size(),
-                                    rollable_extranonce_size: channel_lock
-                                        .get_rollable_extranonce_size(),
-                                    version_rolling: channel_lock.is_version_rolling(),
-                                    shares_accepted: share_accounting.get_shares_accepted(),
-                                    share_work_sum: share_accounting.get_share_work_sum(),
-                                    last_share_sequence_number: share_accounting
-                                        .get_last_share_sequence_number(),
-                                    best_diff: share_accounting.get_best_diff(),
-                                });
-                            }
-                        }
-                    }
+                    let channel_id = extended_channel.get_channel_id();
+                    let target = extended_channel.get_target();
+                    let extranonce_prefix = extended_channel.get_extranonce_prefix();
+                    let user_identity = extended_channel.get_user_identity();
+                    let share_accounting = extended_channel.get_share_accounting();
+
+                    extended_channels.push(ServerExtendedChannelInfo {
+                        channel_id,
+                        user_identity: user_identity.clone(),
+                        nominal_hashrate: extended_channel.get_nominal_hashrate(),
+                        target_hex: hex::encode(target.to_be_bytes()),
+                        extranonce_prefix_hex: hex::encode(extranonce_prefix),
+                        full_extranonce_size: extended_channel.get_full_extranonce_size(),
+                        rollable_extranonce_size: extended_channel.get_rollable_extranonce_size(),
+                        version_rolling: extended_channel.is_version_rolling(),
+                        shares_accepted: share_accounting.get_shares_accepted(),
+                        share_work_sum: share_accounting.get_share_work_sum(),
+                        last_share_sequence_number: share_accounting
+                            .get_last_share_sequence_number(),
+                        best_diff: share_accounting.get_best_diff(),
+                    });
                 }
-            })
-            .unwrap();
+            }
+        }
 
         ServerInfo {
             extended_channels,

miner-apps/translator/src/lib/sv1/downstream/data.rs

@@ -1,4 +1,4 @@
-use std::{cell::RefCell, sync::atomic::AtomicBool, time::Instant};
+use std::time::Instant;
 use stratum_apps::{
     stratum_core::{
         bitcoin::Target,
@@ -16,78 +16,74 @@ use super::SubmitShareWithChannelId;
 #[derive(Debug)]
 pub struct DownstreamData {
     pub channel_id: Option<ChannelId>,
-    pub downstream_id: DownstreamId,
     pub extranonce1: Extranonce<'static>,
     pub extranonce2_len: usize,
+    pub target: Target,
+    pub hashrate: Option<Hashrate>,
     pub version_rolling_mask: Option<HexU32Be>,
     pub version_rolling_min_bit: Option<HexU32Be>,
     pub last_job_version_field: Option<u32>,
     pub authorized_worker_name: String,
     pub user_identity: String,
-    pub target: Target,
-    pub hashrate: Option<Hashrate>,
     pub cached_set_difficulty: Option<json_rpc::Message>,
     pub cached_notify: Option<json_rpc::Message>,
     pub pending_target: Option<Target>,
     pub pending_hashrate: Option<Hashrate>,
-    // Flag to track if SV1 handshake is complete (subscribe + authorize)
-    pub sv1_handshake_complete: AtomicBool,
     // Queue of Sv1 handshake messages received while waiting for SV2 channel to open
     pub queued_sv1_handshake_messages: Vec<json_rpc::Message>,
-    // Flag to indicate we're processing queued Sv1 handshake message responses
-    pub processing_queued_sv1_handshake_responses: AtomicBool,
     // Stores pending shares to be sent to the sv1_server
-    pub pending_share: RefCell<Option<SubmitShareWithChannelId>>,
+    pub pending_share: Option<SubmitShareWithChannelId>,
     // Tracks the upstream target for this downstream, used for vardiff target comparison
     pub upstream_target: Option<Target>,
     // Timestamp of when the last job was received by this downstream, used for keepalive check
     pub last_job_received_time: Option<Instant>,
 }
 
 impl DownstreamData {
-    pub fn new(downstream_id: DownstreamId, target: Target, hashrate: Option<Hashrate>) -> Self {
+    pub fn new(hashrate: Option<Hashrate>, target: Target) -> Self {
         DownstreamData {
             channel_id: None,
-            downstream_id,
             extranonce1: vec![0; 8]
                 .try_into()
                 .expect("8-byte extranonce is always valid"),
             extranonce2_len: 4,
+            target,
+            hashrate,
             version_rolling_mask: None,
             version_rolling_min_bit: None,
             last_job_version_field: None,
             authorized_worker_name: String::new(),
             user_identity: String::new(),
-            target,
-            hashrate,
             cached_set_difficulty: None,
             cached_notify: None,
             pending_target: None,
             pending_hashrate: None,
-            sv1_handshake_complete: AtomicBool::new(false),
             queued_sv1_handshake_messages: Vec::new(),
-            processing_queued_sv1_handshake_responses: AtomicBool::new(false),
-            pending_share: RefCell::new(None),
+            pending_share: None,
             upstream_target: None,
             last_job_received_time: None,
         }
     }
 
-    pub fn set_pending_target(&mut self, new_target: Target) {
+    pub fn set_pending_target(&mut self, new_target: Target, downstream_id: DownstreamId) {
         self.pending_target = Some(new_target);
-        debug!("Downstream {}: Set pending target", self.downstream_id);
+        debug!("Downstream {downstream_id}: Set pending target");
     }
 
-    pub fn set_pending_hashrate(&mut self, new_hashrate: Option<Hashrate>) {
+    pub fn set_pending_hashrate(
+        &mut self,
+        new_hashrate: Option<Hashrate>,
+        downstream_id: DownstreamId,
+    ) {
         self.pending_hashrate = new_hashrate;
-        debug!("Downstream {}: Set pending hashrate", self.downstream_id);
+        debug!("Downstream {downstream_id}: Set pending hashrate");
     }
 
-    pub fn set_upstream_target(&mut self, upstream_target: Target) {
+    pub fn set_upstream_target(&mut self, upstream_target: Target, downstream_id: DownstreamId) {
         self.upstream_target = Some(upstream_target);
         debug!(
-            "Downstream {}: Set upstream target to {:?}",
-            self.downstream_id, upstream_target
+            "Downstream {downstream_id}: Set upstream target to {:?}",
+            upstream_target
         );
     }
 }