pitrou commented on a change in pull request #10968: URL: https://github.com/apache/arrow/pull/10968#discussion_r698481775
########## File path: cpp/src/arrow/util/async_nursery.h ########## @@ -0,0 +1,154 @@ +// 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 <list> + +#include "arrow/result.h" +#include "arrow/status.h" +#include "arrow/util/future.h" +#include "arrow/util/mutex.h" + +namespace arrow { +namespace util { + +class Nursery; +class AsyncCloseablePimpl; + +template <typename T> +struct DestroyingDeleter { + void operator()(T* p) { p->Destroy(); } +}; + +/// An object which should be asynchronously closed before it is destroyed +/// +/// Any AsyncCloseable must be kept alive until its parent is destroyed (this is a given +/// if the parent is a nursery). For shorter lived tasks/objects consider +/// OwnedAsyncCloseable adding a dependent task. +class ARROW_EXPORT AsyncCloseable : public std::enable_shared_from_this<AsyncCloseable> { + public: + AsyncCloseable(); + explicit AsyncCloseable(AsyncCloseable* parent); + virtual ~AsyncCloseable(); + + /// Returns a future that is completed when this object is finished closing + const Future<>& OnClosed(); + + protected: + /// Subclasses should override this and perform any cleanup. Once the future returned + /// by this method finishes then this object is eligible for destruction and any + /// reference to `this` will be invalid + virtual Future<> DoClose() = 0; + + /// This method is called by subclasses to add tasks which must complete before the + /// object can be safely deleted + void AddDependentTask(const Future<>& task); + /// This method can be called by subclasses for error checking purposes. It will + /// return an invalid status if this object has started closing + Status CheckClosed() const; + + Nursery* nursery_; + + private: + void SetNursery(Nursery* nursery); + void Destroy(); + + Future<> on_closed_; + Future<> tasks_finished_; + std::atomic<bool> closed_{false}; + std::atomic<uint32_t> num_tasks_outstanding_{1}; + + friend Nursery; + template <typename T> + friend struct DestroyingDeleter; + friend AsyncCloseablePimpl; +}; + +class ARROW_EXPORT AsyncCloseablePimpl { + protected: + void Init(AsyncCloseable* impl); + + private: + void SetNursery(Nursery* nursery); + void Destroy(); + + AsyncCloseable* impl_; + + friend Nursery; + template <typename T> + friend struct DestroyingDeleter; +}; + +class ARROW_EXPORT Nursery { + public: + template <typename T, typename... Args> + typename std::enable_if<!std::is_array<T>::value, std::shared_ptr<T>>::type + MakeSharedCloseable(Args&&... args) { Review comment: Can you add a docstring explaining what it does? ########## File path: cpp/src/arrow/util/async_nursery.h ########## @@ -0,0 +1,154 @@ +// 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 <list> + +#include "arrow/result.h" +#include "arrow/status.h" +#include "arrow/util/future.h" +#include "arrow/util/mutex.h" + +namespace arrow { +namespace util { + +class Nursery; +class AsyncCloseablePimpl; + +template <typename T> +struct DestroyingDeleter { + void operator()(T* p) { p->Destroy(); } +}; + +/// An object which should be asynchronously closed before it is destroyed +/// +/// Any AsyncCloseable must be kept alive until its parent is destroyed (this is a given +/// if the parent is a nursery). For shorter lived tasks/objects consider +/// OwnedAsyncCloseable adding a dependent task. +class ARROW_EXPORT AsyncCloseable : public std::enable_shared_from_this<AsyncCloseable> { + public: + AsyncCloseable(); + explicit AsyncCloseable(AsyncCloseable* parent); + virtual ~AsyncCloseable(); + + /// Returns a future that is completed when this object is finished closing + const Future<>& OnClosed(); + + protected: + /// Subclasses should override this and perform any cleanup. Once the future returned + /// by this method finishes then this object is eligible for destruction and any + /// reference to `this` will be invalid + virtual Future<> DoClose() = 0; + + /// This method is called by subclasses to add tasks which must complete before the + /// object can be safely deleted + void AddDependentTask(const Future<>& task); + /// This method can be called by subclasses for error checking purposes. It will + /// return an invalid status if this object has started closing + Status CheckClosed() const; + + Nursery* nursery_; + + private: + void SetNursery(Nursery* nursery); + void Destroy(); + + Future<> on_closed_; + Future<> tasks_finished_; + std::atomic<bool> closed_{false}; + std::atomic<uint32_t> num_tasks_outstanding_{1}; + + friend Nursery; + template <typename T> + friend struct DestroyingDeleter; + friend AsyncCloseablePimpl; +}; + +class ARROW_EXPORT AsyncCloseablePimpl { + protected: + void Init(AsyncCloseable* impl); + + private: + void SetNursery(Nursery* nursery); + void Destroy(); + + AsyncCloseable* impl_; + + friend Nursery; + template <typename T> + friend struct DestroyingDeleter; +}; + +class ARROW_EXPORT Nursery { + public: + template <typename T, typename... Args> + typename std::enable_if<!std::is_array<T>::value, std::shared_ptr<T>>::type Review comment: The `enable_if` doesn't seem useful here (especially as you have a `static_assert` below that would catch arrays) ########## File path: cpp/src/arrow/util/async_nursery.h ########## @@ -0,0 +1,154 @@ +// 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 <list> + +#include "arrow/result.h" +#include "arrow/status.h" +#include "arrow/util/future.h" +#include "arrow/util/mutex.h" + +namespace arrow { +namespace util { + +class Nursery; +class AsyncCloseablePimpl; + +template <typename T> +struct DestroyingDeleter { + void operator()(T* p) { p->Destroy(); } +}; + +/// An object which should be asynchronously closed before it is destroyed +/// +/// Any AsyncCloseable must be kept alive until its parent is destroyed (this is a given +/// if the parent is a nursery). For shorter lived tasks/objects consider +/// OwnedAsyncCloseable adding a dependent task. +class ARROW_EXPORT AsyncCloseable : public std::enable_shared_from_this<AsyncCloseable> { + public: + AsyncCloseable(); + explicit AsyncCloseable(AsyncCloseable* parent); + virtual ~AsyncCloseable(); + + /// Returns a future that is completed when this object is finished closing + const Future<>& OnClosed(); + + protected: + /// Subclasses should override this and perform any cleanup. Once the future returned + /// by this method finishes then this object is eligible for destruction and any + /// reference to `this` will be invalid + virtual Future<> DoClose() = 0; Review comment: API ergonomics question: since this is the single point of customization, would it be easier if `AsyncCloseable` took a `std::function<Future<>> close_func` parameter, instead of having to write a subclass? ########## File path: cpp/src/arrow/util/async_nursery.cc ########## @@ -0,0 +1,123 @@ +// 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/async_nursery.h" + +#include "arrow/util/logging.h" + +namespace arrow { +namespace util { + +AsyncCloseable::AsyncCloseable() : on_closed_(Future<>::Make()) {} +AsyncCloseable::AsyncCloseable(AsyncCloseable* parent) : on_closed_(Future<>::Make()) { + parent->AddDependentTask(OnClosed()); +} + +AsyncCloseable::~AsyncCloseable() { + DCHECK_NE(nursery_, nullptr) << "An AsyncCloseable must be created with a nursery " + "using MakeSharedCloseable or MakeUniqueCloseable"; +} + +const Future<>& AsyncCloseable::OnClosed() { return on_closed_; } + +void AsyncCloseable::AddDependentTask(const Future<>& task) { + DCHECK(!closed_); + if (num_tasks_outstanding_.fetch_add(1) == 1) { + tasks_finished_ = Future<>::Make(); + } + task.AddCallback([this](const Status& st) { + if (num_tasks_outstanding_.fetch_sub(1) == 1 && closed_.load()) { + tasks_finished_.MarkFinished(st); + } + }); +} + +void AsyncCloseable::SetNursery(Nursery* nursery) { nursery_ = nursery; } + +void AsyncCloseable::Destroy() { + DCHECK_NE(nursery_, nullptr); + closed_ = true; + nursery_->num_closeables_destroyed_.fetch_add(1); + Future<> finish_fut; + if (tasks_finished_.is_valid()) { + if (num_tasks_outstanding_.fetch_sub(1) > 1) { + finish_fut = AllComplete({DoClose(), tasks_finished_}); + } else { + // Any added tasks have already finished so there is nothing to wait for + finish_fut = DoClose(); + } + } else { + // No dependent tasks were added Review comment: Hmm... is it possible for dependent tasks to be added after this? ########## File path: cpp/src/arrow/util/async_nursery.cc ########## @@ -0,0 +1,123 @@ +// 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/async_nursery.h" + +#include "arrow/util/logging.h" + +namespace arrow { +namespace util { + +AsyncCloseable::AsyncCloseable() : on_closed_(Future<>::Make()) {} +AsyncCloseable::AsyncCloseable(AsyncCloseable* parent) : on_closed_(Future<>::Make()) { + parent->AddDependentTask(OnClosed()); +} + +AsyncCloseable::~AsyncCloseable() { + DCHECK_NE(nursery_, nullptr) << "An AsyncCloseable must be created with a nursery " + "using MakeSharedCloseable or MakeUniqueCloseable"; +} + +const Future<>& AsyncCloseable::OnClosed() { return on_closed_; } + +void AsyncCloseable::AddDependentTask(const Future<>& task) { + DCHECK(!closed_); + if (num_tasks_outstanding_.fetch_add(1) == 1) { + tasks_finished_ = Future<>::Make(); + } + task.AddCallback([this](const Status& st) { + if (num_tasks_outstanding_.fetch_sub(1) == 1 && closed_.load()) { + tasks_finished_.MarkFinished(st); + } + }); +} + +void AsyncCloseable::SetNursery(Nursery* nursery) { nursery_ = nursery; } + +void AsyncCloseable::Destroy() { + DCHECK_NE(nursery_, nullptr); + closed_ = true; + nursery_->num_closeables_destroyed_.fetch_add(1); + Future<> finish_fut; + if (tasks_finished_.is_valid()) { + if (num_tasks_outstanding_.fetch_sub(1) > 1) { + finish_fut = AllComplete({DoClose(), tasks_finished_}); + } else { + // Any added tasks have already finished so there is nothing to wait for + finish_fut = DoClose(); + } + } else { + // No dependent tasks were added + finish_fut = DoClose(); + } + finish_fut.AddCallback([this](const Status& st) { + if (on_closed_.is_valid()) { + on_closed_.MarkFinished(st); + } + nursery_->OnTaskFinished(st); + delete this; Review comment: Ok, so this mandates that this object is heap-allocated using the default C++ allocator, right? Can you mention this somewhere in the docstring? ########## File path: cpp/src/arrow/util/async_nursery.h ########## @@ -0,0 +1,154 @@ +// 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 <list> + +#include "arrow/result.h" +#include "arrow/status.h" +#include "arrow/util/future.h" +#include "arrow/util/mutex.h" + +namespace arrow { +namespace util { + +class Nursery; +class AsyncCloseablePimpl; + +template <typename T> +struct DestroyingDeleter { + void operator()(T* p) { p->Destroy(); } +}; + +/// An object which should be asynchronously closed before it is destroyed +/// +/// Any AsyncCloseable must be kept alive until its parent is destroyed (this is a given +/// if the parent is a nursery). For shorter lived tasks/objects consider +/// OwnedAsyncCloseable adding a dependent task. +class ARROW_EXPORT AsyncCloseable : public std::enable_shared_from_this<AsyncCloseable> { + public: + AsyncCloseable(); + explicit AsyncCloseable(AsyncCloseable* parent); + virtual ~AsyncCloseable(); + + /// Returns a future that is completed when this object is finished closing + const Future<>& OnClosed(); + + protected: + /// Subclasses should override this and perform any cleanup. Once the future returned + /// by this method finishes then this object is eligible for destruction and any + /// reference to `this` will be invalid + virtual Future<> DoClose() = 0; + + /// This method is called by subclasses to add tasks which must complete before the + /// object can be safely deleted + void AddDependentTask(const Future<>& task); + /// This method can be called by subclasses for error checking purposes. It will + /// return an invalid status if this object has started closing + Status CheckClosed() const; + + Nursery* nursery_; + + private: + void SetNursery(Nursery* nursery); + void Destroy(); + + Future<> on_closed_; + Future<> tasks_finished_; + std::atomic<bool> closed_{false}; + std::atomic<uint32_t> num_tasks_outstanding_{1}; + + friend Nursery; + template <typename T> + friend struct DestroyingDeleter; + friend AsyncCloseablePimpl; +}; + +class ARROW_EXPORT AsyncCloseablePimpl { + protected: + void Init(AsyncCloseable* impl); + + private: + void SetNursery(Nursery* nursery); + void Destroy(); + + AsyncCloseable* impl_; + + friend Nursery; + template <typename T> + friend struct DestroyingDeleter; +}; + +class ARROW_EXPORT Nursery { + public: + template <typename T, typename... Args> + typename std::enable_if<!std::is_array<T>::value, std::shared_ptr<T>>::type + MakeSharedCloseable(Args&&... args) { + static_assert(std::is_base_of<AsyncCloseable, T>::value, + "Nursery::MakeSharedCloseable only works with AsyncCloseable types"); + num_closeables_created_.fetch_add(1); + num_tasks_outstanding_.fetch_add(1); + std::shared_ptr<T> shared_closeable(new T(std::forward<Args&&>(args)...), + DestroyingDeleter<T>()); + shared_closeable->SetNursery(this); + return shared_closeable; + } + + template <typename T, typename... Args> + typename std::enable_if<!std::is_array<T>::value, + std::unique_ptr<T, DestroyingDeleter<T>>>::type Review comment: Same remarks here. ########## File path: cpp/src/arrow/util/async_nursery.h ########## @@ -0,0 +1,154 @@ +// 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 <list> + +#include "arrow/result.h" +#include "arrow/status.h" +#include "arrow/util/future.h" +#include "arrow/util/mutex.h" + +namespace arrow { +namespace util { + +class Nursery; +class AsyncCloseablePimpl; + +template <typename T> +struct DestroyingDeleter { + void operator()(T* p) { p->Destroy(); } +}; + +/// An object which should be asynchronously closed before it is destroyed +/// +/// Any AsyncCloseable must be kept alive until its parent is destroyed (this is a given +/// if the parent is a nursery). For shorter lived tasks/objects consider +/// OwnedAsyncCloseable adding a dependent task. +class ARROW_EXPORT AsyncCloseable : public std::enable_shared_from_this<AsyncCloseable> { + public: + AsyncCloseable(); + explicit AsyncCloseable(AsyncCloseable* parent); + virtual ~AsyncCloseable(); + + /// Returns a future that is completed when this object is finished closing + const Future<>& OnClosed(); + + protected: + /// Subclasses should override this and perform any cleanup. Once the future returned + /// by this method finishes then this object is eligible for destruction and any + /// reference to `this` will be invalid + virtual Future<> DoClose() = 0; + + /// This method is called by subclasses to add tasks which must complete before the + /// object can be safely deleted + void AddDependentTask(const Future<>& task); + /// This method can be called by subclasses for error checking purposes. It will + /// return an invalid status if this object has started closing + Status CheckClosed() const; + + Nursery* nursery_; + + private: + void SetNursery(Nursery* nursery); + void Destroy(); + + Future<> on_closed_; + Future<> tasks_finished_; + std::atomic<bool> closed_{false}; + std::atomic<uint32_t> num_tasks_outstanding_{1}; + + friend Nursery; + template <typename T> + friend struct DestroyingDeleter; + friend AsyncCloseablePimpl; +}; + +class ARROW_EXPORT AsyncCloseablePimpl { + protected: + void Init(AsyncCloseable* impl); + + private: + void SetNursery(Nursery* nursery); + void Destroy(); + + AsyncCloseable* impl_; + + friend Nursery; + template <typename T> + friend struct DestroyingDeleter; +}; + +class ARROW_EXPORT Nursery { Review comment: Can you add a docstring explaining what this is/does? ########## File path: cpp/src/arrow/util/async_nursery.h ########## @@ -0,0 +1,154 @@ +// 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 <list> + +#include "arrow/result.h" +#include "arrow/status.h" +#include "arrow/util/future.h" +#include "arrow/util/mutex.h" + +namespace arrow { +namespace util { + +class Nursery; +class AsyncCloseablePimpl; + +template <typename T> +struct DestroyingDeleter { + void operator()(T* p) { p->Destroy(); } +}; + +/// An object which should be asynchronously closed before it is destroyed +/// +/// Any AsyncCloseable must be kept alive until its parent is destroyed (this is a given +/// if the parent is a nursery). For shorter lived tasks/objects consider +/// OwnedAsyncCloseable adding a dependent task. +class ARROW_EXPORT AsyncCloseable : public std::enable_shared_from_this<AsyncCloseable> { + public: + AsyncCloseable(); + explicit AsyncCloseable(AsyncCloseable* parent); + virtual ~AsyncCloseable(); + + /// Returns a future that is completed when this object is finished closing + const Future<>& OnClosed(); + + protected: + /// Subclasses should override this and perform any cleanup. Once the future returned + /// by this method finishes then this object is eligible for destruction and any + /// reference to `this` will be invalid + virtual Future<> DoClose() = 0; + + /// This method is called by subclasses to add tasks which must complete before the + /// object can be safely deleted + void AddDependentTask(const Future<>& task); + /// This method can be called by subclasses for error checking purposes. It will + /// return an invalid status if this object has started closing + Status CheckClosed() const; + + Nursery* nursery_; + + private: + void SetNursery(Nursery* nursery); + void Destroy(); + + Future<> on_closed_; + Future<> tasks_finished_; + std::atomic<bool> closed_{false}; + std::atomic<uint32_t> num_tasks_outstanding_{1}; + + friend Nursery; + template <typename T> + friend struct DestroyingDeleter; + friend AsyncCloseablePimpl; +}; + +class ARROW_EXPORT AsyncCloseablePimpl { + protected: + void Init(AsyncCloseable* impl); + + private: + void SetNursery(Nursery* nursery); + void Destroy(); + + AsyncCloseable* impl_; + + friend Nursery; + template <typename T> + friend struct DestroyingDeleter; +}; + +class ARROW_EXPORT Nursery { + public: + template <typename T, typename... Args> + typename std::enable_if<!std::is_array<T>::value, std::shared_ptr<T>>::type + MakeSharedCloseable(Args&&... args) { + static_assert(std::is_base_of<AsyncCloseable, T>::value, + "Nursery::MakeSharedCloseable only works with AsyncCloseable types"); + num_closeables_created_.fetch_add(1); + num_tasks_outstanding_.fetch_add(1); + std::shared_ptr<T> shared_closeable(new T(std::forward<Args&&>(args)...), + DestroyingDeleter<T>()); + shared_closeable->SetNursery(this); + return shared_closeable; + } + + template <typename T, typename... Args> + typename std::enable_if<!std::is_array<T>::value, + std::unique_ptr<T, DestroyingDeleter<T>>>::type + MakeUniqueCloseable(Args&&... args) { + static_assert(std::is_base_of<AsyncCloseable, T>::value, + "Nursery::MakeUniqueCloseable only works with AsyncCloseable types"); + num_closeables_created_.fetch_add(1); + num_tasks_outstanding_.fetch_add(1); + auto unique_closeable = std::unique_ptr<T, DestroyingDeleter<T>>( + new T(std::forward<Args>(args)...), DestroyingDeleter<T>()); + unique_closeable->SetNursery(this); + return unique_closeable; + } + + template <typename T> + void AddDependentTask(const Future<T>& task) { + num_tasks_outstanding_.fetch_add(1); + task.AddCallback([this](const Result<T>& res) { OnTaskFinished(res.status()); }); + } + + /// Runs `task` within a nursery. This method will not return until + /// all roots added by the task have been closed and destroyed. Review comment: What are "roots" in this context? AsyncCloseable objects? ########## File path: cpp/src/arrow/util/async_nursery.h ########## @@ -0,0 +1,154 @@ +// 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 <list> Review comment: This doesn't seem used? ########## File path: cpp/src/arrow/util/async_nursery.h ########## @@ -0,0 +1,154 @@ +// 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 <list> + +#include "arrow/result.h" +#include "arrow/status.h" +#include "arrow/util/future.h" +#include "arrow/util/mutex.h" + +namespace arrow { +namespace util { + +class Nursery; +class AsyncCloseablePimpl; + +template <typename T> +struct DestroyingDeleter { + void operator()(T* p) { p->Destroy(); } +}; + +/// An object which should be asynchronously closed before it is destroyed +/// +/// Any AsyncCloseable must be kept alive until its parent is destroyed (this is a given +/// if the parent is a nursery). For shorter lived tasks/objects consider +/// OwnedAsyncCloseable adding a dependent task. +class ARROW_EXPORT AsyncCloseable : public std::enable_shared_from_this<AsyncCloseable> { + public: + AsyncCloseable(); + explicit AsyncCloseable(AsyncCloseable* parent); + virtual ~AsyncCloseable(); + + /// Returns a future that is completed when this object is finished closing + const Future<>& OnClosed(); + + protected: + /// Subclasses should override this and perform any cleanup. Once the future returned + /// by this method finishes then this object is eligible for destruction and any + /// reference to `this` will be invalid + virtual Future<> DoClose() = 0; + + /// This method is called by subclasses to add tasks which must complete before the + /// object can be safely deleted + void AddDependentTask(const Future<>& task); + /// This method can be called by subclasses for error checking purposes. It will + /// return an invalid status if this object has started closing + Status CheckClosed() const; + + Nursery* nursery_; + + private: + void SetNursery(Nursery* nursery); + void Destroy(); + + Future<> on_closed_; + Future<> tasks_finished_; + std::atomic<bool> closed_{false}; + std::atomic<uint32_t> num_tasks_outstanding_{1}; + + friend Nursery; + template <typename T> + friend struct DestroyingDeleter; + friend AsyncCloseablePimpl; +}; + +class ARROW_EXPORT AsyncCloseablePimpl { Review comment: What is this supposed to be? Is it useful as a public API? ########## File path: cpp/src/arrow/util/async_nursery_test.cc ########## @@ -0,0 +1,184 @@ +// 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/async_nursery.h" + +#include <gtest/gtest.h> + +#include <thread> + +#include "arrow/result.h" +#include "arrow/testing/gtest_common.h" +#include "arrow/testing/gtest_util.h" + +namespace arrow { +namespace util { + +class GatingDoClose : public AsyncCloseable { + public: + GatingDoClose(AsyncCloseable* parent, Future<> close_future) + : AsyncCloseable(parent), close_future_(std::move(close_future)) {} + explicit GatingDoClose(Future<> close_future) + : close_future_(std::move(close_future)) {} + + Future<> DoClose() override { return close_future_; } + + Future<> close_future_; +}; + +class GatingDoCloseWithUniqueChild : public AsyncCloseable { + public: + GatingDoCloseWithUniqueChild(Nursery* nursery, Future<> close_future) + : child_( + nursery->MakeUniqueCloseable<GatingDoClose>(this, std::move(close_future))) {} + + Future<> DoClose() override { + child_.reset(); + return Future<>::MakeFinished(); + } + + std::unique_ptr<GatingDoClose, DestroyingDeleter<GatingDoClose>> child_; +}; + +class GatingDoCloseWithSharedChild : public AsyncCloseable { + public: + GatingDoCloseWithSharedChild(Nursery* nursery, Future<> close_future) + : child_( + nursery->MakeSharedCloseable<GatingDoClose>(this, std::move(close_future))) {} + + Future<> DoClose() override { + child_.reset(); + return Future<>::MakeFinished(); + } + + std::shared_ptr<GatingDoClose> child_; +}; + +class GatingDoCloseAsDependentTask : public AsyncCloseable { + public: + explicit GatingDoCloseAsDependentTask(Future<> close_future) { + AddDependentTask(std::move(close_future)); + } + + Future<> DoClose() override { return Future<>::MakeFinished(); } +}; + +class MarkWhenDestroyed : public GatingDoClose { + public: + MarkWhenDestroyed(Future<> close_future, bool* destroyed) + : GatingDoClose(std::move(close_future)), destroyed_(destroyed) {} + ~MarkWhenDestroyed() { *destroyed_ = true; } + + private: + bool* destroyed_; +}; + +class EvictsChild : public AsyncCloseable { Review comment: Add a comment explaining what this does/exercises? ########## File path: cpp/src/arrow/util/async_nursery_test.cc ########## @@ -0,0 +1,184 @@ +// 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/async_nursery.h" + +#include <gtest/gtest.h> + +#include <thread> + +#include "arrow/result.h" +#include "arrow/testing/gtest_common.h" +#include "arrow/testing/gtest_util.h" + +namespace arrow { +namespace util { + +class GatingDoClose : public AsyncCloseable { + public: + GatingDoClose(AsyncCloseable* parent, Future<> close_future) + : AsyncCloseable(parent), close_future_(std::move(close_future)) {} + explicit GatingDoClose(Future<> close_future) + : close_future_(std::move(close_future)) {} + + Future<> DoClose() override { return close_future_; } + + Future<> close_future_; +}; + +class GatingDoCloseWithUniqueChild : public AsyncCloseable { + public: + GatingDoCloseWithUniqueChild(Nursery* nursery, Future<> close_future) + : child_( + nursery->MakeUniqueCloseable<GatingDoClose>(this, std::move(close_future))) {} + + Future<> DoClose() override { + child_.reset(); + return Future<>::MakeFinished(); + } + + std::unique_ptr<GatingDoClose, DestroyingDeleter<GatingDoClose>> child_; +}; + +class GatingDoCloseWithSharedChild : public AsyncCloseable { + public: + GatingDoCloseWithSharedChild(Nursery* nursery, Future<> close_future) + : child_( + nursery->MakeSharedCloseable<GatingDoClose>(this, std::move(close_future))) {} + + Future<> DoClose() override { + child_.reset(); + return Future<>::MakeFinished(); + } + + std::shared_ptr<GatingDoClose> child_; +}; + +class GatingDoCloseAsDependentTask : public AsyncCloseable { + public: + explicit GatingDoCloseAsDependentTask(Future<> close_future) { + AddDependentTask(std::move(close_future)); + } + + Future<> DoClose() override { return Future<>::MakeFinished(); } +}; + +class MarkWhenDestroyed : public GatingDoClose { + public: + MarkWhenDestroyed(Future<> close_future, bool* destroyed) + : GatingDoClose(std::move(close_future)), destroyed_(destroyed) {} + ~MarkWhenDestroyed() { *destroyed_ = true; } + + private: + bool* destroyed_; +}; + +class EvictsChild : public AsyncCloseable { + public: + EvictsChild(Nursery* nursery, bool* child_destroyed, Future<> child_future, + Future<> final_future) + : final_close_future_(std::move(final_future)) { + owned_child_ = nursery->MakeSharedCloseable<MarkWhenDestroyed>( + std::move(child_future), child_destroyed); + } + + void EvictChild() { owned_child_.reset(); } + + Future<> DoClose() override { return final_close_future_; } + + private: + Future<> final_close_future_; + std::shared_ptr<GatingDoClose> owned_child_; +}; + +template <typename T> +void AssertDoesNotCloseEarly() { + Future<> gate = Future<>::Make(); + std::atomic<bool> finished{false}; + std::thread thread([&] { + ASSERT_OK(Nursery::RunInNursery( + [&](Nursery* nursery) { nursery->MakeSharedCloseable<T>(gate); })); + finished.store(true); + }); + + SleepABit(); + ASSERT_FALSE(finished.load()); + gate.MarkFinished(); + BusyWait(10, [&] { return finished.load(); }); + thread.join(); +} + +template <typename T> +void AssertDoesNotCloseEarlyWithChild() { + Future<> gate = Future<>::Make(); + std::atomic<bool> finished{false}; + std::thread thread([&] { + ASSERT_OK(Nursery::RunInNursery( + [&](Nursery* nursery) { nursery->MakeSharedCloseable<T>(nursery, gate); })); + finished.store(true); + }); + + SleepABit(); + ASSERT_FALSE(finished.load()); + gate.MarkFinished(); + BusyWait(10, [&] { return finished.load(); }); + thread.join(); +} + +TEST(AsyncNursery, DoClose) { AssertDoesNotCloseEarly<GatingDoClose>(); } + +TEST(AsyncNursery, SharedChildDoClose) { + AssertDoesNotCloseEarlyWithChild<GatingDoCloseWithSharedChild>(); +} + +TEST(AsyncNursery, UniqueChildDoClose) { + AssertDoesNotCloseEarlyWithChild<GatingDoCloseWithUniqueChild>(); +} + +TEST(AsyncNursery, DependentTask) { + AssertDoesNotCloseEarly<GatingDoCloseAsDependentTask>(); +} + +TEST(AsyncNursery, EvictedChild) { + Future<> child_future = Future<>::Make(); + Future<> final_future = Future<>::Make(); + std::atomic<bool> finished{false}; + std::thread thread([&] { + ASSERT_OK(Nursery::RunInNursery([&](Nursery* nursery) { + bool child_destroyed = false; + std::shared_ptr<EvictsChild> evicts_child = + nursery->MakeSharedCloseable<EvictsChild>(nursery, &child_destroyed, + child_future, final_future); + evicts_child->EvictChild(); + // Owner no longer has reference to child here but it's kept alive by nursery + // because it isn't done Review comment: I don't understand this comment, because by reading the source code, I get the impression that the nursery doesn't keep anything alive (it does not have a container of tasks or futures). -- 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. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
