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/02/18 19:20:18 UTC
[GitHub] [arrow] pitrou opened a new pull request #9528: ARROW-8732: [C++] Add basic cancellation API (v2)
pitrou opened a new pull request #9528:
URL: https://github.com/apache/arrow/pull/9528
In this model, a StopToken is instantiated by the consumer, which passes a pointer to producer API(s).
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou closed pull request #9528:
URL: https://github.com/apache/arrow/pull/9528
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
westonpace commented on pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#issuecomment-793064631
Abseil (seems to be Google's equivalent of Folly) has an async-signal-safe spin lock. That might be easier than a doubly linked list. However, it doesn't seem to be easily vendored (just based on the fact that it has 10 abseil includes) :(
https://github.com/abseil/abseil-cpp/blob/master/absl/base/internal/spinlock.h
I'd agree that it seems too unlikely to worry about.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API (v2)
Posted by GitBox <gi...@apache.org>.
pitrou commented on pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#issuecomment-782057195
Hmm, perhaps instead of complicating the Future implementation, it would be better to make the Executor responsible for setting the stop callback.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#issuecomment-790018580
Revision: 848e020cfaf234411d3a0167b58dc39c030823a4
Submitted crossbow builds: [ursacomputing/crossbow @ actions-174](https://github.com/ursacomputing/crossbow/branches/all?query=actions-174)
|Task|Status|
|----|------|
|test-conda-python-3.6|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.6)|
|test-conda-python-3.6-pandas-0.23|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.6-pandas-0.23)|
|test-conda-python-3.7|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.7)|
|test-conda-python-3.7-dask-latest|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.7-dask-latest)|
|test-conda-python-3.7-hdfs-3.2|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.7-hdfs-3.2)|
|test-conda-python-3.7-kartothek-latest|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.7-kartothek-latest)|
|test-conda-python-3.7-kartothek-master|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.7-kartothek-master)|
|test-conda-python-3.7-pandas-latest|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.7-pandas-latest)|
|test-conda-python-3.7-pandas-master|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.7-pandas-master)|
|test-conda-python-3.7-spark-branch-3.0|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.7-spark-branch-3.0)|
|test-conda-python-3.7-turbodbc-latest|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.7-turbodbc-latest)|
|test-conda-python-3.7-turbodbc-master|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.7-turbodbc-master)|
|test-conda-python-3.8|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.8)|
|test-conda-python-3.8-dask-master|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.8-dask-master)|
|test-conda-python-3.8-hypothesis|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.8-hypothesis)|
|test-conda-python-3.8-jpype|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.8-jpype)|
|test-conda-python-3.8-pandas-latest|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.8-pandas-latest)|
|test-conda-python-3.8-pandas-nightly|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.8-pandas-nightly)|
|test-conda-python-3.8-spark-master|[](https://github.com/ursacomputing/crossbow/actions?query=branch:actions-174-github-test-conda-python-3.8-spark-master)|
|test-debian-10-python-3|[](https://dev.azure.com/ursacomputing/crossbow/_build/latest?definitionId=1&branchName=actions-174-azure-test-debian-10-python-3)|
|test-fedora-33-python-3|[](https://dev.azure.com/ursacomputing/crossbow/_build/latest?definitionId=1&branchName=actions-174-azure-test-fedora-33-python-3)|
|test-ubuntu-18.04-python-3|[](https://dev.azure.com/ursacomputing/crossbow/_build/latest?definitionId=1&branchName=actions-174-azure-test-ubuntu-18.04-python-3)|
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r587692069
##########
File path: cpp/src/arrow/util/cancel.cc
##########
@@ -0,0 +1,167 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/util/cancel.h"
+
+#include <atomic>
+#include <mutex>
+#include <sstream>
+#include <utility>
+
+#include "arrow/result.h"
+#include "arrow/util/atomic_shared_ptr.h"
+#include "arrow/util/io_util.h"
+#include "arrow/util/logging.h"
+#include "arrow/util/visibility.h"
+
+namespace arrow {
+
+#if ATOMIC_INT_LOCK_FREE != 2
+#error Lock-free atomic int required for signal safety
+#endif
+
+using internal::ReinstateSignalHandler;
+using internal::SetSignalHandler;
+using internal::SignalHandler;
+
+// NOTE: We care mainly about the making the common case (not cancelled) fast.
+
+struct StopSourceImpl {
+ std::atomic<int> requested_{0}; // will be -1 or signal number if requested
+ std::mutex mutex_;
+ Status cancel_error_;
+};
+
+StopSource::StopSource() : impl_(new StopSourceImpl) {}
+
+StopSource::~StopSource() = default;
+
+void StopSource::RequestStop() { RequestStop(Status::Cancelled("Operation cancelled")); }
+
+void StopSource::RequestStop(Status st) {
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ DCHECK(!st.ok());
+ if (!impl_->requested_) {
+ impl_->requested_ = -1;
+ impl_->cancel_error_ = std::move(st);
+ }
+}
+
+void StopSource::RequestStopFromSignal(int signum) {
+ // Only async-signal-safe code allowed here
+ impl_->requested_.store(signum);
+}
+
+StopToken StopSource::token() { return StopToken(impl_); }
+
+bool StopToken::IsStopRequested() {
+ if (!impl_) {
+ return false;
+ }
+ return impl_->requested_.load() != 0;
+}
+
+Status StopToken::Poll() {
+ if (!impl_) {
+ return Status::OK();
+ }
+ if (!impl_->requested_.load()) {
+ return Status::OK();
+ }
+
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ if (impl_->cancel_error_.ok()) {
+ auto signum = impl_->requested_.load();
+ DCHECK_GT(signum, 0);
+ impl_->cancel_error_ = internal::CancelledFromSignal(signum, "Operation cancelled");
+ }
+ return impl_->cancel_error_;
+}
+
+namespace {
+
+void HandleSignal(int signum);
+
+struct SignalStopState {
+ struct SavedSignalHandler {
+ int signum;
+ SignalHandler handler;
+ };
+
+ Status RegisterHandlers(const std::vector<int>& signals) {
+ if (!saved_handlers.empty()) {
+ return Status::Invalid("Signal handlers already registered");
+ }
+ for (int signum : signals) {
+ ARROW_ASSIGN_OR_RAISE(auto handler,
+ SetSignalHandler(signum, SignalHandler{&HandleSignal}));
+ saved_handlers.push_back({signum, handler});
+ }
+ return Status::OK();
+ }
+
+ void UnregisterHandlers() {
+ auto handlers = std::move(saved_handlers);
+ for (const auto& h : handlers) {
+ ARROW_CHECK_OK(SetSignalHandler(h.signum, h.handler).status());
+ }
+ }
+
+ ~SignalStopState() { UnregisterHandlers(); }
+
+ StopSource stop_source;
+ std::vector<SavedSignalHandler> saved_handlers;
+};
+
+std::shared_ptr<SignalStopState> g_signal_stop_state;
+
+void HandleSignal(int signum) {
+ ReinstateSignalHandler(signum, &HandleSignal);
+ std::shared_ptr<SignalStopState> state = internal::atomic_load(&g_signal_stop_state);
+ if (state) {
+ state->stop_source.RequestStopFromSignal(signum);
+ }
+}
+
+} // namespace
+
+Result<StopSource*> SetSignalStopSource() {
+ if (g_signal_stop_state) {
+ return Status::Invalid("Signal stop source already set up");
+ }
+ internal::atomic_store(&g_signal_stop_state, std::make_shared<SignalStopState>());
+ return &g_signal_stop_state->stop_source;
+}
+
+void ResetSignalStopSource() {
+ internal::atomic_store(&g_signal_stop_state, std::shared_ptr<SignalStopState>{});
Review comment:
Good point, will change.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#issuecomment-793067079
How would a spinlock help the deallocation issue?
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r587691144
##########
File path: cpp/src/arrow/util/cancel.cc
##########
@@ -0,0 +1,167 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/util/cancel.h"
+
+#include <atomic>
+#include <mutex>
+#include <sstream>
+#include <utility>
+
+#include "arrow/result.h"
+#include "arrow/util/atomic_shared_ptr.h"
+#include "arrow/util/io_util.h"
+#include "arrow/util/logging.h"
+#include "arrow/util/visibility.h"
+
+namespace arrow {
+
+#if ATOMIC_INT_LOCK_FREE != 2
+#error Lock-free atomic int required for signal safety
+#endif
+
+using internal::ReinstateSignalHandler;
+using internal::SetSignalHandler;
+using internal::SignalHandler;
+
+// NOTE: We care mainly about the making the common case (not cancelled) fast.
+
+struct StopSourceImpl {
+ std::atomic<int> requested_{0}; // will be -1 or signal number if requested
+ std::mutex mutex_;
+ Status cancel_error_;
+};
+
+StopSource::StopSource() : impl_(new StopSourceImpl) {}
+
+StopSource::~StopSource() = default;
+
+void StopSource::RequestStop() { RequestStop(Status::Cancelled("Operation cancelled")); }
+
+void StopSource::RequestStop(Status st) {
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ DCHECK(!st.ok());
+ if (!impl_->requested_) {
+ impl_->requested_ = -1;
+ impl_->cancel_error_ = std::move(st);
+ }
+}
+
+void StopSource::RequestStopFromSignal(int signum) {
+ // Only async-signal-safe code allowed here
+ impl_->requested_.store(signum);
+}
+
+StopToken StopSource::token() { return StopToken(impl_); }
+
+bool StopToken::IsStopRequested() {
+ if (!impl_) {
+ return false;
+ }
+ return impl_->requested_.load() != 0;
+}
+
+Status StopToken::Poll() {
+ if (!impl_) {
+ return Status::OK();
+ }
+ if (!impl_->requested_.load()) {
+ return Status::OK();
+ }
+
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ if (impl_->cancel_error_.ok()) {
+ auto signum = impl_->requested_.load();
+ DCHECK_GT(signum, 0);
+ impl_->cancel_error_ = internal::CancelledFromSignal(signum, "Operation cancelled");
+ }
Review comment:
If you call Poll() twice, then the cancel_error corresponding to the signal will have been cached.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r588183226
##########
File path: cpp/src/arrow/util/task_group.cc
##########
@@ -67,34 +75,54 @@ class SerialTaskGroup : public TaskGroup {
class ThreadedTaskGroup : public TaskGroup {
public:
- explicit ThreadedTaskGroup(Executor* executor)
- : executor_(executor), nremaining_(0), ok_(true) {}
+ ThreadedTaskGroup(Executor* executor, StopToken stop_token)
+ : executor_(executor),
+ stop_token_(std::move(stop_token)),
+ nremaining_(0),
+ ok_(true) {}
~ThreadedTaskGroup() override {
// Make sure all pending tasks are finished, so that dangling references
// to this don't persist.
ARROW_UNUSED(Finish());
}
- void AppendReal(std::function<Status()> task) override {
+ void AppendReal(FnOnce<Status()> task) override {
DCHECK(!finished_);
+ if (stop_token_.IsStopRequested()) {
+ UpdateStatus(stop_token_.Poll());
+ return;
+ }
+
// The hot path is unlocked thanks to atomics
// Only if an error occurs is the lock taken
if (ok_.load(std::memory_order_acquire)) {
nremaining_.fetch_add(1, std::memory_order_acquire);
auto self = checked_pointer_cast<ThreadedTaskGroup>(shared_from_this());
- Status st = executor_->Spawn(std::bind(
- [](const std::shared_ptr<ThreadedTaskGroup>& self,
- const std::function<Status()>& task) {
- if (self->ok_.load(std::memory_order_acquire)) {
+
+ struct Callable {
+ void operator()() {
+ if (self_->ok_.load(std::memory_order_acquire)) {
+ Status st;
+ if (stop_token_.IsStopRequested()) {
+ st = stop_token_.Poll();
+ } else {
// XXX what about exceptions?
- Status st = task();
- self->UpdateStatus(std::move(st));
+ st = std::move(task_)();
}
- self->OneTaskDone();
- },
- std::move(self), std::move(task)));
+ self_->UpdateStatus(std::move(st));
+ }
+ self_->OneTaskDone();
+ }
+
+ std::shared_ptr<ThreadedTaskGroup> self_;
+ FnOnce<Status()> task_;
+ StopToken stop_token_;
Review comment:
Note to self: can use `self_->stop_token_` instead.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r589478886
##########
File path: cpp/src/arrow/util/cancel.h
##########
@@ -0,0 +1,99 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <functional>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "arrow/status.h"
+#include "arrow/type_fwd.h"
+#include "arrow/util/macros.h"
+#include "arrow/util/visibility.h"
+
+namespace arrow {
+
+class StopToken;
+
+struct StopSourceImpl;
+
+/// EXPERIMENTAL
+class ARROW_EXPORT StopSource {
+ public:
+ StopSource();
+ ~StopSource();
+
+ // Consumer API (the side that stops)
+ void RequestStop();
+ void RequestStop(Status error);
+ void RequestStopFromSignal(int signum);
Review comment:
It can indeed be useful if the application catches signals by itself.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r589478493
##########
File path: cpp/src/arrow/util/cancel.cc
##########
@@ -0,0 +1,167 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/util/cancel.h"
+
+#include <atomic>
+#include <mutex>
+#include <sstream>
+#include <utility>
+
+#include "arrow/result.h"
+#include "arrow/util/atomic_shared_ptr.h"
+#include "arrow/util/io_util.h"
+#include "arrow/util/logging.h"
+#include "arrow/util/visibility.h"
+
+namespace arrow {
+
+#if ATOMIC_INT_LOCK_FREE != 2
+#error Lock-free atomic int required for signal safety
+#endif
+
+using internal::ReinstateSignalHandler;
+using internal::SetSignalHandler;
+using internal::SignalHandler;
+
+// NOTE: We care mainly about the making the common case (not cancelled) fast.
+
+struct StopSourceImpl {
+ std::atomic<int> requested_{0}; // will be -1 or signal number if requested
+ std::mutex mutex_;
+ Status cancel_error_;
+};
+
+StopSource::StopSource() : impl_(new StopSourceImpl) {}
+
+StopSource::~StopSource() = default;
+
+void StopSource::RequestStop() { RequestStop(Status::Cancelled("Operation cancelled")); }
+
+void StopSource::RequestStop(Status st) {
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ DCHECK(!st.ok());
+ if (!impl_->requested_) {
+ impl_->requested_ = -1;
+ impl_->cancel_error_ = std::move(st);
+ }
+}
+
+void StopSource::RequestStopFromSignal(int signum) {
+ // Only async-signal-safe code allowed here
+ impl_->requested_.store(signum);
+}
+
+StopToken StopSource::token() { return StopToken(impl_); }
+
+bool StopToken::IsStopRequested() {
+ if (!impl_) {
+ return false;
+ }
+ return impl_->requested_.load() != 0;
+}
+
+Status StopToken::Poll() {
+ if (!impl_) {
+ return Status::OK();
+ }
+ if (!impl_->requested_.load()) {
+ return Status::OK();
+ }
+
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ if (impl_->cancel_error_.ok()) {
+ auto signum = impl_->requested_.load();
+ DCHECK_GT(signum, 0);
+ impl_->cancel_error_ = internal::CancelledFromSignal(signum, "Operation cancelled");
+ }
+ return impl_->cancel_error_;
+}
+
+namespace {
+
+void HandleSignal(int signum);
+
+struct SignalStopState {
+ struct SavedSignalHandler {
+ int signum;
+ SignalHandler handler;
+ };
+
+ Status RegisterHandlers(const std::vector<int>& signals) {
+ if (!saved_handlers.empty()) {
+ return Status::Invalid("Signal handlers already registered");
+ }
+ for (int signum : signals) {
+ ARROW_ASSIGN_OR_RAISE(auto handler,
+ SetSignalHandler(signum, SignalHandler{&HandleSignal}));
+ saved_handlers.push_back({signum, handler});
+ }
+ return Status::OK();
+ }
+
+ void UnregisterHandlers() {
+ auto handlers = std::move(saved_handlers);
+ for (const auto& h : handlers) {
+ ARROW_CHECK_OK(SetSignalHandler(h.signum, h.handler).status());
+ }
+ }
+
+ ~SignalStopState() { UnregisterHandlers(); }
+
+ StopSource stop_source;
+ std::vector<SavedSignalHandler> saved_handlers;
+};
+
+std::shared_ptr<SignalStopState> g_signal_stop_state;
+
+void HandleSignal(int signum) {
+ ReinstateSignalHandler(signum, &HandleSignal);
+ std::shared_ptr<SignalStopState> state = internal::atomic_load(&g_signal_stop_state);
Review comment:
Ouch, I think you're right.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#issuecomment-789995923
@ursabot crossbow submit -g python
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#issuecomment-792998856
@github-actions crossbow submit -g python
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r589537475
##########
File path: python/pyarrow/_csv.pyx
##########
@@ -34,8 +34,8 @@ from pyarrow.lib cimport (check_status, Field, MemoryPool, Schema,
pyarrow_unwrap_batch, pyarrow_unwrap_table,
pyarrow_wrap_schema, pyarrow_wrap_table,
pyarrow_wrap_data_type, pyarrow_unwrap_data_type,
- Table, RecordBatch)
-from pyarrow.lib import frombytes, tobytes
+ Table, RecordBatch, StopToken)
Review comment:
No JIRA yet. We can open one, or do it when needed.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API (v2)
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r578786958
##########
File path: cpp/src/arrow/util/future.h
##########
@@ -515,15 +548,35 @@ class ARROW_MUST_USE_TYPE Future {
}
void DoMarkFinished(Result<ValueType> res) {
- SetResult(std::move(res));
-
- if (ARROW_PREDICT_TRUE(GetResult()->ok())) {
- impl_->MarkFinished();
+ const FutureState state = StateForResult(res);
+ if (impl_->stop_callback_.has_value()) {
+ struct ResultSetter {
+ Future<T>* fut;
+ Result<ValueType> res;
+
+ void operator()() { fut->SetResult(std::move(res)); }
+ };
+ impl_->MarkFinished(state, ResultSetter{this, std::move(res)});
} else {
- impl_->MarkFailed();
+ // The future cannot be cancelled concurrently as CancelOn()
+ // was not called.
+ SetResult(std::move(res));
+ impl_->MarkFinished(state);
}
Review comment:
No, I'm pointing out that I'm giving a callable to `FutureImpl::MarkFinished` instead of calling it directly.
----------------------------------------------------------------
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 edited a comment on pull request #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou edited a comment on pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#issuecomment-792931920
I pushed an improvement for the deallocation in signal handler problem, but it's still not 100% safe (though probably extremely unlikely). I think solving it entirely would need a lock-free doubly-linked list.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r588183511
##########
File path: cpp/src/arrow/util/thread_pool.cc
##########
@@ -91,12 +101,20 @@ static void WorkerLoop(std::shared_ptr<ThreadPool::State> state,
--state->ready_count_;
DCHECK_GE(state->ready_count_, 0);
{
- FnOnce<void()> task = std::move(state->pending_tasks_.front());
+ Task task = std::move(state->pending_tasks_.front());
state->pending_tasks_.pop_front();
+ StopToken* stop_token = &task.stop_token;
lock.unlock();
- std::move(task)();
+ if (!stop_token->IsStopRequested()) {
+ std::move(task.callable)();
+ } else {
+ if (task.stop_callback) {
+ std::move(task.stop_callback)(stop_token->Poll());
+ }
+ }
+ ARROW_UNUSED(std::move(task)); // release resources before waiting for lock
+ lock.lock();
Review comment:
Note to self: simply re-lock outside of scope? (to let `task` be destroyed naturally)
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r588183511
##########
File path: cpp/src/arrow/util/thread_pool.cc
##########
@@ -91,12 +101,20 @@ static void WorkerLoop(std::shared_ptr<ThreadPool::State> state,
--state->ready_count_;
DCHECK_GE(state->ready_count_, 0);
{
- FnOnce<void()> task = std::move(state->pending_tasks_.front());
+ Task task = std::move(state->pending_tasks_.front());
state->pending_tasks_.pop_front();
+ StopToken* stop_token = &task.stop_token;
lock.unlock();
- std::move(task)();
+ if (!stop_token->IsStopRequested()) {
+ std::move(task.callable)();
+ } else {
+ if (task.stop_callback) {
+ std::move(task.stop_callback)(stop_token->Poll());
+ }
+ }
+ ARROW_UNUSED(std::move(task)); // release resources before waiting for lock
+ lock.lock();
Review comment:
Note to self: better to re-lock outside of scope? (to let `task` be destroyed)
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API (v2)
Posted by GitBox <gi...@apache.org>.
westonpace commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r578780183
##########
File path: cpp/src/arrow/util/future.h
##########
@@ -515,15 +548,35 @@ class ARROW_MUST_USE_TYPE Future {
}
void DoMarkFinished(Result<ValueType> res) {
- SetResult(std::move(res));
-
- if (ARROW_PREDICT_TRUE(GetResult()->ok())) {
- impl_->MarkFinished();
+ const FutureState state = StateForResult(res);
+ if (impl_->stop_callback_.has_value()) {
+ struct ResultSetter {
+ Future<T>* fut;
+ Result<ValueType> res;
+
+ void operator()() { fut->SetResult(std::move(res)); }
+ };
+ impl_->MarkFinished(state, ResultSetter{this, std::move(res)});
} else {
- impl_->MarkFailed();
+ // The future cannot be cancelled concurrently as CancelOn()
+ // was not called.
+ SetResult(std::move(res));
+ impl_->MarkFinished(state);
}
Review comment:
We already had a lock so if I understand correctly you are pointing out that the critical section is getting larger (to include the result writing)?
For the waiters it is unlikely to be a big issue. They are waiting already so they shouldn't be too bothered by a bit more waiting.
For AddCallback/TryAddCallback contention it would have to be a situation where the future the callback being added to was truly asynchronous (spawned work in another thread). If it was a purely synchronous future then mark finished would be finished being called before AddCallback is called. So in this case the person calling AddCallback has already accepted they might be adding a callback early and accepting the cost of their task being added to the scheduler and losing their CPU caches. A little blocking (especially since it results in them being able to run their callback immediately) should already be in the range of costs they are willing to accept.
In other words, you already should not be calling AddCallback on a potentially unfinished future in a performance critical section of code.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
westonpace commented on pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#issuecomment-793078243
It could only deallocate it if the global reference was changed and all attempts to change the global reference should go through the same spinlock. The signal handler should always be given the "second reference" no matter what.
Or, since you're using spinlocks to guard all changes to the global reference, you could just change the global reference to a plain old pointer which might make it more obvious.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API (v2)
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#issuecomment-781577526
https://issues.apache.org/jira/browse/ARROW-8732
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#issuecomment-790509864
@westonpace @bkietz Feel free to review.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
westonpace commented on pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#issuecomment-793073338
The handler would obtain the lock before calling `atomic_load` on the pointer and reset the pointer before it releases the lock. Any call to `ResetSignalStopSource` would grab the lock before modifying the pointer. This should ensure that the pointer obtained by the load cannot go out of scope while the signal handler references it and thus would never have to free.
----------------------------------------------------------------
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] bkietz commented on a change in pull request #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
bkietz commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r587606004
##########
File path: cpp/src/arrow/util/thread_pool_test.cc
##########
@@ -192,32 +208,50 @@ TEST_F(TestThreadPool, StressSpawnThreaded) {
TEST_F(TestThreadPool, SpawnSlow) {
// This checks that Shutdown() waits for all tasks to finish
auto pool = this->MakeThreadPool(2);
- SpawnAdds(pool.get(), 7, [](int x, int y, int* out) {
- return task_slow_add(0.02 /* seconds */, x, y, out);
- });
+ SpawnAdds(pool.get(), 7, task_slow_add<int>{0.02 /* seconds */});
Review comment:
```suggestion
SpawnAdds(pool.get(), 7, task_slow_add<int>{ /*seconds=*/0.02});
```
##########
File path: cpp/src/arrow/util/io_util.cc
##########
@@ -1608,6 +1640,39 @@ Result<SignalHandler> SetSignalHandler(int signum, const SignalHandler& handler)
return Status::OK();
}
+void ReinstateSignalHandler(int signum, SignalHandler::Callback handler) {
+#if !ARROW_HAVE_SIGACTION
+ // Cannot report any errors from signal() (but there shouldn't be any)
+ signal(signum, handler);
+#endif
+}
+
+Status SendSignal(int signum) {
+ if (raise(signum) == 0) {
+ return Status::OK();
+ }
+ if (errno == EINVAL) {
+ return Status::Invalid("Invalid signal number");
+ }
+ return IOErrorFromErrno(errno, "Failed to raise signal");
+}
+
+Status SendSignalToThread(int signum, uint64_t thread_id) {
+#ifdef _WIN32
+ return Status::NotImplemented("Cannot send signal to specific thread on Windows");
+#else
+ // Have to use a C-style cast because pthread_t can be a pointer *or* integer type
+ int r = pthread_kill((pthread_t)thread_id, signum);
+ if (r == 0) {
+ return Status::OK();
+ }
+ if (r == EINVAL) {
+ return Status::Invalid("Invalid signal number");
Review comment:
```suggestion
return Status::Invalid("Invalid signal number ", signum);
```
##########
File path: cpp/src/arrow/csv/reader.cc
##########
@@ -833,9 +843,10 @@ class AsyncThreadedTableReader
AsyncThreadedTableReader(MemoryPool* pool, std::shared_ptr<io::InputStream> input,
const ReadOptions& read_options,
const ParseOptions& parse_options,
- const ConvertOptions& convert_options, Executor* cpu_executor,
- Executor* io_executor)
- : BaseTableReader(pool, input, read_options, parse_options, convert_options),
+ const ConvertOptions& convert_options, StopToken stop_token,
+ Executor* cpu_executor, Executor* io_executor)
Review comment:
Seems we might prefer to rewrite this constructor to take an IOContext
##########
File path: cpp/src/arrow/util/io_util_test.cc
##########
@@ -623,5 +655,46 @@ TEST(FileUtils, LongPaths) {
}
#endif
+static std::atomic<int> signal_received;
+
+static void handle_signal(int signum) {
+ ReinstateSignalHandler(signum, &handle_signal);
+ signal_received.store(signum);
+}
+
+TEST(SendSignal, Generic) {
+ signal_received.store(0);
+ SignalHandlerGuard guard(SIGINT, &handle_signal);
+
+ ASSERT_EQ(signal_received.load(), 0);
+ ASSERT_OK(SendSignal(SIGINT));
+ BusyWait(1.0, [&]() { return signal_received.load() != 0; });
+ ASSERT_EQ(signal_received.load(), SIGINT);
+
+ // Re-try (exercise ReinstateSignalHandler)
+ signal_received.store(0);
+ ASSERT_OK(SendSignal(SIGINT));
+ BusyWait(1.0, [&]() { return signal_received.load() != 0; });
+ ASSERT_EQ(signal_received.load(), SIGINT);
+}
+
+TEST(SendSignal, ToThread) {
+#ifdef _WIN32
+ uint64_t dummy_thread_id = 42;
+ ASSERT_RAISES(NotImplemented, SendSignalToThread(SIGINT, dummy_thread_id));
+#else
+ // Have to use a C-style cast because pthread_t can be a pointer *or* integer type
+ uint64_t thread_id = (uint64_t)(pthread_self());
Review comment:
```suggestion
uint64_t thread_id = (uint64_t)(pthread_self()); // NOLINT readability-casting
```
##########
File path: cpp/src/arrow/util/cancel.cc
##########
@@ -0,0 +1,167 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/util/cancel.h"
+
+#include <atomic>
+#include <mutex>
+#include <sstream>
+#include <utility>
+
+#include "arrow/result.h"
+#include "arrow/util/atomic_shared_ptr.h"
+#include "arrow/util/io_util.h"
+#include "arrow/util/logging.h"
+#include "arrow/util/visibility.h"
+
+namespace arrow {
+
+#if ATOMIC_INT_LOCK_FREE != 2
+#error Lock-free atomic int required for signal safety
+#endif
+
+using internal::ReinstateSignalHandler;
+using internal::SetSignalHandler;
+using internal::SignalHandler;
+
+// NOTE: We care mainly about the making the common case (not cancelled) fast.
+
+struct StopSourceImpl {
+ std::atomic<int> requested_{0}; // will be -1 or signal number if requested
+ std::mutex mutex_;
+ Status cancel_error_;
+};
+
+StopSource::StopSource() : impl_(new StopSourceImpl) {}
+
+StopSource::~StopSource() = default;
+
+void StopSource::RequestStop() { RequestStop(Status::Cancelled("Operation cancelled")); }
+
+void StopSource::RequestStop(Status st) {
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ DCHECK(!st.ok());
+ if (!impl_->requested_) {
+ impl_->requested_ = -1;
+ impl_->cancel_error_ = std::move(st);
+ }
+}
+
+void StopSource::RequestStopFromSignal(int signum) {
+ // Only async-signal-safe code allowed here
+ impl_->requested_.store(signum);
+}
+
+StopToken StopSource::token() { return StopToken(impl_); }
+
+bool StopToken::IsStopRequested() {
+ if (!impl_) {
+ return false;
+ }
+ return impl_->requested_.load() != 0;
+}
+
+Status StopToken::Poll() {
+ if (!impl_) {
+ return Status::OK();
+ }
+ if (!impl_->requested_.load()) {
+ return Status::OK();
+ }
+
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ if (impl_->cancel_error_.ok()) {
+ auto signum = impl_->requested_.load();
+ DCHECK_GT(signum, 0);
+ impl_->cancel_error_ = internal::CancelledFromSignal(signum, "Operation cancelled");
+ }
Review comment:
```suggestion
} else {
DCHECK_EQ(impl_->requested_.load(), -1);
}
```
##########
File path: cpp/src/arrow/util/task_group.cc
##########
@@ -67,34 +75,54 @@ class SerialTaskGroup : public TaskGroup {
class ThreadedTaskGroup : public TaskGroup {
public:
- explicit ThreadedTaskGroup(Executor* executor)
- : executor_(executor), nremaining_(0), ok_(true) {}
+ explicit ThreadedTaskGroup(Executor* executor, StopToken stop_token)
Review comment:
```suggestion
ThreadedTaskGroup(Executor* executor, StopToken stop_token)
```
##########
File path: cpp/src/arrow/util/cancel_test.cc
##########
@@ -0,0 +1,301 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <atomic>
+#include <cmath>
+#include <sstream>
+#include <string>
+#include <thread>
+#include <utility>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+#include <signal.h>
+#ifndef _WIN32
+#include <sys/time.h> // for setitimer()
+#endif
+
+#include "arrow/testing/gtest_util.h"
+#include "arrow/util/cancel.h"
+#include "arrow/util/future.h"
+#include "arrow/util/io_util.h"
+#include "arrow/util/logging.h"
+#include "arrow/util/optional.h"
+
+namespace arrow {
+
+class CancelTest : public ::testing::Test {};
+
+TEST_F(CancelTest, StopBasics) {
+ {
+ StopSource source;
+ StopToken token = source.token();
+ ASSERT_FALSE(token.IsStopRequested());
+ ASSERT_OK(token.Poll());
+
+ source.RequestStop();
+ ASSERT_TRUE(token.IsStopRequested());
+ ASSERT_RAISES(Cancelled, token.Poll());
+ }
+ {
+ StopSource source;
+ StopToken token = source.token();
+ source.RequestStop(Status::IOError("Operation cancelled"));
+ ASSERT_TRUE(token.IsStopRequested());
+ ASSERT_RAISES(IOError, token.Poll());
+ }
+}
+
+TEST_F(CancelTest, StopTokenCopy) {
+ StopSource source;
+ StopToken token = source.token();
+ ASSERT_FALSE(token.IsStopRequested());
+ ASSERT_OK(token.Poll());
+
+ source.RequestStop();
+ ASSERT_TRUE(token.IsStopRequested());
+ ASSERT_RAISES(Cancelled, token.Poll());
+
+ StopToken new_token = token;
+ ASSERT_TRUE(new_token.IsStopRequested());
+ ASSERT_RAISES(Cancelled, new_token.Poll());
+}
+
+TEST_F(CancelTest, RequestStopTwice) {
+ StopSource source;
+ StopToken token = source.token();
+ source.RequestStop();
+ // Second RequestStop() call is ignored
+ source.RequestStop(Status::IOError("Operation cancelled"));
+ ASSERT_TRUE(token.IsStopRequested());
+ ASSERT_RAISES(Cancelled, token.Poll());
+}
+
+TEST_F(CancelTest, Unstoppable) {
+ StopToken token = StopToken::Unstoppable();
+ ASSERT_FALSE(token.IsStopRequested());
+ ASSERT_OK(token.Poll());
+}
+
+TEST_F(CancelTest, SourceVanishes) {
+ {
+ util::optional<StopSource> source{StopSource()};
+ StopToken token = source->token();
+ ASSERT_FALSE(token.IsStopRequested());
+ ASSERT_OK(token.Poll());
+
+ source.reset();
+ ASSERT_FALSE(token.IsStopRequested());
+ ASSERT_OK(token.Poll());
+ }
+ {
+ util::optional<StopSource> source{StopSource()};
+ StopToken token = source->token();
+ source->RequestStop();
+
+ source.reset();
+ ASSERT_TRUE(token.IsStopRequested());
+ ASSERT_RAISES(Cancelled, token.Poll());
+ }
+}
+
+static void noop_signal_handler(int signum) {
+ internal::ReinstateSignalHandler(signum, &noop_signal_handler);
+}
+
+#ifndef _WIN32
+static util::optional<StopSource> signal_stop_source;
+
+static void signal_handler(int signum) {
+ signal_stop_source->RequestStopFromSignal(signum);
+}
+
+static void SetITimer(double seconds) {
Review comment:
```suggestion
// SIGALRM will be received once after the specified wait
static void SetITimer(double seconds) {
```
##########
File path: cpp/src/arrow/util/task_group_test.cc
##########
@@ -114,6 +114,49 @@ void TestTaskGroupErrors(std::shared_ptr<TaskGroup> task_group) {
ASSERT_RAISES(Invalid, task_group->Finish());
}
+void TestTaskGroupCancel(std::shared_ptr<TaskGroup> task_group, StopSource* stop_source) {
+ const int NSUCCESSES = 2;
+ const int NCANCELS = 20;
+
+ std::atomic<int> count(0);
+
+ auto task_group_was_ok = false;
+ task_group->Append([&]() -> Status {
+ for (int i = 0; i < NSUCCESSES; ++i) {
+ task_group->Append([&]() {
+ count++;
+ return Status::OK();
+ });
+ }
+ task_group_was_ok = task_group->ok();
+ for (int i = 0; i < NCANCELS; ++i) {
+ task_group->Append([&]() {
+ SleepFor(1e-2);
+ stop_source->RequestStop();
+ count++;
+ return Status::OK();
+ });
+ }
+
+ return Status::OK();
+ });
+
+ // Cancellation is propagated
+ ASSERT_RAISES(Cancelled, task_group->Finish());
+ ASSERT_TRUE(task_group_was_ok);
+ ASSERT_FALSE(task_group->ok());
+ if (task_group->parallelism() == 1) {
+ // Serial: exactly three successes
+ ASSERT_EQ(count.load(), 3);
+ } else {
+ // Parallel: at least three successes
+ ASSERT_GE(count.load(), 3);
+ ASSERT_LE(count.load(), 2 * task_group->parallelism());
Review comment:
```suggestion
ASSERT_LE(count.load(), NSUCCESSES + task_group->parallelism());
```
##########
File path: cpp/src/arrow/util/cancel.cc
##########
@@ -0,0 +1,167 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/util/cancel.h"
+
+#include <atomic>
+#include <mutex>
+#include <sstream>
+#include <utility>
+
+#include "arrow/result.h"
+#include "arrow/util/atomic_shared_ptr.h"
+#include "arrow/util/io_util.h"
+#include "arrow/util/logging.h"
+#include "arrow/util/visibility.h"
+
+namespace arrow {
+
+#if ATOMIC_INT_LOCK_FREE != 2
+#error Lock-free atomic int required for signal safety
+#endif
+
+using internal::ReinstateSignalHandler;
+using internal::SetSignalHandler;
+using internal::SignalHandler;
+
+// NOTE: We care mainly about the making the common case (not cancelled) fast.
+
+struct StopSourceImpl {
+ std::atomic<int> requested_{0}; // will be -1 or signal number if requested
+ std::mutex mutex_;
+ Status cancel_error_;
+};
+
+StopSource::StopSource() : impl_(new StopSourceImpl) {}
+
+StopSource::~StopSource() = default;
+
+void StopSource::RequestStop() { RequestStop(Status::Cancelled("Operation cancelled")); }
+
+void StopSource::RequestStop(Status st) {
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ DCHECK(!st.ok());
+ if (!impl_->requested_) {
+ impl_->requested_ = -1;
+ impl_->cancel_error_ = std::move(st);
+ }
+}
+
+void StopSource::RequestStopFromSignal(int signum) {
+ // Only async-signal-safe code allowed here
+ impl_->requested_.store(signum);
+}
+
+StopToken StopSource::token() { return StopToken(impl_); }
+
+bool StopToken::IsStopRequested() {
+ if (!impl_) {
+ return false;
+ }
+ return impl_->requested_.load() != 0;
+}
+
+Status StopToken::Poll() {
+ if (!impl_) {
+ return Status::OK();
+ }
+ if (!impl_->requested_.load()) {
+ return Status::OK();
+ }
+
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ if (impl_->cancel_error_.ok()) {
+ auto signum = impl_->requested_.load();
+ DCHECK_GT(signum, 0);
+ impl_->cancel_error_ = internal::CancelledFromSignal(signum, "Operation cancelled");
+ }
+ return impl_->cancel_error_;
+}
+
+namespace {
+
+void HandleSignal(int signum);
+
+struct SignalStopState {
+ struct SavedSignalHandler {
+ int signum;
+ SignalHandler handler;
+ };
+
+ Status RegisterHandlers(const std::vector<int>& signals) {
+ if (!saved_handlers.empty()) {
+ return Status::Invalid("Signal handlers already registered");
+ }
+ for (int signum : signals) {
+ ARROW_ASSIGN_OR_RAISE(auto handler,
+ SetSignalHandler(signum, SignalHandler{&HandleSignal}));
+ saved_handlers.push_back({signum, handler});
+ }
+ return Status::OK();
+ }
+
+ void UnregisterHandlers() {
+ auto handlers = std::move(saved_handlers);
+ for (const auto& h : handlers) {
+ ARROW_CHECK_OK(SetSignalHandler(h.signum, h.handler).status());
+ }
+ }
+
+ ~SignalStopState() { UnregisterHandlers(); }
+
+ StopSource stop_source;
+ std::vector<SavedSignalHandler> saved_handlers;
+};
+
+std::shared_ptr<SignalStopState> g_signal_stop_state;
+
+void HandleSignal(int signum) {
+ ReinstateSignalHandler(signum, &HandleSignal);
+ std::shared_ptr<SignalStopState> state = internal::atomic_load(&g_signal_stop_state);
+ if (state) {
+ state->stop_source.RequestStopFromSignal(signum);
+ }
+}
+
+} // namespace
+
+Result<StopSource*> SetSignalStopSource() {
+ if (g_signal_stop_state) {
+ return Status::Invalid("Signal stop source already set up");
+ }
+ internal::atomic_store(&g_signal_stop_state, std::make_shared<SignalStopState>());
+ return &g_signal_stop_state->stop_source;
+}
+
+void ResetSignalStopSource() {
+ internal::atomic_store(&g_signal_stop_state, std::shared_ptr<SignalStopState>{});
Review comment:
Is this intended to be idempotent? If not,
```suggestion
DCHECK_NE(g_signal_stop_state, nullptr);
internal::atomic_store(&g_signal_stop_state, std::shared_ptr<SignalStopState>{});
```
##########
File path: cpp/src/arrow/util/thread_pool_test.cc
##########
@@ -137,30 +136,47 @@ class TestThreadPool : public ::testing::Test {
return *ThreadPool::Make(threads);
}
- void SpawnAdds(ThreadPool* pool, int nadds, AddTaskFunc add_func) {
- AddTester add_tester(nadds);
+ void SpawnAdds(ThreadPool* pool, int nadds, AddTaskFunc add_func,
+ StopToken stop_token = StopToken::Unstoppable(),
+ StopSource* stop_source = nullptr) {
Review comment:
```suggestion
void SpawnAdds(ThreadPool* pool, int nadds, AddTaskFunc add_func,
StopToken stop_token = StopToken::Unstoppable()) {
SpawnAddsImpl(pool, nadds, add_func, stop_token, nullptr);
}
void SpawnAdds(ThreadPool* pool, int nadds, AddTaskFunc add_func,
StopSource* stop_source) {
SpawnAddsImpl(pool, nadds, add_func, stop_source->token(), stop_source);
}
void SpawnAddsImpl(ThreadPool* pool, int nadds, AddTaskFunc add_func,
StopToken stop_token, StopSource* stop_source) {
```
I'm not sure this is correct for all the cases below; there are a few which use
a StopSource's token but don't pass the StopSource.
If this *isn't* correct, I think that it'd be best to separate the may-be-cancelled
and the won't-be-cancelled cases into separate overloads of SpawnAdds, with
comments clarifying the intent of each.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#issuecomment-793076144
Hmm, but resetting the pointer may still deallocate the underlying object... which is certainly not async-signal-safe.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#issuecomment-792931920
I pushed an improvement for the deallocation in signal handler problem, but it's still not 100% safe. I think solving it entirely would need a lock-free doubly-linked list.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
westonpace commented on pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#issuecomment-793085784
Yep. I think what you have, with the comment, is fine.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API (v2)
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r578685677
##########
File path: cpp/src/arrow/util/future.h
##########
@@ -515,15 +548,35 @@ class ARROW_MUST_USE_TYPE Future {
}
void DoMarkFinished(Result<ValueType> res) {
- SetResult(std::move(res));
-
- if (ARROW_PREDICT_TRUE(GetResult()->ok())) {
- impl_->MarkFinished();
+ const FutureState state = StateForResult(res);
+ if (impl_->stop_callback_.has_value()) {
+ struct ResultSetter {
+ Future<T>* fut;
+ Result<ValueType> res;
+
+ void operator()() { fut->SetResult(std::move(res)); }
+ };
+ impl_->MarkFinished(state, ResultSetter{this, std::move(res)});
} else {
- impl_->MarkFailed();
+ // The future cannot be cancelled concurrently as CancelOn()
+ // was not called.
+ SetResult(std::move(res));
+ impl_->MarkFinished(state);
}
Review comment:
Unfortunately, having the ability to cancel from one thread while marking a future finished from another thread introduces some execution costs inside the Future class. Here, we see that if some cancellation may occur, we must synchronize result writing inside `FutureImpl::MarkFinished`.
cc @westonpace
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r587693382
##########
File path: cpp/src/arrow/csv/reader.cc
##########
@@ -833,9 +843,10 @@ class AsyncThreadedTableReader
AsyncThreadedTableReader(MemoryPool* pool, std::shared_ptr<io::InputStream> input,
const ReadOptions& read_options,
const ParseOptions& parse_options,
- const ConvertOptions& convert_options, Executor* cpu_executor,
- Executor* io_executor)
- : BaseTableReader(pool, input, read_options, parse_options, convert_options),
+ const ConvertOptions& convert_options, StopToken stop_token,
+ Executor* cpu_executor, Executor* io_executor)
Review comment:
We must support the legacy `TableReader::Make` taking both a `MemoryPool*` and an `IOContext`.
----------------------------------------------------------------
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 removed a comment on pull request #9528: ARROW-8732: [C++] Add basic cancellation API (v2)
Posted by GitBox <gi...@apache.org>.
pitrou removed a comment on pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#issuecomment-782057195
Hmm, perhaps instead of complicating the Future implementation, it would be better to make the Executor responsible for setting the stop callback.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API (v2)
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r580484289
##########
File path: cpp/src/arrow/util/future.h
##########
@@ -515,15 +548,35 @@ class ARROW_MUST_USE_TYPE Future {
}
void DoMarkFinished(Result<ValueType> res) {
- SetResult(std::move(res));
-
- if (ARROW_PREDICT_TRUE(GetResult()->ok())) {
- impl_->MarkFinished();
+ const FutureState state = StateForResult(res);
+ if (impl_->stop_callback_.has_value()) {
+ struct ResultSetter {
+ Future<T>* fut;
+ Result<ValueType> res;
+
+ void operator()() { fut->SetResult(std::move(res)); }
+ };
+ impl_->MarkFinished(state, ResultSetter{this, std::move(res)});
} else {
- impl_->MarkFailed();
+ // The future cannot be cancelled concurrently as CancelOn()
+ // was not called.
+ SetResult(std::move(res));
+ impl_->MarkFinished(state);
}
Review comment:
Ok, I removed this and other changes to `future.h`.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
westonpace commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r588547359
##########
File path: cpp/src/arrow/csv/reader.cc
##########
@@ -934,22 +946,34 @@ class AsyncThreadedTableReader
Result<std::shared_ptr<TableReader>> MakeTableReader(
MemoryPool* pool, io::IOContext io_context, std::shared_ptr<io::InputStream> input,
const ReadOptions& read_options, const ParseOptions& parse_options,
- const ConvertOptions& convert_options) {
+ const ConvertOptions& convert_options, StopToken stop_token) {
Review comment:
Why is the stop_token passed in explicitly here and it is grabbed from the `io_context` in `MakeStreamingReader`? Ah, I see it is grabbed out of the `io_context` earlier but it could be pulled out here and save an extra parameter.
##########
File path: cpp/src/arrow/testing/gtest_util.cc
##########
@@ -576,6 +594,13 @@ void SleepFor(double seconds) {
std::chrono::nanoseconds(static_cast<int64_t>(seconds * 1e9)));
}
+void BusyWait(double seconds, std::function<bool()> predicate) {
Review comment:
Nice. I was just thinking of writing something like this.
##########
File path: cpp/src/arrow/csv/reader.cc
##########
@@ -848,6 +859,7 @@ class AsyncThreadedTableReader
}
Status Init() override {
+ // TODO pass stop token to input stream iterator?
Review comment:
For an iterator, no. In the future, if this becomes a generator, possibly. The only way it could be useful is if the I/O could be cancelled somehow. Generally that's not possible. With some non-blocking I/O schemes you can at least give up on the user-land side of the I/O. For networked I/O it may be possible but I seem to recall you saying S3 had no such mechanism.
You could maybe make use of the stop token in the readahead to stop the readahead but that doesn't seem too urgent. It will fill up the readahead queue, and then, there will be no active references, and it will all be cleaned up. Although it may be nice to add a unit test for that scenario. I wonder if I could get a consumer side reference count of some kind and abort it as soon as all consumer references are lost :thinking: . I'll add a JIRA for it.
##########
File path: cpp/src/arrow/util/cancel.h
##########
@@ -0,0 +1,99 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <functional>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "arrow/status.h"
+#include "arrow/type_fwd.h"
+#include "arrow/util/macros.h"
+#include "arrow/util/visibility.h"
+
+namespace arrow {
+
+class StopToken;
+
+struct StopSourceImpl;
+
+/// EXPERIMENTAL
+class ARROW_EXPORT StopSource {
+ public:
+ StopSource();
+ ~StopSource();
+
+ // Consumer API (the side that stops)
+ void RequestStop();
+ void RequestStop(Status error);
+ void RequestStopFromSignal(int signum);
Review comment:
Seems unfortunate this has to be public. Is it possible to make it private and friend it?
Nevermind, I see that this could be useful if a user wants fine-grained control over stopping in which case they might need to register their own signal handlers (since they can't use our signal stop source if they are using their own stop source).
##########
File path: cpp/src/arrow/util/cancel.cc
##########
@@ -0,0 +1,167 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/util/cancel.h"
+
+#include <atomic>
+#include <mutex>
+#include <sstream>
+#include <utility>
+
+#include "arrow/result.h"
+#include "arrow/util/atomic_shared_ptr.h"
+#include "arrow/util/io_util.h"
+#include "arrow/util/logging.h"
+#include "arrow/util/visibility.h"
+
+namespace arrow {
+
+#if ATOMIC_INT_LOCK_FREE != 2
+#error Lock-free atomic int required for signal safety
+#endif
+
+using internal::ReinstateSignalHandler;
+using internal::SetSignalHandler;
+using internal::SignalHandler;
+
+// NOTE: We care mainly about the making the common case (not cancelled) fast.
+
+struct StopSourceImpl {
+ std::atomic<int> requested_{0}; // will be -1 or signal number if requested
+ std::mutex mutex_;
+ Status cancel_error_;
+};
+
+StopSource::StopSource() : impl_(new StopSourceImpl) {}
+
+StopSource::~StopSource() = default;
+
+void StopSource::RequestStop() { RequestStop(Status::Cancelled("Operation cancelled")); }
+
+void StopSource::RequestStop(Status st) {
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ DCHECK(!st.ok());
+ if (!impl_->requested_) {
+ impl_->requested_ = -1;
+ impl_->cancel_error_ = std::move(st);
+ }
+}
+
+void StopSource::RequestStopFromSignal(int signum) {
+ // Only async-signal-safe code allowed here
+ impl_->requested_.store(signum);
Review comment:
I think you can get rid of this mutex. If you changed this to something like...
```
if (!impl_->requested_.fetch_or(-1)) {
impl_->signum = signum;
}
```
`signum` would be a new (non-atomic) int member that replaces the mutex. Then `RequestStop` becomes...
```
if(!impl->requested_.fetch_or(-1)) {
impl->cancel_error_ = std::move(st);
}
```
The main advantage is there would be no way to lose your thread when calling `Poll`.
##########
File path: python/pyarrow/_csv.pyx
##########
@@ -34,8 +34,8 @@ from pyarrow.lib cimport (check_status, Field, MemoryPool, Schema,
pyarrow_unwrap_batch, pyarrow_unwrap_table,
pyarrow_wrap_schema, pyarrow_wrap_table,
pyarrow_wrap_data_type, pyarrow_unwrap_data_type,
- Table, RecordBatch)
-from pyarrow.lib import frombytes, tobytes
+ Table, RecordBatch, StopToken)
Review comment:
Is there a JIRA to expose `StopSource` to python at some point? There are reasons other than signals someone may want to cancel an operation. For example, a GUI-based application may have a cancel button. A web server may want to cancel if the TCP connection for some analysis request is lost.
##########
File path: cpp/src/arrow/util/cancel.cc
##########
@@ -0,0 +1,167 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/util/cancel.h"
+
+#include <atomic>
+#include <mutex>
+#include <sstream>
+#include <utility>
+
+#include "arrow/result.h"
+#include "arrow/util/atomic_shared_ptr.h"
+#include "arrow/util/io_util.h"
+#include "arrow/util/logging.h"
+#include "arrow/util/visibility.h"
+
+namespace arrow {
+
+#if ATOMIC_INT_LOCK_FREE != 2
+#error Lock-free atomic int required for signal safety
+#endif
+
+using internal::ReinstateSignalHandler;
+using internal::SetSignalHandler;
+using internal::SignalHandler;
+
+// NOTE: We care mainly about the making the common case (not cancelled) fast.
+
+struct StopSourceImpl {
+ std::atomic<int> requested_{0}; // will be -1 or signal number if requested
+ std::mutex mutex_;
+ Status cancel_error_;
+};
+
+StopSource::StopSource() : impl_(new StopSourceImpl) {}
+
+StopSource::~StopSource() = default;
+
+void StopSource::RequestStop() { RequestStop(Status::Cancelled("Operation cancelled")); }
+
+void StopSource::RequestStop(Status st) {
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ DCHECK(!st.ok());
+ if (!impl_->requested_) {
+ impl_->requested_ = -1;
+ impl_->cancel_error_ = std::move(st);
+ }
+}
+
+void StopSource::RequestStopFromSignal(int signum) {
+ // Only async-signal-safe code allowed here
+ impl_->requested_.store(signum);
+}
+
+StopToken StopSource::token() { return StopToken(impl_); }
+
+bool StopToken::IsStopRequested() {
+ if (!impl_) {
+ return false;
+ }
+ return impl_->requested_.load() != 0;
+}
+
+Status StopToken::Poll() {
+ if (!impl_) {
+ return Status::OK();
+ }
+ if (!impl_->requested_.load()) {
+ return Status::OK();
+ }
+
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ if (impl_->cancel_error_.ok()) {
+ auto signum = impl_->requested_.load();
+ DCHECK_GT(signum, 0);
+ impl_->cancel_error_ = internal::CancelledFromSignal(signum, "Operation cancelled");
+ }
+ return impl_->cancel_error_;
+}
+
+namespace {
+
+void HandleSignal(int signum);
+
+struct SignalStopState {
+ struct SavedSignalHandler {
+ int signum;
+ SignalHandler handler;
+ };
+
+ Status RegisterHandlers(const std::vector<int>& signals) {
+ if (!saved_handlers.empty()) {
+ return Status::Invalid("Signal handlers already registered");
+ }
+ for (int signum : signals) {
+ ARROW_ASSIGN_OR_RAISE(auto handler,
+ SetSignalHandler(signum, SignalHandler{&HandleSignal}));
+ saved_handlers.push_back({signum, handler});
+ }
+ return Status::OK();
+ }
+
+ void UnregisterHandlers() {
+ auto handlers = std::move(saved_handlers);
+ for (const auto& h : handlers) {
+ ARROW_CHECK_OK(SetSignalHandler(h.signum, h.handler).status());
+ }
+ }
+
+ ~SignalStopState() { UnregisterHandlers(); }
+
+ StopSource stop_source;
+ std::vector<SavedSignalHandler> saved_handlers;
+};
+
+std::shared_ptr<SignalStopState> g_signal_stop_state;
+
+void HandleSignal(int signum) {
+ ReinstateSignalHandler(signum, &HandleSignal);
+ std::shared_ptr<SignalStopState> state = internal::atomic_load(&g_signal_stop_state);
Review comment:
If someone calls `ResetSignalStopSource` while you are handling the signal wouldn't this become the sole reference and, on exit, attempt to destroy it (which would not be async-signal-safe)?
Admittedly, probably not a common occurrence.
##########
File path: cpp/src/arrow/util/cancel_test.cc
##########
@@ -0,0 +1,302 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <atomic>
+#include <cmath>
+#include <sstream>
+#include <string>
+#include <thread>
+#include <utility>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+#include <signal.h>
+#ifndef _WIN32
+#include <sys/time.h> // for setitimer()
+#endif
+
+#include "arrow/testing/gtest_util.h"
+#include "arrow/util/cancel.h"
+#include "arrow/util/future.h"
+#include "arrow/util/io_util.h"
+#include "arrow/util/logging.h"
+#include "arrow/util/optional.h"
+
+namespace arrow {
+
+class CancelTest : public ::testing::Test {};
+
+TEST_F(CancelTest, StopBasics) {
+ {
+ StopSource source;
+ StopToken token = source.token();
+ ASSERT_FALSE(token.IsStopRequested());
+ ASSERT_OK(token.Poll());
+
+ source.RequestStop();
+ ASSERT_TRUE(token.IsStopRequested());
+ ASSERT_RAISES(Cancelled, token.Poll());
+ }
+ {
+ StopSource source;
+ StopToken token = source.token();
+ source.RequestStop(Status::IOError("Operation cancelled"));
+ ASSERT_TRUE(token.IsStopRequested());
+ ASSERT_RAISES(IOError, token.Poll());
+ }
+}
+
+TEST_F(CancelTest, StopTokenCopy) {
+ StopSource source;
+ StopToken token = source.token();
+ ASSERT_FALSE(token.IsStopRequested());
+ ASSERT_OK(token.Poll());
+
+ source.RequestStop();
+ ASSERT_TRUE(token.IsStopRequested());
+ ASSERT_RAISES(Cancelled, token.Poll());
+
+ StopToken new_token = token;
Review comment:
Minor nit, but maybe also test the case where you copy the token before stopping.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r587691144
##########
File path: cpp/src/arrow/util/cancel.cc
##########
@@ -0,0 +1,167 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/util/cancel.h"
+
+#include <atomic>
+#include <mutex>
+#include <sstream>
+#include <utility>
+
+#include "arrow/result.h"
+#include "arrow/util/atomic_shared_ptr.h"
+#include "arrow/util/io_util.h"
+#include "arrow/util/logging.h"
+#include "arrow/util/visibility.h"
+
+namespace arrow {
+
+#if ATOMIC_INT_LOCK_FREE != 2
+#error Lock-free atomic int required for signal safety
+#endif
+
+using internal::ReinstateSignalHandler;
+using internal::SetSignalHandler;
+using internal::SignalHandler;
+
+// NOTE: We care mainly about the making the common case (not cancelled) fast.
+
+struct StopSourceImpl {
+ std::atomic<int> requested_{0}; // will be -1 or signal number if requested
+ std::mutex mutex_;
+ Status cancel_error_;
+};
+
+StopSource::StopSource() : impl_(new StopSourceImpl) {}
+
+StopSource::~StopSource() = default;
+
+void StopSource::RequestStop() { RequestStop(Status::Cancelled("Operation cancelled")); }
+
+void StopSource::RequestStop(Status st) {
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ DCHECK(!st.ok());
+ if (!impl_->requested_) {
+ impl_->requested_ = -1;
+ impl_->cancel_error_ = std::move(st);
+ }
+}
+
+void StopSource::RequestStopFromSignal(int signum) {
+ // Only async-signal-safe code allowed here
+ impl_->requested_.store(signum);
+}
+
+StopToken StopSource::token() { return StopToken(impl_); }
+
+bool StopToken::IsStopRequested() {
+ if (!impl_) {
+ return false;
+ }
+ return impl_->requested_.load() != 0;
+}
+
+Status StopToken::Poll() {
+ if (!impl_) {
+ return Status::OK();
+ }
+ if (!impl_->requested_.load()) {
+ return Status::OK();
+ }
+
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ if (impl_->cancel_error_.ok()) {
+ auto signum = impl_->requested_.load();
+ DCHECK_GT(signum, 0);
+ impl_->cancel_error_ = internal::CancelledFromSignal(signum, "Operation cancelled");
+ }
Review comment:
If you call Poll() twice, then the cancel_error corresponding to the signal will have been cached. So you can have both a cancel_error and a positive signal number.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#issuecomment-789996441
@github-actions crossbow submit -g python
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r589475289
##########
File path: cpp/src/arrow/csv/reader.cc
##########
@@ -934,22 +946,34 @@ class AsyncThreadedTableReader
Result<std::shared_ptr<TableReader>> MakeTableReader(
MemoryPool* pool, io::IOContext io_context, std::shared_ptr<io::InputStream> input,
const ReadOptions& read_options, const ParseOptions& parse_options,
- const ConvertOptions& convert_options) {
+ const ConvertOptions& convert_options, StopToken stop_token) {
Review comment:
Agreed, I didn't think too much about this. Will change.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r587690293
##########
File path: cpp/src/arrow/util/thread_pool_test.cc
##########
@@ -137,30 +136,47 @@ class TestThreadPool : public ::testing::Test {
return *ThreadPool::Make(threads);
}
- void SpawnAdds(ThreadPool* pool, int nadds, AddTaskFunc add_func) {
- AddTester add_tester(nadds);
+ void SpawnAdds(ThreadPool* pool, int nadds, AddTaskFunc add_func,
+ StopToken stop_token = StopToken::Unstoppable(),
+ StopSource* stop_source = nullptr) {
Review comment:
Passing the token without the source tests a source that's not stopped, so it's deliberate. I could try to add overloads though.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r587690293
##########
File path: cpp/src/arrow/util/thread_pool_test.cc
##########
@@ -137,30 +136,47 @@ class TestThreadPool : public ::testing::Test {
return *ThreadPool::Make(threads);
}
- void SpawnAdds(ThreadPool* pool, int nadds, AddTaskFunc add_func) {
- AddTester add_tester(nadds);
+ void SpawnAdds(ThreadPool* pool, int nadds, AddTaskFunc add_func,
+ StopToken stop_token = StopToken::Unstoppable(),
+ StopSource* stop_source = nullptr) {
Review comment:
Passing the token with not the source tests a source that's not stopped, so it's deliberate. I could try to add overloads though.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r589478493
##########
File path: cpp/src/arrow/util/cancel.cc
##########
@@ -0,0 +1,167 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/util/cancel.h"
+
+#include <atomic>
+#include <mutex>
+#include <sstream>
+#include <utility>
+
+#include "arrow/result.h"
+#include "arrow/util/atomic_shared_ptr.h"
+#include "arrow/util/io_util.h"
+#include "arrow/util/logging.h"
+#include "arrow/util/visibility.h"
+
+namespace arrow {
+
+#if ATOMIC_INT_LOCK_FREE != 2
+#error Lock-free atomic int required for signal safety
+#endif
+
+using internal::ReinstateSignalHandler;
+using internal::SetSignalHandler;
+using internal::SignalHandler;
+
+// NOTE: We care mainly about the making the common case (not cancelled) fast.
+
+struct StopSourceImpl {
+ std::atomic<int> requested_{0}; // will be -1 or signal number if requested
+ std::mutex mutex_;
+ Status cancel_error_;
+};
+
+StopSource::StopSource() : impl_(new StopSourceImpl) {}
+
+StopSource::~StopSource() = default;
+
+void StopSource::RequestStop() { RequestStop(Status::Cancelled("Operation cancelled")); }
+
+void StopSource::RequestStop(Status st) {
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ DCHECK(!st.ok());
+ if (!impl_->requested_) {
+ impl_->requested_ = -1;
+ impl_->cancel_error_ = std::move(st);
+ }
+}
+
+void StopSource::RequestStopFromSignal(int signum) {
+ // Only async-signal-safe code allowed here
+ impl_->requested_.store(signum);
+}
+
+StopToken StopSource::token() { return StopToken(impl_); }
+
+bool StopToken::IsStopRequested() {
+ if (!impl_) {
+ return false;
+ }
+ return impl_->requested_.load() != 0;
+}
+
+Status StopToken::Poll() {
+ if (!impl_) {
+ return Status::OK();
+ }
+ if (!impl_->requested_.load()) {
+ return Status::OK();
+ }
+
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ if (impl_->cancel_error_.ok()) {
+ auto signum = impl_->requested_.load();
+ DCHECK_GT(signum, 0);
+ impl_->cancel_error_ = internal::CancelledFromSignal(signum, "Operation cancelled");
+ }
+ return impl_->cancel_error_;
+}
+
+namespace {
+
+void HandleSignal(int signum);
+
+struct SignalStopState {
+ struct SavedSignalHandler {
+ int signum;
+ SignalHandler handler;
+ };
+
+ Status RegisterHandlers(const std::vector<int>& signals) {
+ if (!saved_handlers.empty()) {
+ return Status::Invalid("Signal handlers already registered");
+ }
+ for (int signum : signals) {
+ ARROW_ASSIGN_OR_RAISE(auto handler,
+ SetSignalHandler(signum, SignalHandler{&HandleSignal}));
+ saved_handlers.push_back({signum, handler});
+ }
+ return Status::OK();
+ }
+
+ void UnregisterHandlers() {
+ auto handlers = std::move(saved_handlers);
+ for (const auto& h : handlers) {
+ ARROW_CHECK_OK(SetSignalHandler(h.signum, h.handler).status());
+ }
+ }
+
+ ~SignalStopState() { UnregisterHandlers(); }
+
+ StopSource stop_source;
+ std::vector<SavedSignalHandler> saved_handlers;
+};
+
+std::shared_ptr<SignalStopState> g_signal_stop_state;
+
+void HandleSignal(int signum) {
+ ReinstateSignalHandler(signum, &HandleSignal);
+ std::shared_ptr<SignalStopState> state = internal::atomic_load(&g_signal_stop_state);
Review comment:
Ouch, I think you're right. I'll try to find a way to do this differently...
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r589536904
##########
File path: cpp/src/arrow/util/cancel_test.cc
##########
@@ -0,0 +1,302 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <atomic>
+#include <cmath>
+#include <sstream>
+#include <string>
+#include <thread>
+#include <utility>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+#include <signal.h>
+#ifndef _WIN32
+#include <sys/time.h> // for setitimer()
+#endif
+
+#include "arrow/testing/gtest_util.h"
+#include "arrow/util/cancel.h"
+#include "arrow/util/future.h"
+#include "arrow/util/io_util.h"
+#include "arrow/util/logging.h"
+#include "arrow/util/optional.h"
+
+namespace arrow {
+
+class CancelTest : public ::testing::Test {};
+
+TEST_F(CancelTest, StopBasics) {
+ {
+ StopSource source;
+ StopToken token = source.token();
+ ASSERT_FALSE(token.IsStopRequested());
+ ASSERT_OK(token.Poll());
+
+ source.RequestStop();
+ ASSERT_TRUE(token.IsStopRequested());
+ ASSERT_RAISES(Cancelled, token.Poll());
+ }
+ {
+ StopSource source;
+ StopToken token = source.token();
+ source.RequestStop(Status::IOError("Operation cancelled"));
+ ASSERT_TRUE(token.IsStopRequested());
+ ASSERT_RAISES(IOError, token.Poll());
+ }
+}
+
+TEST_F(CancelTest, StopTokenCopy) {
+ StopSource source;
+ StopToken token = source.token();
+ ASSERT_FALSE(token.IsStopRequested());
+ ASSERT_OK(token.Poll());
+
+ source.RequestStop();
+ ASSERT_TRUE(token.IsStopRequested());
+ ASSERT_RAISES(Cancelled, token.Poll());
+
+ StopToken new_token = token;
Review comment:
Will do.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#issuecomment-793080179
Ah, I see: no need to take a strong reference inside the handler since the spinlock guarantees the global reference isn't mutated. That said, the signal-safe spinlock apparently requires one to block signals.
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r589476690
##########
File path: cpp/src/arrow/util/cancel.cc
##########
@@ -0,0 +1,167 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/util/cancel.h"
+
+#include <atomic>
+#include <mutex>
+#include <sstream>
+#include <utility>
+
+#include "arrow/result.h"
+#include "arrow/util/atomic_shared_ptr.h"
+#include "arrow/util/io_util.h"
+#include "arrow/util/logging.h"
+#include "arrow/util/visibility.h"
+
+namespace arrow {
+
+#if ATOMIC_INT_LOCK_FREE != 2
+#error Lock-free atomic int required for signal safety
+#endif
+
+using internal::ReinstateSignalHandler;
+using internal::SetSignalHandler;
+using internal::SignalHandler;
+
+// NOTE: We care mainly about the making the common case (not cancelled) fast.
+
+struct StopSourceImpl {
+ std::atomic<int> requested_{0}; // will be -1 or signal number if requested
+ std::mutex mutex_;
+ Status cancel_error_;
+};
+
+StopSource::StopSource() : impl_(new StopSourceImpl) {}
+
+StopSource::~StopSource() = default;
+
+void StopSource::RequestStop() { RequestStop(Status::Cancelled("Operation cancelled")); }
+
+void StopSource::RequestStop(Status st) {
+ std::lock_guard<std::mutex> lock(impl_->mutex_);
+ DCHECK(!st.ok());
+ if (!impl_->requested_) {
+ impl_->requested_ = -1;
+ impl_->cancel_error_ = std::move(st);
+ }
+}
+
+void StopSource::RequestStopFromSignal(int signum) {
+ // Only async-signal-safe code allowed here
+ impl_->requested_.store(signum);
Review comment:
No, we need the mutex to protect `cancel_error_` (which we cannot store atomically).
----------------------------------------------------------------
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 #9528: ARROW-8732: [C++] Add basic cancellation API
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #9528:
URL: https://github.com/apache/arrow/pull/9528#discussion_r589473437
##########
File path: cpp/src/arrow/csv/reader.cc
##########
@@ -848,6 +859,7 @@ class AsyncThreadedTableReader
}
Status Init() override {
+ // TODO pass stop token to input stream iterator?
Review comment:
Indeed, the AWS SDK doesn't support cancellation.
Agreed about the readahead part. This can be done later. Thanks!
----------------------------------------------------------------
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