-
Notifications
You must be signed in to change notification settings - Fork 195
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
add io related abstractions to support s3 tm uploads (#3756)
## Description Adds a new `io` module to S3 transfer manager HLL. This is supporting functionality for implementing uploads. High level description of new components: * **InputStream** - this is a new abstraction to replace `ByteStream` for use in transfer manager uploads. The primary driver behind this is once we create a `ByteStream` we have very little control over optimizing how it can be consumed. We lose whether it came from a file or in-memory or elsewhere. Internally this will of course eventually map to a `ByteStream` when we call `UploadPart` or `PutObject` but we want to have control over how bytes are read from the stream (e.g. being able to read a file concurrently). * In the future we may add additional variants to support e.g. unknown streams (e.g. `impl io::Read`) or giving users control over checksums for each part, etc. This is an area of exploration still. * **PartReader / ReadPart** - internal abstractions over `InputStream` that workers will eventually consume. The idea is we know what `InputStream` is and we convert it to something that reads "parts" for multipart upload in the most efficient way possible for whatever the underlying stream type is. * In a subsequent PR workers share a single part reader and read from it concurrently. Workers don't have to care about anything other than asking for the next "part" of data and uploading it. * **PathBodyBuilder / PathBody** - replacements for `FsBuilder` from smithy runtime / SDK. Notable difference is we don't currently allow specifying either an already open file or control over buffer sizes. * We can of course support an open file but the implementation will require additional locking to ensure bytes are read sequentially ---- _By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice._
- Loading branch information
Showing
9 changed files
with
828 additions
and
6 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,79 @@ | ||
| /* | ||
| * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
| use std::error::Error as StdError; | ||
| use std::fmt; | ||
| use std::fmt::Formatter; | ||
| use std::io::{Error as StdIoError, ErrorKind as StdIoErrorKind}; | ||
| use tokio::task::JoinError; | ||
|
|
||
| #[derive(Debug)] | ||
| pub(crate) enum ErrorKind { | ||
| UpperBoundSizeHintRequired, | ||
| OffsetGreaterThanFileSize, | ||
| TaskFailed(JoinError), | ||
| IoError(StdIoError), | ||
| } | ||
|
|
||
| /// An I/O related error occurred | ||
| #[derive(Debug)] | ||
| pub struct Error { | ||
| kind: ErrorKind, | ||
| } | ||
|
|
||
| impl Error { | ||
| pub(crate) fn upper_bound_size_hint_required() -> Error { | ||
| ErrorKind::UpperBoundSizeHintRequired.into() | ||
| } | ||
| } | ||
| impl From<ErrorKind> for Error { | ||
| fn from(kind: ErrorKind) -> Self { | ||
| Self { kind } | ||
| } | ||
| } | ||
|
|
||
| impl From<StdIoError> for Error { | ||
| fn from(err: StdIoError) -> Self { | ||
| ErrorKind::IoError(err).into() | ||
| } | ||
| } | ||
|
|
||
| impl fmt::Display for Error { | ||
| fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { | ||
| match &self.kind { | ||
| ErrorKind::UpperBoundSizeHintRequired => write!( | ||
| f, | ||
| "size hint upper bound (SizeHint::upper) is required but was None" | ||
| ), | ||
| ErrorKind::OffsetGreaterThanFileSize => write!( | ||
| f, | ||
| "offset must be less than or equal to file size but was greater than" | ||
| ), | ||
| ErrorKind::IoError(_) => write!(f, "I/O error"), | ||
| ErrorKind::TaskFailed(_) => write!(f, "task failed"), | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl StdError for Error { | ||
| fn source(&self) -> Option<&(dyn StdError + 'static)> { | ||
| match &self.kind { | ||
| ErrorKind::UpperBoundSizeHintRequired => None, | ||
| ErrorKind::OffsetGreaterThanFileSize => None, | ||
| ErrorKind::IoError(err) => Some(err as _), | ||
| ErrorKind::TaskFailed(err) => Some(err as _), | ||
| } | ||
| } | ||
| } | ||
| impl From<Error> for StdIoError { | ||
| fn from(err: Error) -> Self { | ||
| StdIoError::new(StdIoErrorKind::Other, err) | ||
| } | ||
| } | ||
|
|
||
| impl From<JoinError> for Error { | ||
| fn from(value: JoinError) -> Self { | ||
| ErrorKind::TaskFailed(value).into() | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,17 @@ | ||
| /* | ||
| * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| pub(crate) mod part_reader; | ||
| mod path_body; | ||
| mod stream; | ||
|
|
||
| /// Error types related to I/O abstractions | ||
| pub mod error; | ||
| mod size_hint; | ||
|
|
||
| // re-exports | ||
| pub use self::path_body::PathBodyBuilder; | ||
| pub use self::size_hint::SizeHint; | ||
| pub use self::stream::InputStream; |
Oops, something went wrong.