github-actions[bot] commented on code in PR #69670:
URL: https://github.com/apache/doris/pull/69670#discussion_r4386116556
##########
be/src/util/s3_util.cpp:
##########
@@ -251,9 +257,27 @@ S3ClientFactory::S3ClientFactory() {
}
S3ClientFactory::~S3ClientFactory() {
+ // ExecEnv unregisters the policy before deleting CacheManager. Unit tests
+ // that use the factory without ExecEnv never register a policy.
+ DORIS_CHECK(_cache_policy == nullptr);
+ _cache->close();
+ // ShutdownAPI must not precede SDK client destruction. Outstanding holders
+ // here violate the shutdown contract; clearing the cache alone is
insufficient.
+ DORIS_CHECK(_cache->stats().live == 0);
Review Comment:
[P2] Release packed-file owners before asserting zero live clients
With `enable_graceful_exit_check=true`, start a cloud BE with only HDFS
vaults (or no vaults), then add an S3 vault and use it for packed files. In
that startup path `ExecEnv::_init()` constructs `PackedFileManager` at line 431
before first constructing this factory at line 479.
`PackedFileManager::ensure_file_system()` retains the S3 filesystem in
`_file_systems`, while `stop_background_manager()` only joins its thread;
`ExecEnv::destroy()` never clears those owners. Reverse static destruction
therefore reaches this new check before the packed-file manager releases its
client, and orderly shutdown aborts with `live > 0`. Drain the packed-file
manager's filesystem/context ownership during teardown, or otherwise ensure the
SDK factory outlives it; the default `_exit` path hides this ordering problem.
##########
be/src/util/s3_client_cache.cpp:
##########
@@ -0,0 +1,477 @@
+// 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 "util/s3_client_cache.h"
+
+#include <bvar/bvar.h>
+
+#include <algorithm>
+#include <atomic>
+#include <condition_variable>
+#include <exception>
+#include <future>
+#include <iterator>
+#include <list>
+#include <mutex>
+#include <optional>
+#include <unordered_map>
+#include <utility>
+
+#include "core/custom_allocator.h"
+#include "runtime/memory/cache_manager.h"
+#include "runtime/memory/cache_policy.h"
+#include "runtime/memory/mem_tracker_limiter.h"
+#include "runtime/thread_context.h"
+#include "util/s3_util.h"
+#include "util/time.h"
+
+namespace doris {
+
+struct S3ClientCache::Impl {
+ struct Metrics {
+ // Atomic counters do not allocate per-thread bvar agents while a cache
+ // or manager mutex is held (especially important during memory GC).
+ struct Counter {
+ std::atomic<int64_t> value {0};
+ bvar::PassiveStatus<int64_t> metric {
+ [](void* p) {
+ return
static_cast<Counter*>(p)->value.load(std::memory_order_relaxed);
+ },
+ this};
+ void add(int64_t delta) { value.fetch_add(delta,
std::memory_order_relaxed); }
+ int64_t get_value() const { return
value.load(std::memory_order_relaxed); }
+ };
+ Counter hits;
+ Counter weak_hits;
+ Counter misses;
+ Counter waits;
+ Counter created;
+ Counter destroyed;
+ Counter live;
+ Counter failures;
+ Counter evictions;
+ };
+
+ struct ClientLifetime {
+ ClientLifetime(std::shared_ptr<MemTrackerLimiter> mem_tracker,
+ std::shared_ptr<Metrics> client_metrics)
+ : tracker(std::move(mem_tracker)),
metrics(std::move(client_metrics)) {
+ metrics->created.add(1);
+ metrics->live.add(1);
+ }
+ ~ClientLifetime() {
+ SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(tracker);
+ metrics->destroyed.add(1);
+ metrics->live.add(-1);
+ }
+ std::shared_ptr<MemTrackerLimiter> tracker;
+ std::shared_ptr<Metrics> metrics;
+ };
+
+ struct Retained {
+ // Map-key addresses survive rehash. Only retained nodes dereference
+ // this pointer; detached release batches never inspect it.
+ const S3ClientConf* key;
+ Client client;
+ int64_t last_acquire;
+ };
+ using RetainedList = std::list<Retained, CustomStdAllocator<Retained>>;
+
+ struct Entry {
+ std::weak_ptr<ObjStorageClient> live;
+ std::optional<RetainedList::iterator> retained;
+ };
+ struct Pending {
+ std::promise<Result<Client>> promise;
+ std::shared_future<Result<Client>> future =
promise.get_future().share();
+ };
+ template <typename Value>
+ using Map =
+ std::unordered_map<S3ClientConf, Value, S3ClientConfHash,
std::equal_to<S3ClientConf>,
+ CustomStdAllocator<std::pair<const
S3ClientConf, Value>>>;
+
+ struct ReleaseBatch {
+ explicit ReleaseBatch(std::shared_ptr<MemTrackerLimiter> mem_tracker)
+ : tracker(std::move(mem_tracker)) {}
+ ~ReleaseBatch() {
+ SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(tracker);
+ clients.clear();
+ }
+ std::shared_ptr<MemTrackerLimiter> tracker;
+ RetainedList clients;
+ };
+
+ Impl(size_t cache_capacity, uint32_t idle_ttl, Clock clock_source)
+ : capacity(cache_capacity),
+ ttl(idle_ttl),
+ clock(clock_source ? std::move(clock_source)
+ : Clock([] { return MonotonicSeconds(); })),
+
tracker(MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::CACHE,
+ "S3ClientCache")),
+ metrics(std::make_shared<Metrics>()),
+ entries_metric(std::make_unique<bvar::PassiveStatus<size_t>>(
+ [](void* p) { return
static_cast<Impl*>(p)->snapshot().entries; }, this)),
+ indexed_metric(std::make_unique<bvar::PassiveStatus<size_t>>(
+ [](void* p) { return
static_cast<Impl*>(p)->snapshot().indexed; }, this)),
+ inflight_metric(std::make_unique<bvar::PassiveStatus<size_t>>(
+ [](void* p) { return
static_cast<Impl*>(p)->snapshot().inflight; }, this)) {
+ DORIS_CHECK(ttl > 0);
+ }
+
+ ~Impl() {
+ // Destroy the samplers, not just their exposure, before releasing the
+ // maps they inspect. Hiding a bvar alone does not stop its series
sampler.
+ entries_metric.reset();
+ indexed_metric.reset();
+ inflight_metric.reset();
+ SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(tracker);
+ // Free bucket arrays under the same tracker used for insertion, too.
+ Map<Entry>().swap(entries);
+ Map<std::shared_ptr<Pending>>().swap(pending);
+ }
+
+ Client track(Client client) {
+ // Do not introduce an aliasing control block: list_objects() uses
+ // shared_from_this(), and its iterator must keep the cache's weak
entry
+ // alive too. A client-owned guard observes actual client destruction
+ // without owning the client or referring back to Impl.
+ client->set_lifetime_guard(std::make_shared<ClientLifetime>(tracker,
metrics));
+ return client;
+ }
+
+ // mutex held. Expiring retention does not invalidate a live weak entry.
+ void evict(RetainedList::iterator it, ReleaseBatch& release) {
+ auto entry = entries.find(*it->key);
+ DCHECK(entry != entries.end());
+ entry->second.retained.reset();
+ release.clients.splice(release.clients.end(), retained, it);
+ metrics->evictions.add(1);
+ }
+
+ void sweep(int64_t now, bool all, ReleaseBatch& release, bool clean_index
= true) {
+ while (!retained.empty()) {
+ auto oldest = std::prev(retained.end());
+ // Entries are ordered by factory acquisition time, not IO
activity.
+ // A long scan can outlive TTL and continues through its own owner.
+ if (!all && retained.size() <= capacity && now -
oldest->last_acquire < ttl) {
+ break;
+ }
+ evict(oldest, release);
+ }
+ if (!clean_index) {
+ return; // Acquisitions must not scan all live configurations.
+ }
+ for (auto it = entries.begin(); it != entries.end();) {
+ // Once the last caller and retention release their owners, the key
+ // and weak control block must also be reclaimed.
+ if (it->second.live.expired()) {
+ DCHECK(!it->second.retained.has_value());
+ it = entries.erase(it);
+ } else {
+ ++it;
+ }
+ }
+ // Amortize weak-index scans over growth. This also reclaims dead keys
+ // when periodic memory GC is disabled, without scanning all active
+ // configurations on every acquisition.
+ next_index_sweep = entries.size() * 2 + 64;
+ }
+
+ void retain(typename Map<Entry>::iterator entry, const Client& client,
int64_t now,
+ ReleaseBatch& release) {
+ if (capacity == 0) {
+ return; // Explicitly disable warm retention, while preserving
weak reuse.
+ }
+ if (entry->second.retained.has_value()) {
+ auto it = *entry->second.retained;
+ it->last_acquire = now;
+ retained.splice(retained.begin(), retained, it);
+ } else {
+ retained.push_front({&entry->first, client, now});
+ entry->second.retained = retained.begin();
+ }
+ while (retained.size() > capacity) {
+ evict(std::prev(retained.end()), release);
+ }
+ }
+
+ Result<Client> acquire(const S3ClientConf& conf, const Creator& creator) {
+ SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(tracker);
+ ReleaseBatch release(tracker); // Destruction is outside the mutex on
every exit.
+ // A successful weak lock can become the last owner while retain()
+ // allocates a list node. Keep it outside the lock's scope so
allocation
+ // failure cannot run its SDK destructor under the cache mutex.
+ Client cached;
+ std::shared_ptr<Pending> task;
+ bool leader = false;
+ {
+ std::lock_guard lock(mutex);
+ if (closed) {
+ return ResultError(Status::InternalError("S3 client cache is
closed"));
+ }
+ const auto now = clock();
+ // Scan the full weak index only after sufficient growth; otherwise
+ // only inspect the bounded retention LRU and the requested key.
+ sweep(now, false, release, entries.size() >= next_index_sweep);
+ auto it = entries.find(conf);
+ if (it != entries.end()) {
+ cached = it->second.live.lock();
+ if (cached) {
+ metrics->hits.add(1);
+ if (!it->second.retained.has_value()) {
+ metrics->weak_hits.add(1);
+ }
+ retain(it, cached, now, release);
+ return cached;
+ }
+ // The final external owner can disappear after sweep().
+ entries.erase(it);
+ }
+ metrics->misses.add(1);
+ auto creating = pending.find(conf);
+ if (creating != pending.end()) {
+ metrics->waits.add(1);
+ task = creating->second;
+ } else {
+ task = std::make_shared<Pending>();
+ pending.emplace(conf, task);
+ leader = true;
+ }
+ }
+ if (!leader) {
+ return task->future.get(); // No cache/holder/manager lock is held
while waiting.
+ }
+ return create_and_publish(conf, creator, task, release);
+ }
+
+ Result<Client> create_and_publish(const S3ClientConf& conf, const Creator&
creator,
+ const std::shared_ptr<Pending>& task,
ReleaseBatch& release) {
+ try {
+ auto result = creator(conf);
+ if (result.has_value()) {
+ DORIS_CHECK(result.value() != nullptr);
+ result = track(std::move(result).value());
+ } else {
+ metrics->failures.add(1);
+ }
+ {
+ std::lock_guard lock(mutex);
+ // close() waits for this publication. The caller still
receives
+ // its result, but no new retention is introduced during close.
+ if (result.has_value() && !closed) {
+ auto [entry, inserted] = entries.try_emplace(conf);
+ DORIS_CHECK(inserted);
+ try {
+ retain(entry, result.value(), clock(), release);
+ } catch (...) {
+ // Allocation of the retention node can fail. Do not
+ // expose a half-published client while unwinding to
the
+ // common failure notification below.
+ entries.erase(entry);
+ throw;
+ }
+ entry->second.live = result.value();
+ }
+ task->promise.set_value(result);
+ pending.erase(conf);
Review Comment:
[P2] Make cache mutation safe when configuration hashing throws
`S3ClientConf::get_hash()` allocates `ak + sk` for ordinary AWS credential
lengths. If that allocation throws in `pending.erase(conf)` after
`set_value(result)`, the catch calls `set_exception()` on an already-satisfied
promise and throws `future_error` before removing the pending task. That task
remains forever, so `close()` waits indefinitely even after memory recovers.
The same throwing hash also breaks retention rollback: publishing B into a full
LRU pushes B before `evict(A)` hashes A; if that allocation throws,
`entries.erase(entry)` removes B's key but leaves its retained node pointing to
freed storage for a later sweep. Make these hashes/cleanup nonthrowing, or
fully roll back the map/LRU/pending state without fulfilling a promise twice.
The creator-exception tests do not exercise either allocation point.
--
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]
