wgtmac commented on code in PR #1033:
URL: https://github.com/apache/iceberg-cpp/pull/1033#discussion_r4320397332
##########
src/iceberg/catalog/rest/rest_catalog.h:
##########
@@ -63,9 +63,13 @@ class ICEBERG_REST_EXPORT RestCatalog final
/// 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.
+ /// \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.
static Result<std::shared_ptr<RestCatalog>> Make(const
RestCatalogProperties& config,
Executor* metrics_executor
= nullptr,
- ClientFactory
client_factory = {});
+ ClientFactory
client_factory = {},
+ Executor* auth_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,143 @@
+/*
+ * 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 (auto state = state_.lock()) state->Cancel(id_);
+ state_.reset();
+ id_ = 0;
+}
+
+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 ScheduledTask(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();
Review Comment:
The reported queue-pop window is addressed in the current implementation.
Run removes a task from the shared queue, but the owner registration remains in
ScheduledExecutorState::scheduled_. Cancel removes that registration, and
Dispatch checks/removes it under the same mutex before calling Submit.
Cancellation therefore still wins after queue removal if dispatch has not
begun. Once Dispatch has claimed the registration, submission has begun and
cancellation is not an interrupt/join operation. OAuth Close separately gates
callbacks on closed_.
##########
src/iceberg/util/scheduled_executor.cc:
##########
@@ -0,0 +1,143 @@
+/*
+ * 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 (auto state = state_.lock()) state->Cancel(id_);
+ state_.reset();
+ id_ = 0;
+}
+
+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 ScheduledTask(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();
Review Comment:
This does not leave stale scheduled_ entries. In the referenced revision,
Cancel already erased the ID from scheduled_ before inserting it into
cancelled_, so discarding that cancelled queue item did not need to erase it a
second time. The current implementation also removes the owner registration on
Cancel and directly removes the shared queue entry; the cancelled_ set no
longer exists.
##########
src/iceberg/util/scheduled_executor.cc:
##########
@@ -0,0 +1,143 @@
+/*
+ * 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 (auto state = state_.lock()) state->Cancel(id_);
+ state_.reset();
+ id_ = 0;
+}
+
+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 ScheduledTask(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);
Review Comment:
Addressed by the shared-timer implementation. The queue is now a std::map
ordered by (deadline, id), and Run moves the mapped Task before erasing the
iterator. There is no priority_queue::top() or const_cast in the current code.
--
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]
