wgtmac commented on code in PR #1033:
URL: https://github.com/apache/iceberg-cpp/pull/1033#discussion_r4320401116
##########
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:
The source-level change is intentional. This PR explicitly adopts
Make(config, client_factory, auth_executor, metrics_executor), and all
repository call sites have been updated. We accept the positional API change
rather than retain a second compatibility overload or add an options type.
Resolving this as an accepted API decision, not as a claim that the old
positional signature remains compatible.
##########
src/iceberg/util/scheduled_executor.cc:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+class SharedTimerState;
+
+/// 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;
+
+ ScheduledTask Schedule(std::chrono::steady_clock::time_point deadline,
+ ExecutorTask task,
+ std::optional<ScheduledTaskRejection> on_rejected);
+ void Cancel(uint64_t id);
+ void Shutdown();
+
+ private:
+ friend class SharedTimerState;
+
+ void Dispatch(uint64_t id, ExecutorTask task,
+ std::optional<ScheduledTaskRejection> on_rejected);
+
+ Executor& executor_;
+ std::shared_ptr<SharedTimer> timer_;
+ std::mutex mutex_;
+ std::condition_variable cv_;
+ std::unordered_set<uint64_t> scheduled_;
+ size_t active_dispatches_ = 0;
+ bool shutdown_ = false;
+};
+
+// 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;
+ }
Review Comment:
This inner state has no externally callable shutdown operation. Its Shutdown
is invoked only by SharedTimer destruction (and its own final destruction).
Every live ScheduledExecutorState retains a shared_ptr<SharedTimer>, and
Dispatch retains the owner state, so an existing scheduler cannot call this
inner Schedule after the timer owner has been destroyed. The externally
reachable per-owner Schedule already checks shutdown_. The separate comments
about constructing a new scheduler after the singleton holder itself has
undergone static destruction describe a different issue and remain open; adding
a sentinel to this inner method would not repair access to a destroyed holder.
##########
src/iceberg/util/scheduled_executor.cc:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+class SharedTimerState;
+
+/// 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;
+
+ ScheduledTask Schedule(std::chrono::steady_clock::time_point deadline,
+ ExecutorTask task,
+ std::optional<ScheduledTaskRejection> on_rejected);
+ void Cancel(uint64_t id);
+ void Shutdown();
+
+ private:
+ friend class SharedTimerState;
+
+ void Dispatch(uint64_t id, ExecutorTask task,
+ std::optional<ScheduledTaskRejection> on_rejected);
+
+ Executor& executor_;
+ std::shared_ptr<SharedTimer> timer_;
+ std::mutex mutex_;
+ std::condition_variable cv_;
+ std::unordered_set<uint64_t> scheduled_;
+ size_t active_dispatches_ = 0;
+ bool shutdown_ = false;
+};
+
+// 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() {
+ while (true) {
+ std::optional<Task> task;
+ uint64_t id;
+ {
+ std::unique_lock lock(mutex_);
+ cv_.wait(lock, [this] { return shutdown_ || !tasks_.empty(); });
+ if (shutdown_) return;
+
+ auto key = tasks_.begin()->first;
+ if (cv_.wait_until(lock, key.first, [this, key] {
+ return shutdown_ || tasks_.empty() || tasks_.begin()->first !=
key;
+ })) {
+ continue;
+ }
+ auto it = tasks_.begin();
+ id = it->first.second;
+ task.emplace(std::move(it->second));
+ by_id_.erase(id);
+ tasks_.erase(it);
+ }
+ if (auto state = task->state.lock()) {
+ state->Dispatch(id, std::move(task->task),
std::move(task->on_rejected));
+ }
+ }
+ }
+
+ std::mutex mutex_;
+ std::condition_variable cv_;
+ Queue tasks_;
+ std::unordered_map<uint64_t, Queue::iterator> by_id_;
+ uint64_t next_id_ = 1;
+ bool shutdown_ = false;
+ std::thread timer_thread_;
+};
+
+class SharedTimer {
+ public:
+ static std::shared_ptr<SharedTimer> Instance() {
+ static auto timer = std::make_shared<SharedTimer>();
+ return timer;
+ }
+
+ SharedTimer() : state_(std::make_shared<SharedTimerState>()) {
state_->Start(); }
+
+ ~SharedTimer() { state_->Shutdown(); }
+
+ uint64_t Schedule(std::weak_ptr<ScheduledExecutorState> state,
+ std::chrono::steady_clock::time_point deadline,
ExecutorTask task,
+ std::optional<ScheduledTaskRejection> on_rejected) {
+ return state_->Schedule(std::move(state), deadline, std::move(task),
+ std::move(on_rejected));
+ }
+
+ void Cancel(uint64_t id) { state_->Cancel(id); }
+
+ bool IsTimerThread() const { return state_->IsTimerThread(); }
+
+ private:
+ std::shared_ptr<SharedTimerState> state_;
+};
+
+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), timer_(SharedTimer::Instance()) {}
+
+ScheduledExecutorState::~ScheduledExecutorState() { Shutdown(); }
+
+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 = timer_->Schedule(weak_from_this(), deadline, std::move(task),
+ std::move(on_rejected));
+ scheduled_.insert(id);
+ return {weak_from_this(), id};
+}
+
+void ScheduledExecutorState::Cancel(uint64_t id) {
+ {
+ std::lock_guard lock(mutex_);
+ if (scheduled_.erase(id) == 0) return;
+ }
+ timer_->Cancel(id);
+}
+
+void ScheduledExecutorState::Shutdown() {
+ std::unordered_set<uint64_t> pending;
+ {
+ std::lock_guard lock(mutex_);
+ shutdown_ = true;
+ scheduled_.swap(pending);
+ }
+ for (auto id : pending) timer_->Cancel(id);
+ if (!timer_->IsTimerThread()) {
+ std::unique_lock lock(mutex_);
+ cv_.wait(lock, [this] { return active_dispatches_ == 0; });
+ }
+}
+
+void ScheduledExecutorState::Dispatch(uint64_t id, ExecutorTask task,
+ std::optional<ScheduledTaskRejection>
on_rejected) {
+ {
+ std::lock_guard lock(mutex_);
+ if (shutdown_ || scheduled_.erase(id) == 0) return;
+ ++active_dispatches_;
+ }
+ struct DispatchGuard {
+ ScheduledExecutorState& state;
+ ~DispatchGuard() {
+ std::lock_guard lock(state.mutex_);
+ --state.active_dispatches_;
Review Comment:
This is legal and does not identify a correctness defect. The mutex protects
the wait predicate, and only shutdown waiters are notified here. Moving the
notification outside the lock is an optional micro-optimization, with no
demonstrated contention issue on this path. We are retaining the current short
predicate-update/notification scope rather than adding a change solely for a
speculative optimization.
##########
src/iceberg/catalog/rest/auth/auth_session_internal.h:
##########
@@ -217,29 +224,88 @@ class OAuth2Session final : public AuthSession,
auto next_backoff =
std::min(std::chrono::duration_cast<std::chrono::milliseconds>(backoff * 2),
kMaxBackoff);
+ auto generation = BeginScheduledTask();
std::weak_ptr<OAuth2Session> weak_self = shared_from_this();
- auto retry_id = TokenRefreshScheduler::Instance().Schedule(
- backoff, [weak_self = std::move(weak_self), next_attempt = attempt +
1,
- next_backoff, refresh_started_at] {
+ auto weak_for_rejection = weak_self;
+ auto retry_task = config_.scheduler->Schedule(
+ backoff,
+ ExecutorTask([weak_self = std::move(weak_self), generation,
+ next_attempt = attempt + 1, next_backoff,
refresh_started_at] {
if (auto self = weak_self.lock()) {
+ self->ScheduledTaskStarted(generation);
self->DoRefreshAttempt(next_attempt, next_backoff,
refresh_started_at);
}
- });
- scheduled_task_id_.store(retry_id);
+ }),
+ ::iceberg::internal::ScheduledTaskRejection(
+ [weak_self = std::move(weak_for_rejection), generation](Status) {
+ if (auto self = weak_self.lock()) {
+ self->ScheduledTaskRejected(generation, 0);
+ }
+ }));
+ StoreScheduledTask(std::move(retry_task), generation);
}
}
- void ScheduleRefresh() {
- if (!config_.keep_refreshed || closed_.load()) return;
- auto delay = CalculateRefreshDelay();
+ void ScheduleRefresh() { ScheduleRefreshAfter(std::nullopt, 0); }
+
+ void ScheduleRefreshAfter(std::optional<std::chrono::milliseconds>
override_delay,
+ int dispatch_attempt) {
+ if (!config_.keep_refreshed || config_.scheduler == nullptr ||
closed_.load()) return;
+ auto delay = override_delay.value_or(CalculateRefreshDelay());
if (delay < std::chrono::milliseconds::zero()) return;
+ auto generation = BeginScheduledTask();
std::weak_ptr<OAuth2Session> weak_self = shared_from_this();
- auto new_id = TokenRefreshScheduler::Instance().Schedule(
- delay, [weak_self = std::move(weak_self)] {
- if (auto self = weak_self.lock()) self->DoRefresh();
- });
- scheduled_task_id_.store(new_id);
+ auto weak_for_rejection = weak_self;
+ auto task = config_.scheduler->Schedule(
+ delay, ExecutorTask([weak_self = std::move(weak_self), generation] {
+ if (auto self = weak_self.lock()) {
+ self->ScheduledTaskStarted(generation);
+ self->DoRefresh();
+ }
+ }),
+ ::iceberg::internal::ScheduledTaskRejection(
+ [weak_self = std::move(weak_for_rejection), generation,
+ dispatch_attempt](Status) {
+ if (auto self = weak_self.lock()) {
+ self->ScheduledTaskRejected(generation, dispatch_attempt);
+ }
+ }));
+ StoreScheduledTask(std::move(task), generation);
+ }
+
+ uint64_t BeginScheduledTask() {
+ std::lock_guard lock(scheduled_mutex_);
+ scheduled_task_.Cancel();
+ return ++schedule_generation_;
+ }
+
+ void ScheduledTaskStarted(uint64_t generation) {
+ std::lock_guard lock(scheduled_mutex_);
+ if (generation == schedule_generation_) {
+ ++schedule_generation_;
+ scheduled_task_.Cancel();
+ }
+ }
+
+ void ScheduledTaskRejected(uint64_t generation, int dispatch_attempt) {
+ ScheduledTaskStarted(generation);
+ static constexpr int kMaxDispatchRetries = 5;
+ static constexpr auto kInitialDispatchBackoff =
std::chrono::milliseconds(200);
+ if (dispatch_attempt >= kMaxDispatchRetries || closed_.load()) return;
+ auto multiplier = 1 << dispatch_attempt;
+ auto delay =
+ std::min(kInitialDispatchBackoff * multiplier,
std::chrono::milliseconds(10'000));
+ ScheduleRefreshAfter(delay, dispatch_attempt + 1);
Review Comment:
Addressed in 780956480 by removing generation-based registration entirely.
ScheduleRefresh now holds refresh_mutex_ across cancellation, enqueueing and
handle installation; refresh entry and any rejection-driven reschedule must
acquire that same mutex before progressing. There is only one refresh chain,
and a submission either runs its task or reports rejection, so the old
late-registration/stale-rejection replacement window is no longer present. The
rejection callback also retains the original HTTP attempt/backoff budget. The
new DispatchRejectionDoesNotResetHttpRetryBudget test covers interleaved HTTP
failures and submission rejection.
--
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]
