From dabd53b9292beefa9b5b6b92ffb1c42c3b843605 Mon Sep 17 00:00:00 2001 From: edouardparis Date: Wed, 19 Mar 2025 14:33:24 +0100 Subject: [PATCH] Use tokio mpsc channel sender and receiver in Tokio, when you call .recv().await on a Receiver, it will asynchronously wait for a message to be sent to the channel. This means that the task will yield control back to the Tokio runtime until a message is available, allowing other tasks to run concurrently. If the Sender side of the channel is dropped or closed, recv().await will return None, indicating that no more messages will be sent. --- liana-gui/src/app/state/export.rs | 52 +++++++++++++++++++------ liana-gui/src/export.rs | 65 +++++++++++++++++-------------- 2 files changed, 75 insertions(+), 42 deletions(-) diff --git a/liana-gui/src/app/state/export.rs b/liana-gui/src/app/state/export.rs index 2dbee270..422a835d 100644 --- a/liana-gui/src/app/state/export.rs +++ b/liana-gui/src/app/state/export.rs @@ -158,13 +158,27 @@ impl ExportModal { &mut self.import_export_type { if let Some(sender) = labels.take() { - if sender.send(true).is_err() { - tracing::error!("ExportModal.update(): fail to send labels ACK"); - } + return Task::perform( + async move { + if sender.send(true).await.is_err() { + tracing::error!( + "ExportModal.update(): fail to send labels NACK" + ); + } + }, + |_| ImportExportMessage::Ignore.into(), + ); } else if let Some(sender) = aliases.take() { - if sender.send(true).is_err() { - tracing::error!("ExportModal.update(): fail to send aliases ACK"); - } + return Task::perform( + async move { + if sender.send(true).await.is_err() { + tracing::error!( + "ExportModal.update(): fail to send aliases NACK" + ); + } + }, + |_| ImportExportMessage::Ignore.into(), + ); } } } @@ -173,13 +187,27 @@ impl ExportModal { &mut self.import_export_type { if let Some(sender) = labels.take() { - if sender.send(false).is_err() { - tracing::error!("ExportModal.update(): fail to send labels NACK"); - } + return Task::perform( + async move { + if sender.send(false).await.is_err() { + tracing::error!( + "ExportModal.update(): fail to send labels NACK" + ); + } + }, + |_| ImportExportMessage::Ignore.into(), + ); } else if let Some(sender) = aliases.take() { - if sender.send(false).is_err() { - tracing::error!("ExportModal.update(): fail to send aliases NACK"); - } + return Task::perform( + async move { + if sender.send(false).await.is_err() { + tracing::error!( + "ExportModal.update(): fail to send aliases NACK" + ); + } + }, + |_| ImportExportMessage::Ignore.into(), + ); } } } diff --git a/liana-gui/src/export.rs b/liana-gui/src/export.rs index 7b0c0584..b138f3e9 100644 --- a/liana-gui/src/export.rs +++ b/liana-gui/src/export.rs @@ -5,13 +5,12 @@ use std::{ io::{Read, Write}, path::{Path, PathBuf}, str::FromStr, - sync::{ - mpsc::{channel, sync_channel, Receiver, Sender, SyncSender}, - Arc, Mutex, - }, + sync::{Arc, Mutex}, time, }; +use tokio::sync::mpsc::{channel, unbounded_channel, Sender, UnboundedReceiver, UnboundedSender}; + use async_hwi::bitbox::api::btc::Fingerprint; use chrono::{DateTime, Duration, Utc}; use liana::{ @@ -147,8 +146,8 @@ pub enum ImportExportType { ExportPsbt(String), ExportBackup(String), ImportBackup( - Option>, /*overwrite_labels*/ - Option>, /*overwrite_aliases*/ + Option>, /*overwrite_labels*/ + Option>, /*overwrite_aliases*/ ), WalletFromBackup, Descriptor(LianaDescriptor), @@ -228,8 +227,8 @@ pub enum Progress { None, Psbt(Psbt), Descriptor(LianaDescriptor), - LabelsConflict(SyncSender), - KeyAliasesConflict(SyncSender), + LabelsConflict(Sender), + KeyAliasesConflict(Sender), UpdateAliases(HashMap), WalletFromBackup( ( @@ -242,8 +241,8 @@ pub enum Progress { } pub struct Export { - pub receiver: Receiver, - pub sender: Option>, + pub receiver: UnboundedReceiver, + pub sender: Option>, pub handle: Option>>>, pub daemon: Option>, pub path: Box, @@ -256,7 +255,7 @@ impl Export { path: Box, export_type: ImportExportType, ) -> Self { - let (sender, receiver) = channel(); + let (sender, receiver) = unbounded_channel(); Export { receiver, sender: Some(sender), @@ -269,7 +268,7 @@ impl Export { pub async fn export_logic( export_type: ImportExportType, - sender: Sender, + sender: UnboundedSender, daemon: Option>, path: PathBuf, ) { @@ -349,8 +348,8 @@ pub fn export_subscription( continue; } Err(e) => match e { - std::sync::mpsc::TryRecvError::Empty => false, - std::sync::mpsc::TryRecvError::Disconnected => true, + tokio::sync::mpsc::error::TryRecvError::Empty => false, + tokio::sync::mpsc::error::TryRecvError::Disconnected => true, }, }; @@ -387,7 +386,7 @@ pub fn export_subscription( } pub async fn export_transactions( - sender: &Sender, + sender: &UnboundedSender, daemon: Option>, path: PathBuf, ) -> Result<(), Error> { @@ -526,7 +525,7 @@ pub async fn export_transactions( } pub async fn export_descriptor( - sender: &Sender, + sender: &UnboundedSender, path: PathBuf, descriptor: LianaDescriptor, ) -> Result<(), Error> { @@ -541,7 +540,7 @@ pub async fn export_descriptor( } pub async fn export_string( - sender: &Sender, + sender: &UnboundedSender, path: PathBuf, psbt: String, ) -> Result<(), Error> { @@ -552,7 +551,7 @@ pub async fn export_string( Ok(()) } -pub async fn import_psbt(sender: &Sender, path: PathBuf) -> Result<(), Error> { +pub async fn import_psbt(sender: &UnboundedSender, path: PathBuf) -> Result<(), Error> { let mut file = File::open(&path)?; let mut psbt_str = String::new(); @@ -565,7 +564,10 @@ pub async fn import_psbt(sender: &Sender, path: PathBuf) -> Result<(), Ok(()) } -pub async fn import_descriptor(sender: &Sender, path: PathBuf) -> Result<(), Error> { +pub async fn import_descriptor( + sender: &UnboundedSender, + path: PathBuf, +) -> Result<(), Error> { let mut file = File::open(path)?; let mut descr_str = String::new(); @@ -589,7 +591,7 @@ pub async fn import_descriptor(sender: &Sender, path: PathBuf) -> Resu /// - import labels if no conflict or user ACK /// - update aliases if no conflict or user ACK pub async fn import_backup( - sender: &Sender, + sender: &UnboundedSender, path: PathBuf, daemon: Option>, ) -> Result<(), Error> { @@ -679,7 +681,7 @@ pub async fn import_backup( let backup_labels_map = labels.clone().into_map(); // if there is a conflict, we ask user to ACK before overwrite - let (ack_sender, ack_receiver) = sync_channel(0); + let (ack_sender, mut ack_receiver) = channel(0); let mut conflict = false; for (k, l) in &backup_labels_map { if let Some(lab) = labels_map.get(k) { @@ -691,9 +693,9 @@ pub async fn import_backup( } } if conflict { - write_labels = match ack_receiver.recv() { - Ok(b) => b, - Err(_) => { + write_labels = match ack_receiver.recv().await { + Some(b) => b, + None => { return Err(Error::BackupImport("Fail to receive labels ACK".into())); } } @@ -743,7 +745,7 @@ pub async fn import_backup( } }; - let (ack_sender, ack_receiver) = sync_channel(0); + let (ack_sender, mut ack_receiver) = channel(0); let mut conflict = false; for (fg, key) in &account.keys { if let Some(k) = settings_aliases.get(fg) { @@ -757,9 +759,9 @@ pub async fn import_backup( } if conflict { // wait for the user ACK/NACK - write_aliases = match ack_receiver.recv() { - Ok(a) => a, - Err(_) => { + write_aliases = match ack_receiver.recv().await { + Some(a) => a, + None => { return Err(Error::BackupImport("Fail to receive aliases ACK".into())); } }; @@ -902,7 +904,10 @@ impl From for RestoreBackupError { /// - extract descriptor /// - extract network /// - extract aliases -pub async fn wallet_from_backup(sender: &Sender, path: PathBuf) -> Result<(), Error> { +pub async fn wallet_from_backup( + sender: &UnboundedSender, + path: PathBuf, +) -> Result<(), Error> { // Load backup from file let mut file = File::open(path)?; @@ -1076,7 +1081,7 @@ pub async fn import_backup_at_launch( } pub async fn export_labels( - sender: &Sender, + sender: &UnboundedSender, daemon: Option>, path: PathBuf, ) -> Result<(), Error> {