Coverage Report

Created: 2026-03-13 12:15

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/io/cache/file_block.cpp
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
#include "io/cache/file_block.h"
19
20
#include <glog/logging.h>
21
// IWYU pragma: no_include <bits/chrono.h>
22
#include <chrono> // IWYU pragma: keep
23
#include <sstream>
24
#include <string>
25
#include <thread>
26
27
#include "common/status.h"
28
#include "cpp/sync_point.h"
29
#include "io/cache/block_file_cache.h"
30
31
namespace doris {
32
namespace io {
33
34
1.28k
std::ostream& operator<<(std::ostream& os, const FileBlock::State& value) {
35
1.28k
    os << FileBlock::state_to_string(value);
36
1.28k
    return os;
37
1.28k
}
38
39
FileBlock::FileBlock(const FileCacheKey& key, size_t size, BlockFileCache* mgr,
40
                     State download_state)
41
79.5k
        : _block_range(key.offset, key.offset + size - 1),
42
79.5k
          _download_state(download_state),
43
79.5k
          _mgr(mgr),
44
79.5k
          _key(key) {
45
    /// On creation, file block state can be EMPTY, DOWNLOADED, SKIP_CACHE.
46
79.5k
    switch (_download_state) {
47
0
    case State::DOWNLOADING: {
48
0
        DCHECK(false) << "Can create cell with either EMPTY, DOWNLOADED, SKIP_CACHE ";
49
0
        break;
50
0
    }
51
79.5k
    default: {
52
79.5k
        break;
53
0
    }
54
79.5k
    }
55
79.5k
}
56
57
1.56M
FileBlock::State FileBlock::state() const {
58
1.56M
    std::lock_guard block_lock(_mutex);
59
1.56M
    return _download_state;
60
1.56M
}
61
62
840k
FileBlock::State FileBlock::state_unsafe() const {
63
840k
    return _download_state;
64
840k
}
65
66
908k
uint64_t FileBlock::get_caller_id() {
67
908k
    uint64_t id;
68
#if defined(__APPLE__)
69
    // On macOS, use pthread_threadid_np to get the thread ID
70
    pthread_threadid_np(nullptr, &id);
71
#else
72
908k
    id = static_cast<uint64_t>(pthread_self());
73
908k
#endif
74
908k
    DCHECK(id != 0);
75
908k
    return id;
76
908k
}
77
78
31.8k
uint64_t FileBlock::get_or_set_downloader() {
79
31.8k
    std::lock_guard block_lock(_mutex);
80
81
31.8k
    if (_downloader_id == 0 && _download_state != State::DOWNLOADED) {
82
31.8k
        DCHECK(_download_state != State::DOWNLOADING);
83
31.8k
        _downloader_id = get_caller_id();
84
31.8k
        _download_state = State::DOWNLOADING;
85
31.8k
    } else if (_downloader_id == get_caller_id()) {
86
4
        LOG(INFO) << "Attempt to set the same downloader for block " << range().to_string()
87
4
                  << " for the second time";
88
4
    }
89
90
31.8k
    return _downloader_id;
91
31.8k
}
92
93
19
void FileBlock::reset_downloader(std::lock_guard<std::mutex>& block_lock) {
94
19
    DCHECK(_downloader_id != 0) << "There is no downloader";
95
96
19
    DCHECK(get_caller_id() == _downloader_id) << "Downloader can be reset only by downloader";
97
98
19
    reset_downloader_impl(block_lock);
99
19
}
100
101
19
void FileBlock::reset_downloader_impl(std::lock_guard<std::mutex>& block_lock) {
102
19
    if (_downloaded_size == range().size()) {
103
2
        Status st = set_downloaded(block_lock);
104
2
        if (!st.ok()) {
105
1
            LOG(WARNING) << "reset downloader error" << st;
106
1
        }
107
17
    } else {
108
17
        _downloaded_size = 0;
109
17
        _download_state = State::EMPTY;
110
17
        _downloader_id = 0;
111
17
    }
112
19
}
113
114
31.8k
Status FileBlock::set_downloaded(std::lock_guard<std::mutex>& /* block_lock */) {
115
31.8k
    DCHECK(_download_state != State::DOWNLOADED);
116
31.8k
    if (_downloaded_size == 0) {
117
1
        _download_state = State::EMPTY;
118
1
        _downloader_id = 0;
119
1
        return Status::InternalError("Try to set empty block {} as downloaded",
120
1
                                     _block_range.to_string());
121
1
    }
122
31.8k
    Status status = _mgr->_storage->finalize(_key, this->_block_range.size());
123
31.8k
    if (status.ok()) [[likely]] {
124
31.8k
        _download_state = State::DOWNLOADED;
125
31.8k
    } else {
126
29
        _download_state = State::EMPTY;
127
29
        _downloaded_size = 0;
128
29
    }
129
31.8k
    _downloader_id = 0;
130
31.8k
    return status;
131
31.8k
}
132
133
16
uint64_t FileBlock::get_downloader() const {
134
16
    std::lock_guard block_lock(_mutex);
135
16
    return _downloader_id;
136
16
}
137
138
26.5k
bool FileBlock::is_downloader() const {
139
26.5k
    std::lock_guard block_lock(_mutex);
140
26.5k
    return get_caller_id() == _downloader_id;
141
26.5k
}
142
143
840k
bool FileBlock::is_downloader_impl(std::lock_guard<std::mutex>& /* block_lock */) const {
144
840k
    return get_caller_id() == _downloader_id;
145
840k
}
146
147
31.8k
Status FileBlock::append(Slice data) {
148
31.8k
    DCHECK(data.size != 0) << "Writing zero size is not allowed";
149
31.8k
    RETURN_IF_ERROR(_mgr->_storage->append(_key, data));
150
31.8k
    _downloaded_size += data.size;
151
31.8k
    return Status::OK();
152
31.8k
}
153
154
31.8k
Status FileBlock::finalize() {
155
31.8k
    if (_downloaded_size == 0) {
156
1
        std::lock_guard block_lock(_mutex);
157
1
        _download_state = State::EMPTY;
158
1
        _downloader_id = 0;
159
1
        _cv.notify_all();
160
1
        return Status::InternalError("Try to finalize an empty file block {}",
161
1
                                     _block_range.to_string());
162
1
    }
163
31.8k
    if (_downloaded_size != _block_range.size()) {
164
10.8k
        SCOPED_CACHE_LOCK(_mgr->_mutex, _mgr);
165
10.8k
        size_t old_size = _block_range.size();
166
10.8k
        _block_range.right = _block_range.left + _downloaded_size - 1;
167
10.8k
        size_t new_size = _block_range.size();
168
10.8k
        DCHECK(new_size < old_size);
169
10.8k
        _mgr->reset_range(_key.hash, _block_range.left, old_size, new_size, cache_lock);
170
10.8k
    }
171
31.8k
    std::lock_guard block_lock(_mutex);
172
31.8k
    Status st = set_downloaded(block_lock);
173
31.8k
    _cv.notify_all();
174
31.8k
    return st;
175
31.8k
}
176
177
600k
Status FileBlock::read(Slice buffer, size_t read_offset) {
178
600k
    return _mgr->_storage->read(_key, read_offset, buffer);
179
600k
}
180
181
790
Status FileBlock::change_cache_type(FileCacheType new_type) {
182
790
    SCOPED_CACHE_LOCK(_mgr->_mutex, _mgr);
183
790
    return change_cache_type_lock(new_type, cache_lock);
184
790
}
185
186
Status FileBlock::change_cache_type_lock(FileCacheType new_type,
187
791
                                         std::lock_guard<std::mutex>& cache_lock) {
188
791
    std::lock_guard block_lock(_mutex);
189
190
791
    if (new_type == _key.meta.type) {
191
1
        return Status::OK();
192
1
    }
193
790
    if (_download_state == State::DOWNLOADED) {
194
6
        Status st;
195
6
        TEST_SYNC_POINT_CALLBACK("FileBlock::change_cache_type", &st);
196
6
        RETURN_IF_ERROR(_mgr->_storage->change_key_meta_type(_key, new_type, _block_range.size()));
197
6
    }
198
790
    _mgr->change_cache_type(_key.hash, _block_range.left, new_type, cache_lock);
199
790
    _key.meta.type = new_type;
200
790
    return Status::OK();
201
790
}
202
203
3.93k
FileBlock::State FileBlock::wait() {
204
3.93k
    std::unique_lock block_lock(_mutex);
205
206
3.93k
    if (_downloader_id == 0) {
207
1
        return _download_state;
208
1
    }
209
210
3.93k
    if (_download_state == State::DOWNLOADING) {
211
3.93k
        DCHECK(_downloader_id != 0 && _downloader_id != get_caller_id());
212
3.93k
        _cv.wait_for(block_lock, std::chrono::milliseconds(config::block_cache_wait_timeout_ms));
213
3.93k
    }
214
215
3.93k
    return _download_state;
216
3.93k
}
217
218
840k
void FileBlock::complete_unlocked(std::lock_guard<std::mutex>& block_lock) {
219
840k
    if (is_downloader_impl(block_lock)) {
220
19
        reset_downloader(block_lock);
221
19
        _cv.notify_all();
222
19
    }
223
840k
}
224
225
1
std::string FileBlock::get_info_for_log() const {
226
1
    std::lock_guard block_lock(_mutex);
227
1
    return get_info_for_log_impl(block_lock);
228
1
}
229
230
1
std::string FileBlock::get_info_for_log_impl(std::lock_guard<std::mutex>& block_lock) const {
231
1
    std::stringstream info;
232
1
    info << "File block: " << range().to_string() << ", ";
233
1
    info << "state: " << state_to_string(_download_state) << ", ";
234
1
    info << "size: " << _block_range.size() << ", ";
235
1
    info << "downloader id: " << _downloader_id << ", ";
236
1
    info << "caller id: " << get_caller_id();
237
238
1
    return info.str();
239
1
}
240
241
1.02M
FileBlock::State FileBlock::state_unlock(std::lock_guard<std::mutex>&) const {
242
1.02M
    return _download_state;
243
1.02M
}
244
245
1.29k
std::string FileBlock::state_to_string(FileBlock::State state) {
246
1.29k
    switch (state) {
247
1.28k
    case FileBlock::State::DOWNLOADED:
248
1.28k
        return "DOWNLOADED";
249
2
    case FileBlock::State::EMPTY:
250
2
        return "EMPTY";
251
2
    case FileBlock::State::DOWNLOADING:
252
2
        return "DOWNLOADING";
253
1
    case FileBlock::State::SKIP_CACHE:
254
1
        return "SKIP_CACHE";
255
0
    default:
256
0
        DCHECK(false);
257
0
        return "";
258
1.29k
    }
259
1.29k
}
260
261
1
std::string FileBlock::get_cache_file() const {
262
1
    return _mgr->_storage->get_local_file(this->_key);
263
1
}
264
265
758k
FileBlocksHolder::~FileBlocksHolder() {
266
1.59M
    for (auto file_block_it = file_blocks.begin(); file_block_it != file_blocks.end();) {
267
840k
        auto current_file_block_it = file_block_it;
268
840k
        auto& file_block = *current_file_block_it;
269
840k
        BlockFileCache* _mgr = file_block->_mgr;
270
840k
        {
271
840k
            bool should_remove = false;
272
840k
            {
273
840k
                std::lock_guard block_lock(file_block->_mutex);
274
840k
                file_block->complete_unlocked(block_lock);
275
840k
                if (file_block.use_count() == 2 &&
276
840k
                    (file_block->is_deleting() ||
277
817k
                     file_block->state_unlock(block_lock) == FileBlock::State::EMPTY)) {
278
46.5k
                    should_remove = true;
279
46.5k
                }
280
840k
            }
281
840k
            if (should_remove) {
282
46.5k
                SCOPED_CACHE_LOCK(_mgr->_mutex, _mgr);
283
46.5k
                std::lock_guard block_lock(file_block->_mutex);
284
46.5k
                if (file_block.use_count() == 2) {
285
46.5k
                    DCHECK(file_block->state_unlock(block_lock) != FileBlock::State::DOWNLOADING);
286
                    // one in cache, one in here
287
46.5k
                    if (file_block->is_deleting() ||
288
46.5k
                        file_block->state_unlock(block_lock) == FileBlock::State::EMPTY) {
289
46.5k
                        _mgr->remove(file_block, cache_lock, block_lock, false);
290
46.5k
                    }
291
46.5k
                }
292
46.5k
            }
293
840k
        }
294
840k
        file_block_it = file_blocks.erase(current_file_block_it);
295
840k
    }
296
758k
}
297
298
} // namespace io
299
} // namespace doris