async: Use tokio::time::sleep in stead of std::thread::sleep

The `std::thread::sleep()` is blocking the thread which prevent the
handling of `Control+C` signal in CLI.

Changing to async sleep `tokio::time::sleep()` will fix it.

We still have blocking sleep `nm_dbus` code, cannot change it yet as we
are using the blocking method of DBUS, need to migrate from zbus 1.x to
2.x+ for async support there.

It is hard to have auto test case evaluation `Control+C` got instant
feedback or not. Hence no test case, manually tested.

Signed-off-by: Gris Ge <[email protected]>

rust/src/cli/apply.rs

@@ -99,6 +99,7 @@ where
     );
     let rt = tokio::runtime::Builder::new_current_thread()
         .enable_io()
+        .enable_time()
         .build()
         .map_err(|e| {
             CliError::from(format!("tokio::runtime::Builder failed with {e}"))

rust/src/lib/nm/query_apply/apply.rs

@@ -35,7 +35,7 @@ use crate::{
 // There is plan to simply the `add_net_state`, `chg_net_state`, `del_net_state`
 // `cur_net_state`, `des_net_state` into single struct. Suppress the clippy
 // warning for now
-pub(crate) fn nm_apply(
+pub(crate) async fn nm_apply(
     merged_state: &MergedNetworkState,
     checkpoint: &str,
     timeout: u32,
@@ -178,7 +178,7 @@ pub(crate) fn nm_apply(
         )?;
     }
 
-    activate_nm_profiles(&mut nm_api, nm_conns_to_activate.as_slice())?;
+    activate_nm_profiles(&mut nm_api, nm_conns_to_activate.as_slice()).await?;
 
     deactivate_nm_profiles(&mut nm_api, nm_conns_to_deactivate.as_slice())?;
 

rust/src/lib/nm/query_apply/profile.rs

@@ -115,8 +115,8 @@ pub(crate) fn save_nm_profiles(
     Ok(())
 }
 
-pub(crate) fn activate_nm_profiles(
-    nm_api: &mut NmApi,
+pub(crate) async fn activate_nm_profiles(
+    nm_api: &mut NmApi<'_>,
     nm_conns: &[NmConnection],
 ) -> Result<(), NmstateError> {
     let mut nm_conns = nm_conns.to_vec();
@@ -150,7 +150,7 @@ pub(crate) fn activate_nm_profiles(
                 nm_api
                     .extend_timeout_if_required()
                     .map_err(nm_error_to_nmstate)?;
-                std::thread::sleep(std::time::Duration::from_secs(1));
+                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
             }
         } else {
             break;

rust/src/lib/query_apply/net_state.rs

@@ -100,6 +100,7 @@ impl NetworkState {
     pub fn apply(&self) -> Result<(), NmstateError> {
         let rt = tokio::runtime::Builder::new_current_thread()
             .enable_io()
+            .enable_time()
             .build()
             .map_err(|e| {
                 NmstateError::new(
@@ -147,9 +148,10 @@ impl NetworkState {
         if let Err(e) = cur_net_state.retrieve_async().await {
             if e.kind().can_retry() {
                 log::info!("Retrying on: {}", e);
-                std::thread::sleep(std::time::Duration::from_millis(
+                tokio::time::sleep(std::time::Duration::from_millis(
                     RETRY_NM_INTERVAL_MILLISECONDS,
-                ));
+                ))
+                .await;
                 cur_net_state.retrieve_async().await?;
             } else {
                 return Err(e);
@@ -189,9 +191,10 @@ impl NetworkState {
             Err(e) => {
                 if e.kind().can_retry() {
                     log::info!("Retrying on: {}", e);
-                    std::thread::sleep(std::time::Duration::from_millis(
+                    tokio::time::sleep(std::time::Duration::from_millis(
                         RETRY_NM_INTERVAL_MILLISECONDS,
-                    ));
+                    ))
+                    .await;
                     nm_checkpoint_create(timeout)?
                 } else {
                     return Err(e);
@@ -268,7 +271,7 @@ impl NetworkState {
         // we try to apply the state again if so.
         with_retry(RETRY_NM_INTERVAL_MILLISECONDS, RETRY_NM_COUNT, || async {
             nm_checkpoint_timeout_extend(checkpoint, timeout)?;
-            nm_apply(merged_state, checkpoint, timeout)?;
+            nm_apply(merged_state, checkpoint, timeout).await?;
             if merged_state.ovsdb.is_changed && ovsdb_is_running() {
                 ovsdb_apply(merged_state)?;
             }
@@ -440,9 +443,10 @@ where
                 }
             } else {
                 log::info!("Retrying on: {}", e);
-                std::thread::sleep(std::time::Duration::from_millis(
+                tokio::time::sleep(std::time::Duration::from_millis(
                     interval_ms,
-                ));
+                ))
+                .await;
                 cur_count += 1;
                 continue;
             }