Блог

Я хочу загружать файл по частям одновременно, но с ограничением, например, максимум на 4 задачи, для этого я использую FuturesUnordered как в следующем фрагменте кода:

 let mut tasks = FuturesUnordered::new();

let bin_parts = db_parts.iter().values();
for bin_part in bin_parts {
    if let Ok(p) = bin_part {
        let part: Part = from_reader(amp;p[..])?;
        tasks.push(async { upload_part(s3, key, file, amp;upload_id, sdb, part).await });

        // limit to N threads (4 for example)
        if tasks.len() == threads {
            while let Some(r) = tasks.next().await {
                if r.is_ok() {
                    pb.inc(1)
                }
            }
        }
    }
}

// consume remaining tasks
loop {
    if let Some(r) = tasks.next().await {
        if r.is_ok() {
            pb.inc(1)
        }
    } else {
        pb.finish();
        break;
    }
}

if !db_parts.is_empty() {
    return Err(anyhow!("could not upload all parts"));
}

// Complete Multipart Upload
let uploaded = sdb.uploaded_parts()?;
let action = actions::CompleteMultipartUpload::new(key, amp;upload_id, uploaded);
let rs = action.request(s3).await?;
  

Код работает, но я не очень хорошо понимаю вывод из clippy , я запускаю его следующим образом:

 cargo clippy --all-targets --all-features -- -D clippy::nursery -D warnings
  

И это ошибка:

 error: future cannot be sent between threads safely
   --> src/s3m/multipart_upload.rs:36:6
    |
36  | ) -> Result<String> {
    |      ^^^^^^^^^^^^^^ future returned by `multipart_upload` is not `Send`
    |
    = note: `-D clippy::future-not-send` implied by `-D clippy::nursery`
note: future is not `Send` as this value is used across an await
   --> src/s3m/multipart_upload.rs:116:14
    |
80  |     let bin_parts = db_parts.iter().values();
    |         --------- has type `impl std::iter::DoubleEndedIterator` which is not `Send`
...
116 |     let rs = action.request(s3).await?;
    |              ^^^^^^^^^^^^^^^^^^^^^^^^ await occurs here, with `bin_parts` maybe used later
...
125 | }
    | - `bin_parts` is later dropped here
    = note: `*const crossbeam_epoch::internal::Local` doesn't implement `std::marker::Send`
    = help: for further information visit https://rust-lang.github.io/rust-clippy/master/index.html#future_not_send
  

Поиск по теме, кажется, я обнаружил, что это потому, что bin_part переменная не удалена и что в некоторых случаях это можно сделать, используя внутреннюю область с {} , но я все еще не могу найти, как это улучшить / исправить.

Я попытался использовать область:

 {
let bin_parts = db_parts.iter().values();
    for bin_part in bin_parts {
        if let Ok(p) = bin_part {
            let part: Part = from_reader(amp;p[..])?;
            tasks.push(async { upload_part(s3, key, file, amp;upload_id, sdb, part).await });

            // limit to N threads (4 for example)
            if tasks.len() == threads {
                while let Some(r) = tasks.next().await {
                    if r.is_ok() {
                        pb.inc(1)
                    }
                }
            }
        }
    }
}
  

Но получаю ту же ошибку:

 error: future cannot be sent between threads safely
  --> src/s3m/multipart_upload.rs:36:6
   |
36 | ) -> Result<String> {
   |      ^^^^^^^^^^^^^^ future returned by `multipart_upload` is not `Send`
   |
   = note: `-D clippy::future-not-send` implied by `-D clippy::nursery`
note: future is not `Send` as this value is used across an await
  --> src/s3m/multipart_upload.rs:89:41
   |
81 |         let bin_parts = db_parts.iter().values();
   |             --------- has type `impl std::iter::DoubleEndedIterator` which is not `Send`
...
89 |                     while let Some(r) = tasks.next().await {
   |                                         ^^^^^^^^^^^^^^^^^^ await occurs here, with `bin_parts` maybe used later
...
97 |     }
   |     - `bin_parts` is later dropped here

  

Я попытался drop после цикла:

 drop(bin_parts);
  

И это ошибка, которую я получаю:

 error[E0382]: use of moved value: `bin_parts`
  --> src/s3m/multipart_upload.rs:96:10
   |
80 |     let bin_parts = db_parts.iter().values();
   |         --------- move occurs because `bin_parts` has type `impl std::iter::DoubleEndedIterator`, which does not implement the `Copy` trait
81 |     for bin_part in bin_parts {
   |                     ---------
   |                     |
   |                     value moved here
   |                     help: consider borrowing to avoid moving into the for loop: `amp;bin_parts`
...
96 |     drop(bin_parts);
   |          ^^^^^^^^^ value used here after move

error[E0382]: use of moved value: `bin_parts`
  --> src/s3m/multipart_upload.rs:96:10
   |
80 |     let bin_parts = db_parts.iter().values();
   |         --------- move occurs because `bin_parts` has type `impl std::iter::DoubleEndedIterator`, which does not implement the `Copy` trait
81 |     for bin_part in bin_parts {
   |                     ---------
   |                     |
   |                     value moved here
   |                     help: consider borrowing to avoid moving into the for loop: `amp;bin_parts`
...
96 |     drop(bin_parts);
   |          ^^^^^^^^^ value used here after move
  

Я не могу удалить значение, потому что оно было перемещено.