You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by GitBox <gi...@apache.org> on 2021/03/11 17:36:05 UTC
[GitHub] [arrow] pitrou opened a new pull request #9678: ARROW-11907: [C++] Use our own executor in S3FileSystem
pitrou opened a new pull request #9678:
URL: https://github.com/apache/arrow/pull/9678
The async APIs in the AWS SDK merely spawn a new thread each time
they are called.
By using our own executor, we schedule requests on our IO thread pool,
and we allow for potential cancellation.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a change in pull request #9678: ARROW-11907: [C++] Use our own executor in S3FileSystem
Posted by GitBox <gi...@apache.org>.
westonpace commented on a change in pull request #9678:
URL: https://github.com/apache/arrow/pull/9678#discussion_r594965503
##########
File path: cpp/src/arrow/filesystem/s3fs.cc
##########
@@ -1537,16 +1568,14 @@ class S3FileSystem::Impl {
}
req.SetBucket(ToAwsString(bucket));
req.SetDelete(std::move(del));
- delete_handlers.emplace_back();
- futures.push_back(&delete_handlers.back().future);
- client_->DeleteObjectsAsync(req, delete_handlers.back());
+ ARROW_ASSIGN_OR_RAISE(auto fut, io_context_.executor()->Submit(
+ io_context_.stop_token(), [client, req]() {
+ return client->DeleteObjects(req);
+ }));
+ futures.push_back(std::move(fut).Then(delete_cb));
}
- WaitForAll(futures);
- for (const auto* fut : futures) {
- RETURN_NOT_OK(fut->status());
- }
- return Status::OK();
+ return AllComplete(futures).status();
Review comment:
Blocking at the end sort of defeats the purpose but it can be a fine first step. If I do something like this I'll rename the method DeleteObjectsAsync that returns `Future<>` and then add back in `DeleteObjects` which calls `DeleteObjectsAsync` and waits on the status. This ensures backwards compatibility but makes it even easier to consume the async version in the future.
##########
File path: cpp/src/arrow/filesystem/s3fs.cc
##########
@@ -1035,6 +1025,26 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
+ static void HandleUploadOutcome(const std::shared_ptr<UploadState>& state,
+ int part_number, const S3Model::UploadPartRequest& req,
+ const Result<S3Model::UploadPartOutcome>& result) {
+ std::unique_lock<std::mutex> lock(state->mutex);
+ if (!result.ok()) {
+ state->status &= result.status();
+ } else {
+ const auto& outcome = *result;
+ if (!outcome.IsSuccess()) {
+ state->status &= UploadPartError(req, outcome);
+ } else {
+ AddCompletedPart(state, part_number, outcome.GetResult());
+ }
+ }
+ // Notify completion, regardless of success / error status
+ if (--state->parts_in_progress == 0) {
Review comment:
Similar to the other comment, if you use `Flush()` it blocks. You could create a `FlushAsync` and update the code here so that instead of just `notify_all()` it marked a future finished. Or you could use a threaded task group which already has this logic.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] pitrou commented on pull request #9678: ARROW-11907: [C++] Use our own executor in S3FileSystem
Posted by GitBox <gi...@apache.org>.
pitrou commented on pull request #9678:
URL: https://github.com/apache/arrow/pull/9678#issuecomment-796913424
@westonpace Do you want to review this?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] pitrou commented on pull request #9678: ARROW-11907: [C++] Use our own executor in S3FileSystem
Posted by GitBox <gi...@apache.org>.
pitrou commented on pull request #9678:
URL: https://github.com/apache/arrow/pull/9678#issuecomment-800425458
Ok, I think the Future change was a mistake, since the condition we're waiting for may be signalled multiple times.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] pitrou commented on a change in pull request #9678: ARROW-11907: [C++] Use our own executor in S3FileSystem
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9678:
URL: https://github.com/apache/arrow/pull/9678#discussion_r595053426
##########
File path: cpp/src/arrow/filesystem/s3fs.cc
##########
@@ -1035,6 +1025,26 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
+ static void HandleUploadOutcome(const std::shared_ptr<UploadState>& state,
+ int part_number, const S3Model::UploadPartRequest& req,
+ const Result<S3Model::UploadPartOutcome>& result) {
+ std::unique_lock<std::mutex> lock(state->mutex);
+ if (!result.ok()) {
+ state->status &= result.status();
+ } else {
+ const auto& outcome = *result;
+ if (!outcome.IsSuccess()) {
+ state->status &= UploadPartError(req, outcome);
+ } else {
+ AddCompletedPart(state, part_number, outcome.GetResult());
+ }
+ }
+ // Notify completion, regardless of success / error status
+ if (--state->parts_in_progress == 0) {
Review comment:
For now we're not exposing any `FlushAsync` method on `OutputStream`. I don't know if that would later be necessary. But we can indeed prepare for it by using a future.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on pull request #9678: ARROW-11907: [C++] Use our own executor in S3FileSystem
Posted by GitBox <gi...@apache.org>.
westonpace commented on pull request #9678:
URL: https://github.com/apache/arrow/pull/9678#issuecomment-797434279
Sure, it might take me a day or two. I want to learn the S3 stuff so I'll try running some of this. I'll try and get to it by the end of Monday.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] pitrou commented on pull request #9678: ARROW-11907: [C++] Use our own executor in S3FileSystem
Posted by GitBox <gi...@apache.org>.
pitrou commented on pull request #9678:
URL: https://github.com/apache/arrow/pull/9678#issuecomment-800409902
Ok, there are genuine failures now. Will take a look.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] pitrou commented on pull request #9678: ARROW-11907: [C++] Use our own executor in S3FileSystem
Posted by GitBox <gi...@apache.org>.
pitrou commented on pull request #9678:
URL: https://github.com/apache/arrow/pull/9678#issuecomment-800289462
AppVeyor succeeded here: https://ci.appveyor.com/project/pitrou/arrow/builds/38247813
However, there was a weird hang in another AppVeyor build, which I've restarted.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] pitrou closed pull request #9678: ARROW-11907: [C++] Use our own executor in S3FileSystem
Posted by GitBox <gi...@apache.org>.
pitrou closed pull request #9678:
URL: https://github.com/apache/arrow/pull/9678
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] pitrou commented on a change in pull request #9678: ARROW-11907: [C++] Use our own executor in S3FileSystem
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9678:
URL: https://github.com/apache/arrow/pull/9678#discussion_r595049739
##########
File path: cpp/src/arrow/filesystem/s3fs.cc
##########
@@ -1537,16 +1568,14 @@ class S3FileSystem::Impl {
}
req.SetBucket(ToAwsString(bucket));
req.SetDelete(std::move(del));
- delete_handlers.emplace_back();
- futures.push_back(&delete_handlers.back().future);
- client_->DeleteObjectsAsync(req, delete_handlers.back());
+ ARROW_ASSIGN_OR_RAISE(auto fut, io_context_.executor()->Submit(
+ io_context_.stop_token(), [client, req]() {
+ return client->DeleteObjects(req);
+ }));
+ futures.push_back(std::move(fut).Then(delete_cb));
}
- WaitForAll(futures);
- for (const auto* fut : futures) {
- RETURN_NOT_OK(fut->status());
- }
- return Status::OK();
+ return AllComplete(futures).status();
Review comment:
Well, this is implementing a sync method. The FileSystem interface doesn't have an async deletion method. We may add one, but I don't think that's necessary for now.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] pitrou commented on pull request #9678: ARROW-11907: [C++] Use our own executor in S3FileSystem
Posted by GitBox <gi...@apache.org>.
pitrou commented on pull request #9678:
URL: https://github.com/apache/arrow/pull/9678#issuecomment-800153533
> Will that always be appropriate?
Probably not, but you can pass a custom executor when creating a filesystem. The hard question being of course how to choose the right size...
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] github-actions[bot] commented on pull request #9678: ARROW-11907: [C++] Use our own executor in S3FileSystem
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #9678:
URL: https://github.com/apache/arrow/pull/9678#issuecomment-796919200
https://issues.apache.org/jira/browse/ARROW-11907
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org