Copilot commented on code in PR #1033: URL: https://github.com/apache/iceberg-cpp/pull/1033#discussion_r4315620981
########## 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(); Review Comment: Cancellation can be lost in the race where `Run()` pops a due task (erasing from `scheduled_`) and then `Cancel(id)` runs before the task is dispatched. In that case `scheduled_.erase(id)` returns 0, so the ID is never recorded in `cancelled_`, and the task may still be submitted/executed even though the handle was cancelled. To make cancellation robust, avoid gating cancellation on `scheduled_` membership (e.g., always record cancellation and have `Run()` check cancellation immediately before submission, or use a per-task shared cancellation flag/record that is valid through the dispatch boundary). ########## 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(); Review Comment: `scheduled_` is not updated when a cancelled task is discarded here. This leaves stale IDs in `scheduled_`, which can cause incorrect cancellation bookkeeping and unbounded growth over time. Consider erasing `tasks_.top().id` from `scheduled_` before popping the task (and keep the sets consistent for all task removal paths)." ########## 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); Review Comment: Moving out of `std::priority_queue::top()` via `const_cast` is brittle and easy to misreason about (and can become unsafe if `Task` fields affecting heap ordering ever change). A more maintainable approach is to store moveable pointers (e.g., `std::unique_ptr<Task>`) in the queue, or switch to a heap on `std::vector<Task>` where you can `std::pop_heap` + `std::move` without casting away const. ########## src/iceberg/catalog/rest/rest_catalog.h: ########## @@ -63,9 +63,13 @@ class ICEBERG_REST_EXPORT RestCatalog final /// are reported synchronously. /// \param client_factory Optional HTTP client factory, called with the initial config /// and then the merged server config. When empty, uses the default HTTP client. + /// \param auth_executor Optional non-owning executor for OAuth2 token refresh. When + /// null, background token refresh is disabled. When non-null, it must outlive the + /// catalog and all authentication sessions created by it. static Result<std::shared_ptr<RestCatalog>> Make(const RestCatalogProperties& config, Executor* metrics_executor = nullptr, - ClientFactory client_factory = {}); + ClientFactory client_factory = {}, + Executor* auth_executor = nullptr); Review Comment: This reorders parameters of an exported/public factory (`Make`), which is a source-breaking change for existing callers (especially those relying on the previous `(config, metrics_executor, client_factory)` ordering). To preserve API compatibility, consider adding an overload with the old signature that forwards into the new one (or keep the original order and append `auth_executor` as the last optional parameter), and optionally deprecate the old overload after a transition period. ########## 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: New behavior is introduced for rejection handling (`on_rejected`) but the migrated tests in `ScheduledExecutorTest.*` don't cover it. Add a unit test using an `Executor` implementation that deterministically rejects `Submit()` to verify: (1) `on_rejected` is invoked exactly once per due task, and (2) scheduling/cancellation/shutdown interactions don’t accidentally call `on_rejected` after shutdown. -- 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]
