Switch to one single runtime for S3 storage

Changed S3Backend::start_storage_transaction from creating a new Runtime on each call to using a lazily-initialized global runtime for all S3 instances to use.
Since Tokio v0.1's Runtime has no mechanism to allow for blocking that doesn't require an &mut self, the runtime is stored inside of a mutex to allow for
mutable access. We're already transitively dependent on parking_lot v0.10.2, so this is just adding it as a direct dependency (Removing tokio v0.2 will remove the other)
version of parking_lot we depend on). When upgrading to tokio v0.2 this should be changed since v0.2 has the Runtime::handle method which allows runtime access from only &self.

Cargo.toml

@@ -46,6 +46,7 @@ path-slash = "0.1.3"
 once_cell = { version = "1.4.0", features = ["parking_lot"] }
 base64 = "0.12.1"
 strum = { version = "0.18.0", features = ["derive"] }
+parking_lot = "0.10.2"
 
 # Data serialization and deserialization
 serde = { version = "1.0", features = ["derive"] }

src/storage/s3.rs

@@ -4,6 +4,8 @@ use failure::Error;
 use futures::stream::{FuturesUnordered, Stream};
 use futures::Future;
 use log::{error, warn};
+use once_cell::sync::Lazy;
+use parking_lot::Mutex;
 use rusoto_core::region::Region;
 use rusoto_credential::DefaultCredentialsProvider;
 use rusoto_s3::{GetObjectRequest, PutObjectRequest, S3Client, S3};
@@ -16,17 +18,21 @@ mod test;
 pub(crate) use test::TestS3;
 
 pub(crate) static S3_BUCKET_NAME: &str = "rust-docs-rs";
+static S3_RUNTIME: Lazy<Mutex<Runtime>> =
+    Lazy::new(|| Mutex::new(Runtime::new().expect("Failed to create S3 runtime")));
 
 pub(crate) struct S3Backend {
     client: S3Client,
     bucket: String,
+    runtime: &'static Mutex<Runtime>,
 }
 
 impl S3Backend {
     pub(crate) fn new(client: S3Client, bucket: &str) -> Self {
         Self {
             client,
             bucket: bucket.into(),
+            runtime: &*S3_RUNTIME,
         }
     }
 
@@ -63,16 +69,12 @@ impl S3Backend {
     }
 
     pub(super) fn start_storage_transaction(&self) -> Result<S3StorageTransaction, Error> {
-        Ok(S3StorageTransaction {
-            s3: self,
-            runtime: Runtime::new()?,
-        })
+        Ok(S3StorageTransaction { s3: self })
     }
 }
 
 pub(super) struct S3StorageTransaction<'a> {
     s3: &'a S3Backend,
-    runtime: Runtime,
 }
 
 impl<'a> StorageTransaction for S3StorageTransaction<'a> {
@@ -100,7 +102,7 @@ impl<'a> StorageTransaction for S3StorageTransaction<'a> {
             }
             attempts += 1;
 
-            match self.runtime.block_on(futures.map(drop).collect()) {
+            match self.s3.runtime.lock().block_on(futures.map(drop).collect()) {
                 // this batch was successful, start another batch if there are still more files
                 Ok(_) => break,
                 Err(err) => {