bounded channel for mem control (#406)

* bounded channel for mem control

* make constant

Author: 0g-peterzhb

node/miner/src/mine.rs

@@ -18,9 +18,11 @@ use crate::{
 
 use std::sync::Arc;
 
+const MINE_ANSWER_CHANNEL_CAPACITY: usize = 20;
+
 pub struct PoraService {
     mine_context_receiver: broadcast::Receiver<MineContextMessage>,
-    mine_answer_sender: mpsc::UnboundedSender<AnswerWithoutProof>,
+    mine_answer_sender: mpsc::Sender<AnswerWithoutProof>,
     msg_recv: broadcast::Receiver<MinerMessage>,
     loader: Arc<dyn PoraLoader>,
 
@@ -123,9 +125,9 @@ impl PoraService {
         loader: Arc<dyn PoraLoader>,
         config: &MinerConfig,
         miner_id: H256,
-    ) -> mpsc::UnboundedReceiver<AnswerWithoutProof> {
+    ) -> mpsc::Receiver<AnswerWithoutProof> {
         let (mine_answer_sender, mine_answer_receiver) =
-            mpsc::unbounded_channel::<AnswerWithoutProof>();
+            mpsc::channel::<AnswerWithoutProof>(MINE_ANSWER_CHANNEL_CAPACITY);
         let mine_range = MineRangeConfig {
             start_position: Some(0),
             end_position: Some(u64::MAX),
@@ -218,7 +220,7 @@ impl PoraService {
 
                     if let Some(answer) = miner.batch_iteration(nonce, self.iter_batch).await {
                         info!("Hit Pora answer {:?}", answer);
-                        if self.mine_answer_sender.send(answer).is_err() {
+                        if self.mine_answer_sender.send(answer).await.is_err() {
                             warn!("Mine submitter channel closed");
                         }
                     } else if cpu_percent < 100 {

node/miner/src/submitter.rs

@@ -18,7 +18,7 @@ use crate::watcher::MineContextMessage;
 use zgs_spec::{BYTES_PER_SEAL, SECTORS_PER_SEAL};
 
 pub struct Submitter {
-    mine_answer_receiver: mpsc::UnboundedReceiver<AnswerWithoutProof>,
+    mine_answer_receiver: mpsc::Receiver<AnswerWithoutProof>,
     mine_context_receiver: broadcast::Receiver<MineContextMessage>,
     mine_contract: PoraMine<MineServiceMiddleware>,
     flow_contract: ZgsFlow<Provider<RetryClient<Http>>>,
@@ -29,7 +29,7 @@ pub struct Submitter {
 impl Submitter {
     pub fn spawn(
         executor: TaskExecutor,
-        mine_answer_receiver: mpsc::UnboundedReceiver<AnswerWithoutProof>,
+        mine_answer_receiver: mpsc::Receiver<AnswerWithoutProof>,
         mine_context_receiver: broadcast::Receiver<MineContextMessage>,
         provider: Arc<Provider<RetryClient<Http>>>,
         signing_provider: Arc<MineServiceMiddleware>,