Copilot commented on code in PR #1033: URL: https://github.com/apache/iceberg-cpp/pull/1033#discussion_r4321203941
########## src/iceberg/util/scheduled_executor.cc: ########## @@ -0,0 +1,240 @@ +/* + * 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 <map> +#include <mutex> +#include <thread> +#include <unordered_map> +#include <unordered_set> +#include <utility> + +#include "iceberg/util/scheduled_executor_internal.h" + +namespace iceberg::internal { + +class SharedTimer; + +/// Per-executor registrations; all instances share the same timer thread. +class ScheduledExecutorState + : public std::enable_shared_from_this<ScheduledExecutorState> { + public: + explicit ScheduledExecutorState(Executor& executor); + ~ScheduledExecutorState(); + + ScheduledExecutorState(const ScheduledExecutorState&) = delete; + ScheduledExecutorState& operator=(const ScheduledExecutorState&) = delete; + + ScheduledTaskHandle Schedule( + std::chrono::steady_clock::time_point deadline, ExecutorTask task, + std::optional<TaskSubmissionFailureCallback> on_submit_failure); + void Cancel(uint64_t id); + void Shutdown(); + + private: + friend class SharedTimer; + + void Dispatch(uint64_t id, ExecutorTask task, + std::optional<TaskSubmissionFailureCallback> on_submit_failure); + + Executor& executor_; + SharedTimer& timer_; + std::mutex mutex_; + std::condition_variable cv_; + std::unordered_set<uint64_t> scheduled_; + size_t active_dispatches_ = 0; + bool shutdown_ = false; +}; + +// Process-lifetime storage keeps scheduling safe even from static destructors. +class SharedTimer { + public: + static SharedTimer& Instance() { + static auto* timer = new SharedTimer(); + return *timer; + } + + uint64_t Schedule(std::weak_ptr<ScheduledExecutorState> state, + std::chrono::steady_clock::time_point deadline, ExecutorTask task, + std::optional<TaskSubmissionFailureCallback> on_submit_failure) { + 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_submit_failure = std::move(on_submit_failure)}); Review Comment: `inserted` is never used. With common warning levels (and especially with `-Werror`), this may fail the build. Consider avoiding the structured binding entirely (e.g., store only the iterator from `emplace(...).first`) or marking the variable as intentionally 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 public `RestCatalog::Make` parameter order in a way that breaks existing positional call sites (previously `Make(config, metrics_executor, client_factory)`). To preserve source compatibility, consider adding an overload matching the old signature (and forwarding internally to the new one), or keep the original ordering and append `auth_executor` as an additional trailing optional parameter. ########## src/iceberg/util/scheduled_executor.cc: ########## @@ -0,0 +1,240 @@ +/* + * 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 <map> +#include <mutex> +#include <thread> +#include <unordered_map> +#include <unordered_set> +#include <utility> + +#include "iceberg/util/scheduled_executor_internal.h" + +namespace iceberg::internal { + +class SharedTimer; + +/// Per-executor registrations; all instances share the same timer thread. +class ScheduledExecutorState + : public std::enable_shared_from_this<ScheduledExecutorState> { + public: + explicit ScheduledExecutorState(Executor& executor); + ~ScheduledExecutorState(); + + ScheduledExecutorState(const ScheduledExecutorState&) = delete; + ScheduledExecutorState& operator=(const ScheduledExecutorState&) = delete; + + ScheduledTaskHandle Schedule( + std::chrono::steady_clock::time_point deadline, ExecutorTask task, + std::optional<TaskSubmissionFailureCallback> on_submit_failure); + void Cancel(uint64_t id); + void Shutdown(); + + private: + friend class SharedTimer; + + void Dispatch(uint64_t id, ExecutorTask task, + std::optional<TaskSubmissionFailureCallback> on_submit_failure); + + Executor& executor_; + SharedTimer& timer_; + std::mutex mutex_; + std::condition_variable cv_; + std::unordered_set<uint64_t> scheduled_; + size_t active_dispatches_ = 0; + bool shutdown_ = false; +}; + +// Process-lifetime storage keeps scheduling safe even from static destructors. +class SharedTimer { + public: + static SharedTimer& Instance() { + static auto* timer = new SharedTimer(); + return *timer; + } + + uint64_t Schedule(std::weak_ptr<ScheduledExecutorState> state, + std::chrono::steady_clock::time_point deadline, ExecutorTask task, + std::optional<TaskSubmissionFailureCallback> on_submit_failure) { + 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_submit_failure = std::move(on_submit_failure)}); + 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. Review Comment: `cancelled` is not referenced, which can trigger an unused-variable warning. If the intent is to ensure captured resources are released after the lock is dropped, make that explicit in a warning-free way (e.g., reset/destroy it outside the critical section, or mark it `[[maybe_unused]]` with a brief comment). -- 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]
