Coverage Report

Created: 2026-10-30 11:52

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/io/cache/block_file_cache.h
Line
Count
Source
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#pragma once
19
20
#include <bvar/bvar.h>
21
#include <concurrentqueue.h>
22
23
#include <algorithm>
24
#include <array>
25
#include <atomic>
26
#include <boost/lockfree/spsc_queue.hpp>
27
#include <condition_variable>
28
#include <functional>
29
#include <limits>
30
#include <map>
31
#include <memory>
32
#include <mutex>
33
#include <optional>
34
#include <thread>
35
#include <unordered_map>
36
#include <unordered_set>
37
#include <vector>
38
39
#include "io/cache/block_file_cache_ttl_mgr.h"
40
#include "io/cache/cache_lru_dumper.h"
41
#include "io/cache/file_block.h"
42
#include "io/cache/file_cache_common.h"
43
#include "io/cache/file_cache_storage.h"
44
#include "io/cache/lru_queue_recorder.h"
45
#include "runtime/runtime_profile.h"
46
#include "util/threadpool.h"
47
48
namespace doris::io {
49
using RecycleFileCacheKeys = moodycamel::ConcurrentQueue<FileCacheKey>;
50
51
class LockScopedTimer {
52
public:
53
1.89M
    LockScopedTimer() : start_(std::chrono::steady_clock::now()) {}
54
1.89M
    ~LockScopedTimer() {
55
1.89M
        auto end = std::chrono::steady_clock::now();
56
1.89M
        auto duration_us =
57
1.89M
                std::chrono::duration_cast<std::chrono::microseconds>(end - start_).count();
58
1.89M
        if (duration_us > config::cache_lock_held_long_tail_threshold_us) {
59
0
            LOG(WARNING) << "Lock held time " << std::to_string(duration_us) << "us. "
60
0
                         << get_stack_trace();
61
0
        }
62
1.89M
    }
63
64
private:
65
    std::chrono::time_point<std::chrono::steady_clock> start_;
66
};
67
68
// Note: the cache_lock is scoped, so do not add do...while(0) here.
69
#define SCOPED_CACHE_LOCK(MUTEX, cache)                                                           \
70
1.88M
    std::chrono::time_point<std::chrono::steady_clock> start_time =                               \
71
1.88M
            std::chrono::steady_clock::now();                                                     \
72
1.88M
    std::lock_guard cache_lock(MUTEX);                                                            \
73
1.88M
    std::chrono::time_point<std::chrono::steady_clock> acq_time =                                 \
74
1.88M
            std::chrono::steady_clock::now();                                                     \
75
1.88M
    auto duration_us =                                                                            \
76
1.88M
            std::chrono::duration_cast<std::chrono::microseconds>(acq_time - start_time).count(); \
77
1.88M
    *(cache->_cache_lock_wait_time_us) << duration_us;                                            \
78
1.88M
    if (duration_us > config::cache_lock_wait_long_tail_threshold_us) {                           \
79
0
        LOG(WARNING) << "Lock wait time " << std::to_string(duration_us) << "us. "                \
80
0
                     << get_stack_trace() << std::endl;                                           \
81
0
    }                                                                                             \
82
1.88M
    LockScopedTimer cache_lock_timer;
83
84
class AsyncCacheWriteManager;
85
class FSFileCacheStorage;
86
class InflightWriteBufferIndex;
87
88
struct FileBlocksProbeResult {
89
    explicit FileBlocksProbeResult(std::vector<FileBlockSPtr> file_blocks_)
90
295
            : file_blocks(std::move(file_blocks_)) {}
91
25
    FileBlocksProbeResult(FileBlocksProbeResult&&) noexcept = default;
92
    FileBlocksProbeResult& operator=(FileBlocksProbeResult&&) noexcept = delete;
93
    FileBlocksProbeResult(const FileBlocksProbeResult&) = delete;
94
    FileBlocksProbeResult& operator=(const FileBlocksProbeResult&) = delete;
95
    ~FileBlocksProbeResult();
96
97
    /// One entry per cache-block-sized input slot, in offset order. A null entry is a cache miss;
98
    /// a non-null entry covers the whole slot. Its right boundary can exceed the final short slot
99
    /// while a file writer still owns a full-size preallocated tail block. Retaining and releasing
100
    /// a probe result never acquires or completes downloader ownership.
101
    std::vector<FileBlockSPtr> file_blocks;
102
};
103
104
// NeedUpdateLRUBlocks keeps FileBlockSPtr entries that require LRU updates in a
105
// deduplicated, sharded container. Entries are keyed by the raw FileBlock
106
// pointer so that multiple shared_ptr copies of the same block are treated as a
107
// single pending update. The structure is thread-safe and optimized for high
108
// contention insert/drain workloads in the background update thread.
109
// Note that Blocks are updated in batch, internal order is not important.
110
class NeedUpdateLRUBlocks {
111
public:
112
428
    NeedUpdateLRUBlocks() = default;
113
114
    // Insert a block into the pending set. Returns true only when the block
115
    // was not already queued. Null inputs are ignored.
116
    bool insert(FileBlockSPtr block, size_t max_queue_size = std::numeric_limits<size_t>::max());
117
118
    // Drain up to `limit` unique blocks into `output`. The method returns how
119
    // many blocks were actually drained and shrinks the internal size
120
    // accordingly.
121
    size_t drain(size_t limit, std::vector<FileBlockSPtr>* output);
122
123
    // Remove every pending block from the structure and reset the size.
124
    void clear();
125
126
    // Thread-safe approximate size of queued unique blocks.
127
312k
    size_t size() const { return _size.load(std::memory_order_relaxed); }
128
129
private:
130
    static constexpr size_t kShardCount = 64;
131
    static constexpr size_t kShardMask = kShardCount - 1;
132
133
    struct Shard {
134
        std::mutex mutex;
135
        std::unordered_map<FileBlock*, FileBlockSPtr> entries;
136
    };
137
138
    size_t shard_index(FileBlock* ptr) const;
139
    void decrease_size(size_t delta);
140
141
    std::array<Shard, kShardCount> _shards;
142
    std::atomic<size_t> _size {0};
143
};
144
145
// The BlockFileCache is responsible for the management of the blocks
146
// The current strategies are lru and ttl.
147
148
struct FileBlockCell {
149
    friend class FileBlock;
150
151
    FileBlockSPtr file_block;
152
    /// Iterator is put here on first reservation attempt, if successful.
153
    std::optional<LRUQueue::Iterator> queue_iterator;
154
155
    mutable int64_t atime {0};
156
1.38M
    void update_atime() const {
157
1.38M
        atime = std::chrono::duration_cast<std::chrono::seconds>(
158
1.38M
                        std::chrono::steady_clock::now().time_since_epoch())
159
1.38M
                        .count();
160
1.38M
    }
161
162
    /// Pointer to file block is always hold by the cache itself. Every other strong reference
163
    /// means somebody is still using the block. CachedRemoteFileReader keeps only weak references
164
    /// in its direct-read map, so however many readers know about a block, they do not keep it
165
    /// busy.
166
1.43M
    bool releasable() const { return file_block.use_count() == 1; }
167
168
2.71M
    size_t size() const { return file_block->_block_range.size(); }
169
170
    FileBlockCell() = default;
171
    FileBlockCell(FileBlockSPtr file_block, std::lock_guard<std::mutex>& cache_lock);
172
    FileBlockCell(FileBlockCell&& other) noexcept
173
829k
            : file_block(std::move(other.file_block)),
174
829k
              queue_iterator(other.queue_iterator),
175
829k
              atime(other.atime) {
176
829k
        file_block->cell = this;
177
829k
    }
178
179
    FileBlockCell& operator=(const FileBlockCell&) = delete;
180
    FileBlockCell(const FileBlockCell&) = delete;
181
182
0
    size_t dowloading_size() const { return file_block->_downloaded_size; }
183
};
184
185
class BlockFileCache {
186
    friend class FSFileCacheStorage;
187
    friend class MemFileCacheStorage;
188
    friend class FileBlock;
189
    friend struct FileBlocksHolder;
190
    friend class CacheLRUDumper;
191
    friend class LRUQueueRecorder;
192
    friend struct FileBlockCell;
193
    friend class BlockFileCacheTest;
194
195
public:
196
    // hash the file_name to uint128
197
    static UInt128Wrapper hash(const std::string& path);
198
199
    BlockFileCache(const std::string& cache_base_path, const FileCacheSettings& cache_settings);
200
201
    virtual ~BlockFileCache();
202
203
    /// Restore cache from local filesystem.
204
    Status initialize();
205
206
    /// Cache capacity in bytes.
207
14
    [[nodiscard]] size_t capacity() const { return _capacity; }
208
209
    // try to release all releasable block
210
    // it maybe hang the io/system
211
    size_t try_release();
212
213
5.95k
    [[nodiscard]] const std::string& get_base_path() const { return _cache_base_path; }
214
215
    // Get storage for inspection
216
1
    FileCacheStorage* get_storage() const { return _storage.get(); }
217
218
    /**
219
         * Given an `offset` and `size` representing [offset, offset + size) bytes interval,
220
         * return list of cached non-overlapping non-empty
221
         * file blocks `[block1, ..., blockN]` which intersect with given interval.
222
         *
223
         * blocks in returned list are ordered in ascending order and represent a full contiguous
224
         * interval (no holes). Each block in returned list has state: DOWNLOADED, DOWNLOADING or EMPTY.
225
         *
226
         * As long as pointers to returned file blocks are hold
227
         * it is guaranteed that these file blocks are not removed from cache.
228
         */
229
    FileBlocksHolder get_or_set(const UInt128Wrapper& hash, size_t offset, size_t size,
230
                                CacheContext& context);
231
232
    /// Whether existing DOWNLOADED/DOWNLOADING blocks fully cover `[offset, offset + size)`.
233
    /// Handles arbitrary block boundaries, including restored upload blocks. May load metadata,
234
    /// but never reserves cache cells, waits for downloads or touches LRU state.
235
    bool is_range_downloaded_or_downloading(const UInt128Wrapper& hash, size_t offset, size_t size,
236
                                            const CacheContext& context);
237
238
    /// Probe the block-aligned `[offset, offset + size)` range without creating cache cells or
239
    /// touching LRU state. The result contains one ordered slot per cache block; each slot is null
240
    /// on miss or owns an existing block that starts at and covers the slot. A final short slot can
241
    /// be covered by a full-size block preallocated by a file writer. `context` supplies cache
242
    /// metadata when lazy loading is required.
243
    FileBlocksProbeResult probe(const UInt128Wrapper& hash, size_t offset, size_t size,
244
                                const CacheContext& context);
245
246
    /// Touch `block` after a successful local read, provided it is still the cached cell. The
247
    /// supplied `context` controls the target LRU queue and query-level accounting.
248
    void touch_probe_block_if_cached(const FileBlockSPtr& block, const CacheContext& context);
249
250
    /// Check whether `block` is being deleted while taking cache/block locks in canonical order.
251
    bool is_block_deleting(const FileBlockSPtr& block) const;
252
253
    /// Return this cache disk's async-write manager.
254
466
    AsyncCacheWriteManager* async_write_manager() const { return _async_write_manager.get(); }
255
256
    /// Return this cache disk's inflight payload index.
257
149
    InflightWriteBufferIndex* inflight_write_buffer_index() const {
258
149
        return _inflight_write_buffer_index.get();
259
149
    }
260
261
    /**
262
     * Return existing downloaded blocks only if they fully cover [offset, offset + size).
263
     * This lookup is read-only: it does not reserve cache space or create EMPTY blocks.
264
     */
265
    Status get_downloaded_blocks_if_fully_covered(const UInt128Wrapper& hash, size_t offset,
266
                                                  size_t size, const CacheContext& context,
267
                                                  FileBlocks* blocks, bool* fully_covered);
268
269
    /// Diagnostic snapshot of resident DOWNLOADED intervals, clipped to the requested range.
270
    /// Handles irregular block boundaries; never loads metadata, reserves space or touches LRU.
271
    std::vector<FileBlock::Range> downloaded_ranges(const UInt128Wrapper& hash, size_t offset,
272
                                                    size_t size);
273
274
    /**
275
     * record blocks read directly by CachedRemoteFileReader
276
     */
277
    void add_need_update_lru_block(FileBlockSPtr block);
278
279
    /**
280
     * Clear all cached data for this cache instance async
281
     *
282
     * @returns summary message
283
     */
284
    std::string clear_file_cache_async();
285
    std::string clear_file_cache_sync();
286
287
    /**
288
     * Reset the cache capacity. If the new_capacity is smaller than _capacity, the redundant data will be remove async.
289
     *
290
     * @returns summary message
291
     */
292
    std::string reset_capacity(size_t new_capacity);
293
294
    std::map<size_t, FileBlockSPtr> get_blocks_by_key(const UInt128Wrapper& hash);
295
296
    /// For debug and UT
297
    std::string dump_structure(const UInt128Wrapper& hash);
298
    std::string dump_single_cache_type(const UInt128Wrapper& hash, size_t offset);
299
300
    void dump_lru_queues(bool force);
301
302
    [[nodiscard]] size_t get_used_cache_size(FileCacheType type) const;
303
304
    [[nodiscard]] size_t get_file_blocks_num(FileCacheType type) const;
305
306
    // change the block cache type
307
    void change_cache_type(const UInt128Wrapper& hash, size_t offset, FileCacheType new_type,
308
                           std::lock_guard<std::mutex>& cache_lock);
309
310
    // remove all blocks that belong to the key
311
    void remove_if_cached(const UInt128Wrapper& key);
312
    void remove_if_cached_async(const UInt128Wrapper& key);
313
314
    // Reset the block size and keep FileBlock, LRU queue, and cache counters consistent.
315
    void reset_range(const UInt128Wrapper&, size_t offset, size_t old_size, size_t new_size,
316
                     std::lock_guard<std::mutex>& cache_lock);
317
318
    // get the hotest blocks message by key
319
    // The tuple is composed of <offset, size, cache_type, expiration_time>
320
    [[nodiscard]] std::vector<std::tuple<size_t, size_t, FileCacheType, uint64_t>>
321
    get_hot_blocks_meta(const UInt128Wrapper& hash) const;
322
323
    [[nodiscard]] bool get_async_open_success() const { return _async_open_done; }
324
325
    // The manager that keeps cached blocks in the cache type their tablet's TTL asks for.
326
    // Exposed so that tests can drive it deterministically rather than race its threads.
327
    BlockFileCacheTtlMgr* get_ttl_mgr() { return _ttl_mgr.get(); }
328
329
    BlockFileCache& operator=(const BlockFileCache&) = delete;
330
    BlockFileCache(const BlockFileCache&) = delete;
331
332
    // try to reserve the new space for the new block if the cache is full
333
    bool try_reserve(const UInt128Wrapper& hash, const CacheContext& context, size_t offset,
334
                     size_t size, std::lock_guard<std::mutex>& cache_lock);
335
336
    /**
337
     * Proactively evict cache blocks to free up space before cache is full.
338
     * 
339
     * This function attempts to evict blocks from both NORMAL and TTL queues to maintain 
340
     * cache size below high watermark. Unlike try_reserve() which blocks until space is freed,
341
     * this function initiates asynchronous eviction in background.
342
     * 
343
     * @param size Number of bytes to try to evict
344
     * @param cache_lock Lock that must be held while accessing cache data structures
345
     * 
346
     * @pre Caller must hold cache_lock
347
     * @pre _need_evict_cache_in_advance must be true
348
     * @pre _recycle_keys queue must have capacity for evicted blocks
349
     */
350
    void try_evict_in_advance(size_t size, std::lock_guard<std::mutex>& cache_lock);
351
352
    void update_ttl_atime(const UInt128Wrapper& hash);
353
354
    std::map<std::string, double> get_stats();
355
356
    // for be UTs
357
    std::map<std::string, double> get_stats_unsafe();
358
    [[nodiscard]] size_t need_update_lru_blocks_size_unsafe() const {
359
        return _need_update_lru_blocks.size();
360
    }
361
362
    using AccessRecord =
363
            std::unordered_map<AccessKeyAndOffset, LRUQueue::Iterator, KeyAndOffsetHash>;
364
365
    /// Used to track and control the cache access of each query.
366
    /// Through it, we can realize the processing of different queries by the cache layer.
367
    struct QueryFileCacheContext {
368
        LRUQueue lru_queue;
369
        AccessRecord records;
370
371
5
        QueryFileCacheContext(size_t max_cache_size) : lru_queue(max_cache_size, 0, 0) {}
372
373
        void remove(const UInt128Wrapper& hash, size_t offset,
374
                    std::lock_guard<std::mutex>& cache_lock);
375
376
        void reserve(const UInt128Wrapper& hash, size_t offset, size_t size,
377
                     std::lock_guard<std::mutex>& cache_lock);
378
379
69
        size_t get_max_cache_size() const { return lru_queue.get_max_size(); }
380
381
48
        size_t get_cache_size(std::lock_guard<std::mutex>& cache_lock) const {
382
48
            return lru_queue.get_capacity(cache_lock);
383
48
        }
384
385
50
        LRUQueue& queue() { return lru_queue; }
386
    };
387
388
    using QueryFileCacheContextPtr = std::shared_ptr<QueryFileCacheContext>;
389
    using QueryFileCacheContextMap = std::unordered_map<TUniqueId, QueryFileCacheContextPtr>;
390
391
    QueryFileCacheContextPtr get_query_context(const TUniqueId& query_id,
392
                                               std::lock_guard<std::mutex>&);
393
394
    void remove_query_context(const TUniqueId& query_id);
395
396
    QueryFileCacheContextPtr get_or_set_query_context(const TUniqueId& query_id,
397
                                                      std::lock_guard<std::mutex>& cache_lock,
398
                                                      int file_cache_query_limit_percent);
399
400
    /// Save a query context information, and adopt different cache policies
401
    /// for different queries through the context cache layer.
402
    struct QueryFileCacheContextHolder {
403
        QueryFileCacheContextHolder(const TUniqueId& query_id, BlockFileCache* mgr,
404
                                    QueryFileCacheContextPtr context)
405
7
                : query_id(query_id), mgr(mgr), context(context) {}
406
407
        QueryFileCacheContextHolder& operator=(const QueryFileCacheContextHolder&) = delete;
408
        QueryFileCacheContextHolder(const QueryFileCacheContextHolder&) = delete;
409
410
7
        ~QueryFileCacheContextHolder() {
411
            /// If only the query_map and the current holder hold the context_query,
412
            /// the query has been completed and the query_context is released.
413
7
            if (context) {
414
6
                context.reset();
415
6
                mgr->remove_query_context(query_id);
416
6
            }
417
7
        }
418
419
        const TUniqueId& query_id;
420
        BlockFileCache* mgr = nullptr;
421
        QueryFileCacheContextPtr context;
422
    };
423
    using QueryFileCacheContextHolderPtr = std::unique_ptr<QueryFileCacheContextHolder>;
424
    QueryFileCacheContextHolderPtr get_query_context_holder(const TUniqueId& query_id,
425
                                                            int file_cache_query_limit_percent);
426
427
5.93k
    int64_t approximate_available_cache_size() const {
428
5.93k
        return std::max<int64_t>(
429
5.93k
                _cache_capacity_metrics->get_value() - _cur_cache_size_metrics->get_value(), 0);
430
5.93k
    }
431
432
    Status report_file_cache_inconsistency(std::vector<std::string>& results);
433
    Status check_file_cache_consistency(InconsistencyContext& inconsistency_context);
434
435
private:
436
    // Shared scan used by both clear modes. It keeps the FileBlock holder lifecycle intact:
437
    // releasable blocks are removed immediately, while blocks held by readers are only marked
438
    // deleting and are later removed by FileBlocksHolder destruction or, when the last holder
439
    // does not release through FileBlock::release_cache_reference(), by run_background_gc().
440
    std::string clear_file_cache_impl(bool sync_remove);
441
442
    LRUQueue& get_queue(FileCacheType type);
443
    const LRUQueue& get_queue(FileCacheType type) const;
444
445
    template <class T, class U>
446
        requires IsXLock<T> && IsXLock<U>
447
    void remove(FileBlockSPtr file_block, T& cache_lock, U& segment_lock, bool sync = true);
448
449
    FileBlocks get_impl(const UInt128Wrapper& hash, const CacheContext& context,
450
                        const FileBlock::Range& range, std::lock_guard<std::mutex>& cache_lock);
451
452
    template <class T>
453
        requires IsXLock<T>
454
    FileBlockCell* get_cell(const UInt128Wrapper& hash, size_t offset, T& cache_lock);
455
456
    virtual FileBlockCell* add_cell(const UInt128Wrapper& hash, const CacheContext& context,
457
                                    size_t offset, size_t size, FileBlock::State state,
458
                                    std::lock_guard<std::mutex>& cache_lock);
459
460
    Status initialize_unlocked(std::lock_guard<std::mutex>& cache_lock);
461
462
    void update_block_lru(FileBlockSPtr block, std::lock_guard<std::mutex>& cache_lock);
463
464
    void use_cell(const FileBlockCell& cell, FileBlocks* result, bool not_need_move,
465
                  std::lock_guard<std::mutex>& cache_lock);
466
467
    bool try_reserve_for_lru(const UInt128Wrapper& hash, QueryFileCacheContextPtr query_context,
468
                             const CacheContext& context, size_t offset, size_t size,
469
                             std::lock_guard<std::mutex>& cache_lock,
470
                             bool evict_in_advance = false);
471
472
    bool try_reserve_during_async_load(size_t size, std::lock_guard<std::mutex>& cache_lock);
473
474
    std::vector<FileCacheType> get_other_cache_type(FileCacheType cur_cache_type);
475
    std::vector<FileCacheType> get_other_cache_type_without_ttl(FileCacheType cur_cache_type);
476
477
    bool try_reserve_from_other_queue(FileCacheType cur_cache_type, size_t offset, int64_t cur_time,
478
                                      std::lock_guard<std::mutex>& cache_lock,
479
                                      bool evict_in_advance = false);
480
481
    size_t get_available_cache_size(FileCacheType cache_type) const;
482
483
    FileBlocks split_range_into_cells(const UInt128Wrapper& hash, const CacheContext& context,
484
                                      size_t offset, size_t size, FileBlock::State state,
485
                                      std::lock_guard<std::mutex>& cache_lock);
486
487
    std::string dump_structure_unlocked(const UInt128Wrapper& hash,
488
                                        std::lock_guard<std::mutex>& cache_lock);
489
490
    std::string dump_single_cache_type_unlocked(const UInt128Wrapper& hash, size_t offset,
491
                                                std::lock_guard<std::mutex>& cache_lock);
492
493
    void fill_holes_with_empty_file_blocks(FileBlocks& file_blocks, const UInt128Wrapper& hash,
494
                                           const CacheContext& context,
495
                                           const FileBlock::Range& range,
496
                                           std::lock_guard<std::mutex>& cache_lock);
497
498
    size_t get_used_cache_size_unlocked(FileCacheType type,
499
                                        std::lock_guard<std::mutex>& cache_lock) const;
500
501
    void check_disk_resource_limit();
502
    void check_need_evict_cache_in_advance();
503
504
    size_t get_available_cache_size_unlocked(FileCacheType type,
505
                                             std::lock_guard<std::mutex>& cache_lock) const;
506
507
    size_t get_file_blocks_num_unlocked(FileCacheType type,
508
                                        std::lock_guard<std::mutex>& cache_lock) const;
509
510
    bool need_to_move(FileCacheType cell_type, FileCacheType query_type) const;
511
512
    void run_background_monitor();
513
    void run_background_gc();
514
    void run_background_lru_log_replay();
515
    size_t replay_lru_logs_once();
516
    void run_background_lru_dump();
517
    void restore_lru_queues_from_disk(std::lock_guard<std::mutex>& cache_lock);
518
    void run_background_evict_in_advance();
519
    void run_background_block_lru_update();
520
521
    bool try_reserve_from_other_queue_by_time_interval(FileCacheType cur_type,
522
                                                       std::vector<FileCacheType> other_cache_types,
523
                                                       size_t size, int64_t cur_time,
524
                                                       std::lock_guard<std::mutex>& cache_lock,
525
                                                       bool evict_in_advance);
526
527
    bool try_reserve_from_other_queue_by_size(FileCacheType cur_type,
528
                                              std::vector<FileCacheType> other_cache_types,
529
                                              size_t size, std::lock_guard<std::mutex>& cache_lock,
530
                                              bool evict_in_advance);
531
532
    bool is_overflow(size_t removed_size, size_t need_size, size_t cur_cache_size,
533
                     bool evict_in_advance) const;
534
535
    void remove_file_blocks(std::vector<FileBlockCell*>&, std::lock_guard<std::mutex>&, bool sync,
536
                            std::string& reason);
537
538
    void find_evict_candidates(LRUQueue& queue, size_t size, size_t cur_cache_size,
539
                               size_t& removed_size, std::vector<FileBlockCell*>& to_evict,
540
                               std::lock_guard<std::mutex>& cache_lock, size_t& cur_removed_size,
541
                               bool evict_in_advance);
542
543
    Status check_ofstream_status(std::ofstream& out, std::string& filename);
544
    Status dump_one_lru_entry(std::ofstream& out, std::string& filename, const UInt128Wrapper& hash,
545
                              size_t offset, size_t size);
546
    Status finalize_dump(std::ofstream& out, size_t entry_num, std::string& tmp_filename,
547
                         std::string& final_filename, size_t& file_size);
548
    Status check_ifstream_status(std::ifstream& in, std::string& filename);
549
    Status parse_dump_footer(std::ifstream& in, std::string& filename, size_t& entry_num);
550
    Status parse_one_lru_entry(std::ifstream& in, std::string& filename, UInt128Wrapper& hash,
551
                               size_t& offset, size_t& size);
552
    void remove_lru_dump_files();
553
554
    void clear_need_update_lru_blocks();
555
556
    // Marks a cell that cannot be erased yet as deleting and remembers it in _deleting_blocks, so
557
    // that the block is still reclaimed when its last reference is dropped without going through
558
    // FileBlock::release_cache_reference(). This is the only place that sets a block deleting.
559
    template <class T>
560
        requires IsXLock<T>
561
    void mark_cell_deleting(FileBlockCell& cell, T& cache_lock);
562
563
    // Removes up to batch_limit blocks that were marked deleting and are not referenced any
564
    // more. Returns how many of them were removed.
565
    size_t recycle_deleting_blocks(size_t batch_limit);
566
567
    // info
568
    std::string _cache_base_path;
569
    size_t _capacity = 0;
570
    size_t _max_file_block_size = 0;
571
572
    mutable std::mutex _mutex;
573
    bool _close {false};
574
    std::mutex _close_mtx;
575
    std::condition_variable _close_cv;
576
    std::thread _cache_background_monitor_thread;
577
    std::thread _cache_background_gc_thread;
578
    std::thread _cache_background_evict_in_advance_thread;
579
    std::thread _cache_background_lru_dump_thread;
580
    std::thread _cache_background_lru_log_replay_thread;
581
    std::thread _cache_background_block_lru_update_thread;
582
    std::atomic_bool _async_open_done {false};
583
    // disk space or inode is less than the specified value
584
    std::atomic<bool> _disk_resource_limit_mode {false};
585
    std::atomic<bool> _need_evict_cache_in_advance {false};
586
    bool _is_initialized {false};
587
588
    // strategy
589
    using FileBlocksByOffset = std::map<size_t, FileBlockCell>;
590
    using CachedFiles = std::unordered_map<UInt128Wrapper, FileBlocksByOffset, KeyHash>;
591
    CachedFiles _files;
592
    QueryFileCacheContextMap _query_map;
593
    size_t _cur_cache_size = 0;
594
    size_t _cur_ttl_size = 0;
595
    std::multimap<uint64_t, UInt128Wrapper> _time_to_key;
596
    std::unordered_map<UInt128Wrapper, uint64_t, KeyHash> _key_to_time;
597
    // The three queues are level queue.
598
    // It means as level1/level2/level3 queue.
599
    // but the level2 is maximum.
600
    // If some datas are importance, we can cache it into index queue
601
    // If some datas are just use once, we can cache it into disposable queue
602
    // The size proportion is [1:17:2].
603
    LRUQueue _index_queue;
604
    LRUQueue _normal_queue;
605
    LRUQueue _disposable_queue;
606
    LRUQueue _ttl_queue;
607
608
    // keys for async remove
609
    RecycleFileCacheKeys _recycle_keys;
610
611
    // Blocks that were busy when something asked to delete them. They keep sitting in _files
612
    // until their last reference goes away. Only FileBlocksHolder and FileBlocksProbeResult
613
    // release through FileBlock::release_cache_reference(), which is the one path that notices
614
    // is_deleting and removes the block; every other reference holder just drops its shared_ptr
615
    // and would strand the block here forever. run_background_gc() rechecks these candidates and
616
    // removes the ones that became releasable.
617
    //
618
    // Invariant: holds exactly the keys of the cells in _files whose block is deleting. A key is
619
    // added by mark_cell_deleting() and erased by remove() together with its cell, so a block
620
    // created later at the same offset is never taken for the one that was marked.
621
    std::unordered_set<AccessKeyAndOffset, KeyAndOffsetHash> _deleting_blocks;
622
623
    std::unique_ptr<LRUQueueRecorder> _lru_recorder;
624
    std::unique_ptr<CacheLRUDumper> _lru_dumper;
625
    std::unique_ptr<BlockFileCacheTtlMgr> _ttl_mgr;
626
627
    std::unique_ptr<InflightWriteBufferIndex> _inflight_write_buffer_index;
628
    std::unique_ptr<AsyncCacheWriteManager> _async_write_manager;
629
630
    // metrics
631
    std::shared_ptr<bvar::Status<size_t>> _cache_capacity_metrics;
632
    std::shared_ptr<bvar::Status<size_t>> _cur_cache_size_metrics;
633
    std::shared_ptr<bvar::Status<size_t>> _cur_ttl_cache_size_metrics;
634
    std::shared_ptr<bvar::Status<size_t>> _cur_ttl_cache_lru_queue_cache_size_metrics;
635
    std::shared_ptr<bvar::Status<size_t>> _cur_ttl_cache_lru_queue_element_count_metrics;
636
    std::shared_ptr<bvar::Status<size_t>> _cur_normal_queue_element_count_metrics;
637
    std::shared_ptr<bvar::Status<size_t>> _cur_normal_queue_cache_size_metrics;
638
    std::shared_ptr<bvar::Status<size_t>> _cur_index_queue_element_count_metrics;
639
    std::shared_ptr<bvar::Status<size_t>> _cur_index_queue_cache_size_metrics;
640
    std::shared_ptr<bvar::Status<size_t>> _cur_disposable_queue_element_count_metrics;
641
    std::shared_ptr<bvar::Status<size_t>> _cur_disposable_queue_cache_size_metrics;
642
    std::array<std::shared_ptr<bvar::Adder<size_t>>, 4> _queue_evict_size_metrics;
643
    std::shared_ptr<bvar::Adder<size_t>> _total_read_size_metrics;
644
    std::shared_ptr<bvar::Adder<size_t>> _total_hit_size_metrics;
645
    std::shared_ptr<bvar::Adder<size_t>> _total_evict_size_metrics;
646
    std::shared_ptr<bvar::Adder<size_t>> _evict_not_downloaded_size_metrics;
647
    std::shared_ptr<bvar::Adder<size_t>> _evict_not_downloaded_num_metrics;
648
    std::shared_ptr<bvar::Adder<size_t>> _gc_evict_bytes_metrics;
649
    std::shared_ptr<bvar::Adder<size_t>> _gc_evict_count_metrics;
650
    std::shared_ptr<bvar::Adder<size_t>> _evict_by_time_metrics_matrix[4][4];
651
    std::shared_ptr<bvar::Adder<size_t>> _evict_by_size_metrics_matrix[4][4];
652
    std::shared_ptr<bvar::Adder<size_t>> _evict_by_self_lru_metrics_matrix[4];
653
    std::shared_ptr<bvar::Adder<size_t>> _evict_by_try_release;
654
655
    std::shared_ptr<bvar::Window<bvar::Adder<size_t>>> _num_hit_blocks_5m;
656
    std::shared_ptr<bvar::Window<bvar::Adder<size_t>>> _num_read_blocks_5m;
657
    std::shared_ptr<bvar::Window<bvar::Adder<size_t>>> _num_hit_blocks_1h;
658
    std::shared_ptr<bvar::Window<bvar::Adder<size_t>>> _num_read_blocks_1h;
659
660
    std::shared_ptr<bvar::Adder<size_t>> _num_read_blocks;
661
    std::shared_ptr<bvar::Adder<size_t>> _num_hit_blocks;
662
    std::shared_ptr<bvar::Adder<size_t>> _num_removed_blocks;
663
664
    std::shared_ptr<bvar::Adder<size_t>> _no_warmup_num_read_blocks;
665
    std::shared_ptr<bvar::Adder<size_t>> _no_warmup_num_hit_blocks;
666
667
    std::shared_ptr<bvar::Window<bvar::Adder<size_t>>> _no_warmup_num_hit_blocks_5m;
668
    std::shared_ptr<bvar::Window<bvar::Adder<size_t>>> _no_warmup_num_read_blocks_5m;
669
    std::shared_ptr<bvar::Window<bvar::Adder<size_t>>> _no_warmup_num_hit_blocks_1h;
670
    std::shared_ptr<bvar::Window<bvar::Adder<size_t>>> _no_warmup_num_read_blocks_1h;
671
672
    std::shared_ptr<bvar::Status<double>> _hit_ratio;
673
    std::shared_ptr<bvar::Status<double>> _hit_ratio_5m;
674
    std::shared_ptr<bvar::Status<double>> _hit_ratio_1h;
675
    std::shared_ptr<bvar::Status<double>> _no_warmup_hit_ratio;
676
    std::shared_ptr<bvar::Status<double>> _no_warmup_hit_ratio_5m;
677
    std::shared_ptr<bvar::Status<double>> _no_warmup_hit_ratio_1h;
678
    std::shared_ptr<bvar::Status<size_t>> _disk_limit_mode_metrics;
679
    std::shared_ptr<bvar::Status<size_t>> _need_evict_cache_in_advance_metrics;
680
    std::shared_ptr<bvar::Status<size_t>> _meta_store_write_queue_size_metrics;
681
682
    std::shared_ptr<bvar::LatencyRecorder> _cache_lock_wait_time_us;
683
    std::shared_ptr<bvar::LatencyRecorder> _get_or_set_latency_us;
684
    std::shared_ptr<bvar::LatencyRecorder> _probe_latency_us;
685
    std::shared_ptr<bvar::LatencyRecorder> _storage_sync_remove_latency_us;
686
    std::shared_ptr<bvar::LatencyRecorder> _storage_retry_sync_remove_latency_us;
687
    std::shared_ptr<bvar::LatencyRecorder> _storage_async_remove_latency_us;
688
    std::shared_ptr<bvar::LatencyRecorder> _evict_in_advance_latency_us;
689
    std::shared_ptr<bvar::LatencyRecorder> _recycle_keys_length_recorder;
690
    std::shared_ptr<bvar::Status<size_t>> _deleting_block_count_metrics;
691
    std::shared_ptr<bvar::Adder<size_t>> _recycle_deleting_blocks_metrics;
692
    std::shared_ptr<bvar::LatencyRecorder> _update_lru_blocks_latency_us;
693
    std::shared_ptr<bvar::LatencyRecorder> _need_update_lru_blocks_length_recorder;
694
    std::shared_ptr<bvar::Adder<size_t>> _need_update_lru_blocks_produce_metrics;
695
    std::shared_ptr<bvar::Adder<size_t>> _need_update_lru_blocks_consume_metrics;
696
    std::shared_ptr<bvar::LatencyRecorder> _ttl_gc_latency_us;
697
698
    std::shared_ptr<bvar::LatencyRecorder> _shadow_queue_levenshtein_distance;
699
    std::array<std::shared_ptr<bvar::LatencyRecorder>, 4> _lru_recorder_queue_length_recorder;
700
    std::array<std::shared_ptr<bvar::Adder<size_t>>, 4> _lru_recorder_queue_produce_metrics;
701
    std::array<std::shared_ptr<bvar::Adder<size_t>>, 4> _lru_recorder_queue_consume_metrics;
702
    std::array<std::shared_ptr<bvar::Status<size_t>>, 4>
703
            _lru_recorder_shadow_queue_element_count_metrics;
704
    std::shared_ptr<bvar::Adder<size_t>> _lru_recorder_log_replay_idle_metrics;
705
    // keep _storage last so it will deconstruct first
706
    // otherwise, load_cache_info_into_memory might crash
707
    // coz it will use other members of BlockFileCache
708
    // so join this async load thread first
709
    std::unique_ptr<FileCacheStorage> _storage;
710
    std::shared_ptr<bvar::LatencyRecorder> _lru_dump_latency_us;
711
    std::mutex _dump_lru_queues_mtx;
712
    NeedUpdateLRUBlocks _need_update_lru_blocks;
713
};
714
715
} // namespace doris::io