使用Tauri批量下载时遭遇Future线程安全发送错误

错误信息 #

future cannot be sent between threads safely
within impl futures::Future<Output = Result<Vec<std::string::String>, std::string::String>>, the trait std::marker::Send is not implemented for Rc<Node>,
which is required by impl futures::Future<Output = Result<Vec<std::string::String>, std::string::String>>: std::marker::Send

问题代码 #

错误触发在#[tauri::command]装饰的函数处，根源是调用get_total_pages(&document).await：

#[tauri::command] // ERROR here
async fn handle_download_file(source: &str, url: &str, batch_size: usize,) -> Result<Vec<String>, String> {
    source::download_file(source, url, batch_size).await
}

pub async fn download_file(url: &str, batch_size: usize) -> Result<Vec<String>, String> {
    let document = kuchikiki::parse_html().one("<html></html>");

    // ERROR because of this line
    let total_pages = get_total_pages(&document).await;

    Ok(vec![])
}

async fn get_total_pages(document: &NodeRef) -> usize {
    let mut total_pages = 1;

    let last_page_node = match document.select("#list-chapter ul.pagination > li.last a") {
        Ok(mut nodes) => nodes.next(),
        Err(_) => None,
    };

    match last_page_node {
        Some(node) => {
            let last_page_url = node.attributes.borrow().get("href").unwrap().to_string();
            let total_page = last_page_url
                .split("=")
                .last()
                .expect("Couldn't get last split at '='")
                .parse::<usize>()
                .unwrap_or(1);
            total_pages = total_page;
        }
        None => {
            total_pages = 1;
        }
    };

    return total_pages;
}

原因 #

Tauri命令要求异步任务必须实现Send trait（因为Tauri会在不同线程间调度命令），但kuchikiki的NodeRef基于Rc，Rc不是线程安全的，没有实现Send，导致包含它的future无法跨线程发送。

解决方法 #

方案1：使用线程安全的ArcNodeRef #

kuchikiki提供了基于Arc的线程安全节点引用，替换原有的NodeRef即可：

pub async fn download_file(url: &str, batch_size: usize) -> Result<Vec<String>, String> {
    // 使用arc_one创建线程安全的文档引用
    let document = kuchikiki::parse_html().arc_one("<html></html>");

    let total_pages = get_total_pages(&document).await;

    Ok(vec![])
}

async fn get_total_pages(document: &kuchikiki::ArcNodeRef) -> usize {
    let mut total_pages = 1;

    let last_page_node = match document.select("#list-chapter ul.pagination > li.last a") {
        Ok(mut nodes) => nodes.next(),
        Err(_) => None,
    };

    match last_page_node {
        Some(node) => {
            let last_page_url = node.attributes.borrow().get("href").unwrap().to_string();
            let total_page = last_page_url
                .split("=")
                .last()
                .expect("Couldn't get last split at '='")
                .parse::<usize>()
                .unwrap_or(1);
            total_pages = total_page;
        }
        None => {
            total_pages = 1;
        }
    };

    total_pages
}

方案2：将同步操作移出异步上下文 #

get_total_pages内部没有异步逻辑，直接改成同步函数，去掉.await调用：

pub async fn download_file(url: &str, batch_size: usize) -> Result<Vec<String>, String> {
    let document = kuchikiki::parse_html().one("<html></html>");

    // 直接调用同步函数，无需await
    let total_pages = get_total_pages(&document);

    Ok(vec![])
}

// 移除async关键字，改为同步函数
fn get_total_pages(document: &NodeRef) -> usize {
    let mut total_pages = 1;

    let last_page_node = match document.select("#list-chapter ul.pagination > li.last a") {
        Ok(mut nodes) => nodes.next(),
        Err(_) => None,
    };

    match last_page_node {
        Some(node) => {
            let last_page_url = node.attributes.borrow().get("href").unwrap().to_string();
            let total_page = last_page_url
                .split("=")
                .last()
                .expect("Couldn't get last split at '='")
                .parse::<usize>()
                .unwrap_or(1);
            total_pages = total_page;
        }
        None => {
            total_pages = 1;
        }
    };

    total_pages
}

方案3：用spawn_blocking包裹同步逻辑 #

如果必须保留异步调用形式，把涉及Rc的代码放到tokio::task::spawn_blocking中，让它在阻塞线程池执行：

pub async fn download_file(url: &str, batch_size: usize) -> Result<Vec<String>, String> {
    let document = kuchikiki::parse_html().one("<html></html>");

    // 用spawn_blocking包装同步操作，返回Send的future
    let total_pages = tokio::task::spawn_blocking(move || {
        get_total_pages(&document)
    }).await.expect("spawn_blocking执行失败");

    Ok(vec![])
}

// 同步函数
fn get_total_pages(document: &NodeRef) -> usize {
    let mut total_pages = 1;

    let last_page_node = match document.select("#list-chapter ul.pagination > li.last a") {
        Ok(mut nodes) => nodes.next(),
        Err(_) => None,
    };

    match last_page_node {
        Some(node) => {
            let last_page_url = node.attributes.borrow().get("href").unwrap().to_string();
            let total_page = last_page_url
                .split("=")
                .last()
                .expect("Couldn't get last split at '='")
                .parse::<usize>()
                .unwrap_or(1);
            total_pages = total_page;
        }
        None => {
            total_pages = 1;
        }
    };

    total_pages
}

内容的提问来源于stack exchange，提问作者dr_nyt