zjw1111 commented on code in PR #272: URL: https://github.com/apache/paimon-cpp/pull/272#discussion_r3911487889
########## include/paimon/utils/prefetch_cache_config.h: ########## Review Comment: Where is this parameterized constructor used? ########## src/paimon/common/utils/file_block_cache_test.cpp: ########## @@ -0,0 +1,286 @@ +/* + * 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 "paimon/common/utils/file_block_cache.h" + +#include <chrono> +#include <fstream> +#include <memory> +#include <string> +#include <string_view> +#include <thread> +#include <utility> + +#include "gtest/gtest.h" +#include "paimon/common/factories/io_hook.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/fs/file_system.h" +#include "paimon/fs/file_system_factory.h" +#include "paimon/testing/utils/gated_async_input_stream.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { + +namespace { + +constexpr char kContent[] = "abcdefghijklmnopqrstuvwxyz"; +// 26 bytes in blocks of 8: block 0 = [18, 26), block 1 = [10, 18), +// block 2 = [2, 10) and block 3 = [0, 2), truncated at the start of the file. +constexpr uint64_t kBlockSize = 8; + +// Write the test content into a fresh directory and open it for reading. The +// directory is returned so that it outlives the stream. +std::shared_ptr<InputStream> OpenTestFile(std::unique_ptr<UniqueTestDirectory>* dir) { + *dir = UniqueTestDirectory::Create(); + EXPECT_TRUE(*dir); + std::string path = (*dir)->Str() + "/data_file"; + std::ofstream file(path, std::ios::binary); + EXPECT_TRUE(file.is_open()); + file.write(kContent, sizeof(kContent) - 1); + EXPECT_FALSE(file.fail()); + file.close(); + + Result<std::unique_ptr<FileSystem>> fs = FileSystemFactory::Get("local", path, {}); + EXPECT_TRUE(fs.ok()); + Result<std::unique_ptr<InputStream>> in = fs.value()->Open(path); + EXPECT_TRUE(in.ok()); + return std::move(in).value(); Review Comment: code style ########## include/paimon/utils/prefetch_cache_config.h: ########## Review Comment: Consider moving the implementation of `CacheConfig` into a separate `.cpp` file or into the `.h` file. It is currently placed in `read_ahead_cache.cpp`, which looks a bit odd. ########## src/paimon/testing/utils/gated_async_input_stream.h: ########## @@ -0,0 +1,114 @@ +/* + * 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. + */ + +#pragma once + +#include <functional> +#include <memory> +#include <mutex> +#include <string> +#include <utility> +#include <vector> + +#include "paimon/fs/file_system.h" + +namespace paimon::test { + +/// An InputStream wrapper that holds the ReadAsync callbacks until ReleaseAll() +/// is called, letting tests observe a cache while its fetches are in flight. +class GatedAsyncInputStream : public InputStream { + public: + explicit GatedAsyncInputStream(std::shared_ptr<InputStream> inner) : inner_(std::move(inner)) {} + + Status Close() override { + return inner_->Close(); + } + Status Seek(int64_t offset, SeekOrigin origin) override { + return inner_->Seek(offset, origin); + } + Result<int64_t> GetPos() const override { + return inner_->GetPos(); + } + Result<int64_t> Read(char* buffer, int64_t size) override { + return inner_->Read(buffer, size); + } + Result<int64_t> Read(char* buffer, int64_t size, int64_t offset) override { + return inner_->Read(buffer, size, offset); + } + void ReadAsync(char* buffer, int64_t size, int64_t offset, + std::function<void(Status)>&& callback) override { + std::lock_guard<std::mutex> lock(mutex_); + async_read_count_++; + pending_.push_back({buffer, size, offset, std::move(callback)}); + } + Result<std::string> GetUri() const override { + return inner_->GetUri(); + } + Result<int64_t> Length() const override { + return inner_->Length(); + } + + int AsyncReadCount() { + std::lock_guard<std::mutex> lock(mutex_); + return async_read_count_; + } + + /// Complete all held fetches against the underlying stream. + void ReleaseAll() { + std::vector<PendingRead> taken; + { + std::lock_guard<std::mutex> lock(mutex_); + taken = std::move(pending_); + pending_.clear(); + } + for (auto& read : taken) { + Result<int64_t> res = inner_->Read(read.buffer, read.size, read.offset); + read.callback(res.ok() ? Status::OK() : res.status()); + } + } + + /// Fail all held fetches without touching the underlying stream, so that a + /// test can observe how a cache reports a failed fetch. + void FailAll(const Status& status) { + std::vector<PendingRead> taken; + { + std::lock_guard<std::mutex> lock(mutex_); + taken = std::move(pending_); + pending_.clear(); + } + for (auto& read : taken) { + read.callback(status); + } + } + + private: + struct PendingRead { + char* buffer; + int64_t size; + int64_t offset; + std::function<void(Status)> callback; + }; + + std::shared_ptr<InputStream> inner_; + std::mutex mutex_; + std::vector<PendingRead> pending_; + int async_read_count_ = 0; Review Comment: code style ########## src/paimon/common/utils/read_ahead_cache.cpp: ########## @@ -147,6 +152,9 @@ class ReadAheadCache::Impl { std::vector<std::atomic<bool>> is_cached_; std::vector<ByteRange> pending_ranges_; bool is_initialized_ = false; + // Caches the reads that no registered range covers, or null when the block + // cache is disabled. Owns its own locking and counters. + std::unique_ptr<FileBlockCache> block_cache_; // Statistics of the Read() requests issued to the cache, aggregated over // all streams sharing this cache. std::atomic<uint64_t> read_count_{0}; Review Comment: TODO: metrics refactor to a struct ########## include/paimon/utils/prefetch_cache_config.h: ########## Review Comment: maybe can remove it? ########## src/paimon/common/utils/file_block_cache.h: ########## @@ -0,0 +1,145 @@ +/* + * 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. + */ + +#pragma once + +#include <atomic> +#include <cstdint> +#include <future> +#include <memory> +#include <mutex> +#include <unordered_map> + +#include "paimon/common/utils/read_ahead_cache.h" +#include "paimon/fs/file_system.h" +#include "paimon/memory/bytes.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/status.h" +#include "paimon/visibility.h" + +namespace paimon { + +/// A cache of fixed-size blocks of one file, serving the reads that no +/// prefetched range covers: a parquet reader reads the footer and the page index +/// before any range can be registered, and every reader of the file reads the +/// same bytes. +/// +/// Blocks are aligned to the END of the file: block 0 is +/// [file_size - block_size, file_size). The metadata of a parquet/orc file lives +/// in its tail and arrow reads exactly the last 64 KiB as the footer, so an +/// end-aligned block matches that read instead of straddling two blocks. It also +/// keeps every block inside the file, as long as the given file size is the size +/// of the file: a block fetch failing because the file is shorter than that only +/// costs the caching of that block, see Read(). +/// +/// A block is published before its fetch is dispatched, so concurrent readers of +/// the same block wait for that one fetch instead of issuing their own. +/// +/// Blocks are never evicted: once the capacity is reached Read() declines +/// instead of replacing a block. That keeps every dispatched fetch reachable +/// through its block, so Release() and the destructor can wait for the fetches +/// still writing into the block buffers. +class PAIMON_EXPORT FileBlockCache { + public: + /// Requests served by a block and fetches issued for the blocks themselves, + /// reported by the owner of this cache through its own metrics. + struct Counters { + uint64_t hits = 0; + uint64_t hit_bytes = 0; + uint64_t fetches = 0; + uint64_t fetch_bytes = 0; + }; + + /// @param stream The stream the blocks are fetched from. + /// @param file_size Size of the file behind `stream`, which the blocks are + /// aligned to the end of. Must not be zero. + /// @param block_size Granularity of the blocks. Must not be zero. + /// @param capacity Maximum total size of the cached blocks, in bytes. + /// @param memory_pool The pool the block buffers are allocated from. + FileBlockCache(const std::shared_ptr<InputStream>& stream, uint64_t file_size, + uint64_t block_size, uint64_t capacity, + const std::shared_ptr<MemoryPool>& memory_pool); + ~FileBlockCache(); + + /// Serve the given range out of its block, fetching that block first if it + /// is not cached yet. + /// @param range The byte range to read. + /// @param dest Destination buffer with at least `range.length` bytes. + /// @return true if the range was served and `dest` was filled; false when + /// one block cannot serve the range (it straddles two blocks, is larger than + /// a block or reaches past EOF), when the capacity is exhausted or when the + /// fetch of the block failed, leaving `dest` untouched so the caller can read + /// the bytes itself. A fetch failure is never reported to the caller: the + /// block reads more than the caller asked for, so the caller reads its own + /// bytes instead and reports the failure itself if they cannot be read + /// either. A block whose fetch failed is not fetched again. + bool Read(const ByteRange& range, char* dest); + + /// Drop all cached blocks, waiting for the fetches still writing into their + /// buffers. The counters are kept readable for the owner's metrics. + void Release(); + + /// Zero the counters while keeping the cached blocks, which cache the file + /// rather than a round of reads. + void ResetCounters(); + + Counters GetCounters() const; + + private: + /// A cached block. Blocks are handed out as shared_ptr so that a reader + /// keeps its block alive once it has released the lock. + struct Block { + ByteRange range; + std::shared_ptr<Bytes> buffer; + std::shared_ptr<std::promise<Status>> promise; + // shared_future, as every reader of the block waits on it. + std::shared_future<Status> future; + }; + + /// Whether one block can serve the given range. + bool CanServe(const ByteRange& range) const; + /// Index of the block holding `offset`, counted from the END of the file, so + /// that block 0 is the last block. `offset` must be inside the file. + uint64_t IndexOf(uint64_t offset) const; + /// Range of the block with the given index, clamped at the start of the file. + ByteRange RangeOf(uint64_t index) const; + /// Fetch the block into its buffer and resolve its promise with the outcome. + /// Must be called after the block has been published, and only by the reader + /// that published it. + void Fetch(const std::shared_ptr<Block>& block); + + std::shared_ptr<InputStream> stream_; + uint64_t file_size_; + uint64_t block_size_; + uint64_t capacity_; + std::shared_ptr<MemoryPool> memory_pool_; + // Blocks are aligned, so keying them by index keeps them disjoint by + // construction and needs no ordering. A plain mutex is enough: only the + // reads that no prefetched range covers touch the map, and they are few. + mutable std::mutex mutex_; + std::unordered_map<uint64_t, std::shared_ptr<Block>> blocks_; + // Bytes held by blocks_, guarded by mutex_ and bounded by capacity_. + uint64_t cached_bytes_ = 0; + std::atomic<uint64_t> hits_{0}; + std::atomic<uint64_t> hit_bytes_{0}; + std::atomic<uint64_t> fetches_{0}; + std::atomic<uint64_t> fetch_bytes_{0}; Review Comment: A broader question: at the moment there are many layers of metrics across different levels of the read path — `ParquetMetrics`, `PrefetchMetrics`, `ReadAheadCache` metrics, `FileBlockCache` metrics. Have you looked into whether the meanings of these metrics overlap or conflict with each other, and whether the metric collection is accurate? ########## src/paimon/common/utils/file_block_cache.h: ########## @@ -0,0 +1,145 @@ +/* + * 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. + */ + +#pragma once + +#include <atomic> +#include <cstdint> +#include <future> +#include <memory> +#include <mutex> +#include <unordered_map> + +#include "paimon/common/utils/read_ahead_cache.h" +#include "paimon/fs/file_system.h" +#include "paimon/memory/bytes.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/status.h" +#include "paimon/visibility.h" + +namespace paimon { + +/// A cache of fixed-size blocks of one file, serving the reads that no +/// prefetched range covers: a parquet reader reads the footer and the page index +/// before any range can be registered, and every reader of the file reads the +/// same bytes. +/// +/// Blocks are aligned to the END of the file: block 0 is +/// [file_size - block_size, file_size). The metadata of a parquet/orc file lives +/// in its tail and arrow reads exactly the last 64 KiB as the footer, so an +/// end-aligned block matches that read instead of straddling two blocks. It also +/// keeps every block inside the file, as long as the given file size is the size +/// of the file: a block fetch failing because the file is shorter than that only +/// costs the caching of that block, see Read(). +/// +/// A block is published before its fetch is dispatched, so concurrent readers of +/// the same block wait for that one fetch instead of issuing their own. +/// +/// Blocks are never evicted: once the capacity is reached Read() declines +/// instead of replacing a block. That keeps every dispatched fetch reachable +/// through its block, so Release() and the destructor can wait for the fetches +/// still writing into the block buffers. +class PAIMON_EXPORT FileBlockCache { Review Comment: Please do a self-test and analysis of whether there are any lifecycle issues between `bytes` and `pool` inside `ReadAheadCache` and `FileBlockCache`, especially in asynchronous S3/OSS scenarios. Please do not use `GetDefaultPool()`, since it is a global static pool and may hide lifecycle problems even if they exist. ########## src/paimon/common/utils/read_ahead_cache_test.cpp: ########## @@ -62,7 +65,7 @@ TestCacheEnv CreateTestFileAndCache(const std::string& filename, const std::stri auto in = std::move(in_result).value(); Review Comment: code style ########## include/paimon/utils/prefetch_cache_config.h: ########## Review Comment: Add default value descriptions for these parameters. ########## src/paimon/testing/utils/gated_async_input_stream.h: ########## @@ -0,0 +1,114 @@ +/* + * 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. + */ + +#pragma once + +#include <functional> +#include <memory> +#include <mutex> +#include <string> +#include <utility> +#include <vector> + +#include "paimon/fs/file_system.h" + +namespace paimon::test { + +/// An InputStream wrapper that holds the ReadAsync callbacks until ReleaseAll() +/// is called, letting tests observe a cache while its fetches are in flight. +class GatedAsyncInputStream : public InputStream { + public: + explicit GatedAsyncInputStream(std::shared_ptr<InputStream> inner) : inner_(std::move(inner)) {} + + Status Close() override { + return inner_->Close(); + } + Status Seek(int64_t offset, SeekOrigin origin) override { + return inner_->Seek(offset, origin); + } + Result<int64_t> GetPos() const override { + return inner_->GetPos(); + } + Result<int64_t> Read(char* buffer, int64_t size) override { + return inner_->Read(buffer, size); + } + Result<int64_t> Read(char* buffer, int64_t size, int64_t offset) override { + return inner_->Read(buffer, size, offset); + } + void ReadAsync(char* buffer, int64_t size, int64_t offset, + std::function<void(Status)>&& callback) override { + std::lock_guard<std::mutex> lock(mutex_); + async_read_count_++; + pending_.push_back({buffer, size, offset, std::move(callback)}); + } + Result<std::string> GetUri() const override { + return inner_->GetUri(); + } + Result<int64_t> Length() const override { + return inner_->Length(); + } + + int AsyncReadCount() { Review Comment: code style ########## src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp: ########## @@ -236,7 +236,11 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl if (io_metrics) { input_stream = std::make_shared<MetricsInputStream>(input_stream, io_metrics); } - cache = std::make_shared<ReadAheadCache>(input_stream, cache_config, pool); + // The file size lets the cache align its blocks to the end of the file, + // where the metadata the readers read before any range is registered + // lives. A non-positive size means unknown and disables the block cache. + const uint64_t file_size = data_file_size > 0 ? static_cast<uint64_t>(data_file_size) : 0; Review Comment: Under what circumstances could a negative value appear here? If we're concerned about a negative `data_file_size`, we should validate it globally at the very beginning. ########## src/paimon/common/utils/file_block_cache_test.cpp: ########## @@ -0,0 +1,286 @@ +/* + * 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 "paimon/common/utils/file_block_cache.h" + +#include <chrono> +#include <fstream> +#include <memory> +#include <string> +#include <string_view> +#include <thread> +#include <utility> + +#include "gtest/gtest.h" +#include "paimon/common/factories/io_hook.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/fs/file_system.h" +#include "paimon/fs/file_system_factory.h" +#include "paimon/testing/utils/gated_async_input_stream.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { + +namespace { + +constexpr char kContent[] = "abcdefghijklmnopqrstuvwxyz"; +// 26 bytes in blocks of 8: block 0 = [18, 26), block 1 = [10, 18), +// block 2 = [2, 10) and block 3 = [0, 2), truncated at the start of the file. +constexpr uint64_t kBlockSize = 8; + +// Write the test content into a fresh directory and open it for reading. The +// directory is returned so that it outlives the stream. +std::shared_ptr<InputStream> OpenTestFile(std::unique_ptr<UniqueTestDirectory>* dir) { + *dir = UniqueTestDirectory::Create(); + EXPECT_TRUE(*dir); + std::string path = (*dir)->Str() + "/data_file"; + std::ofstream file(path, std::ios::binary); + EXPECT_TRUE(file.is_open()); + file.write(kContent, sizeof(kContent) - 1); + EXPECT_FALSE(file.fail()); + file.close(); + + Result<std::unique_ptr<FileSystem>> fs = FileSystemFactory::Get("local", path, {}); + EXPECT_TRUE(fs.ok()); + Result<std::unique_ptr<InputStream>> in = fs.value()->Open(path); + EXPECT_TRUE(in.ok()); + return std::move(in).value(); +} + +// Assert that the range is served out of a block with the expected content. +void AssertServed(const ByteRange& range, const std::string& expected, FileBlockCache* cache) { + std::string dest(range.length, 'X'); + ASSERT_TRUE(cache->Read(range, dest.data())) << expected; + EXPECT_EQ(expected, std::string_view(dest.data(), range.length)); Review Comment: code style ########## src/paimon/common/utils/read_ahead_cache_test.cpp: ########## Review Comment: where uses `path` and `pool` in TestCacheEnv? ########## src/paimon/common/utils/file_block_cache.h: ########## @@ -0,0 +1,145 @@ +/* + * 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. + */ + +#pragma once + +#include <atomic> +#include <cstdint> +#include <future> +#include <memory> +#include <mutex> +#include <unordered_map> + +#include "paimon/common/utils/read_ahead_cache.h" +#include "paimon/fs/file_system.h" +#include "paimon/memory/bytes.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/status.h" +#include "paimon/visibility.h" + +namespace paimon { + +/// A cache of fixed-size blocks of one file, serving the reads that no +/// prefetched range covers: a parquet reader reads the footer and the page index +/// before any range can be registered, and every reader of the file reads the +/// same bytes. +/// +/// Blocks are aligned to the END of the file: block 0 is +/// [file_size - block_size, file_size). The metadata of a parquet/orc file lives +/// in its tail and arrow reads exactly the last 64 KiB as the footer, so an +/// end-aligned block matches that read instead of straddling two blocks. It also +/// keeps every block inside the file, as long as the given file size is the size +/// of the file: a block fetch failing because the file is shorter than that only +/// costs the caching of that block, see Read(). +/// +/// A block is published before its fetch is dispatched, so concurrent readers of +/// the same block wait for that one fetch instead of issuing their own. +/// +/// Blocks are never evicted: once the capacity is reached Read() declines +/// instead of replacing a block. That keeps every dispatched fetch reachable +/// through its block, so Release() and the destructor can wait for the fetches +/// still writing into the block buffers. +class PAIMON_EXPORT FileBlockCache { + public: + /// Requests served by a block and fetches issued for the blocks themselves, + /// reported by the owner of this cache through its own metrics. + struct Counters { + uint64_t hits = 0; + uint64_t hit_bytes = 0; + uint64_t fetches = 0; + uint64_t fetch_bytes = 0; + }; + + /// @param stream The stream the blocks are fetched from. + /// @param file_size Size of the file behind `stream`, which the blocks are + /// aligned to the end of. Must not be zero. + /// @param block_size Granularity of the blocks. Must not be zero. + /// @param capacity Maximum total size of the cached blocks, in bytes. + /// @param memory_pool The pool the block buffers are allocated from. + FileBlockCache(const std::shared_ptr<InputStream>& stream, uint64_t file_size, + uint64_t block_size, uint64_t capacity, + const std::shared_ptr<MemoryPool>& memory_pool); + ~FileBlockCache(); + + /// Serve the given range out of its block, fetching that block first if it + /// is not cached yet. + /// @param range The byte range to read. + /// @param dest Destination buffer with at least `range.length` bytes. + /// @return true if the range was served and `dest` was filled; false when + /// one block cannot serve the range (it straddles two blocks, is larger than + /// a block or reaches past EOF), when the capacity is exhausted or when the + /// fetch of the block failed, leaving `dest` untouched so the caller can read + /// the bytes itself. A fetch failure is never reported to the caller: the + /// block reads more than the caller asked for, so the caller reads its own + /// bytes instead and reports the failure itself if they cannot be read + /// either. A block whose fetch failed is not fetched again. + bool Read(const ByteRange& range, char* dest); + + /// Drop all cached blocks, waiting for the fetches still writing into their + /// buffers. The counters are kept readable for the owner's metrics. + void Release(); + + /// Zero the counters while keeping the cached blocks, which cache the file + /// rather than a round of reads. + void ResetCounters(); + + Counters GetCounters() const; + + private: + /// A cached block. Blocks are handed out as shared_ptr so that a reader + /// keeps its block alive once it has released the lock. + struct Block { + ByteRange range; + std::shared_ptr<Bytes> buffer; + std::shared_ptr<std::promise<Status>> promise; + // shared_future, as every reader of the block waits on it. + std::shared_future<Status> future; + }; + + /// Whether one block can serve the given range. + bool CanServe(const ByteRange& range) const; + /// Index of the block holding `offset`, counted from the END of the file, so + /// that block 0 is the last block. `offset` must be inside the file. + uint64_t IndexOf(uint64_t offset) const; + /// Range of the block with the given index, clamped at the start of the file. + ByteRange RangeOf(uint64_t index) const; + /// Fetch the block into its buffer and resolve its promise with the outcome. + /// Must be called after the block has been published, and only by the reader + /// that published it. + void Fetch(const std::shared_ptr<Block>& block); + + std::shared_ptr<InputStream> stream_; + uint64_t file_size_; + uint64_t block_size_; + uint64_t capacity_; + std::shared_ptr<MemoryPool> memory_pool_; + // Blocks are aligned, so keying them by index keeps them disjoint by + // construction and needs no ordering. A plain mutex is enough: only the + // reads that no prefetched range covers touch the map, and they are few. + mutable std::mutex mutex_; + std::unordered_map<uint64_t, std::shared_ptr<Block>> blocks_; + // Bytes held by blocks_, guarded by mutex_ and bounded by capacity_. + uint64_t cached_bytes_ = 0; + std::atomic<uint64_t> hits_{0}; Review Comment: TODO: metrics refactor to a struct -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
