Coverage Report

Created: 2026-06-11 17:27

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 <functional>
28
#include <limits>
29
#include <memory>
30
#include <mutex>
31
#include <optional>
32
#include <thread>
33
#include <unordered_map>
34
#include <vector>
35
36
#include "io/cache/block_file_cache_ttl_mgr.h"
37
#include "io/cache/cache_lru_dumper.h"
38
#include "io/cache/file_block.h"
39
#include "io/cache/file_cache_common.h"
40
#include "io/cache/file_cache_storage.h"
41
#include "io/cache/lru_queue_recorder.h"
42
#include "runtime/runtime_profile.h"
43
#include "util/threadpool.h"
44
45
namespace doris::io {
46
using RecycleFileCacheKeys = moodycamel::ConcurrentQueue<FileCacheKey>;
47
48
class LockScopedTimer {
49
public:
50
7.54k
    LockScopedTimer() : start_(std::chrono::steady_clock::now()) {}
51
7.54k
    ~LockScopedTimer() {
52
7.54k
        auto end = std::chrono::steady_clock::now();
53
7.54k
        auto duration_us =
54
7.54k
                std::chrono::duration_cast<std::chrono::microseconds>(end - start_).count();
55
7.54k
        if (duration_us > config::cache_lock_held_long_tail_threshold_us) {
56
0
            LOG(WARNING) << "Lock held time " << std::to_string(duration_us) << "us. "
57
0
                         << get_stack_trace();
58
0
        }
59
7.54k
    }
60
61
private:
62
    std::chrono::time_point<std::chrono::steady_clock> start_;
63
};
64
65
// Note: the cache_lock is scoped, so do not add do...while(0) here.
66
#define SCOPED_CACHE_LOCK(MUTEX, cache)                                                           \
67
7.54k
    std::chrono::time_point<std::chrono::steady_clock> start_time =                               \
68
7.54k
            std::chrono::steady_clock::now();                                                     \
69
7.54k
    std::lock_guard cache_lock(MUTEX);                                                            \
70
7.54k
    std::chrono::time_point<std::chrono::steady_clock> acq_time =                                 \
71
7.54k
            std::chrono::steady_clock::now();                                                     \
72
7.54k
    auto duration_us =                                                                            \
73
7.54k
            std::chrono::duration_cast<std::chrono::microseconds>(acq_time - start_time).count(); \
74
7.54k
    *(cache->_cache_lock_wait_time_us) << duration_us;                                            \
75
7.54k
    if (duration_us > config::cache_lock_wait_long_tail_threshold_us) {                           \
76
0
        LOG(WARNING) << "Lock wait time " << std::to_string(duration_us) << "us. "                \
77
0
                     << get_stack_trace() << std::endl;                                           \
78
0
    }                                                                                             \
79
7.54k
    LockScopedTimer cache_lock_timer;
80
81
class FSFileCacheStorage;
82
83
// NeedUpdateLRUBlocks keeps FileBlockSPtr entries that require LRU updates in a
84
// deduplicated, sharded container. Entries are keyed by the raw FileBlock
85
// pointer so that multiple shared_ptr copies of the same block are treated as a
86
// single pending update. The structure is thread-safe and optimized for high
87
// contention insert/drain workloads in the background update thread.
88
// Note that Blocks are updated in batch, internal order is not important.
89
class NeedUpdateLRUBlocks {
90
public:
91
185
    NeedUpdateLRUBlocks() = default;
92
93
    // Insert a block into the pending set. Returns true only when the block
94
    // was not already queued. Null inputs are ignored.
95
    bool insert(FileBlockSPtr block, size_t max_queue_size = std::numeric_limits<size_t>::max());
96
97
    // Drain up to `limit` unique blocks into `output`. The method returns how
98
    // many blocks were actually drained and shrinks the internal size
99
    // accordingly.
100
    size_t drain(size_t limit, std::vector<FileBlockSPtr>* output);
101
102
    // Remove every pending block from the structure and reset the size.
103
    void clear();
104
105
    // Thread-safe approximate size of queued unique blocks.
106
1.68k
    size_t size() const { return _size.load(std::memory_order_relaxed); }
107
108
private:
109
    static constexpr size_t kShardCount = 64;
110
    static constexpr size_t kShardMask = kShardCount - 1;
111
112
    struct Shard {
113
        std::mutex mutex;
114
        std::unordered_map<FileBlock*, FileBlockSPtr> entries;
115
    };
116
117
    size_t shard_index(FileBlock* ptr) const;
118
    void decrease_size(size_t delta);
119
120
    std::array<Shard, kShardCount> _shards;
121
    std::atomic<size_t> _size {0};
122
};
123
124
// The BlockFileCache is responsible for the management of the blocks
125
// The current strategies are lru and ttl.
126
127
struct FileBlockCell {
128
    friend class FileBlock;
129
130
    FileBlockSPtr file_block;
131
    /// Iterator is put here on first reservation attempt, if successful.
132
    std::optional<LRUQueue::Iterator> queue_iterator;
133
134
    mutable int64_t atime {0};
135
730k
    void update_atime() const {
136
730k
        atime = std::chrono::duration_cast<std::chrono::seconds>(
137
730k
                        std::chrono::steady_clock::now().time_since_epoch())
138
730k
                        .count();
139
730k
    }
140
141
    /// Pointer to file block is always hold by the cache itself.
142
    /// Apart from pointer in cache, it can be hold by cache users, when they call
143
    /// getorSet(), but cache users always hold it via FileBlocksHolder.
144
1.90k
    bool releasable() const {
145
1.90k
        return (file_block.use_count() == 1 ||
146
1.90k
                (file_block.use_count() == 2 && file_block->_owned_by_cached_reader));
147
1.90k
    }
148
149
544k
    size_t size() const { return file_block->_block_range.size(); }
150
151
70
    FileBlockCell() = default;
152
    FileBlockCell(FileBlockSPtr file_block, std::lock_guard<std::mutex>& cache_lock);
153
    FileBlockCell(FileBlockCell&& other) noexcept
154
27.2k
            : file_block(std::move(other.file_block)),
155
27.2k
              queue_iterator(other.queue_iterator),
156
27.2k
              atime(other.atime) {
157
27.2k
        file_block->cell = this;
158
27.2k
    }
159
160
    FileBlockCell& operator=(const FileBlockCell&) = delete;
161
    FileBlockCell(const FileBlockCell&) = delete;
162
163
0
    size_t dowloading_size() const { return file_block->_downloaded_size; }
164
};
165
166
class BlockFileCache {
167
    friend class FSFileCacheStorage;
168
    friend class MemFileCacheStorage;
169
    friend class FileBlock;
170
    friend struct FileBlocksHolder;
171
    friend class CacheLRUDumper;
172
    friend class LRUQueueRecorder;
173
    friend struct FileBlockCell;
174
    friend class BlockFileCacheTest;
175
176
public:
177
    // hash the file_name to uint128
178
    static UInt128Wrapper hash(const std::string& path);
179
180
    BlockFileCache(const std::string& cache_base_path, const FileCacheSettings& cache_settings);
181
182
180
    virtual ~BlockFileCache() {
183
180
        {
184
180
            std::lock_guard lock(_close_mtx);
185
180
            _close = true;
186
180
        }
187
180
        _close_cv.notify_all();
188
180
        if (_cache_background_monitor_thread.joinable()) {
189
155
            _cache_background_monitor_thread.join();
190
155
        }
191
180
        if (_cache_background_gc_thread.joinable()) {
192
155
            _cache_background_gc_thread.join();
193
155
        }
194
180
        if (_cache_background_evict_in_advance_thread.joinable()) {
195
155
            _cache_background_evict_in_advance_thread.join();
196
155
        }
197
180
        if (_cache_background_lru_dump_thread.joinable()) {
198
155
            _cache_background_lru_dump_thread.join();
199
155
        }
200
180
        if (_cache_background_lru_log_replay_thread.joinable()) {
201
155
            _cache_background_lru_log_replay_thread.join();
202
155
        }
203
180
        if (_cache_background_block_lru_update_thread.joinable()) {
204
155
            _cache_background_block_lru_update_thread.join();
205
155
        }
206
180
        if (_ttl_mgr) {
207
133
            _ttl_mgr.reset();
208
133
        }
209
180
    }
210
211
    /// Restore cache from local filesystem.
212
    Status initialize();
213
214
    /// Cache capacity in bytes.
215
14
    [[nodiscard]] size_t capacity() const { return _capacity; }
216
217
    // try to release all releasable block
218
    // it maybe hang the io/system
219
    size_t try_release();
220
221
147
    [[nodiscard]] const std::string& get_base_path() const { return _cache_base_path; }
222
223
    // Get storage for inspection
224
0
    FileCacheStorage* get_storage() const { return _storage.get(); }
225
226
    /**
227
         * Given an `offset` and `size` representing [offset, offset + size) bytes interval,
228
         * return list of cached non-overlapping non-empty
229
         * file blocks `[block1, ..., blockN]` which intersect with given interval.
230
         *
231
         * blocks in returned list are ordered in ascending order and represent a full contiguous
232
         * interval (no holes). Each block in returned list has state: DOWNLOADED, DOWNLOADING or EMPTY.
233
         *
234
         * As long as pointers to returned file blocks are hold
235
         * it is guaranteed that these file blocks are not removed from cache.
236
         */
237
    FileBlocksHolder get_or_set(const UInt128Wrapper& hash, size_t offset, size_t size,
238
                                CacheContext& context);
239
240
    /**
241
     * record blocks read directly by CachedRemoteFileReader
242
     */
243
    void add_need_update_lru_block(FileBlockSPtr block);
244
245
    /**
246
     * Clear all cached data for this cache instance async
247
     *
248
     * @returns summary message
249
     */
250
    std::string clear_file_cache_async();
251
    std::string clear_file_cache_directly();
252
253
    /**
254
     * Reset the cache capacity. If the new_capacity is smaller than _capacity, the redundant data will be remove async.
255
     *
256
     * @returns summary message
257
     */
258
    std::string reset_capacity(size_t new_capacity);
259
260
    std::map<size_t, FileBlockSPtr> get_blocks_by_key(const UInt128Wrapper& hash);
261
262
    /// For debug and UT
263
    std::string dump_structure(const UInt128Wrapper& hash);
264
    std::string dump_single_cache_type(const UInt128Wrapper& hash, size_t offset);
265
266
    void dump_lru_queues(bool force);
267
268
    [[nodiscard]] size_t get_used_cache_size(FileCacheType type) const;
269
270
    [[nodiscard]] size_t get_file_blocks_num(FileCacheType type) const;
271
272
    // change the block cache type
273
    void change_cache_type(const UInt128Wrapper& hash, size_t offset, FileCacheType new_type,
274
                           std::lock_guard<std::mutex>& cache_lock);
275
276
    // remove all blocks that belong to the key
277
    void remove_if_cached(const UInt128Wrapper& key);
278
    void remove_if_cached_async(const UInt128Wrapper& key);
279
280
    // Reset the block size and keep FileBlock, LRU queue, and cache counters consistent.
281
    void reset_range(const UInt128Wrapper&, size_t offset, size_t old_size, size_t new_size,
282
                     std::lock_guard<std::mutex>& cache_lock);
283
284
    // get the hotest blocks message by key
285
    // The tuple is composed of <offset, size, cache_type, expiration_time>
286
    [[nodiscard]] std::vector<std::tuple<size_t, size_t, FileCacheType, uint64_t>>
287
    get_hot_blocks_meta(const UInt128Wrapper& hash) const;
288
289
559
    [[nodiscard]] bool get_async_open_success() const { return _async_open_done; }
290
291
    BlockFileCache& operator=(const BlockFileCache&) = delete;
292
    BlockFileCache(const BlockFileCache&) = delete;
293
294
    // try to reserve the new space for the new block if the cache is full
295
    bool try_reserve(const UInt128Wrapper& hash, const CacheContext& context, size_t offset,
296
                     size_t size, std::lock_guard<std::mutex>& cache_lock);
297
298
    /**
299
     * Proactively evict cache blocks to free up space before cache is full.
300
     * 
301
     * This function attempts to evict blocks from both NORMAL and TTL queues to maintain 
302
     * cache size below high watermark. Unlike try_reserve() which blocks until space is freed,
303
     * this function initiates asynchronous eviction in background.
304
     * 
305
     * @param size Number of bytes to try to evict
306
     * @param cache_lock Lock that must be held while accessing cache data structures
307
     * 
308
     * @pre Caller must hold cache_lock
309
     * @pre _need_evict_cache_in_advance must be true
310
     * @pre _recycle_keys queue must have capacity for evicted blocks
311
     */
312
    void try_evict_in_advance(size_t size, std::lock_guard<std::mutex>& cache_lock);
313
314
    void update_ttl_atime(const UInt128Wrapper& hash);
315
316
    void pause_ttl_manager();
317
    void resume_ttl_manager();
318
319
    std::map<std::string, double> get_stats();
320
321
    // for be UTs
322
    std::map<std::string, double> get_stats_unsafe();
323
5
    [[nodiscard]] size_t need_update_lru_blocks_size_unsafe() const {
324
5
        return _need_update_lru_blocks.size();
325
5
    }
326
327
    using AccessRecord =
328
            std::unordered_map<AccessKeyAndOffset, LRUQueue::Iterator, KeyAndOffsetHash>;
329
330
    /// Used to track and control the cache access of each query.
331
    /// Through it, we can realize the processing of different queries by the cache layer.
332
    struct QueryFileCacheContext {
333
        LRUQueue lru_queue;
334
        AccessRecord records;
335
336
5
        QueryFileCacheContext(size_t max_cache_size) : lru_queue(max_cache_size, 0, 0) {}
337
338
        void remove(const UInt128Wrapper& hash, size_t offset,
339
                    std::lock_guard<std::mutex>& cache_lock);
340
341
        void reserve(const UInt128Wrapper& hash, size_t offset, size_t size,
342
                     std::lock_guard<std::mutex>& cache_lock);
343
344
69
        size_t get_max_cache_size() const { return lru_queue.get_max_size(); }
345
346
48
        size_t get_cache_size(std::lock_guard<std::mutex>& cache_lock) const {
347
48
            return lru_queue.get_capacity(cache_lock);
348
48
        }
349
350
50
        LRUQueue& queue() { return lru_queue; }
351
    };
352
353
    using QueryFileCacheContextPtr = std::shared_ptr<QueryFileCacheContext>;
354
    using QueryFileCacheContextMap = std::unordered_map<TUniqueId, QueryFileCacheContextPtr>;
355
356
    QueryFileCacheContextPtr get_query_context(const TUniqueId& query_id,
357
                                               std::lock_guard<std::mutex>&);
358
359
    void remove_query_context(const TUniqueId& query_id);
360
361
    QueryFileCacheContextPtr get_or_set_query_context(const TUniqueId& query_id,
362
                                                      std::lock_guard<std::mutex>& cache_lock,
363
                                                      int file_cache_query_limit_percent);
364
365
    /// Save a query context information, and adopt different cache policies
366
    /// for different queries through the context cache layer.
367
    struct QueryFileCacheContextHolder {
368
        QueryFileCacheContextHolder(const TUniqueId& query_id, BlockFileCache* mgr,
369
                                    QueryFileCacheContextPtr context)
370
7
                : query_id(query_id), mgr(mgr), context(context) {}
371
372
        QueryFileCacheContextHolder& operator=(const QueryFileCacheContextHolder&) = delete;
373
        QueryFileCacheContextHolder(const QueryFileCacheContextHolder&) = delete;
374
375
7
        ~QueryFileCacheContextHolder() {
376
            /// If only the query_map and the current holder hold the context_query,
377
            /// the query has been completed and the query_context is released.
378
7
            if (context) {
379
6
                context.reset();
380
6
                mgr->remove_query_context(query_id);
381
6
            }
382
7
        }
383
384
        const TUniqueId& query_id;
385
        BlockFileCache* mgr = nullptr;
386
        QueryFileCacheContextPtr context;
387
    };
388
    using QueryFileCacheContextHolderPtr = std::unique_ptr<QueryFileCacheContextHolder>;
389
    QueryFileCacheContextHolderPtr get_query_context_holder(const TUniqueId& query_id,
390
                                                            int file_cache_query_limit_percent);
391
392
0
    int64_t approximate_available_cache_size() const {
393
0
        return std::max<int64_t>(
394
0
                _cache_capacity_metrics->get_value() - _cur_cache_size_metrics->get_value(), 0);
395
0
    }
396
397
    Status report_file_cache_inconsistency(std::vector<std::string>& results);
398
    Status check_file_cache_consistency(InconsistencyContext& inconsistency_context);
399
400
private:
401
    LRUQueue& get_queue(FileCacheType type);
402
    const LRUQueue& get_queue(FileCacheType type) const;
403
404
    template <class T, class U>
405
        requires IsXLock<T> && IsXLock<U>
406
    void remove(FileBlockSPtr file_block, T& cache_lock, U& segment_lock, bool sync = true);
407
408
    FileBlocks get_impl(const UInt128Wrapper& hash, const CacheContext& context,
409
                        const FileBlock::Range& range, std::lock_guard<std::mutex>& cache_lock);
410
411
    template <class T>
412
        requires IsXLock<T>
413
    FileBlockCell* get_cell(const UInt128Wrapper& hash, size_t offset, T& cache_lock);
414
415
    virtual FileBlockCell* add_cell(const UInt128Wrapper& hash, const CacheContext& context,
416
                                    size_t offset, size_t size, FileBlock::State state,
417
                                    std::lock_guard<std::mutex>& cache_lock);
418
419
    Status initialize_unlocked(std::lock_guard<std::mutex>& cache_lock);
420
421
    void update_block_lru(FileBlockSPtr block, std::lock_guard<std::mutex>& cache_lock);
422
423
    void use_cell(const FileBlockCell& cell, FileBlocks* result, bool not_need_move,
424
                  std::lock_guard<std::mutex>& cache_lock);
425
426
    bool try_reserve_for_lru(const UInt128Wrapper& hash, QueryFileCacheContextPtr query_context,
427
                             const CacheContext& context, size_t offset, size_t size,
428
                             std::lock_guard<std::mutex>& cache_lock,
429
                             bool evict_in_advance = false);
430
431
    bool try_reserve_during_async_load(size_t size, std::lock_guard<std::mutex>& cache_lock);
432
433
    std::vector<FileCacheType> get_other_cache_type(FileCacheType cur_cache_type);
434
    std::vector<FileCacheType> get_other_cache_type_without_ttl(FileCacheType cur_cache_type);
435
436
    bool try_reserve_from_other_queue(FileCacheType cur_cache_type, size_t offset, int64_t cur_time,
437
                                      std::lock_guard<std::mutex>& cache_lock,
438
                                      bool evict_in_advance = false);
439
440
    size_t get_available_cache_size(FileCacheType cache_type) const;
441
442
    FileBlocks split_range_into_cells(const UInt128Wrapper& hash, const CacheContext& context,
443
                                      size_t offset, size_t size, FileBlock::State state,
444
                                      std::lock_guard<std::mutex>& cache_lock);
445
446
    std::string dump_structure_unlocked(const UInt128Wrapper& hash,
447
                                        std::lock_guard<std::mutex>& cache_lock);
448
449
    std::string dump_single_cache_type_unlocked(const UInt128Wrapper& hash, size_t offset,
450
                                                std::lock_guard<std::mutex>& cache_lock);
451
452
    void fill_holes_with_empty_file_blocks(FileBlocks& file_blocks, const UInt128Wrapper& hash,
453
                                           const CacheContext& context,
454
                                           const FileBlock::Range& range,
455
                                           std::lock_guard<std::mutex>& cache_lock);
456
457
    size_t get_used_cache_size_unlocked(FileCacheType type,
458
                                        std::lock_guard<std::mutex>& cache_lock) const;
459
460
    void check_disk_resource_limit();
461
    void check_need_evict_cache_in_advance();
462
463
    size_t get_available_cache_size_unlocked(FileCacheType type,
464
                                             std::lock_guard<std::mutex>& cache_lock) const;
465
466
    size_t get_file_blocks_num_unlocked(FileCacheType type,
467
                                        std::lock_guard<std::mutex>& cache_lock) const;
468
469
    bool need_to_move(FileCacheType cell_type, FileCacheType query_type) const;
470
471
    void run_background_monitor();
472
    void run_background_gc();
473
    void run_background_lru_log_replay();
474
    size_t replay_lru_logs_once();
475
    void run_background_lru_dump();
476
    void restore_lru_queues_from_disk(std::lock_guard<std::mutex>& cache_lock);
477
    void run_background_evict_in_advance();
478
    void run_background_block_lru_update();
479
480
    bool try_reserve_from_other_queue_by_time_interval(FileCacheType cur_type,
481
                                                       std::vector<FileCacheType> other_cache_types,
482
                                                       size_t size, int64_t cur_time,
483
                                                       std::lock_guard<std::mutex>& cache_lock,
484
                                                       bool evict_in_advance);
485
486
    bool try_reserve_from_other_queue_by_size(FileCacheType cur_type,
487
                                              std::vector<FileCacheType> other_cache_types,
488
                                              size_t size, std::lock_guard<std::mutex>& cache_lock,
489
                                              bool evict_in_advance);
490
491
    bool is_overflow(size_t removed_size, size_t need_size, size_t cur_cache_size,
492
                     bool evict_in_advance) const;
493
494
    void remove_file_blocks(std::vector<FileBlockCell*>&, std::lock_guard<std::mutex>&, bool sync,
495
                            std::string& reason);
496
497
    void find_evict_candidates(LRUQueue& queue, size_t size, size_t cur_cache_size,
498
                               size_t& removed_size, std::vector<FileBlockCell*>& to_evict,
499
                               std::lock_guard<std::mutex>& cache_lock, size_t& cur_removed_size,
500
                               bool evict_in_advance);
501
502
    Status check_ofstream_status(std::ofstream& out, std::string& filename);
503
    Status dump_one_lru_entry(std::ofstream& out, std::string& filename, const UInt128Wrapper& hash,
504
                              size_t offset, size_t size);
505
    Status finalize_dump(std::ofstream& out, size_t entry_num, std::string& tmp_filename,
506
                         std::string& final_filename, size_t& file_size);
507
    Status check_ifstream_status(std::ifstream& in, std::string& filename);
508
    Status parse_dump_footer(std::ifstream& in, std::string& filename, size_t& entry_num);
509
    Status parse_one_lru_entry(std::ifstream& in, std::string& filename, UInt128Wrapper& hash,
510
                               size_t& offset, size_t& size);
511
    void remove_lru_dump_files();
512
513
    void clear_need_update_lru_blocks();
514
515
    // info
516
    std::string _cache_base_path;
517
    size_t _capacity = 0;
518
    size_t _max_file_block_size = 0;
519
520
    mutable std::mutex _mutex;
521
    bool _close {false};
522
    std::mutex _close_mtx;
523
    std::condition_variable _close_cv;
524
    std::thread _cache_background_monitor_thread;
525
    std::thread _cache_background_gc_thread;
526
    std::thread _cache_background_evict_in_advance_thread;
527
    std::thread _cache_background_lru_dump_thread;
528
    std::thread _cache_background_lru_log_replay_thread;
529
    std::thread _cache_background_block_lru_update_thread;
530
    std::atomic_bool _async_open_done {false};
531
    // disk space or inode is less than the specified value
532
    bool _disk_resource_limit_mode {false};
533
    bool _need_evict_cache_in_advance {false};
534
    bool _is_initialized {false};
535
536
    // strategy
537
    using FileBlocksByOffset = std::map<size_t, FileBlockCell>;
538
    using CachedFiles = std::unordered_map<UInt128Wrapper, FileBlocksByOffset, KeyHash>;
539
    CachedFiles _files;
540
    QueryFileCacheContextMap _query_map;
541
    size_t _cur_cache_size = 0;
542
    size_t _cur_ttl_size = 0;
543
    std::multimap<uint64_t, UInt128Wrapper> _time_to_key;
544
    std::unordered_map<UInt128Wrapper, uint64_t, KeyHash> _key_to_time;
545
    // The three queues are level queue.
546
    // It means as level1/level2/level3 queue.
547
    // but the level2 is maximum.
548
    // If some datas are importance, we can cache it into index queue
549
    // If some datas are just use once, we can cache it into disposable queue
550
    // The size proportion is [1:17:2].
551
    LRUQueue _index_queue;
552
    LRUQueue _normal_queue;
553
    LRUQueue _disposable_queue;
554
    LRUQueue _ttl_queue;
555
556
    // keys for async remove
557
    RecycleFileCacheKeys _recycle_keys;
558
559
    std::unique_ptr<LRUQueueRecorder> _lru_recorder;
560
    std::unique_ptr<CacheLRUDumper> _lru_dumper;
561
    std::unique_ptr<BlockFileCacheTtlMgr> _ttl_mgr;
562
563
    // metrics
564
    std::shared_ptr<bvar::Status<size_t>> _cache_capacity_metrics;
565
    std::shared_ptr<bvar::Status<size_t>> _cur_cache_size_metrics;
566
    std::shared_ptr<bvar::Status<size_t>> _cur_ttl_cache_size_metrics;
567
    std::shared_ptr<bvar::Status<size_t>> _cur_ttl_cache_lru_queue_cache_size_metrics;
568
    std::shared_ptr<bvar::Status<size_t>> _cur_ttl_cache_lru_queue_element_count_metrics;
569
    std::shared_ptr<bvar::Status<size_t>> _cur_normal_queue_element_count_metrics;
570
    std::shared_ptr<bvar::Status<size_t>> _cur_normal_queue_cache_size_metrics;
571
    std::shared_ptr<bvar::Status<size_t>> _cur_index_queue_element_count_metrics;
572
    std::shared_ptr<bvar::Status<size_t>> _cur_index_queue_cache_size_metrics;
573
    std::shared_ptr<bvar::Status<size_t>> _cur_disposable_queue_element_count_metrics;
574
    std::shared_ptr<bvar::Status<size_t>> _cur_disposable_queue_cache_size_metrics;
575
    std::array<std::shared_ptr<bvar::Adder<size_t>>, 4> _queue_evict_size_metrics;
576
    std::shared_ptr<bvar::Adder<size_t>> _total_read_size_metrics;
577
    std::shared_ptr<bvar::Adder<size_t>> _total_hit_size_metrics;
578
    std::shared_ptr<bvar::Adder<size_t>> _total_evict_size_metrics;
579
    std::shared_ptr<bvar::Adder<size_t>> _gc_evict_bytes_metrics;
580
    std::shared_ptr<bvar::Adder<size_t>> _gc_evict_count_metrics;
581
    std::shared_ptr<bvar::Adder<size_t>> _evict_by_time_metrics_matrix[4][4];
582
    std::shared_ptr<bvar::Adder<size_t>> _evict_by_size_metrics_matrix[4][4];
583
    std::shared_ptr<bvar::Adder<size_t>> _evict_by_self_lru_metrics_matrix[4];
584
    std::shared_ptr<bvar::Adder<size_t>> _evict_by_try_release;
585
586
    std::shared_ptr<bvar::Window<bvar::Adder<size_t>>> _num_hit_blocks_5m;
587
    std::shared_ptr<bvar::Window<bvar::Adder<size_t>>> _num_read_blocks_5m;
588
    std::shared_ptr<bvar::Window<bvar::Adder<size_t>>> _num_hit_blocks_1h;
589
    std::shared_ptr<bvar::Window<bvar::Adder<size_t>>> _num_read_blocks_1h;
590
591
    std::shared_ptr<bvar::Adder<size_t>> _num_read_blocks;
592
    std::shared_ptr<bvar::Adder<size_t>> _num_hit_blocks;
593
    std::shared_ptr<bvar::Adder<size_t>> _num_removed_blocks;
594
595
    std::shared_ptr<bvar::Adder<size_t>> _no_warmup_num_read_blocks;
596
    std::shared_ptr<bvar::Adder<size_t>> _no_warmup_num_hit_blocks;
597
598
    std::shared_ptr<bvar::Window<bvar::Adder<size_t>>> _no_warmup_num_hit_blocks_5m;
599
    std::shared_ptr<bvar::Window<bvar::Adder<size_t>>> _no_warmup_num_read_blocks_5m;
600
    std::shared_ptr<bvar::Window<bvar::Adder<size_t>>> _no_warmup_num_hit_blocks_1h;
601
    std::shared_ptr<bvar::Window<bvar::Adder<size_t>>> _no_warmup_num_read_blocks_1h;
602
603
    std::shared_ptr<bvar::Status<double>> _hit_ratio;
604
    std::shared_ptr<bvar::Status<double>> _hit_ratio_5m;
605
    std::shared_ptr<bvar::Status<double>> _hit_ratio_1h;
606
    std::shared_ptr<bvar::Status<double>> _no_warmup_hit_ratio;
607
    std::shared_ptr<bvar::Status<double>> _no_warmup_hit_ratio_5m;
608
    std::shared_ptr<bvar::Status<double>> _no_warmup_hit_ratio_1h;
609
    std::shared_ptr<bvar::Status<size_t>> _disk_limit_mode_metrics;
610
    std::shared_ptr<bvar::Status<size_t>> _need_evict_cache_in_advance_metrics;
611
    std::shared_ptr<bvar::Status<size_t>> _meta_store_write_queue_size_metrics;
612
613
    std::shared_ptr<bvar::LatencyRecorder> _cache_lock_wait_time_us;
614
    std::shared_ptr<bvar::LatencyRecorder> _get_or_set_latency_us;
615
    std::shared_ptr<bvar::LatencyRecorder> _storage_sync_remove_latency_us;
616
    std::shared_ptr<bvar::LatencyRecorder> _storage_retry_sync_remove_latency_us;
617
    std::shared_ptr<bvar::LatencyRecorder> _storage_async_remove_latency_us;
618
    std::shared_ptr<bvar::LatencyRecorder> _evict_in_advance_latency_us;
619
    std::shared_ptr<bvar::LatencyRecorder> _recycle_keys_length_recorder;
620
    std::shared_ptr<bvar::LatencyRecorder> _update_lru_blocks_latency_us;
621
    std::shared_ptr<bvar::LatencyRecorder> _need_update_lru_blocks_length_recorder;
622
    std::shared_ptr<bvar::Adder<size_t>> _need_update_lru_blocks_produce_metrics;
623
    std::shared_ptr<bvar::Adder<size_t>> _need_update_lru_blocks_consume_metrics;
624
    std::shared_ptr<bvar::LatencyRecorder> _ttl_gc_latency_us;
625
626
    std::shared_ptr<bvar::LatencyRecorder> _shadow_queue_levenshtein_distance;
627
    std::array<std::shared_ptr<bvar::LatencyRecorder>, 4> _lru_recorder_queue_length_recorder;
628
    std::array<std::shared_ptr<bvar::Adder<size_t>>, 4> _lru_recorder_queue_produce_metrics;
629
    std::array<std::shared_ptr<bvar::Adder<size_t>>, 4> _lru_recorder_queue_consume_metrics;
630
    std::shared_ptr<bvar::Adder<size_t>> _lru_recorder_log_replay_idle_metrics;
631
    // keep _storage last so it will deconstruct first
632
    // otherwise, load_cache_info_into_memory might crash
633
    // coz it will use other members of BlockFileCache
634
    // so join this async load thread first
635
    std::unique_ptr<FileCacheStorage> _storage;
636
    std::shared_ptr<bvar::LatencyRecorder> _lru_dump_latency_us;
637
    std::mutex _dump_lru_queues_mtx;
638
    NeedUpdateLRUBlocks _need_update_lru_blocks;
639
};
640
641
} // namespace doris::io