wgtmac commented on code in PR #1033: URL: https://github.com/apache/iceberg-cpp/pull/1033#discussion_r4320695204
########## src/iceberg/util/scheduled_executor.cc: ########## @@ -0,0 +1,143 @@ +/* + * 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 <utility> + +#include "iceberg/util/scheduled_executor_internal.h" + +namespace iceberg::internal { + +void ScheduledTask::Cancel() { + if (auto state = state_.lock()) state->Cancel(id_); + state_.reset(); + id_ = 0; +} + +ScheduledExecutorState::ScheduledExecutorState(Executor& executor) + : executor_(executor) {} + +ScheduledExecutorState::~ScheduledExecutorState() { Shutdown(); } + +void ScheduledExecutorState::Start() { + auto self = shared_from_this(); + timer_thread_ = std::thread([self = std::move(self)] { self->Run(); }); +} + +ScheduledTask ScheduledExecutorState::Schedule( + std::chrono::steady_clock::time_point deadline, ExecutorTask task, + std::optional<ScheduledTaskRejection> on_rejected) { + std::lock_guard lock(mutex_); + if (shutdown_) return {}; + auto id = next_id_++; + tasks_.push(Task{.id = id, + .deadline = deadline, + .task = std::move(task), + .on_rejected = std::move(on_rejected)}); + scheduled_.insert(id); + cv_.notify_one(); + return ScheduledTask(weak_from_this(), id); +} + +void ScheduledExecutorState::Cancel(uint64_t id) { + if (id == 0) return; + { + std::lock_guard lock(mutex_); + if (scheduled_.erase(id) != 0) cancelled_.insert(id); + } + cv_.notify_one(); +} + +void ScheduledExecutorState::Shutdown() { + { + std::lock_guard lock(mutex_); + if (shutdown_) return; + shutdown_ = true; + while (!tasks_.empty()) tasks_.pop(); + cancelled_.clear(); + scheduled_.clear(); + } + cv_.notify_one(); + if (!timer_thread_.joinable()) return; + if (timer_thread_.get_id() == std::this_thread::get_id()) { + timer_thread_.detach(); + } else { + timer_thread_.join(); + } +} + +void ScheduledExecutorState::Run() { + while (true) { + std::optional<Task> task; + { + std::unique_lock lock(mutex_); + cv_.wait(lock, [this] { return shutdown_ || !tasks_.empty(); }); + if (shutdown_) return; + + while (!tasks_.empty() && cancelled_.contains(tasks_.top().id)) { + cancelled_.erase(tasks_.top().id); + tasks_.pop(); + } + if (tasks_.empty()) continue; + + auto deadline = tasks_.top().deadline; + if (cv_.wait_until(lock, deadline, [this, deadline] { + return shutdown_ || tasks_.empty() || tasks_.top().deadline != deadline || + cancelled_.contains(tasks_.top().id); + })) { + continue; + } + + task.emplace(std::move(const_cast<Task&>(tasks_.top()))); + tasks_.pop(); + scheduled_.erase(task->id); + if (cancelled_.erase(task->id) != 0) continue; + } + + auto status = executor_.Submit(std::move(task->task)); + if (!status.has_value() && task->on_rejected.has_value()) { + std::move (*task->on_rejected)(std::move(status)); + } + } Review Comment: Addressed in bb30175e9 by extending the existing cancellation and shutdown tests with a rejecting executor and explicit rejection counters. Cancelling a pending task, shutting down its scheduler, and attempting to schedule after shutdown leave both Submit and on_rejected counts at zero. The existing due-task rejection test now also checks exactly one submission. All seven scheduler tests pass with ASAN/UBSAN; no separate duplicate suite was added. ########## src/iceberg/util/scheduled_executor.cc: ########## @@ -0,0 +1,145 @@ +/* + * 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 <utility> + +#include "iceberg/util/scheduled_executor_internal.h" + +namespace iceberg::internal { + +void ScheduledTask::Cancel() { + if (id_.has_value()) { + if (auto state = state_.lock()) state->Cancel(*id_); + } + state_.reset(); + id_.reset(); +} + +ScheduledExecutorState::ScheduledExecutorState(Executor& executor) + : executor_(executor) {} + +ScheduledExecutorState::~ScheduledExecutorState() { Shutdown(); } + +void ScheduledExecutorState::Start() { + auto self = shared_from_this(); + timer_thread_ = std::thread([self = std::move(self)] { self->Run(); }); +} + +ScheduledTask ScheduledExecutorState::Schedule( + std::chrono::steady_clock::time_point deadline, ExecutorTask task, + std::optional<ScheduledTaskRejection> on_rejected) { + std::lock_guard lock(mutex_); + if (shutdown_) return {}; + auto id = next_id_++; + tasks_.push(Task{.id = id, + .deadline = deadline, + .task = std::move(task), + .on_rejected = std::move(on_rejected)}); + scheduled_.insert(id); + cv_.notify_one(); + return {weak_from_this(), id}; +} + +void ScheduledExecutorState::Cancel(uint64_t id) { + if (id == 0) return; + { + std::lock_guard lock(mutex_); + if (scheduled_.erase(id) != 0) cancelled_.insert(id); + } + cv_.notify_one(); +} + +void ScheduledExecutorState::Shutdown() { + { + std::lock_guard lock(mutex_); + if (shutdown_) return; + shutdown_ = true; + while (!tasks_.empty()) tasks_.pop(); + cancelled_.clear(); + scheduled_.clear(); + } + cv_.notify_one(); + if (!timer_thread_.joinable()) return; + if (timer_thread_.get_id() == std::this_thread::get_id()) { + timer_thread_.detach(); + } else { + timer_thread_.join(); + } +} + +void ScheduledExecutorState::Run() { + while (true) { + std::optional<Task> task; + { + std::unique_lock lock(mutex_); + cv_.wait(lock, [this] { return shutdown_ || !tasks_.empty(); }); + if (shutdown_) return; + + while (!tasks_.empty() && cancelled_.contains(tasks_.top().id)) { + cancelled_.erase(tasks_.top().id); + tasks_.pop(); + } + if (tasks_.empty()) continue; + + auto deadline = tasks_.top().deadline; + if (cv_.wait_until(lock, deadline, [this, deadline] { + return shutdown_ || tasks_.empty() || tasks_.top().deadline != deadline || + cancelled_.contains(tasks_.top().id); + })) { + continue; + } + + task.emplace(std::move(const_cast<Task&>(tasks_.top()))); + tasks_.pop(); + scheduled_.erase(task->id); + if (cancelled_.erase(task->id) != 0) continue; + } + + auto status = executor_.Submit(std::move(task->task)); + if (!status.has_value() && task->on_rejected.has_value()) { + std::move (*task->on_rejected)(std::move(status)); + } + } +} + +ScheduledExecutor::ScheduledExecutor(Executor& executor) + : state_(std::make_shared<ScheduledExecutorState>(executor)) { + state_->Start(); +} + +ScheduledExecutor::~ScheduledExecutor() { Shutdown(); } + +ScheduledTask ScheduledExecutor::Schedule( + std::chrono::steady_clock::time_point deadline, ExecutorTask task, + std::optional<ScheduledTaskRejection> on_rejected) { + return state_->Schedule(deadline, std::move(task), std::move(on_rejected)); +} + +ScheduledTask ScheduledExecutor::Schedule( + std::chrono::milliseconds delay, ExecutorTask task, + std::optional<ScheduledTaskRejection> on_rejected) { + return Schedule(std::chrono::steady_clock::now() + delay, std::move(task), + std::move(on_rejected)); +} Review Comment: Fixed in bb30175e9. The process-wide shared_ptr holder is now intentionally non-destructing, following the existing process-lifetime holder pattern in the logging code; per-scheduler shutdown remains unchanged. The new subprocess test releases all scheduler owners, then constructs and dispatches another scheduler in an atexit callback registered before timer initialization. With the pre-fix implementation ASAN reports heap-use-after-free in SharedTimer::Instance; the fixed implementation passes. ########## src/iceberg/util/scheduled_executor.cc: ########## @@ -0,0 +1,145 @@ +/* + * 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 <utility> + +#include "iceberg/util/scheduled_executor_internal.h" + +namespace iceberg::internal { + +void ScheduledTask::Cancel() { + if (id_.has_value()) { + if (auto state = state_.lock()) state->Cancel(*id_); + } + state_.reset(); + id_.reset(); +} + +ScheduledExecutorState::ScheduledExecutorState(Executor& executor) + : executor_(executor) {} + +ScheduledExecutorState::~ScheduledExecutorState() { Shutdown(); } + +void ScheduledExecutorState::Start() { + auto self = shared_from_this(); + timer_thread_ = std::thread([self = std::move(self)] { self->Run(); }); +} + +ScheduledTask ScheduledExecutorState::Schedule( + std::chrono::steady_clock::time_point deadline, ExecutorTask task, + std::optional<ScheduledTaskRejection> on_rejected) { + std::lock_guard lock(mutex_); + if (shutdown_) return {}; + auto id = next_id_++; + tasks_.push(Task{.id = id, + .deadline = deadline, + .task = std::move(task), + .on_rejected = std::move(on_rejected)}); + scheduled_.insert(id); + cv_.notify_one(); + return {weak_from_this(), id}; +} + +void ScheduledExecutorState::Cancel(uint64_t id) { + if (id == 0) return; + { + std::lock_guard lock(mutex_); + if (scheduled_.erase(id) != 0) cancelled_.insert(id); + } + cv_.notify_one(); +} + +void ScheduledExecutorState::Shutdown() { + { + std::lock_guard lock(mutex_); + if (shutdown_) return; + shutdown_ = true; + while (!tasks_.empty()) tasks_.pop(); + cancelled_.clear(); + scheduled_.clear(); + } + cv_.notify_one(); + if (!timer_thread_.joinable()) return; + if (timer_thread_.get_id() == std::this_thread::get_id()) { + timer_thread_.detach(); + } else { + timer_thread_.join(); + } +} + +void ScheduledExecutorState::Run() { + while (true) { + std::optional<Task> task; + { + std::unique_lock lock(mutex_); + cv_.wait(lock, [this] { return shutdown_ || !tasks_.empty(); }); + if (shutdown_) return; + + while (!tasks_.empty() && cancelled_.contains(tasks_.top().id)) { + cancelled_.erase(tasks_.top().id); + tasks_.pop(); + } + if (tasks_.empty()) continue; + + auto deadline = tasks_.top().deadline; + if (cv_.wait_until(lock, deadline, [this, deadline] { + return shutdown_ || tasks_.empty() || tasks_.top().deadline != deadline || + cancelled_.contains(tasks_.top().id); + })) { + continue; + } + + task.emplace(std::move(const_cast<Task&>(tasks_.top()))); + tasks_.pop(); + scheduled_.erase(task->id); + if (cancelled_.erase(task->id) != 0) continue; + } + + auto status = executor_.Submit(std::move(task->task)); + if (!status.has_value() && task->on_rejected.has_value()) { + std::move (*task->on_rejected)(std::move(status)); + } + } +} + +ScheduledExecutor::ScheduledExecutor(Executor& executor) + : state_(std::make_shared<ScheduledExecutorState>(executor)) { + state_->Start(); +} + +ScheduledExecutor::~ScheduledExecutor() { Shutdown(); } + +ScheduledTask ScheduledExecutor::Schedule( + std::chrono::steady_clock::time_point deadline, ExecutorTask task, + std::optional<ScheduledTaskRejection> on_rejected) { + return state_->Schedule(deadline, std::move(task), std::move(on_rejected)); +} + +ScheduledTask ScheduledExecutor::Schedule( + std::chrono::milliseconds delay, ExecutorTask task, + std::optional<ScheduledTaskRejection> on_rejected) { + return Schedule(std::chrono::steady_clock::now() + delay, std::move(task), + std::move(on_rejected)); +} + +void ScheduledExecutor::Shutdown() { + if (state_) state_->Shutdown(); +} + Review Comment: Fixed in bb30175e9 along with the duplicate lifetime thread. The process-wide holder now survives static destruction, while individual schedulers still cancel their own work and wait for active submissions. A subprocess static-destruction regression reproduces an ASAN heap-use-after-free with the old holder and passes after this 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. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
