fix sync_block_process

Author: jackzhhuang

sync/src/parallel/worker_scheduler.rs

@@ -1,5 +1,6 @@
 use std::{sync::atomic::AtomicU64, time::Duration};
 
+use starcoin_logger::prelude::debug;
 use tokio::sync::RwLock;
 
 #[derive(Debug)]
@@ -29,17 +30,20 @@ impl WorkerScheduler {
     }
 
     pub async fn tell_worker_to_stop(&self) {
+        debug!("tell_worker_to_stop");
         let mut state = self.state.write().await;
         *state = WorkerSchedulerState::Inactive;
     }
 
     pub async fn tell_worker_to_start(&self) {
+        debug!("tell_worker_to_start");
         let mut state = self.state.write().await;
         *state = WorkerSchedulerState::Active;
     }
 
     pub async fn check_if_stop(&self) -> bool {
         let state = self.state.read().await;
+        debug!("check the state of the workers thread: {:?}", *state);
         match *state {
             WorkerSchedulerState::Inactive => true,
             WorkerSchedulerState::Active => false,
@@ -67,6 +71,7 @@ impl WorkerScheduler {
             } else {
                 tokio::task::yield_now().await;
                 tokio::time::sleep(Duration::from_millis(200)).await;
+                debug!("wait for worker to exit");
             }
         }
     }

sync/src/tasks/tests_dag.rs

@@ -20,7 +20,10 @@ async fn sync_block_process(
     local_node: Arc<SyncNodeMocker>,
     registry: &ServiceRef<RegistryService>,
 ) -> Result<(Arc<SyncNodeMocker>, Arc<SyncNodeMocker>)> {
+    let worker_scheduler = Arc::new(WorkerScheduler::new());
     loop {
+        worker_scheduler.tell_worker_to_stop().await;
+        worker_scheduler.wait_for_worker().await;
         let target = target_node.sync_target();
 
         let storage = local_node.chain().get_storage();
@@ -36,7 +39,6 @@ async fn sync_block_process(
             registry.service_ref::<BlockConnectorService<MockTxPoolService>>(),
         )?;
 
-        let worker_scheduler = Arc::new(WorkerScheduler::new());
         let (sync_task, _task_handle, task_event_counter) = full_sync_task(
             current_block_id,
             target.clone(),
@@ -54,6 +56,7 @@ async fn sync_block_process(
             local_node.sync_dag_store.clone(),
             worker_scheduler.clone(),
         )?;
+        worker_scheduler.tell_worker_to_start().await;
         let branch = sync_task.await?;
         info!("checking branch in sync service is the same as target's branch");
         assert_eq!(branch.current_header().id(), target.target_id.id());