Copilot commented on code in PR #1033: URL: https://github.com/apache/iceberg-cpp/pull/1033#discussion_r4315826741
########## 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); Review Comment: This introduces locals that are never used (`inserted`, and `cancelled` is only written then discarded). If the repo builds with `-Werror` / unused-variable warnings, this can fail compilation. Consider rewriting `emplace` handling to only keep the iterator (e.g., use `.first`), and replace `cancelled` with a narrower scope or a comment + `[[maybe_unused]]` to explicitly document the lifetime-intent without triggering warnings. ########## src/iceberg/catalog/rest/rest_catalog.h: ########## @@ -58,14 +58,18 @@ class ICEBERG_REST_EXPORT RestCatalog final std::function<Result<std::shared_ptr<HttpClient>>(const RestCatalogProperties&)>; /// \brief Create a RestCatalog instance. + /// \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. /// \param metrics_executor Optional non-owning executor for asynchronous REST metrics. /// It must outlive the catalog and all tables created by it. When null, REST metrics /// 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. static Result<std::shared_ptr<RestCatalog>> Make(const RestCatalogProperties& config, - Executor* metrics_executor = nullptr, - ClientFactory client_factory = {}); + ClientFactory client_factory = {}, + Executor* auth_executor = nullptr, + Executor* metrics_executor = nullptr); std::string_view name() const override; Review Comment: This changes the parameter ordering of a public exported factory (`ICEBERG_REST_EXPORT`), which is a source-breaking API change for callers passing `metrics_executor` positionally (e.g., `Make(config, metrics_executor, factory)`). If preserving API compatibility is a goal, consider adding an overload with the previous signature (or introducing an options struct) and forwarding internally to the new implementation. ########## 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_); Review Comment: This introduces locals that are never used (`inserted`, and `cancelled` is only written then discarded). If the repo builds with `-Werror` / unused-variable warnings, this can fail compilation. Consider rewriting `emplace` handling to only keep the iterator (e.g., use `.first`), and replace `cancelled` with a narrower scope or a comment + `[[maybe_unused]]` to explicitly document the lifetime-intent without triggering warnings. ########## 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: The previous `TokenRefreshScheduler::Instance()` was intentionally leaked to avoid destruction-order races at process exit. `SharedTimer::Instance()` is a function-local static `shared_ptr` that will be destroyed during static teardown, which can reintroduce shutdown-order hazards if any remaining static destructors (or other late shutdown paths) try to schedule/cancel work after the timer is destroyed. To preserve the earlier safety property, consider making the shared timer intentionally non-destructing (e.g., allocate once and never delete / `NoDestructor`-style), or otherwise guarantee no scheduled-executor users can run during static teardown. ########## 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: The previous `TokenRefreshScheduler::Instance()` was intentionally leaked to avoid destruction-order races at process exit. `SharedTimer::Instance()` is a function-local static `shared_ptr` that will be destroyed during static teardown, which can reintroduce shutdown-order hazards if any remaining static destructors (or other late shutdown paths) try to schedule/cancel work after the timer is destroyed. To preserve the earlier safety property, consider making the shared timer intentionally non-destructing (e.g., allocate once and never delete / `NoDestructor`-style), or otherwise guarantee no scheduled-executor users can run during static teardown. ########## src/iceberg/util/scheduled_executor_internal.h: ########## @@ -0,0 +1,125 @@ +/* + * 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 <chrono> +#include <condition_variable> +#include <cstdint> +#include <memory> +#include <mutex> +#include <optional> +#include <queue> +#include <thread> +#include <unordered_set> +#include <vector> + +#include "iceberg/iceberg_export.h" +#include "iceberg/result.h" +#include "iceberg/util/executor.h" + +namespace iceberg::internal { + +class ScheduledExecutorState; + +using ScheduledTaskRejection = FnOnce<void(Status)>; + +class ICEBERG_EXPORT ScheduledTask { + public: + ScheduledTask() = default; + ScheduledTask(std::weak_ptr<ScheduledExecutorState> state, uint64_t id) + : state_(std::move(state)), id_(id) {} + + void Cancel(); + bool IsValid() const { return id_.has_value() && !state_.expired(); } + + private: + std::weak_ptr<ScheduledExecutorState> state_; + std::optional<uint64_t> id_; Review Comment: `ScheduledTask` is currently copyable (implicit copy ctor/assign). Copying will produce multiple handles with the same `id_`, and cancelling one handle does not invalidate the others (so `IsValid()` may remain true on copies even though the task was already cancelled). If the intended semantics are a unique cancellation token, consider making `ScheduledTask` move-only (delete copy, default move) to prevent misleading validity/cancellation behavior. -- 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]
