生锈的 AWS 多部分上传使用 rusoto，多线程(人造丝)对"没有反应堆运行......"感到恐慌

Question

我试图在rust中上传一个文件到aws，因为我使用的是rusoto_s3的s3 rust客户端，当这些部分从单个发送时，我设法获得多部分上传代码工作。线程，然而，这不是我想要的，我想上传大文件，我希望能够发送这些部分在多个为此，我做了一点谷歌搜索，我遇到了人造丝。

多部分上传的工作方式如下:

1. 启动multipart ->aws将返回一个ID
2. 使用此ID发送不同的部件，传递文件块，以及部件号->aws将返回Etag
3. 一旦你发送了所有的零件，发送一个完整的上传请求与所有完成的零件作为一个数组包含Etag和零件号。

我是rust的新手，来自c++和Java背景，以下是我的代码:

#[tokio::test]
async fn if_multipart_then_upload_multiparts_dicom() {
let now = Instant::now();
dotenv().ok();
let local_filename = "./files/test_big.DCM";
let destination_filename = "24_time_test.dcm";
let mut file = std::fs::File::open(local_filename).unwrap();
const CHUNK_SIZE: usize = 7_000_000;
let mut buffer = Vec::with_capacity(CHUNK_SIZE);
let client = super::get_client().await;
let create_multipart_request = CreateMultipartUploadRequest {
bucket: client.bucket_name.to_owned(),
key: destination_filename.to_owned(),
..Default::default()
};
// Start the multipart upload and note the upload_id generated
let response = client
.s3
.create_multipart_upload(create_multipart_request)
.await
.expect("Couldn't create multipart upload");
let upload_id = response.upload_id.unwrap();
// Create upload parts
let create_upload_part = |body: Vec<u8>, part_number: i64| -> UploadPartRequest {
UploadPartRequest {
body: Some(body.into()),
bucket: client.bucket_name.to_owned(),
key: destination_filename.to_owned(),
upload_id: upload_id.to_owned(),
part_number: part_number,
..Default::default()
}
};
let completed_parts = Arc::new(Mutex::new(vec![]));
rayon::scope(|scope| {
let mut part_number = 1;
loop {
let maximum_bytes_to_read = CHUNK_SIZE - buffer.len();
println!("maximum_bytes_to_read: {}", maximum_bytes_to_read);
file.by_ref()
.take(maximum_bytes_to_read as u64)
.read_to_end(&mut buffer)
.unwrap();
println!("length: {}", buffer.len());
println!("part_number: {}", part_number);
if buffer.len() == 0 {
// The file has ended.
break;
}
let next_buffer = Vec::with_capacity(CHUNK_SIZE);
let data_to_send = buffer;
let completed_parts_cloned = completed_parts.clone();
scope.spawn(move |_| {
let part = create_upload_part(data_to_send.to_vec(), part_number);
{
let part_number = part.part_number;
let client = executor::block_on(super::get_client());
let response = executor::block_on(client.s3.upload_part(part));
completed_parts_cloned.lock().unwrap().push(CompletedPart {
e_tag: response
.expect("Couldn't complete multipart upload")
.e_tag
.clone(),
part_number: Some(part_number),
});
}
});
buffer = next_buffer;
part_number = part_number + 1;
}
});
let completed_upload = CompletedMultipartUpload {
parts: Some(completed_parts.lock().unwrap().to_vec()),
};
let complete_req = CompleteMultipartUploadRequest {
bucket: client.bucket_name.to_owned(),
key: destination_filename.to_owned(),
upload_id: upload_id.to_owned(),
multipart_upload: Some(completed_upload),
..Default::default()
};
client
.s3
.complete_multipart_upload(complete_req)
.await
.expect("Couldn't complete multipart upload");
println!(
"time taken: {}, with chunk:: {}",
now.elapsed().as_secs(),
CHUNK_SIZE
);
}

下面是我得到的输出和错误示例:

maximum_bytes_to_read: 7000000
length: 7000000
part_number: 1
maximum_bytes_to_read: 7000000
length: 7000000
part_number: 2
maximum_bytes_to_read: 7000000
thread '<unnamed>' panicked at 'there is no reactor running, must be called from the context of a Tokio 1.x runtime', C:UsersDNDT.cargoregistrysrcgithub.com-1ecc6299db9ec823tokio-1.2.0srcruntimeblockingpool.rs:85:33
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
thread '<unnamed>' panicked at 'there is no reactor running, must be called from the context of a Tokio 1.x runtime', C:UsersDNDT.cargoregistrysrcgithub.com-1ecc6299db9ec823tokio-1.2.0srcruntimeblockingpool.rs:85:33
length: 7000000

我在谷歌上搜索了这个错误，但我不清楚它到底是什么:

there is no reactor running, must be called from the context of Tokio runtime” 

以下是我的发现:另一个同样错误的问题

和另一个问题

这似乎是一些兼容性问题，因为s3可能使用的tokio版本与我拥有的tokio版本不兼容。

下面是一些相关的依赖项:

tokio = { version = "1", features = ["full"] }
tokio-compat-02 = "0.1.2"
rusoto_s3 = "0.46.0"
rusoto_core = "0.46.0"
rusoto_credential = "0.46.0"
rayon = "1.5.0"

