Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fix for background download in fs cache #55252

Merged
merged 4 commits into from Oct 8, 2023
Merged

Fix for background download in fs cache #55252

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
d13588e
Fix
kssenii Oct 5, 2023
c57a92f
Better
kssenii Oct 5, 2023
289020f
Merge remote-tracking branch 'upstream/master' into fix-background-do…
kssenii Oct 7, 2023
1a839ad
Add one more try-catch
kssenii Oct 7, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
3 changes: 2 additions & 1 deletion src/Interpreters/Cache/FileCache.cpp
View file
Open in desktop
Expand Up @@ -521,7 +521,7 @@ KeyMetadata::iterator FileCache::addFileSegment(
result_state = state;
}

auto file_segment = std::make_shared<FileSegment>(key, offset, size, result_state, settings, this, locked_key.getKeyMetadata());
auto file_segment = std::make_shared<FileSegment>(key, offset, size, result_state, settings, background_download_threads > 0, this, locked_key.getKeyMetadata());
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

0 usually means unlimited threads, but here it means we don't need to download a segment. Probably requires a comment somewhere.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added

auto file_segment_metadata = std::make_shared<FileSegmentMetadata>(std::move(file_segment));

auto [file_segment_metadata_it, inserted] = locked_key.getKeyMetadata()->emplace(offset, file_segment_metadata);
Expand Down Expand Up @@ -1018,6 +1018,7 @@ void FileCache::loadMetadataForKeys(const fs::path & keys_dir)
auto file_segment = std::make_shared<FileSegment>(key, offset, size,
FileSegment::State::DOWNLOADED,
CreateFileSegmentSettings(segment_kind),
false,
this,
key_metadata,
cache_it);
Expand Down
2 changes: 1 addition & 1 deletion src/Interpreters/Cache/FileCache.h
View file
Open in desktop
Expand Up @@ -159,7 +159,7 @@ class FileCache : private boost::noncopyable
const size_t max_file_segment_size;
const size_t bypass_cache_threshold = 0;
const size_t boundary_alignment;
const size_t background_download_threads;
const size_t background_download_threads; /// 0 means background download is disabled.
const size_t metadata_download_threads;

Poco::Logger * log;
Expand Down
10 changes: 9 additions & 1 deletion src/Interpreters/Cache/FileSegment.cpp
View file
Open in desktop
Expand Up @@ -44,13 +44,15 @@ FileSegment::FileSegment(
size_t size_,
State download_state_,
const CreateFileSegmentSettings & settings,
bool background_download_enabled_,
FileCache * cache_,
std::weak_ptr<KeyMetadata> key_metadata_,
Priority::Iterator queue_iterator_)
: file_key(key_)
, segment_range(offset_, offset_ + size_ - 1)
, segment_kind(settings.kind)
, is_unbound(settings.unbounded)
, background_download_enabled(background_download_enabled_)
, download_state(download_state_)
, key_metadata(key_metadata_)
, queue_iterator(queue_iterator_)
Expand Down Expand Up @@ -539,6 +541,12 @@ void FileSegment::setDownloadedUnlocked(const FileSegmentGuard::Lock &)
chassert(fs::file_size(getPathInLocalCache()) == downloaded_size);
}

void FileSegment::setDownloadFailed()
{
auto lock = lockFileSegment();
setDownloadFailedUnlocked(lock);
}

