Copilot commented on code in PR #1033: URL: https://github.com/apache/iceberg-cpp/pull/1033#discussion_r4316236569
########## src/iceberg/util/scheduled_executor.cc: ########## @@ -0,0 +1,249 @@ +/* + * 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 <map> +#include <thread> +#include <unordered_map> +#include <utility> + +#include "iceberg/util/scheduled_executor_internal.h" + +namespace iceberg::internal { + +// The timer thread retains its state until Run returns. This also allows the last +// timer owner to be released on the timer thread during process teardown. +class SharedTimerState : public std::enable_shared_from_this<SharedTimerState> { + public: + ~SharedTimerState() { Shutdown(); } + + void Start() { + timer_thread_ = std::thread([self = shared_from_this()] { self->Run(); }); + } + + uint64_t Schedule(std::weak_ptr<ScheduledExecutorState> state, + std::chrono::steady_clock::time_point deadline, ExecutorTask task, + std::optional<ScheduledTaskRejection> on_rejected) { + std::lock_guard lock(mutex_); + auto id = next_id_++; + auto [it, inserted] = tasks_.emplace(std::pair{deadline, id}, + Task{.state = std::move(state), + .task = std::move(task), + .on_rejected = std::move(on_rejected)}); + by_id_.emplace(id, it); + cv_.notify_one(); + return id; + } + + void Cancel(uint64_t id) { + std::optional<Task> cancelled; + { + std::lock_guard lock(mutex_); + auto it = by_id_.find(id); + if (it == by_id_.end()) return; + cancelled.emplace(std::move(it->second->second)); + tasks_.erase(it->second); + by_id_.erase(it); + } + cv_.notify_one(); + // Release callback captures outside the queue lock. + } + + bool IsTimerThread() const { + return timer_thread_.get_id() == std::this_thread::get_id(); + } + + void Shutdown() { + Queue discarded; + { + std::lock_guard lock(mutex_); + if (shutdown_) return; + shutdown_ = true; + tasks_.swap(discarded); + by_id_.clear(); + } + cv_.notify_one(); + if (!timer_thread_.joinable()) return; + if (IsTimerThread()) { + timer_thread_.detach(); + } else { Review Comment: The structured binding element `inserted` is unused. With common warning sets (e.g., `-Wall -Wextra`, possibly with `-Werror`), this can fail builds. Consider rewriting this to avoid the unused binding (e.g., capture only the iterator), or explicitly mark the unused value `[[maybe_unused]]`. ########## src/iceberg/catalog/rest/rest_catalog.h: ########## @@ -58,14 +58,19 @@ 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. It must enqueue work + /// asynchronously without blocking the shared timer thread. /// \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); Review Comment: This changes the exported `RestCatalog::Make` signature by reordering parameters and adding `auth_executor`, which is a source-breaking API change for downstream callers. If maintaining compatibility is important, consider adding an overload matching the previous signature (e.g., `(config, metrics_executor, client_factory)`), delegating to the new one, and optionally deprecating the old overload in documentation. ########## src/iceberg/util/scheduled_executor.cc: ########## @@ -0,0 +1,249 @@ +/* + * 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 <map> +#include <thread> +#include <unordered_map> +#include <utility> + +#include "iceberg/util/scheduled_executor_internal.h" + +namespace iceberg::internal { + +// The timer thread retains its state until Run returns. This also allows the last +// timer owner to be released on the timer thread during process teardown. +class SharedTimerState : public std::enable_shared_from_this<SharedTimerState> { + public: + ~SharedTimerState() { Shutdown(); } + + void Start() { + timer_thread_ = std::thread([self = shared_from_this()] { self->Run(); }); + } + + uint64_t Schedule(std::weak_ptr<ScheduledExecutorState> state, + std::chrono::steady_clock::time_point deadline, ExecutorTask task, + std::optional<ScheduledTaskRejection> on_rejected) { + std::lock_guard lock(mutex_); + auto id = next_id_++; + auto [it, inserted] = tasks_.emplace(std::pair{deadline, id}, + Task{.state = std::move(state), + .task = std::move(task), + .on_rejected = std::move(on_rejected)}); + by_id_.emplace(id, it); + cv_.notify_one(); + return id; + } + + void Cancel(uint64_t id) { + std::optional<Task> cancelled; + { + std::lock_guard lock(mutex_); + auto it = by_id_.find(id); + if (it == by_id_.end()) return; + cancelled.emplace(std::move(it->second->second)); + tasks_.erase(it->second); + by_id_.erase(it); + } + cv_.notify_one(); + // Release callback captures outside the queue lock. + } + + bool IsTimerThread() const { + return timer_thread_.get_id() == std::this_thread::get_id(); + } + + void Shutdown() { + Queue discarded; + { + std::lock_guard lock(mutex_); + if (shutdown_) return; + shutdown_ = true; + tasks_.swap(discarded); + by_id_.clear(); + } + cv_.notify_one(); + if (!timer_thread_.joinable()) return; + if (IsTimerThread()) { + timer_thread_.detach(); + } else { + timer_thread_.join(); + } + } + + private: + struct Task { + std::weak_ptr<ScheduledExecutorState> state; + ExecutorTask task; + std::optional<ScheduledTaskRejection> on_rejected; + }; + + using Key = std::pair<std::chrono::steady_clock::time_point, uint64_t>; + using Queue = std::map<Key, Task>; + + void Run() { Review Comment: `cancelled` is intentionally used only for its lifetime (to release captures outside the lock), but some toolchains may warn about it being set-but-not-used (`-Wunused-but-set-variable`). Consider making the intent explicit (e.g., `[[maybe_unused]]` on the variable, or a `(void)cancelled;` after the critical section) to prevent warning-as-error build failures. ########## src/iceberg/util/scheduled_executor_internal.h: ########## @@ -0,0 +1,114 @@ +/* + * 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 <unordered_set> + +#include "iceberg/iceberg_export.h" +#include "iceberg/result.h" +#include "iceberg/util/executor.h" + +namespace iceberg::internal { + +class SharedTimer; +class SharedTimerState; +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: Review Comment: `ScheduledTask` is implicitly copyable, but `IsValid()` only checks that the handle still has an id and that the executor state is alive—not that the underlying task is still scheduled. If the handle is copied, cancelling one copy will not invalidate the others, and `IsValid()` for those copies will keep returning `true`, which is misleading for a cancellation handle. Consider making `ScheduledTask` move-only (delete copy ctor/assignment) to enforce single ownership, or change `IsValid()` semantics/implementation to reflect actual registration status (e.g., query state for whether `id_` is still scheduled). -- 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]
