be/src/io/cache/file_block.cpp

Source
// 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 "io/cache/file_block.h"

#include <glog/logging.h>
// IWYU pragma: no_include <bits/chrono.h>
#include <chrono> // IWYU pragma: keep
#include <sstream>
#include <string>
#include <thread>

#include "common/status.h"
#include "cpp/sync_point.h"
#include "io/cache/block_file_cache.h"

namespace doris {
namespace io {

std::ostream& operator<<(std::ostream& os, const FileBlock::State& value) {
    os << FileBlock::state_to_string(value);
    return os;
}

FileBlock::FileBlock(const FileCacheKey& key, size_t size, BlockFileCache* mgr,
                     State download_state)
        : _block_range(key.offset, key.offset + size - 1),
          _download_state(download_state),
          _mgr(mgr),
          _key(key) {
    /// On creation, file block state can be EMPTY, DOWNLOADED, SKIP_CACHE.
    switch (_download_state) {
    case State::DOWNLOADING: {
        DCHECK(false) << "Can create cell with either EMPTY, DOWNLOADED, SKIP_CACHE ";
        break;
    }
    default: {
        break;
    }
    }
}

FileBlock::State FileBlock::state() const {
    std::lock_guard block_lock(_mutex);
    return _download_state;
}

FileBlock::State FileBlock::state_unsafe() const {
    return _download_state;
}

uint64_t FileBlock::get_caller_id() {
    uint64_t id;
#if defined(__APPLE__)
    // On macOS, use pthread_threadid_np to get the thread ID
    pthread_threadid_np(nullptr, &id);
#else
    id = static_cast<uint64_t>(pthread_self());
#endif
    DCHECK(id != 0);
    return id;
}

uint64_t FileBlock::get_or_set_downloader() {
    std::lock_guard block_lock(_mutex);

    if (_downloader_id == 0 && _download_state != State::DOWNLOADED) {
        DCHECK(_download_state != State::DOWNLOADING);
        _downloader_id = get_caller_id();
        _download_state = State::DOWNLOADING;
    } else if (_downloader_id == get_caller_id()) {
        LOG(INFO) << "Attempt to set the same downloader for block " << range().to_string()
                  << " for the second time";
    }

    return _downloader_id;
}

void FileBlock::reset_downloader(std::lock_guard<std::mutex>& block_lock) {
    DCHECK(_downloader_id != 0) << "There is no downloader";

    DCHECK(get_caller_id() == _downloader_id) << "Downloader can be reset only by downloader";

    reset_downloader_impl(block_lock);
}

void FileBlock::reset_downloader_impl(std::lock_guard<std::mutex>& block_lock) {
    if (_downloaded_size == range().size()) {
        Status st = set_downloaded(block_lock);
        if (!st.ok()) {
            LOG(WARNING) << "reset downloader error" << st;
        }
    } else {
        _downloaded_size = 0;
        _download_state = State::EMPTY;
        _downloader_id = 0;
    }
}

Status FileBlock::set_downloaded(std::lock_guard<std::mutex>& /* block_lock */) {
    DCHECK(_download_state != State::DOWNLOADED);
    if (_downloaded_size == 0) {
        _download_state = State::EMPTY;
        _downloader_id = 0;
        return Status::InternalError("Try to set empty block {} as downloaded",
                                     _block_range.to_string());
    }
    Status status = _mgr->_storage->finalize(_key, this->_block_range.size());
    if (status.ok()) [[likely]] {
        _download_state = State::DOWNLOADED;
    } else {
        _download_state = State::EMPTY;
        _downloaded_size = 0;
    }
    _downloader_id = 0;
    return status;
}

uint64_t FileBlock::get_downloader() const {
    std::lock_guard block_lock(_mutex);
    return _downloader_id;
}

bool FileBlock::is_downloader() const {
    std::lock_guard block_lock(_mutex);
    return get_caller_id() == _downloader_id;
}

bool FileBlock::is_downloader_impl(std::lock_guard<std::mutex>& /* block_lock */) const {
    return get_caller_id() == _downloader_id;
}

Status FileBlock::append(Slice data) {
    DCHECK(data.size != 0) << "Writing zero size is not allowed";
    RETURN_IF_ERROR(_mgr->_storage->append(_key, data));
    _downloaded_size += data.size;
    return Status::OK();
}

Status FileBlock::finalize() {
    if (_downloaded_size == 0) {
        std::lock_guard block_lock(_mutex);
        _download_state = State::EMPTY;
        _downloader_id = 0;
        _cv.notify_all();
        return Status::InternalError("Try to finalize an empty file block {}",
                                     _block_range.to_string());
    }
    if (_downloaded_size != _block_range.size()) {
        SCOPED_CACHE_LOCK(_mgr->_mutex, _mgr);
        size_t old_size = _block_range.size();
        _block_range.right = _block_range.left + _downloaded_size - 1;
        size_t new_size = _block_range.size();
        DCHECK(new_size < old_size);
        _mgr->reset_range(_key.hash, _block_range.left, old_size, new_size, cache_lock);
    }
    std::lock_guard block_lock(_mutex);
    Status st = set_downloaded(block_lock);
    _cv.notify_all();
    return st;
}

Status FileBlock::read(Slice buffer, size_t read_offset) {
    return _mgr->_storage->read(_key, read_offset, buffer);
}

Status FileBlock::change_cache_type(FileCacheType new_type) {
    SCOPED_CACHE_LOCK(_mgr->_mutex, _mgr);
    return change_cache_type_lock(new_type, cache_lock);
}

Status FileBlock::change_cache_type_lock(FileCacheType new_type,
                                         std::lock_guard<std::mutex>& cache_lock) {
    std::lock_guard block_lock(_mutex);

    if (new_type == _key.meta.type) {
        return Status::OK();
    }
    if (_download_state == State::DOWNLOADED) {
        Status st;
        TEST_SYNC_POINT_CALLBACK("FileBlock::change_cache_type", &st);
        RETURN_IF_ERROR(_mgr->_storage->change_key_meta_type(_key, new_type, _block_range.size()));
    }
    _mgr->change_cache_type(_key.hash, _block_range.left, new_type, cache_lock);
    _key.meta.type = new_type;
    return Status::OK();
}

FileBlock::State FileBlock::wait() {
    std::unique_lock block_lock(_mutex);

    if (_downloader_id == 0) {
        return _download_state;
    }

    if (_download_state == State::DOWNLOADING) {
        DCHECK(_downloader_id != 0 && _downloader_id != get_caller_id());
        _cv.wait_for(block_lock, std::chrono::milliseconds(config::block_cache_wait_timeout_ms));
    }

    return _download_state;
}

void FileBlock::complete_unlocked(std::lock_guard<std::mutex>& block_lock) {
    if (is_downloader_impl(block_lock)) {
        reset_downloader(block_lock);
        _cv.notify_all();
    }
}

std::string FileBlock::get_info_for_log() const {
    std::lock_guard block_lock(_mutex);
    return get_info_for_log_impl(block_lock);
}

std::string FileBlock::get_info_for_log_impl(std::lock_guard<std::mutex>& block_lock) const {
    std::stringstream info;
    info << "File block: " << range().to_string() << ", ";
    info << "state: " << state_to_string(_download_state) << ", ";
    info << "size: " << _block_range.size() << ", ";
    info << "downloader id: " << _downloader_id << ", ";
    info << "caller id: " << get_caller_id();

    return info.str();
}

FileBlock::State FileBlock::state_unlock(std::lock_guard<std::mutex>&) const {
    return _download_state;
}

std::string FileBlock::state_to_string(FileBlock::State state) {
    switch (state) {
    case FileBlock::State::DOWNLOADED:
        return "DOWNLOADED";
    case FileBlock::State::EMPTY:
        return "EMPTY";
    case FileBlock::State::DOWNLOADING:
        return "DOWNLOADING";
    case FileBlock::State::SKIP_CACHE:
        return "SKIP_CACHE";
    default:
        DCHECK(false);
        return "";
    }
}

std::string FileBlock::get_cache_file() const {
    return _mgr->_storage->get_local_file(this->_key);
}

FileBlocksHolder::~FileBlocksHolder() {
    for (auto file_block_it = file_blocks.begin(); file_block_it != file_blocks.end();) {
        auto current_file_block_it = file_block_it;
        auto& file_block = *current_file_block_it;
        BlockFileCache* _mgr = file_block->_mgr;
        {
            bool should_remove = false;
            {
                std::lock_guard block_lock(file_block->_mutex);
                file_block->complete_unlocked(block_lock);
                if (file_block.use_count() == 2 &&
                    (file_block->is_deleting() ||
                     file_block->state_unlock(block_lock) == FileBlock::State::EMPTY)) {
                    should_remove = true;
                }
            }
            if (should_remove) {
                SCOPED_CACHE_LOCK(_mgr->_mutex, _mgr);
                std::lock_guard block_lock(file_block->_mutex);
                if (file_block.use_count() == 2) {
                    DCHECK(file_block->state_unlock(block_lock) != FileBlock::State::DOWNLOADING);
                    // one in cache, one in here
                    if (file_block->is_deleting() ||
                        file_block->state_unlock(block_lock) == FileBlock::State::EMPTY) {
                        _mgr->remove(file_block, cache_lock, block_lock, false);
                    }
                }
            }
        }
        file_block_it = file_blocks.erase(current_file_block_it);
    }
}

} // namespace io
} // namespace doris