void FileSegment::setDownloadFailedUnlocked(const FileSegmentGuard::Lock & lock)
{
LOG_INFO(log, "Setting download as failed: {}", getInfoForLogUnlocked(lock));
Expand Down Expand Up @@ -652,7 +660,7 @@ void FileSegment::complete()

if (is_last_holder)
{
if (remote_file_reader)
if (background_download_enabled && remote_file_reader)
{
LOG_TEST(
log, "Submitting file segment for background download "
Expand Down
4 changes: 4 additions & 0 deletions src/Interpreters/Cache/FileSegment.h
View file
Open in desktop
Expand Up @@ -110,6 +110,7 @@ friend class FileCache; /// Because of reserved_size in tryReserve().
size_t size_,
State download_state_,
const CreateFileSegmentSettings & create_settings = {},
bool background_download_enabled_ = false,
FileCache * cache_ = nullptr,
std::weak_ptr<KeyMetadata> key_metadata_ = std::weak_ptr<KeyMetadata>(),
Priority::Iterator queue_iterator_ = Priority::Iterator{});
Expand Down Expand Up @@ -262,6 +263,8 @@ friend class FileCache; /// Because of reserved_size in tryReserve().

void setDownloadedSize(size_t delta);

void setDownloadFailed();

private:
String getDownloaderUnlocked(const FileSegmentGuard::Lock &) const;
bool isDownloaderUnlocked(const FileSegmentGuard::Lock & segment_lock) const;
Expand Down Expand Up @@ -290,6 +293,7 @@ friend class FileCache; /// Because of reserved_size in tryReserve().
/// Size of the segment is not known until it is downloaded and
/// can be bigger than max_file_segment_size.
const bool is_unbound = false;
const bool background_download_enabled;

std::atomic<State> download_state;
DownloaderId downloader_id; /// The one who prepares the download
Expand Down
83 changes: 46 additions & 37 deletions src/Interpreters/Cache/Metadata.cpp
View file
Open in desktop
Expand Up @@ -514,65 +514,74 @@ void CacheMetadata::downloadThreadFunc()

CurrentMetrics::sub(CurrentMetrics::FilesystemCacheDownloadQueueElements);

FileSegmentsHolderPtr holder;
try
{
FileSegmentsHolderPtr holder;
try
{
auto locked_key = lockKeyMetadata(key, KeyNotFoundPolicy::RETURN_NULL);
if (!locked_key)
continue;
{
auto locked_key = lockKeyMetadata(key, KeyNotFoundPolicy::RETURN_NULL);
if (!locked_key)
continue;

auto file_segment_metadata = locked_key->tryGetByOffset(offset);
if (!file_segment_metadata || file_segment_metadata->evicting())
continue;
auto file_segment_metadata = locked_key->tryGetByOffset(offset);
if (!file_segment_metadata || file_segment_metadata->evicting())
continue;

auto file_segment = file_segment_weak.lock();

if (!file_segment
|| file_segment != file_segment_metadata->file_segment
|| file_segment->state() != FileSegment::State::PARTIALLY_DOWNLOADED)
continue;

holder = std::make_unique<FileSegmentsHolder>(FileSegments{file_segment});
}

auto file_segment = file_segment_weak.lock();
auto & file_segment = holder->front();

if (!file_segment
|| file_segment != file_segment_metadata->file_segment
|| file_segment->state() != FileSegment::State::PARTIALLY_DOWNLOADED)
if (file_segment.getOrSetDownloader() != FileSegment::getCallerId())
continue;

holder = std::make_unique<FileSegmentsHolder>(FileSegments{file_segment});
}
chassert(file_segment.getDownloadedSize() != file_segment.range().size());
chassert(file_segment.assertCorrectness());

downloadImpl(holder->front(), memory);
}
catch (...)
{
if (holder)
{
const auto & file_segment = holder->front();
LOG_ERROR(
log, "Error during background download of {}:{} ({}): {}",
file_segment.key(), file_segment.offset(),
file_segment.getInfoForLog(), getCurrentExceptionMessage(true));
downloadImpl(file_segment, memory);
}
else
catch (...)
{
tryLogCurrentException(__PRETTY_FUNCTION__);
chassert(false);
if (holder)
{
auto & file_segment = holder->front();
file_segment.setDownloadFailed();

LOG_ERROR(
log, "Error during background download of {}:{} ({}): {}",
file_segment.key(), file_segment.offset(),
file_segment.getInfoForLog(), getCurrentExceptionMessage(true));
}
else
{
tryLogCurrentException(__PRETTY_FUNCTION__);
chassert(false);
}
}
}
catch (...)
{
tryLogCurrentException(__PRETTY_FUNCTION__);
chassert(false);
}
}
}

void CacheMetadata::downloadImpl(FileSegment & file_segment, std::optional<Memory<>> & memory)
{
chassert(file_segment.assertCorrectness());

if (file_segment.getOrSetDownloader() != FileSegment::getCallerId())
return;

if (file_segment.getDownloadedSize() == file_segment.range().size())
throw Exception(ErrorCodes::LOGICAL_ERROR, "File segment is already fully downloaded");

LOG_TEST(
log, "Downloading {} bytes for file segment {}",
file_segment.range().size() - file_segment.getDownloadedSize(), file_segment.getInfoForLog());

auto reader = file_segment.getRemoteFileReader();

if (!reader)
{
throw Exception(
Expand Down Expand Up @@ -802,7 +811,7 @@ void LockedKey::shrinkFileSegmentToDownloadedSize(

metadata->file_segment = std::make_shared<FileSegment>(
getKey(), offset, downloaded_size, FileSegment::State::DOWNLOADED,
CreateFileSegmentSettings(file_segment->getKind()),
CreateFileSegmentSettings(file_segment->getKind()), false,
file_segment->cache, key_metadata, file_segment->queue_iterator);

if (diff)
Expand Down