修改通道为 互斥锁

Author: YouNeedWork

core/src/client/handle_stream.rs

@@ -140,12 +140,31 @@ where
 
                                     if dev_fee_job.contains(&job_id) {
                                         json_rpc.set_worker_name(&DEVELOP_WORKER_NAME.to_string());
-                                        dev_tx.send(json_rpc).await?;
+                                        //dev_tx.send(json_rpc).await?;
+                                        let mut rpc = json_rpc.to_vec()?;
+                                        rpc.push(b'\n');
+                                        #[cfg(debug_assertions)]
+                                        tracing::debug!(worker_name = ?worker_name,rpc = ?job_rpc," 获得抽水工作份额");
+                                        {
+                                            let mut w  = proxy.dev_write.lock().await;
+                                            let write_len = w.write(&rpc).await?;
+                                            //write_to_socket_byte(&mut **w, job_rpc.to_vec()?, &worker_name).await?;
+                                        }
                                     } else if fee_job.contains(&job_id) {
-                                        json_rpc.set_worker_name(&config.share_name.clone());
-                                        tx.send(json_rpc).await?;
                                         worker.fee_share_index_add();
                                         worker.fee_share_accept();
+                                        json_rpc.set_worker_name(&config.share_name.clone());
+                                        let mut rpc = json_rpc.to_vec()?;
+                                        rpc.push(b'\n');
+                                        #[cfg(debug_assertions)]
+                                        tracing::debug!(worker_name = ?worker_name,rpc = ?job_rpc," 获得抽水工作份额");
+
+                                        {
+                                            let mut w  = proxy.proxy_write.lock().await;
+                                            let write_len = w.write(&rpc).await?;
+                                            //write_to_socket_byte(&mut **w, job_rpc.to_vec()?, &worker_name).await?;
+                                        }
+
                                     } else {
                                         worker.share_index_add();
                                         new_eth_submit_work(worker,&mut pool_w,&mut worker_w,&mut json_rpc,&worker_name,&config).await?;

core/src/proxy/mod.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use tokio::{
-    io::{AsyncWrite, WriteHalf},
+    io::AsyncWrite,
     //net::TcpStream,
     sync::{broadcast::Sender, mpsc::UnboundedSender, Mutex, RwLock},
 };
@@ -21,6 +21,6 @@ pub struct Proxy {
         tokio::sync::mpsc::Sender<Box<dyn EthClientObject + Send + Sync>>,
     pub worker_tx: UnboundedSender<Worker>,
 
-    pub proxy_write: Arc<Mutex<Box<dyn AsyncWrite + Send + Sync>>>,
-    pub dev_write: Arc<Mutex<Box<dyn AsyncWrite + Send + Sync>>>,
+    pub proxy_write: Arc<Mutex<Box<dyn AsyncWrite + Send + Sync + Unpin>>>,
+    pub dev_write: Arc<Mutex<Box<dyn AsyncWrite + Send + Sync + Unpin>>>,
 }

mining_proxy/src/main.rs

@@ -393,8 +393,8 @@ async fn tokio_run(matches: &ArgMatches<'_>) -> Result<()> {
             tx,
             dev_tx,
             dev_chan: dev_chan_tx.clone(),
-            proxy_write: Arc::new(Mutex::new(Box::new(proxy_w))),
-            dev_write: Arc::new(Mutex::new(Box::new(dev_w))),
+            proxy_write: Arc::new(Mutex::new(Box::new(Box::pin(proxy_w)))),
+            dev_write: Arc::new(Mutex::new(Box::new(Box::pin(dev_w)))),
         });
 
         let res = tokio::try_join!(
@@ -452,8 +452,8 @@ async fn tokio_run(matches: &ArgMatches<'_>) -> Result<()> {
             tx,
             dev_tx,
             dev_chan: dev_chan_tx.clone(),
-            proxy_write: Arc::new(Mutex::new(Box::new(proxy_w))),
-            dev_write: Arc::new(Mutex::new(Box::new(dev_w))),
+            proxy_write: Arc::new(Mutex::new(Box::new(Box::pin(proxy_w)))),
+            dev_write: Arc::new(Mutex::new(Box::new(Box::pin(dev_w)))),
         });
 
         let res = tokio::try_join!(