Coverage Report

Created: 2026-05-28 10:49

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.91k	std::ostream& operator<<(std::ostream& os, const FileBlock::State& value) {
35	1.91k	os << FileBlock::state_to_string(value);
36	1.91k	return os;
37	1.91k	}
38
39		FileBlock::FileBlock(const FileCacheKey& key, size_t size, BlockFileCache* mgr,
40		State download_state)
41	363k	: _block_range(key.offset, key.offset + size - 1),
42	363k	_download_state(download_state),
43	363k	_mgr(mgr),
44	363k	_key(key) {
45		/// On creation, file block state can be EMPTY, DOWNLOADED, SKIP_CACHE.
46	363k	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	363k	default: {
52	363k	break;
53	0	}
54	363k	}
55	363k	}
56
57	7.60M	FileBlock::State FileBlock::state() const {
58	7.60M	std::lock_guard block_lock(_mutex);
59	7.60M	return _download_state;
60	7.60M	}
61
62	4.08M	FileBlock::State FileBlock::state_unsafe() const {
63	4.08M	return _download_state;
64	4.08M	}
65
66	4.33M	uint64_t FileBlock::get_caller_id() {
67	4.33M	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	4.33M	id = static_cast<uint64_t>(pthread_self());
73	4.33M	#endif
74	4.33M	DCHECK(id != 0);
75	4.33M	return id;
76	4.33M	}
77
78	122k	uint64_t FileBlock::get_or_set_downloader() {
79	122k	std::lock_guard block_lock(_mutex);
80
81	122k	if (_downloader_id == 0 && _download_state != State::DOWNLOADED) {
82	122k	DCHECK(_download_state != State::DOWNLOADING);
83	122k	_downloader_id = get_caller_id();
84	122k	_download_state = State::DOWNLOADING;
85	122k	} 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	122k	return _downloader_id;
91	122k	}
92
93	21	void FileBlock::reset_downloader(std::lock_guard<std::mutex>& block_lock) {
94	21	DCHECK(_downloader_id != 0) << "There is no downloader";
95
96	21	DCHECK(get_caller_id() == _downloader_id) << "Downloader can be reset only by downloader";
97
98	21	reset_downloader_impl(block_lock);
99	21	}
100
101	21	void FileBlock::reset_downloader_impl(std::lock_guard<std::mutex>& block_lock) {
102	21	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	19	} else {
108	19	_downloaded_size = 0;
109	19	_download_state = State::EMPTY;
110	19	_downloader_id = 0;
111	19	}
112	21	}
113
114	122k	Status FileBlock::set_downloaded(std::lock_guard<std::mutex>& /* block_lock */) {
115	122k	DCHECK(_download_state != State::DOWNLOADED);
116	122k	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	122k	Status status = _mgr->_storage->finalize(_key, this->_block_range.size());
123	122k	if (status.ok()) [[likely]] {
124	121k	_download_state = State::DOWNLOADED;
125	121k	} else {
126	796	_download_state = State::EMPTY;
127	796	_downloaded_size = 0;
128	796	}
129	122k	_downloader_id = 0;
130	122k	return status;
131	122k	}
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	117k	bool FileBlock::is_downloader() const {
139	117k	std::lock_guard block_lock(_mutex);
140	117k	return get_caller_id() == _downloader_id;
141	117k	}
142
143	4.07M	bool FileBlock::is_downloader_impl(std::lock_guard<std::mutex>& /* block_lock */) const {
144	4.07M	return get_caller_id() == _downloader_id;
145	4.07M	}
146
147	122k	Status FileBlock::append(Slice data) {
148	122k	DCHECK(data.size != 0) << "Writing zero size is not allowed";
149	122k	RETURN_IF_ERROR(_mgr->_storage->append(_key, data));
150	122k	_downloaded_size += data.size;
151	122k	return Status::OK();
152	122k	}
153
154	122k	Status FileBlock::finalize() {
155	122k	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	122k	if (_downloaded_size != _block_range.size()) {
164	57.2k	SCOPED_CACHE_LOCK(_mgr->_mutex, _mgr);
165	57.2k	size_t old_size = _block_range.size();
166	57.2k	_block_range.right = _block_range.left + _downloaded_size - 1;
167	57.2k	size_t new_size = _block_range.size();
168	57.2k	DCHECK(new_size < old_size);
169	57.2k	_mgr->reset_range(_key.hash, _block_range.left, old_size, new_size, cache_lock);
170	57.2k	}
171	122k	std::lock_guard block_lock(_mutex);
172	122k	Status st = set_downloaded(block_lock);
173	122k	_cv.notify_all();
174	122k	return st;
175	122k	}
176
177	2.78M	Status FileBlock::read(Slice buffer, size_t read_offset) {
178	2.78M	return _mgr->_storage->read(_key, read_offset, buffer);
179	2.78M	}
180
181	2.72k	Status FileBlock::change_cache_type(FileCacheType new_type) {
182	2.72k	SCOPED_CACHE_LOCK(_mgr->_mutex, _mgr);
183	2.72k	return change_cache_type_lock(new_type, cache_lock);
184	2.72k	}
185
186		Status FileBlock::change_cache_type_lock(FileCacheType new_type,
187	2.72k	std::lock_guard<std::mutex>& cache_lock) {
188	2.72k	std::lock_guard block_lock(_mutex);
189
190	2.72k	if (new_type == _key.meta.type) {
191	1	return Status::OK();
192	1	}
193	2.72k	if (_download_state == State::DOWNLOADED) {
194	7	Status st;
195	7	TEST_SYNC_POINT_CALLBACK("FileBlock::change_cache_type", &st);
196	7	RETURN_IF_ERROR(_mgr->_storage->change_key_meta_type(_key, new_type, _block_range.size()));
197	7	}
198	2.72k	_mgr->change_cache_type(_key.hash, _block_range.left, new_type, cache_lock);
199	2.72k	_key.meta.type = new_type;
200	2.72k	return Status::OK();
201	2.72k	}
202
203	7.09k	FileBlock::State FileBlock::wait() {
204	7.09k	std::unique_lock block_lock(_mutex);
205
206	7.09k	if (_downloader_id == 0) {
207	3	return _download_state;
208	3	}
209
210	7.09k	if (_download_state == State::DOWNLOADING) {
211	7.09k	DCHECK(_downloader_id != 0 && _downloader_id != get_caller_id());
212	7.09k	_cv.wait_for(block_lock, std::chrono::milliseconds(config::block_cache_wait_timeout_ms));
213	7.09k	}
214
215	7.09k	return _download_state;
216	7.09k	}
217
218	4.07M	void FileBlock::complete_unlocked(std::lock_guard<std::mutex>& block_lock) {
219	4.07M	if (is_downloader_impl(block_lock)) {
220	21	reset_downloader(block_lock);
221	21	_cv.notify_all();
222	21	}
223	4.07M	}
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	5.09M	FileBlock::State FileBlock::state_unlock(std::lock_guard<std::mutex>&) const {
242	5.09M	return _download_state;
243	5.09M	}
244
245	1.91k	std::string FileBlock::state_to_string(FileBlock::State state) {
246	1.91k	switch (state) {
247	1.90k	case FileBlock::State::DOWNLOADED:
248	1.90k	return "DOWNLOADED";
249	3	case FileBlock::State::EMPTY:
250	3	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.91k	}
259	1.91k	}
260
261	1	std::string FileBlock::get_cache_file() const {
262	1	return _mgr->_storage->get_local_file(this->_key);
263	1	}
264
265	3.85M	FileBlocksHolder::~FileBlocksHolder() {
266	7.93M	for (auto file_block_it = file_blocks.begin(); file_block_it != file_blocks.end();) {
267	4.07M	auto current_file_block_it = file_block_it;
268	4.07M	auto& file_block = *current_file_block_it;
269	4.07M	BlockFileCache* _mgr = file_block->_mgr;
270	4.07M	{
271	4.07M	bool should_remove = false;
272	4.07M	{
273	4.07M	std::lock_guard block_lock(file_block->_mutex);
274	4.07M	file_block->complete_unlocked(block_lock);
275	4.07M	if (file_block.use_count() == 2 &&
276	4.07M	(file_block->is_deleting() \|\|
277	4.04M	file_block->state_unlock(block_lock) == FileBlock::State::EMPTY)) {
278	234k	should_remove = true;
279	234k	}
280	4.07M	}
281	4.07M	if (should_remove) {
282	234k	SCOPED_CACHE_LOCK(_mgr->_mutex, _mgr);
283	234k	std::lock_guard block_lock(file_block->_mutex);
284	234k	if (file_block.use_count() == 2) {
285	234k	DCHECK(file_block->state_unlock(block_lock) != FileBlock::State::DOWNLOADING);
286		// one in cache, one in here
287	234k	if (file_block->is_deleting() \|\|
288	234k	file_block->state_unlock(block_lock) == FileBlock::State::EMPTY) {
289	234k	_mgr->remove(file_block, cache_lock, block_lock, false);
290	234k	}
291	234k	}
292	234k	}
293	4.07M	}
294	4.07M	file_block_it = file_blocks.erase(current_file_block_it);
295	4.07M	}
296	3.85M	}
297
298		} // namespace io
299		} // namespace doris