Copilot commented on code in PR #1033: URL: https://github.com/apache/iceberg-cpp/pull/1033#discussion_r4326262406
########## src/iceberg/util/scheduled_executor.cc: ########## @@ -0,0 +1,213 @@ +/* + * 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 <condition_variable> +#include <exception> +#include <iterator> +#include <map> +#include <mutex> +#include <thread> +#include <utility> + +#include "iceberg/util/scheduled_executor_internal.h" + +namespace iceberg::internal { + +// Kept alive so static destructors can still schedule tasks. +class SharedTimer { + public: + static SharedTimer& Instance() { + static auto* timer = new SharedTimer(); + return *timer; + } + + void Schedule(std::weak_ptr<ScheduledExecutorState> owner, + std::chrono::steady_clock::time_point deadline, ExecutorTask task, + std::optional<TaskSubmissionCallback> on_submission) { + std::lock_guard lock(mutex_); + // Equal deadlines keep insertion order. + auto it = tasks_.emplace(deadline, Task{.owner = std::move(owner), + .task = std::move(task), + .on_submission = std::move(on_submission)}); Review Comment: `std::multimap` does not guarantee preserving insertion order among equivalent keys, but the implementation and public docs claim 'Equal deadlines keep insertion order' / 'Equal deadlines are submitted in scheduling order.' If ordering for equal deadlines is required, encode a monotonic sequence into the key (e.g., `(deadline, seq)`), or store a FIFO per-deadline (e.g., `std::map<time_point, std::deque<Task>>`). If ordering is not required, the comment/docs should be updated to avoid promising it. ########## 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 Borrowed OAuth2 refresh executor. With OAuth2 credentials, + /// token-refresh-enabled=true (default) requires it; null returns InvalidArgument. + /// Must enqueue asynchronously and outlive the catalog, its auth sessions and their + /// tasks. /// \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: Reordering `RestCatalog::Make` parameters is a source-breaking change and can also silently misroute arguments for existing call sites that pass a metrics executor as the 2nd positional parameter (it now binds to `client_factory`). Consider adding an overload that preserves the previous signature (e.g., `Make(config, Executor* metrics_executor, ClientFactory client_factory = {})`) and forwards internally, or introduce a small options struct / tag types to prevent ambiguous/accidental binding. ########## src/iceberg/util/scheduled_executor.cc: ########## @@ -0,0 +1,213 @@ +/* + * 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 <condition_variable> +#include <exception> +#include <iterator> +#include <map> +#include <mutex> +#include <thread> +#include <utility> + +#include "iceberg/util/scheduled_executor_internal.h" + +namespace iceberg::internal { + +// Kept alive so static destructors can still schedule tasks. +class SharedTimer { + public: + static SharedTimer& Instance() { + static auto* timer = new SharedTimer(); + return *timer; + } + + void Schedule(std::weak_ptr<ScheduledExecutorState> owner, + std::chrono::steady_clock::time_point deadline, ExecutorTask task, + std::optional<TaskSubmissionCallback> on_submission) { + std::lock_guard lock(mutex_); + // Equal deadlines keep insertion order. + auto it = tasks_.emplace(deadline, Task{.owner = std::move(owner), + .task = std::move(task), + .on_submission = std::move(on_submission)}); + // Wake the timer only if its next deadline changed. + if (it == tasks_.begin()) { + cv_.notify_one(); + } + } + + void CancelAll(const std::weak_ptr<ScheduledExecutorState>& owner) { + Queue cancelled; // Release task captures outside the queue lock. + { + std::lock_guard lock(mutex_); + for (auto it = tasks_.begin(); it != tasks_.end();) { + auto next = std::next(it); + if (SameOwner(it->second.owner, owner)) { + cancelled.insert(tasks_.extract(it)); + } + it = next; + } + } + if (!cancelled.empty()) { + cv_.notify_one(); + } + } + + bool IsTimerThread() const { + return timer_thread_.get_id() == std::this_thread::get_id(); + } + + private: + SharedTimer() : timer_thread_([this] { Run(); }) {} + + struct Task { + std::weak_ptr<ScheduledExecutorState> owner; + ExecutorTask task; + std::optional<TaskSubmissionCallback> on_submission; + }; + + using Queue = std::multimap<std::chrono::steady_clock::time_point, Task>; Review Comment: `std::multimap` does not guarantee preserving insertion order among equivalent keys, but the implementation and public docs claim 'Equal deadlines keep insertion order' / 'Equal deadlines are submitted in scheduling order.' If ordering for equal deadlines is required, encode a monotonic sequence into the key (e.g., `(deadline, seq)`), or store a FIFO per-deadline (e.g., `std::map<time_point, std::deque<Task>>`). If ordering is not required, the comment/docs should be updated to avoid promising it. ########## src/iceberg/util/scheduled_executor.cc: ########## @@ -0,0 +1,213 @@ +/* + * 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 <condition_variable> +#include <exception> +#include <iterator> +#include <map> +#include <mutex> +#include <thread> +#include <utility> + +#include "iceberg/util/scheduled_executor_internal.h" + +namespace iceberg::internal { + +// Kept alive so static destructors can still schedule tasks. +class SharedTimer { + public: + static SharedTimer& Instance() { + static auto* timer = new SharedTimer(); + return *timer; + } + + void Schedule(std::weak_ptr<ScheduledExecutorState> owner, + std::chrono::steady_clock::time_point deadline, ExecutorTask task, + std::optional<TaskSubmissionCallback> on_submission) { + std::lock_guard lock(mutex_); + // Equal deadlines keep insertion order. + auto it = tasks_.emplace(deadline, Task{.owner = std::move(owner), + .task = std::move(task), + .on_submission = std::move(on_submission)}); + // Wake the timer only if its next deadline changed. + if (it == tasks_.begin()) { + cv_.notify_one(); + } + } + + void CancelAll(const std::weak_ptr<ScheduledExecutorState>& owner) { + Queue cancelled; // Release task captures outside the queue lock. + { + std::lock_guard lock(mutex_); + for (auto it = tasks_.begin(); it != tasks_.end();) { + auto next = std::next(it); + if (SameOwner(it->second.owner, owner)) { + cancelled.insert(tasks_.extract(it)); + } + it = next; + } + } + if (!cancelled.empty()) { + cv_.notify_one(); + } + } + + bool IsTimerThread() const { + return timer_thread_.get_id() == std::this_thread::get_id(); + } + + private: + SharedTimer() : timer_thread_([this] { Run(); }) {} + + struct Task { + std::weak_ptr<ScheduledExecutorState> owner; + ExecutorTask task; + std::optional<TaskSubmissionCallback> on_submission; + }; + + using Queue = std::multimap<std::chrono::steady_clock::time_point, Task>; + + static bool SameOwner(const std::weak_ptr<ScheduledExecutorState>& a, + const std::weak_ptr<ScheduledExecutorState>& b) { + return !a.owner_before(b) && !b.owner_before(a); + } + + void Run(); + + std::mutex mutex_; + std::condition_variable cv_; + Queue tasks_; + std::thread timer_thread_; +}; + +class ScheduledExecutorState + : public std::enable_shared_from_this<ScheduledExecutorState> { + public: + explicit ScheduledExecutorState(Executor& executor) + : executor_(executor), timer_(SharedTimer::Instance()) {} + + ScheduledExecutorState(const ScheduledExecutorState&) = delete; + ScheduledExecutorState& operator=(const ScheduledExecutorState&) = delete; + + void Schedule(std::chrono::steady_clock::time_point deadline, ExecutorTask task, + std::optional<TaskSubmissionCallback> on_submission) { + std::lock_guard lock(mutex_); + if (shutdown_) { + return; + } + timer_.Schedule(weak_from_this(), deadline, std::move(task), + std::move(on_submission)); + } + + void Shutdown() { + { + std::lock_guard lock(mutex_); + shutdown_ = true; + } + timer_.CancelAll(weak_from_this()); + if (!timer_.IsTimerThread()) { + std::unique_lock lock(mutex_); + cv_.wait(lock, [this] { return active_dispatches_ == 0; }); + } + } + + private: + friend class SharedTimer; + + void Dispatch(ExecutorTask task, std::optional<TaskSubmissionCallback> on_submission) { + { + std::lock_guard lock(mutex_); + // Shutdown may race with popping a task. + if (shutdown_) { + return; + } + ++active_dispatches_; + } + struct DispatchGuard { + ScheduledExecutorState& state; + ~DispatchGuard() { + std::lock_guard lock(state.mutex_); + if (--state.active_dispatches_ == 0) { + state.cv_.notify_all(); + } + } + } guard{*this}; + + Status status; + try { + status = executor_.Submit(std::move(task)); + } catch (const std::exception& e) { + status = UnknownError("Executor::Submit threw: {}", e.what()); + } catch (...) { + status = UnknownError("Executor::Submit threw a non-standard exception"); + } + if (!on_submission.has_value()) { + return; + } + try { + std::move (*on_submission)(std::move(status)); Review Comment: Minor readability/style issue: `std::move (*on_submission)` is unusual spacing and easy to misread. Prefer `std::move(*on_submission)(...)` for consistency with typical C++ formatting and to reduce reviewer confusion (same pattern appears elsewhere in tests). -- 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]