我认为主要问题在于实际上想要在rayon线程中运行async代码。我试着用executor::block_on改变我的async代码阻塞代码，我也花了一些时间试图让编译器高兴，我有多个线程，他们都想写let completed_parts = Arc::new(Mutex::new(vec![]));，所以我在这里做了一些克隆，让编译器高兴。

如果我用过的图标很重要，下面是它们:

extern crate dotenv;
extern crate tokio;
use bytes::Bytes;
use dotenv::dotenv;
use futures::executor;
use futures::*;
use rusoto_core::credential::{EnvironmentProvider, ProvideAwsCredentials};
use rusoto_s3::util::{PreSignedRequest, PreSignedRequestOption};
use rusoto_s3::PutObjectRequest;
use rusoto_s3::StreamingBody;
use rusoto_s3::{
CompleteMultipartUploadRequest, CompletedMultipartUpload, CompletedPart,
CreateMultipartUploadRequest, UploadPartRequest, S3,
};
use std::io::Read;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use std::time::Instant;
use tokio::fs;

新的生锈，所以有很多移动的部件来得到这个正确的!

Answer 1

感谢@Jmb的讨论，我摆脱了线程，我spawn一个tokio任务如下:

创建一个持有期货的向量，以便我们可以等待它们:

let mut multiple_parts_futures = Vec::new();

生成async任务:

loop { // loop file chuncks
...
let send_part_task_future = tokio::task::spawn(async move {
// Upload part
...
}

然后等待所有的期货:

let _results = futures::future::join_all(multiple_parts_futures).await;

值得一提的是，完成的部分需要排序:

let mut completed_parts_vector = completed_parts.lock().unwrap().to_vec();
completed_parts_vector.sort_by_key(|part| part.part_number);

整个代码是:

#[tokio::test]
async fn if_multipart_then_upload_multiparts_dicom() {
let now = Instant::now();
dotenv().ok();
let local_filename = "./files/test_big.DCM";
let destination_filename = generate_unique_name();
let destination_filename_clone = destination_filename.clone();
let mut file = std::fs::File::open(local_filename).unwrap();
const CHUNK_SIZE: usize = 6_000_000;
let mut buffer = Vec::with_capacity(CHUNK_SIZE);
let client = super::get_client().await;
let create_multipart_request = CreateMultipartUploadRequest {
bucket: client.bucket_name.to_owned(),
key: destination_filename.to_owned(),
..Default::default()
};
// Start the multipart upload and note the upload_id generated
let response = client
.s3
.create_multipart_upload(create_multipart_request)
.await
.expect("Couldn't create multipart upload");
let upload_id = response.upload_id.unwrap();
let upload_id_clone = upload_id.clone();
// Create upload parts
let create_upload_part = move |body: Vec<u8>, part_number: i64| -> UploadPartRequest {
UploadPartRequest {
body: Some(body.into()),
bucket: client.bucket_name.to_owned(),
key: destination_filename_clone.to_owned(),
upload_id: upload_id_clone.to_owned(),
part_number: part_number,
..Default::default()
}
};
let create_upload_part_arc = Arc::new(create_upload_part);
let completed_parts = Arc::new(Mutex::new(vec![]));
let mut part_number = 1;
let mut multiple_parts_futures = Vec::new();
loop {
let maximum_bytes_to_read = CHUNK_SIZE - buffer.len();
println!("maximum_bytes_to_read: {}", maximum_bytes_to_read);
file.by_ref()
.take(maximum_bytes_to_read as u64)
.read_to_end(&mut buffer)
.unwrap();
println!("length: {}", buffer.len());
println!("part_number: {}", part_number);
if buffer.len() == 0 {
// The file has ended.
break;
}
let next_buffer = Vec::with_capacity(CHUNK_SIZE);
let data_to_send = buffer;
let completed_parts_cloned = completed_parts.clone();
let create_upload_part_arc_cloned = create_upload_part_arc.clone();
let send_part_task_future = tokio::task::spawn(async move {
let part = create_upload_part_arc_cloned(data_to_send.to_vec(), part_number);
{
let part_number = part.part_number;
let client = super::get_client().await;
let response = client.s3.upload_part(part).await;
completed_parts_cloned.lock().unwrap().push(CompletedPart {
e_tag: response
.expect("Couldn't complete multipart upload")
.e_tag
.clone(),
part_number: Some(part_number),
});
}
});
multiple_parts_futures.push(send_part_task_future);
buffer = next_buffer;
part_number = part_number + 1;
}
let client = super::get_client().await;
println!("waiting for futures");
let _results = futures::future::join_all(multiple_parts_futures).await;
let mut completed_parts_vector = completed_parts.lock().unwrap().to_vec();
completed_parts_vector.sort_by_key(|part| part.part_number);
println!("futures done");
let completed_upload = CompletedMultipartUpload {
parts: Some(completed_parts_vector),
};
let complete_req = CompleteMultipartUploadRequest {
bucket: client.bucket_name.to_owned(),
key: destination_filename.to_owned(),
upload_id: upload_id.to_owned(),
multipart_upload: Some(completed_upload),
..Default::default()
};
client
.s3
.complete_multipart_upload(complete_req)
.await
.expect("Couldn't complete multipart upload");
println!(
"time taken: {}, with chunk:: {}",
now.elapsed().as_secs(),
CHUNK_SIZE
);
}