ClickHouse/src/Interpreters/tests/gtest_lru_file_cache.cpp

Ignoring revisions in .git-blame-ignore-revs. Click here to bypass and see the normal blame view.

520 lines
22 KiB
C++
Raw Normal View History

2022-01-13 11:57:56 +00:00
#include <iomanip>
#include <iostream>
#include <gtest/gtest.h>
2022-08-31 19:07:39 +00:00
#include <Interpreters/Cache/FileCache.h>
2022-08-26 17:23:46 +00:00
#include <Interpreters/Cache/FileSegment.h>
2022-01-21 15:39:34 +00:00
#include <Common/CurrentThread.h>
2022-01-13 11:57:56 +00:00
#include <Common/filesystemHelpers.h>
2022-09-06 11:30:02 +00:00
#include <Interpreters/Cache/FileCacheSettings.h>
2022-01-24 09:32:39 +00:00
#include <Common/tests/gtest_global_context.h>
2022-01-21 15:39:34 +00:00
#include <Common/SipHash.h>
#include <Common/hex.h>
#include <Interpreters/Context.h>
#include <IO/ReadHelpers.h>
2022-01-13 11:57:56 +00:00
#include <IO/WriteHelpers.h>
#include <filesystem>
#include <thread>
2022-11-21 17:45:59 +00:00
2022-01-13 11:57:56 +00:00
namespace fs = std::filesystem;
fs::path caches_dir = fs::current_path() / "lru_cache_test";
String cache_base_path = caches_dir / "cache1" / "";
2022-01-13 11:57:56 +00:00
void assertRange(
2022-01-22 22:56:24 +00:00
[[maybe_unused]] size_t assert_n, DB::FileSegmentPtr file_segment,
2022-01-13 11:57:56 +00:00
const DB::FileSegment::Range & expected_range, DB::FileSegment::State expected_state)
{
auto range = file_segment->range();
2022-01-23 16:51:18 +00:00
std::cerr << fmt::format("\nAssert #{} : {} == {} (state: {} == {})\n", assert_n,
range.toString(), expected_range.toString(),
toString(file_segment->state()), toString(expected_state));
2022-01-13 11:57:56 +00:00
ASSERT_EQ(range.left, expected_range.left);
ASSERT_EQ(range.right, expected_range.right);
ASSERT_EQ(file_segment->state(), expected_state);
}
2022-01-13 11:57:56 +00:00
void printRanges(const auto & segments)
{
std::cerr << "\nHaving file segments: ";
for (const auto & segment : segments)
std::cerr << '\n' << segment->range().toString() << " (state: " + DB::FileSegment::stateToString(segment->state()) + ")" << "\n";
2022-01-13 11:57:56 +00:00
}
std::vector<DB::FileSegmentPtr> fromHolder(const DB::FileSegmentsHolder & holder)
{
return std::vector<DB::FileSegmentPtr>(holder.file_segments.begin(), holder.file_segments.end());
}
2022-08-10 05:50:30 +00:00
String getFileSegmentPath(const String & base_path, const DB::FileCache::Key & key, size_t offset)
2022-01-21 15:39:34 +00:00
{
2022-05-14 12:26:04 +00:00
auto key_str = key.toString();
2022-01-21 15:39:34 +00:00
return fs::path(base_path) / key_str.substr(0, 3) / key_str / DB::toString(offset);
2022-01-13 11:57:56 +00:00
}
void download(DB::FileSegmentPtr file_segment)
2022-01-13 11:57:56 +00:00
{
const auto & key = file_segment->key();
size_t size = file_segment->range().size();
2022-05-14 12:26:04 +00:00
auto key_str = key.toString();
2022-01-21 15:39:34 +00:00
auto subdir = fs::path(cache_base_path) / key_str.substr(0, 3) / key_str;
2022-01-13 11:57:56 +00:00
if (!fs::exists(subdir))
fs::create_directories(subdir);
std::string data(size, '0');
2022-09-06 11:30:02 +00:00
file_segment->write(data.data(), size, file_segment->getCurrentWriteOffset());
2022-01-21 15:39:34 +00:00
}
void prepareAndDownload(DB::FileSegmentPtr file_segment)
2022-01-21 15:39:34 +00:00
{
// std::cerr << "Reserving: " << file_segment->range().size() << " for: " << file_segment->range().toString() << "\n";
2022-01-21 15:39:34 +00:00
ASSERT_TRUE(file_segment->reserve(file_segment->range().size()));
download(file_segment);
2022-01-13 11:57:56 +00:00
}
void complete(const DB::FileSegmentsHolder & holder)
2022-01-13 11:57:56 +00:00
{
for (const auto & file_segment : holder.file_segments)
{
ASSERT_TRUE(file_segment->getOrSetDownloader() == DB::FileSegment::getCallerId());
prepareAndDownload(file_segment);
file_segment->completeWithoutState();
2022-01-13 11:57:56 +00:00
}
}
2022-11-23 14:48:51 +00:00
TEST(FileCache, get)
2022-11-23 14:48:51 +00:00
{
if (fs::exists(cache_base_path))
fs::remove_all(cache_base_path);
fs::create_directories(cache_base_path);
2022-01-21 15:39:34 +00:00
DB::ThreadStatus thread_status;
/// To work with cache need query_id and query context.
2022-06-06 08:48:30 +00:00
std::string query_id = "query_id";
2022-01-24 09:32:39 +00:00
auto query_context = DB::Context::createCopy(getContext().context);
2022-01-21 15:39:34 +00:00
query_context->makeQueryContext();
2022-06-06 08:48:30 +00:00
query_context->setCurrentQueryId(query_id);
2022-01-21 15:39:34 +00:00
DB::CurrentThread::QueryScope query_scope_holder(query_context);
2022-03-25 17:31:15 +00:00
DB::FileCacheSettings settings;
settings.max_size = 30;
settings.max_elements = 5;
2022-01-13 11:57:56 +00:00
{
2022-08-31 16:07:40 +00:00
auto cache = DB::FileCache(cache_base_path, settings);
cache.initialize();
auto key = cache.hash("key1");
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
{
2022-09-06 11:30:02 +00:00
auto holder = cache.getOrSet(key, 0, 10, {}); /// Add range [0, 9]
2022-08-31 16:07:40 +00:00
auto segments = fromHolder(holder);
/// Range was not present in cache. It should be added in cache as one while file segment.
ASSERT_EQ(segments.size(), 1);
2022-01-21 15:39:34 +00:00
2022-08-31 16:07:40 +00:00
assertRange(1, segments[0], DB::FileSegment::Range(0, 9), DB::FileSegment::State::EMPTY);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
/// Exception because space not reserved.
/// EXPECT_THROW(download(segments[0]), DB::Exception);
/// Exception because space can be reserved only by downloader
/// EXPECT_THROW(segments[0]->reserve(segments[0]->range().size()), DB::Exception);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
ASSERT_TRUE(segments[0]->getOrSetDownloader() == DB::FileSegment::getCallerId());
ASSERT_TRUE(segments[0]->reserve(segments[0]->range().size()));
assertRange(2, segments[0], DB::FileSegment::Range(0, 9), DB::FileSegment::State::DOWNLOADING);
2022-01-13 11:57:56 +00:00
download(segments[0]);
segments[0]->completeWithoutState();
2022-08-31 16:07:40 +00:00
assertRange(3, segments[0], DB::FileSegment::Range(0, 9), DB::FileSegment::State::DOWNLOADED);
}
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
/// Current cache: [__________]
/// ^ ^
/// 0 9
ASSERT_EQ(cache.getFileSegmentsNum(), 1);
ASSERT_EQ(cache.getUsedCacheSize(), 10);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
{
/// Want range [5, 14], but [0, 9] already in cache, so only [10, 14] will be put in cache.
2022-09-06 11:30:02 +00:00
auto holder = cache.getOrSet(key, 5, 10, {});
2022-08-31 16:07:40 +00:00
auto segments = fromHolder(holder);
ASSERT_EQ(segments.size(), 2);
assertRange(4, segments[0], DB::FileSegment::Range(0, 9), DB::FileSegment::State::DOWNLOADED);
assertRange(5, segments[1], DB::FileSegment::Range(10, 14), DB::FileSegment::State::EMPTY);
ASSERT_TRUE(segments[1]->getOrSetDownloader() == DB::FileSegment::getCallerId());
prepareAndDownload(segments[1]);
segments[1]->completeWithoutState();
2022-08-31 16:07:40 +00:00
assertRange(6, segments[1], DB::FileSegment::Range(10, 14), DB::FileSegment::State::DOWNLOADED);
}
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
/// Current cache: [__________][_____]
/// ^ ^^ ^
/// 0 910 14
ASSERT_EQ(cache.getFileSegmentsNum(), 2);
ASSERT_EQ(cache.getUsedCacheSize(), 15);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
{
2022-09-06 11:30:02 +00:00
auto holder = cache.getOrSet(key, 9, 1, {}); /// Get [9, 9]
2022-08-31 16:07:40 +00:00
auto segments = fromHolder(holder);
ASSERT_EQ(segments.size(), 1);
assertRange(7, segments[0], DB::FileSegment::Range(0, 9), DB::FileSegment::State::DOWNLOADED);
}
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
{
2022-09-06 11:30:02 +00:00
auto holder = cache.getOrSet(key, 9, 2, {}); /// Get [9, 10]
2022-08-31 16:07:40 +00:00
auto segments = fromHolder(holder);
ASSERT_EQ(segments.size(), 2);
assertRange(8, segments[0], DB::FileSegment::Range(0, 9), DB::FileSegment::State::DOWNLOADED);
assertRange(9, segments[1], DB::FileSegment::Range(10, 14), DB::FileSegment::State::DOWNLOADED);
}
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
{
2022-09-06 11:30:02 +00:00
auto holder = cache.getOrSet(key, 10, 1, {}); /// Get [10, 10]
2022-08-31 16:07:40 +00:00
auto segments = fromHolder(holder);
ASSERT_EQ(segments.size(), 1);
assertRange(10, segments[0], DB::FileSegment::Range(10, 14), DB::FileSegment::State::DOWNLOADED);
}
2022-01-13 11:57:56 +00:00
complete(cache.getOrSet(key, 17, 4, {})); /// Get [17, 20]
complete(cache.getOrSet(key, 24, 3, {})); /// Get [24, 26]
2022-09-06 11:30:02 +00:00
/// completeWithState(cache.getOrSet(key, 27, 1, false)); /// Get [27, 27]
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
/// Current cache: [__________][_____] [____] [___][]
/// ^ ^^ ^ ^ ^ ^ ^^^
/// 0 910 14 17 20 24 2627
///
ASSERT_EQ(cache.getFileSegmentsNum(), 4);
ASSERT_EQ(cache.getUsedCacheSize(), 22);
2022-08-31 16:07:40 +00:00
{
2022-09-06 11:30:02 +00:00
auto holder = cache.getOrSet(key, 0, 26, {}); /// Get [0, 25]
2022-08-31 16:07:40 +00:00
auto segments = fromHolder(holder);
ASSERT_EQ(segments.size(), 6);
2022-08-31 16:07:40 +00:00
assertRange(11, segments[0], DB::FileSegment::Range(0, 9), DB::FileSegment::State::DOWNLOADED);
assertRange(12, segments[1], DB::FileSegment::Range(10, 14), DB::FileSegment::State::DOWNLOADED);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
/// Missing [15, 16] should be added in cache.
assertRange(13, segments[2], DB::FileSegment::Range(15, 16), DB::FileSegment::State::EMPTY);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
ASSERT_TRUE(segments[2]->getOrSetDownloader() == DB::FileSegment::getCallerId());
prepareAndDownload(segments[2]);
2022-01-13 11:57:56 +00:00
segments[2]->completeWithoutState();
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
assertRange(14, segments[3], DB::FileSegment::Range(17, 20), DB::FileSegment::State::DOWNLOADED);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
/// New [21, 23], but will not be added in cache because of elements limit (5)
assertRange(15, segments[4], DB::FileSegment::Range(21, 23), DB::FileSegment::State::EMPTY);
ASSERT_TRUE(segments[4]->getOrSetDownloader() == DB::FileSegment::getCallerId());
ASSERT_FALSE(segments[4]->reserve(1));
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
assertRange(16, segments[5], DB::FileSegment::Range(24, 26), DB::FileSegment::State::DOWNLOADED);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
/// Current cache: [__________][_____][ ][____] [___]
/// ^ ^ ^
/// 0 20 24
///
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
/// Range [27, 27] must be evicted in previous getOrSet [0, 25].
/// Let's not invalidate pointers to returned segments from range [0, 25] and
/// as max elements size is reached, next attempt to put something in cache should fail.
/// This will also check that [27, 27] was indeed evicted.
2022-01-13 11:57:56 +00:00
2022-09-06 11:30:02 +00:00
auto holder1 = cache.getOrSet(key, 27, 1, {});
2022-08-31 16:07:40 +00:00
auto segments_1 = fromHolder(holder1); /// Get [27, 27]
ASSERT_EQ(segments_1.size(), 1);
assertRange(17, segments_1[0], DB::FileSegment::Range(27, 27), DB::FileSegment::State::EMPTY);
}
2022-08-31 16:07:40 +00:00
{
2022-09-06 11:30:02 +00:00
auto holder = cache.getOrSet(key, 12, 10, {}); /// Get [12, 21]
2022-08-31 16:07:40 +00:00
auto segments = fromHolder(holder);
ASSERT_EQ(segments.size(), 4);
2022-08-31 16:07:40 +00:00
assertRange(18, segments[0], DB::FileSegment::Range(10, 14), DB::FileSegment::State::DOWNLOADED);
assertRange(19, segments[1], DB::FileSegment::Range(15, 16), DB::FileSegment::State::DOWNLOADED);
assertRange(20, segments[2], DB::FileSegment::Range(17, 20), DB::FileSegment::State::DOWNLOADED);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
assertRange(21, segments[3], DB::FileSegment::Range(21, 21), DB::FileSegment::State::EMPTY);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
ASSERT_TRUE(segments[3]->getOrSetDownloader() == DB::FileSegment::getCallerId());
prepareAndDownload(segments[3]);
2022-01-13 11:57:56 +00:00
segments[3]->completeWithoutState();
2022-08-31 16:07:40 +00:00
ASSERT_TRUE(segments[3]->state() == DB::FileSegment::State::DOWNLOADED);
}
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
/// Current cache: [_____][__][____][_] [___]
/// ^ ^ ^ ^ ^
/// 10 17 21 24 26
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
ASSERT_EQ(cache.getFileSegmentsNum(), 5);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
{
2022-09-06 11:30:02 +00:00
auto holder = cache.getOrSet(key, 23, 5, {}); /// Get [23, 28]
2022-08-31 16:07:40 +00:00
auto segments = fromHolder(holder);
ASSERT_EQ(segments.size(), 3);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
assertRange(22, segments[0], DB::FileSegment::Range(23, 23), DB::FileSegment::State::EMPTY);
assertRange(23, segments[1], DB::FileSegment::Range(24, 26), DB::FileSegment::State::DOWNLOADED);
assertRange(24, segments[2], DB::FileSegment::Range(27, 27), DB::FileSegment::State::EMPTY);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
ASSERT_TRUE(segments[0]->getOrSetDownloader() == DB::FileSegment::getCallerId());
ASSERT_TRUE(segments[2]->getOrSetDownloader() == DB::FileSegment::getCallerId());
prepareAndDownload(segments[0]);
prepareAndDownload(segments[2]);
segments[0]->completeWithoutState();
segments[2]->completeWithoutState();
2022-08-31 16:07:40 +00:00
}
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
/// Current cache: [____][_] [][___][__]
/// ^ ^ ^^^ ^^ ^
/// 17 21 2324 26 28
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
{
2022-09-06 11:30:02 +00:00
auto holder5 = cache.getOrSet(key, 2, 3, {}); /// Get [2, 4]
2022-08-31 16:07:40 +00:00
auto s5 = fromHolder(holder5);
ASSERT_EQ(s5.size(), 1);
assertRange(25, s5[0], DB::FileSegment::Range(2, 4), DB::FileSegment::State::EMPTY);
2022-09-06 11:30:02 +00:00
auto holder1 = cache.getOrSet(key, 30, 2, {}); /// Get [30, 31]
2022-08-31 16:07:40 +00:00
auto s1 = fromHolder(holder1);
ASSERT_EQ(s1.size(), 1);
assertRange(26, s1[0], DB::FileSegment::Range(30, 31), DB::FileSegment::State::EMPTY);
ASSERT_TRUE(s5[0]->getOrSetDownloader() == DB::FileSegment::getCallerId());
ASSERT_TRUE(s1[0]->getOrSetDownloader() == DB::FileSegment::getCallerId());
prepareAndDownload(s5[0]);
prepareAndDownload(s1[0]);
s5[0]->completeWithoutState();
s1[0]->completeWithoutState();
2022-08-31 16:07:40 +00:00
/// Current cache: [___] [_][___][_] [__]
/// ^ ^ ^ ^ ^ ^ ^ ^
/// 2 4 23 24 26 27 30 31
2022-09-06 11:30:02 +00:00
auto holder2 = cache.getOrSet(key, 23, 1, {}); /// Get [23, 23]
2022-08-31 16:07:40 +00:00
auto s2 = fromHolder(holder2);
ASSERT_EQ(s2.size(), 1);
2022-09-06 11:30:02 +00:00
auto holder3 = cache.getOrSet(key, 24, 3, {}); /// Get [24, 26]
2022-08-31 16:07:40 +00:00
auto s3 = fromHolder(holder3);
ASSERT_EQ(s3.size(), 1);
2022-09-06 11:30:02 +00:00
auto holder4 = cache.getOrSet(key, 27, 1, {}); /// Get [27, 27]
2022-08-31 16:07:40 +00:00
auto s4 = fromHolder(holder4);
ASSERT_EQ(s4.size(), 1);
/// All cache is now unreleasable because pointers are still hold
2022-09-06 11:30:02 +00:00
auto holder6 = cache.getOrSet(key, 0, 40, {});
2022-08-31 16:07:40 +00:00
auto f = fromHolder(holder6);
ASSERT_EQ(f.size(), 9);
assertRange(27, f[0], DB::FileSegment::Range(0, 1), DB::FileSegment::State::EMPTY);
assertRange(28, f[2], DB::FileSegment::Range(5, 22), DB::FileSegment::State::EMPTY);
assertRange(29, f[6], DB::FileSegment::Range(28, 29), DB::FileSegment::State::EMPTY);
assertRange(30, f[8], DB::FileSegment::Range(32, 39), DB::FileSegment::State::EMPTY);
ASSERT_TRUE(f[0]->getOrSetDownloader() == DB::FileSegment::getCallerId());
ASSERT_TRUE(f[2]->getOrSetDownloader() == DB::FileSegment::getCallerId());
ASSERT_TRUE(f[6]->getOrSetDownloader() == DB::FileSegment::getCallerId());
ASSERT_TRUE(f[8]->getOrSetDownloader() == DB::FileSegment::getCallerId());
ASSERT_FALSE(f[0]->reserve(1));
ASSERT_FALSE(f[2]->reserve(1));
ASSERT_FALSE(f[6]->reserve(1));
ASSERT_FALSE(f[8]->reserve(1));
}
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
{
2022-09-06 11:30:02 +00:00
auto holder = cache.getOrSet(key, 2, 3, {}); /// Get [2, 4]
2022-08-31 16:07:40 +00:00
auto segments = fromHolder(holder);
ASSERT_EQ(segments.size(), 1);
assertRange(31, segments[0], DB::FileSegment::Range(2, 4), DB::FileSegment::State::DOWNLOADED);
}
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
/// Current cache: [___] [_][___][_] [__]
/// ^ ^ ^ ^ ^ ^ ^ ^
/// 2 4 23 24 26 27 30 31
2022-01-13 11:57:56 +00:00
{
2022-09-06 11:30:02 +00:00
auto holder = cache.getOrSet(key, 25, 5, {}); /// Get [25, 29]
2022-08-31 16:07:40 +00:00
auto segments = fromHolder(holder);
2022-01-13 11:57:56 +00:00
ASSERT_EQ(segments.size(), 3);
2022-08-31 16:07:40 +00:00
assertRange(32, segments[0], DB::FileSegment::Range(24, 26), DB::FileSegment::State::DOWNLOADED);
assertRange(33, segments[1], DB::FileSegment::Range(27, 27), DB::FileSegment::State::DOWNLOADED);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
assertRange(34, segments[2], DB::FileSegment::Range(28, 29), DB::FileSegment::State::EMPTY);
ASSERT_TRUE(segments[2]->getOrSetDownloader() == DB::FileSegment::getCallerId());
2022-01-13 11:57:56 +00:00
ASSERT_TRUE(segments[2]->state() == DB::FileSegment::State::DOWNLOADING);
2022-08-31 16:07:40 +00:00
bool lets_start_download = false;
std::mutex mutex;
std::condition_variable cv;
std::thread other_1([&]
2022-01-13 11:57:56 +00:00
{
2022-08-31 16:07:40 +00:00
DB::ThreadStatus thread_status_1;
auto query_context_1 = DB::Context::createCopy(getContext().context);
query_context_1->makeQueryContext();
query_context_1->setCurrentQueryId("query_id_1");
DB::CurrentThread::QueryScope query_scope_holder_1(query_context_1);
thread_status_1.attachQueryContext(query_context_1);
2022-09-06 11:30:02 +00:00
auto holder_2 = cache.getOrSet(key, 25, 5, {}); /// Get [25, 29] once again.
2022-08-31 16:07:40 +00:00
auto segments_2 = fromHolder(holder_2);
ASSERT_EQ(segments.size(), 3);
assertRange(35, segments_2[0], DB::FileSegment::Range(24, 26), DB::FileSegment::State::DOWNLOADED);
assertRange(36, segments_2[1], DB::FileSegment::Range(27, 27), DB::FileSegment::State::DOWNLOADED);
assertRange(37, segments_2[2], DB::FileSegment::Range(28, 29), DB::FileSegment::State::DOWNLOADING);
ASSERT_TRUE(segments[2]->getOrSetDownloader() != DB::FileSegment::getCallerId());
ASSERT_TRUE(segments[2]->state() == DB::FileSegment::State::DOWNLOADING);
{
std::lock_guard lock(mutex);
lets_start_download = true;
}
cv.notify_one();
segments_2[2]->wait();
ASSERT_TRUE(segments_2[2]->state() == DB::FileSegment::State::DOWNLOADED);
});
{
std::unique_lock lock(mutex);
cv.wait(lock, [&]{ return lets_start_download; });
2022-01-13 11:57:56 +00:00
}
prepareAndDownload(segments[2]);
segments[2]->completeWithoutState();
2022-08-31 16:07:40 +00:00
ASSERT_TRUE(segments[2]->state() == DB::FileSegment::State::DOWNLOADED);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
other_1.join();
2022-01-13 11:57:56 +00:00
}
2022-08-31 16:07:40 +00:00
/// Current cache: [___] [___][_][__][__]
/// ^ ^ ^ ^ ^^ ^^ ^
/// 2 4 24 26 27 2930 31
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
{
/// Now let's check the similar case but getting ERROR state after segment->wait(), when
2022-09-06 11:30:02 +00:00
/// state is changed not manually via segment->completeWithState(state) but from destructor of holder
2022-08-31 16:07:40 +00:00
/// and notify_all() is also called from destructor of holder.
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
std::optional<DB::FileSegmentsHolder> holder;
2022-09-06 11:30:02 +00:00
holder.emplace(cache.getOrSet(key, 3, 23, {})); /// Get [3, 25]
2022-01-21 15:39:34 +00:00
2022-08-31 16:07:40 +00:00
auto segments = fromHolder(*holder);
ASSERT_EQ(segments.size(), 3);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
assertRange(38, segments[0], DB::FileSegment::Range(2, 4), DB::FileSegment::State::DOWNLOADED);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
assertRange(39, segments[1], DB::FileSegment::Range(5, 23), DB::FileSegment::State::EMPTY);
ASSERT_TRUE(segments[1]->getOrSetDownloader() == DB::FileSegment::getCallerId());
ASSERT_TRUE(segments[1]->state() == DB::FileSegment::State::DOWNLOADING);
2022-01-21 15:39:34 +00:00
2022-08-31 16:07:40 +00:00
assertRange(40, segments[2], DB::FileSegment::Range(24, 26), DB::FileSegment::State::DOWNLOADED);
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
bool lets_start_download = false;
std::mutex mutex;
std::condition_variable cv;
2022-01-13 11:57:56 +00:00
2022-08-31 16:07:40 +00:00
std::thread other_1([&]
{
DB::ThreadStatus thread_status_1;
auto query_context_1 = DB::Context::createCopy(getContext().context);
query_context_1->makeQueryContext();
query_context_1->setCurrentQueryId("query_id_1");
DB::CurrentThread::QueryScope query_scope_holder_1(query_context_1);
thread_status_1.attachQueryContext(query_context_1);
2022-09-06 11:30:02 +00:00
auto holder_2 = cache.getOrSet(key, 3, 23, {}); /// Get [3, 25] once again
2022-08-31 16:07:40 +00:00
auto segments_2 = fromHolder(*holder);
ASSERT_EQ(segments_2.size(), 3);
assertRange(41, segments_2[0], DB::FileSegment::Range(2, 4), DB::FileSegment::State::DOWNLOADED);
assertRange(42, segments_2[1], DB::FileSegment::Range(5, 23), DB::FileSegment::State::DOWNLOADING);
assertRange(43, segments_2[2], DB::FileSegment::Range(24, 26), DB::FileSegment::State::DOWNLOADED);
ASSERT_TRUE(segments_2[1]->getDownloader() != DB::FileSegment::getCallerId());
ASSERT_TRUE(segments_2[1]->state() == DB::FileSegment::State::DOWNLOADING);
{
std::lock_guard lock(mutex);
lets_start_download = true;
}
cv.notify_one();
segments_2[1]->wait();
printRanges(segments_2);
ASSERT_TRUE(segments_2[1]->state() == DB::FileSegment::State::PARTIALLY_DOWNLOADED);
ASSERT_TRUE(segments_2[1]->getOrSetDownloader() == DB::FileSegment::getCallerId());
prepareAndDownload(segments_2[1]);
segments_2[1]->completeWithoutState();
2022-08-31 16:07:40 +00:00
});
2022-01-13 11:57:56 +00:00
{
2022-08-31 16:07:40 +00:00
std::unique_lock lock(mutex);
cv.wait(lock, [&]{ return lets_start_download; });
2022-01-13 11:57:56 +00:00
}
2022-08-31 16:07:40 +00:00
holder.reset();
other_1.join();
printRanges(segments);
ASSERT_TRUE(segments[1]->state() == DB::FileSegment::State::DOWNLOADED);
2022-01-13 11:57:56 +00:00
}
}
2022-01-21 15:39:34 +00:00
/// Current cache: [___][ ][___][_][__]
/// ^ ^^ ^ ^^ ^ ^
/// 2 45 24 2627 28 29
2022-01-13 11:57:56 +00:00
{
/// Test LRUCache::restore().
auto cache2 = DB::FileCache(cache_base_path, settings);
2022-03-02 13:17:19 +00:00
cache2.initialize();
2022-08-31 16:07:40 +00:00
auto key = cache2.hash("key1");
2022-03-09 09:36:52 +00:00
2022-09-06 11:30:02 +00:00
auto holder1 = cache2.getOrSet(key, 2, 28, {}); /// Get [2, 29]
2022-05-13 16:51:07 +00:00
2022-01-13 11:57:56 +00:00
auto segments1 = fromHolder(holder1);
2022-01-21 15:39:34 +00:00
ASSERT_EQ(segments1.size(), 5);
2022-01-13 11:57:56 +00:00
2022-01-21 15:39:34 +00:00
assertRange(44, segments1[0], DB::FileSegment::Range(2, 4), DB::FileSegment::State::DOWNLOADED);
assertRange(45, segments1[1], DB::FileSegment::Range(5, 23), DB::FileSegment::State::DOWNLOADED);
assertRange(45, segments1[2], DB::FileSegment::Range(24, 26), DB::FileSegment::State::DOWNLOADED);
assertRange(46, segments1[3], DB::FileSegment::Range(27, 27), DB::FileSegment::State::DOWNLOADED);
assertRange(47, segments1[4], DB::FileSegment::Range(28, 29), DB::FileSegment::State::DOWNLOADED);
2022-01-13 11:57:56 +00:00
}
{
/// Test max file segment size
2022-03-25 17:31:15 +00:00
auto settings2 = settings;
2022-03-29 17:57:55 +00:00
settings2.max_file_segment_size = 10;
auto cache2 = DB::FileCache(caches_dir / "cache2", settings2);
2022-03-09 09:36:52 +00:00
cache2.initialize();
2022-08-31 16:07:40 +00:00
auto key = cache2.hash("key1");
2022-03-09 09:36:52 +00:00
2022-09-06 11:30:02 +00:00
auto holder1 = cache2.getOrSet(key, 0, 25, {}); /// Get [0, 24]
auto segments1 = fromHolder(holder1);
ASSERT_EQ(segments1.size(), 3);
assertRange(48, segments1[0], DB::FileSegment::Range(0, 9), DB::FileSegment::State::EMPTY);
assertRange(49, segments1[1], DB::FileSegment::Range(10, 19), DB::FileSegment::State::EMPTY);
assertRange(50, segments1[2], DB::FileSegment::Range(20, 24), DB::FileSegment::State::EMPTY);
}
2022-06-05 16:00:59 +00:00
2022-01-13 11:57:56 +00:00
}