mirror of
https://github.com/facebook/rocksdb.git
synced 2026-07-07 14:47:40 +08:00
propagate request_id from app -> Rocks -> FS (#13616)
Summary: [internal use] Allow the application to pass a request_id per read request to RocksDB and pass it down to the FileSystem (via IODebugContext) Pull Request resolved: https://github.com/facebook/rocksdb/pull/13616 Test Plan: ./db_test --gtest_filter=DBTest.RequestIdPlumbingTest Validates that RocksDB Api calls with request_id set result in request_id being passed to the filesystem through IODebugContext Reviewed By: pdillinger Differential Revision: D74912824 Pulled By: virajthakur fbshipit-source-id: 4f15fef3ff7b5d700563f993f9b211c991020fb6
This commit is contained in:
committed by
Facebook GitHub Bot
parent
9a9a403a89
commit
acab405fc1
@@ -250,7 +250,8 @@ Status BlobFileReader::ReadFromFile(const RandomAccessFileReader* file_reader,
|
|||||||
Status s;
|
Status s;
|
||||||
|
|
||||||
IOOptions io_options;
|
IOOptions io_options;
|
||||||
s = file_reader->PrepareIOOptions(read_options, io_options);
|
IODebugContext dbg;
|
||||||
|
s = file_reader->PrepareIOOptions(read_options, io_options, &dbg);
|
||||||
if (!s.ok()) {
|
if (!s.ok()) {
|
||||||
return s;
|
return s;
|
||||||
}
|
}
|
||||||
@@ -259,13 +260,13 @@ Status BlobFileReader::ReadFromFile(const RandomAccessFileReader* file_reader,
|
|||||||
constexpr char* scratch = nullptr;
|
constexpr char* scratch = nullptr;
|
||||||
|
|
||||||
s = file_reader->Read(io_options, read_offset, read_size, slice, scratch,
|
s = file_reader->Read(io_options, read_offset, read_size, slice, scratch,
|
||||||
aligned_buf);
|
aligned_buf, &dbg);
|
||||||
} else {
|
} else {
|
||||||
buf->reset(new char[read_size]);
|
buf->reset(new char[read_size]);
|
||||||
constexpr AlignedBuf* aligned_scratch = nullptr;
|
constexpr AlignedBuf* aligned_scratch = nullptr;
|
||||||
|
|
||||||
s = file_reader->Read(io_options, read_offset, read_size, slice, buf->get(),
|
s = file_reader->Read(io_options, read_offset, read_size, slice, buf->get(),
|
||||||
aligned_scratch);
|
aligned_scratch, &dbg);
|
||||||
}
|
}
|
||||||
|
|
||||||
if (!s.ok()) {
|
if (!s.ok()) {
|
||||||
@@ -334,7 +335,8 @@ Status BlobFileReader::GetBlob(
|
|||||||
constexpr bool for_compaction = true;
|
constexpr bool for_compaction = true;
|
||||||
|
|
||||||
IOOptions io_options;
|
IOOptions io_options;
|
||||||
s = file_reader_->PrepareIOOptions(read_options, io_options);
|
IODebugContext dbg;
|
||||||
|
s = file_reader_->PrepareIOOptions(read_options, io_options, &dbg);
|
||||||
if (!s.ok()) {
|
if (!s.ok()) {
|
||||||
return s;
|
return s;
|
||||||
}
|
}
|
||||||
@@ -463,10 +465,11 @@ void BlobFileReader::MultiGetBlob(
|
|||||||
PERF_COUNTER_ADD(blob_read_count, num_blobs);
|
PERF_COUNTER_ADD(blob_read_count, num_blobs);
|
||||||
PERF_COUNTER_ADD(blob_read_byte, total_len);
|
PERF_COUNTER_ADD(blob_read_byte, total_len);
|
||||||
IOOptions opts;
|
IOOptions opts;
|
||||||
s = file_reader_->PrepareIOOptions(read_options, opts);
|
IODebugContext dbg;
|
||||||
|
s = file_reader_->PrepareIOOptions(read_options, opts, &dbg);
|
||||||
if (s.ok()) {
|
if (s.ok()) {
|
||||||
s = file_reader_->MultiRead(opts, read_reqs.data(), read_reqs.size(),
|
s = file_reader_->MultiRead(opts, read_reqs.data(), read_reqs.size(),
|
||||||
direct_io ? &aligned_buf : nullptr);
|
direct_io ? &aligned_buf : nullptr, &dbg);
|
||||||
}
|
}
|
||||||
if (!s.ok()) {
|
if (!s.ok()) {
|
||||||
for (auto& req : read_reqs) {
|
for (auto& req : read_reqs) {
|
||||||
|
|||||||
@@ -144,6 +144,103 @@ TEST_F(DBTest, MockEnvTest) {
|
|||||||
delete db;
|
delete db;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
TEST_F(DBTest, RequestIdPlumbingTest) {
|
||||||
|
// test that request_id is passed to the filesystem, from
|
||||||
|
// ReadOptions to IODebugContext
|
||||||
|
Options options = CurrentOptions();
|
||||||
|
options.env = env_;
|
||||||
|
|
||||||
|
// Create a mock environment to capture IODebugContext during reads
|
||||||
|
const std::string* captured_request_id_dbg;
|
||||||
|
|
||||||
|
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
|
||||||
|
"RandomAccessFileReader::Read:IODebugContext", [&](void* arg) {
|
||||||
|
IODebugContext* dbg = static_cast<IODebugContext*>(arg);
|
||||||
|
if (dbg == nullptr) {
|
||||||
|
captured_request_id_dbg = nullptr;
|
||||||
|
} else {
|
||||||
|
captured_request_id_dbg = dbg->request_id;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
|
||||||
|
|
||||||
|
ASSERT_OK(Put("k1", "v1"));
|
||||||
|
ASSERT_OK(Flush());
|
||||||
|
|
||||||
|
// test request_id plumbing during a get
|
||||||
|
{
|
||||||
|
const std::string test_request_id = "test_request_id_123";
|
||||||
|
ReadOptions read_opts;
|
||||||
|
read_opts.request_id = &test_request_id;
|
||||||
|
std::string value;
|
||||||
|
ASSERT_OK(db_->Get(read_opts, "k1", &value));
|
||||||
|
|
||||||
|
// Verify the request_id was propagated to the file system
|
||||||
|
ASSERT_NE(captured_request_id_dbg, nullptr);
|
||||||
|
ASSERT_EQ(*captured_request_id_dbg, test_request_id);
|
||||||
|
}
|
||||||
|
|
||||||
|
captured_request_id_dbg = nullptr;
|
||||||
|
|
||||||
|
// test request_id plumbing during iterator seek
|
||||||
|
ASSERT_OK(Put("k2", "v2"));
|
||||||
|
ASSERT_OK(Flush());
|
||||||
|
{
|
||||||
|
ReadOptions read_opts;
|
||||||
|
const std::string request_id = "test_request_id_456";
|
||||||
|
read_opts.request_id = &request_id;
|
||||||
|
|
||||||
|
std::unique_ptr<Iterator> iter(db_->NewIterator(read_opts));
|
||||||
|
iter->Seek("k2");
|
||||||
|
ASSERT_TRUE(iter->Valid());
|
||||||
|
|
||||||
|
// Verify the request_id was propagated to the file system
|
||||||
|
ASSERT_NE(captured_request_id_dbg, nullptr);
|
||||||
|
ASSERT_EQ(*captured_request_id_dbg, request_id);
|
||||||
|
}
|
||||||
|
|
||||||
|
// test request_id plumbing during multiget
|
||||||
|
captured_request_id_dbg = nullptr;
|
||||||
|
|
||||||
|
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
|
||||||
|
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
|
||||||
|
"RandomAccessFileReader::MultiRead:IODebugContext", [&](void* arg) {
|
||||||
|
IODebugContext* dbg = static_cast<IODebugContext*>(arg);
|
||||||
|
if (dbg == nullptr) {
|
||||||
|
captured_request_id_dbg = nullptr;
|
||||||
|
} else {
|
||||||
|
captured_request_id_dbg = dbg->request_id;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
|
||||||
|
|
||||||
|
ASSERT_OK(Put("k3", "v3"));
|
||||||
|
ASSERT_OK(Put("k4", "v4"));
|
||||||
|
ASSERT_OK(Flush());
|
||||||
|
|
||||||
|
{
|
||||||
|
ReadOptions read_opts;
|
||||||
|
const std::string multiget_request_id = "test_request_id_789";
|
||||||
|
read_opts.request_id = &multiget_request_id;
|
||||||
|
|
||||||
|
std::vector<std::string> values;
|
||||||
|
std::vector<Slice> keys = {Slice("k3"), Slice("k4")};
|
||||||
|
|
||||||
|
values.resize(keys.size());
|
||||||
|
|
||||||
|
std::vector<ColumnFamilyHandle*> cfhs(keys.size(),
|
||||||
|
db_->DefaultColumnFamily());
|
||||||
|
db_->MultiGet(read_opts, cfhs, keys, &values);
|
||||||
|
|
||||||
|
ASSERT_NE(captured_request_id_dbg, nullptr);
|
||||||
|
ASSERT_EQ(*captured_request_id_dbg, multiget_request_id);
|
||||||
|
}
|
||||||
|
|
||||||
|
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
|
||||||
|
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
|
||||||
|
}
|
||||||
|
|
||||||
TEST_F(DBTest, MemEnvTest) {
|
TEST_F(DBTest, MemEnvTest) {
|
||||||
std::unique_ptr<Env> env{NewMemEnv(Env::Default())};
|
std::unique_ptr<Env> env{NewMemEnv(Env::Default())};
|
||||||
Options options;
|
Options options;
|
||||||
|
|||||||
+4
-3
@@ -247,15 +247,16 @@ IOStatus GenerateOneFileChecksum(
|
|||||||
Slice slice;
|
Slice slice;
|
||||||
uint64_t offset = 0;
|
uint64_t offset = 0;
|
||||||
IOOptions opts;
|
IOOptions opts;
|
||||||
io_s = reader->PrepareIOOptions(read_options, opts);
|
IODebugContext dbg;
|
||||||
|
io_s = reader->PrepareIOOptions(read_options, opts, &dbg);
|
||||||
if (!io_s.ok()) {
|
if (!io_s.ok()) {
|
||||||
return io_s;
|
return io_s;
|
||||||
}
|
}
|
||||||
while (size > 0) {
|
while (size > 0) {
|
||||||
size_t bytes_to_read =
|
size_t bytes_to_read =
|
||||||
static_cast<size_t>(std::min(uint64_t{readahead_size}, size));
|
static_cast<size_t>(std::min(uint64_t{readahead_size}, size));
|
||||||
io_s =
|
io_s = reader->Read(opts, offset, bytes_to_read, &slice, buf.get(), nullptr,
|
||||||
reader->Read(opts, offset, bytes_to_read, &slice, buf.get(), nullptr);
|
&dbg);
|
||||||
if (!io_s.ok()) {
|
if (!io_s.ok()) {
|
||||||
return IOStatus::Corruption("file read failed with error: " +
|
return IOStatus::Corruption("file read failed with error: " +
|
||||||
io_s.ToString());
|
io_s.ToString());
|
||||||
|
|||||||
+8
-1
@@ -86,7 +86,14 @@ IOStatus GenerateOneFileChecksum(
|
|||||||
const ReadOptions& read_options, Statistics* stats, SystemClock* clock);
|
const ReadOptions& read_options, Statistics* stats, SystemClock* clock);
|
||||||
|
|
||||||
inline IOStatus PrepareIOFromReadOptions(const ReadOptions& ro,
|
inline IOStatus PrepareIOFromReadOptions(const ReadOptions& ro,
|
||||||
SystemClock* clock, IOOptions& opts) {
|
SystemClock* clock, IOOptions& opts,
|
||||||
|
IODebugContext* dbg = nullptr) {
|
||||||
|
if (ro.request_id != nullptr) {
|
||||||
|
if (dbg != nullptr && dbg->request_id == nullptr) {
|
||||||
|
dbg->SetRequestId(ro.request_id);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if (ro.deadline.count()) {
|
if (ro.deadline.count()) {
|
||||||
std::chrono::microseconds now =
|
std::chrono::microseconds now =
|
||||||
std::chrono::microseconds(clock->NowMicros());
|
std::chrono::microseconds(clock->NowMicros());
|
||||||
|
|||||||
@@ -106,11 +106,14 @@ IOStatus RandomAccessFileReader::Create(
|
|||||||
|
|
||||||
IOStatus RandomAccessFileReader::Read(const IOOptions& opts, uint64_t offset,
|
IOStatus RandomAccessFileReader::Read(const IOOptions& opts, uint64_t offset,
|
||||||
size_t n, Slice* result, char* scratch,
|
size_t n, Slice* result, char* scratch,
|
||||||
AlignedBuf* aligned_buf) const {
|
AlignedBuf* aligned_buf,
|
||||||
|
IODebugContext* dbg) const {
|
||||||
(void)aligned_buf;
|
(void)aligned_buf;
|
||||||
const Env::IOPriority rate_limiter_priority = opts.rate_limiter_priority;
|
const Env::IOPriority rate_limiter_priority = opts.rate_limiter_priority;
|
||||||
|
|
||||||
TEST_SYNC_POINT_CALLBACK("RandomAccessFileReader::Read", nullptr);
|
TEST_SYNC_POINT_CALLBACK("RandomAccessFileReader::Read", nullptr);
|
||||||
|
TEST_SYNC_POINT_CALLBACK("RandomAccessFileReader::Read:IODebugContext",
|
||||||
|
const_cast<void*>(static_cast<void*>(dbg)));
|
||||||
|
|
||||||
// To be paranoid: modify scratch a little bit, so in case underlying
|
// To be paranoid: modify scratch a little bit, so in case underlying
|
||||||
// FileSystem doesn't fill the buffer but return success and `scratch` returns
|
// FileSystem doesn't fill the buffer but return success and `scratch` returns
|
||||||
@@ -175,7 +178,7 @@ IOStatus RandomAccessFileReader::Read(const IOOptions& opts, uint64_t offset,
|
|||||||
// the opts.timeout before calling file_->Read
|
// the opts.timeout before calling file_->Read
|
||||||
assert(!opts.timeout.count() || allowed == read_size);
|
assert(!opts.timeout.count() || allowed == read_size);
|
||||||
io_s = file_->Read(aligned_offset + buf.CurrentSize(), allowed, opts,
|
io_s = file_->Read(aligned_offset + buf.CurrentSize(), allowed, opts,
|
||||||
&tmp, buf.Destination(), nullptr);
|
&tmp, buf.Destination(), dbg);
|
||||||
}
|
}
|
||||||
if (ShouldNotifyListeners()) {
|
if (ShouldNotifyListeners()) {
|
||||||
auto finish_ts = FileOperationInfo::FinishNow();
|
auto finish_ts = FileOperationInfo::FinishNow();
|
||||||
@@ -237,7 +240,7 @@ IOStatus RandomAccessFileReader::Read(const IOOptions& opts, uint64_t offset,
|
|||||||
// the opts.timeout before calling file_->Read
|
// the opts.timeout before calling file_->Read
|
||||||
assert(!opts.timeout.count() || allowed == n);
|
assert(!opts.timeout.count() || allowed == n);
|
||||||
io_s = file_->Read(offset + pos, allowed, opts, &tmp_result,
|
io_s = file_->Read(offset + pos, allowed, opts, &tmp_result,
|
||||||
scratch + pos, nullptr);
|
scratch + pos, dbg);
|
||||||
}
|
}
|
||||||
if (ShouldNotifyListeners()) {
|
if (ShouldNotifyListeners()) {
|
||||||
auto finish_ts = FileOperationInfo::FinishNow();
|
auto finish_ts = FileOperationInfo::FinishNow();
|
||||||
@@ -311,7 +314,8 @@ bool TryMerge(FSReadRequest* dest, const FSReadRequest& src) {
|
|||||||
IOStatus RandomAccessFileReader::MultiRead(const IOOptions& opts,
|
IOStatus RandomAccessFileReader::MultiRead(const IOOptions& opts,
|
||||||
FSReadRequest* read_reqs,
|
FSReadRequest* read_reqs,
|
||||||
size_t num_reqs,
|
size_t num_reqs,
|
||||||
AlignedBuf* aligned_buf) const {
|
AlignedBuf* aligned_buf,
|
||||||
|
IODebugContext* dbg) const {
|
||||||
(void)aligned_buf; // suppress warning of unused variable in LITE mode
|
(void)aligned_buf; // suppress warning of unused variable in LITE mode
|
||||||
assert(num_reqs > 0);
|
assert(num_reqs > 0);
|
||||||
|
|
||||||
@@ -420,8 +424,10 @@ IOStatus RandomAccessFileReader::MultiRead(const IOOptions& opts,
|
|||||||
remaining_bytes -= request_bytes;
|
remaining_bytes -= request_bytes;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
io_s = file_->MultiRead(fs_reqs, num_fs_reqs, opts,
|
TEST_SYNC_POINT_CALLBACK(
|
||||||
/*IODebugContext*=*/nullptr);
|
"RandomAccessFileReader::MultiRead:IODebugContext",
|
||||||
|
const_cast<void*>(static_cast<void*>(dbg)));
|
||||||
|
io_s = file_->MultiRead(fs_reqs, num_fs_reqs, opts, dbg);
|
||||||
RecordInHistogram(stats_, MULTIGET_IO_BATCH_SIZE, num_fs_reqs);
|
RecordInHistogram(stats_, MULTIGET_IO_BATCH_SIZE, num_fs_reqs);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -475,18 +481,21 @@ IOStatus RandomAccessFileReader::MultiRead(const IOOptions& opts,
|
|||||||
}
|
}
|
||||||
|
|
||||||
IOStatus RandomAccessFileReader::PrepareIOOptions(const ReadOptions& ro,
|
IOStatus RandomAccessFileReader::PrepareIOOptions(const ReadOptions& ro,
|
||||||
IOOptions& opts) const {
|
IOOptions& opts,
|
||||||
|
IODebugContext* dbg) const {
|
||||||
if (clock_ != nullptr) {
|
if (clock_ != nullptr) {
|
||||||
return PrepareIOFromReadOptions(ro, clock_, opts);
|
return PrepareIOFromReadOptions(ro, clock_, opts, dbg);
|
||||||
} else {
|
} else {
|
||||||
return PrepareIOFromReadOptions(ro, SystemClock::Default().get(), opts);
|
return PrepareIOFromReadOptions(ro, SystemClock::Default().get(), opts,
|
||||||
|
dbg);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
IOStatus RandomAccessFileReader::ReadAsync(
|
IOStatus RandomAccessFileReader::ReadAsync(
|
||||||
FSReadRequest& req, const IOOptions& opts,
|
FSReadRequest& req, const IOOptions& opts,
|
||||||
std::function<void(FSReadRequest&, void*)> cb, void* cb_arg,
|
std::function<void(FSReadRequest&, void*)> cb, void* cb_arg,
|
||||||
void** io_handle, IOHandleDeleter* del_fn, AlignedBuf* aligned_buf) {
|
void** io_handle, IOHandleDeleter* del_fn, AlignedBuf* aligned_buf,
|
||||||
|
IODebugContext* dbg) {
|
||||||
IOStatus s;
|
IOStatus s;
|
||||||
// Create a callback and populate info.
|
// Create a callback and populate info.
|
||||||
auto read_async_callback =
|
auto read_async_callback =
|
||||||
@@ -532,14 +541,14 @@ IOStatus RandomAccessFileReader::ReadAsync(
|
|||||||
(stats_ != nullptr) ? &elapsed : nullptr, true /*overwrite*/,
|
(stats_ != nullptr) ? &elapsed : nullptr, true /*overwrite*/,
|
||||||
true /*delay_enabled*/);
|
true /*delay_enabled*/);
|
||||||
s = file_->ReadAsync(aligned_req, opts, read_async_callback,
|
s = file_->ReadAsync(aligned_req, opts, read_async_callback,
|
||||||
read_async_info, io_handle, del_fn, nullptr /*dbg*/);
|
read_async_info, io_handle, del_fn, dbg);
|
||||||
} else {
|
} else {
|
||||||
StopWatch sw(clock_, stats_, hist_type_,
|
StopWatch sw(clock_, stats_, hist_type_,
|
||||||
GetFileReadHistograms(stats_, opts.io_activity),
|
GetFileReadHistograms(stats_, opts.io_activity),
|
||||||
(stats_ != nullptr) ? &elapsed : nullptr, true /*overwrite*/,
|
(stats_ != nullptr) ? &elapsed : nullptr, true /*overwrite*/,
|
||||||
true /*delay_enabled*/);
|
true /*delay_enabled*/);
|
||||||
s = file_->ReadAsync(req, opts, read_async_callback, read_async_info,
|
s = file_->ReadAsync(req, opts, read_async_callback, read_async_info,
|
||||||
io_handle, del_fn, nullptr /*dbg*/);
|
io_handle, del_fn, dbg);
|
||||||
}
|
}
|
||||||
RecordTick(stats_, READ_ASYNC_MICROS, elapsed);
|
RecordTick(stats_, READ_ASYNC_MICROS, elapsed);
|
||||||
|
|
||||||
|
|||||||
@@ -164,7 +164,8 @@ class RandomAccessFileReader {
|
|||||||
// the internally allocated buffer on return, and the result refers to a
|
// the internally allocated buffer on return, and the result refers to a
|
||||||
// region in aligned_buf.
|
// region in aligned_buf.
|
||||||
IOStatus Read(const IOOptions& opts, uint64_t offset, size_t n, Slice* result,
|
IOStatus Read(const IOOptions& opts, uint64_t offset, size_t n, Slice* result,
|
||||||
char* scratch, AlignedBuf* aligned_buf) const;
|
char* scratch, AlignedBuf* aligned_buf,
|
||||||
|
IODebugContext* dbg = nullptr) const;
|
||||||
|
|
||||||
// REQUIRES:
|
// REQUIRES:
|
||||||
// num_reqs > 0, reqs do not overlap, and offsets in reqs are increasing.
|
// num_reqs > 0, reqs do not overlap, and offsets in reqs are increasing.
|
||||||
@@ -172,10 +173,12 @@ class RandomAccessFileReader {
|
|||||||
// In direct IO mode, aligned_buf stores the aligned buffer allocated inside
|
// In direct IO mode, aligned_buf stores the aligned buffer allocated inside
|
||||||
// MultiRead, the result Slices in reqs refer to aligned_buf.
|
// MultiRead, the result Slices in reqs refer to aligned_buf.
|
||||||
IOStatus MultiRead(const IOOptions& opts, FSReadRequest* reqs,
|
IOStatus MultiRead(const IOOptions& opts, FSReadRequest* reqs,
|
||||||
size_t num_reqs, AlignedBuf* aligned_buf) const;
|
size_t num_reqs, AlignedBuf* aligned_buf,
|
||||||
|
IODebugContext* dbg = nullptr) const;
|
||||||
|
|
||||||
IOStatus Prefetch(const IOOptions& opts, uint64_t offset, size_t n) const {
|
IOStatus Prefetch(const IOOptions& opts, uint64_t offset, size_t n,
|
||||||
return file_->Prefetch(offset, n, opts, nullptr);
|
IODebugContext* dbg = nullptr) const {
|
||||||
|
return file_->Prefetch(offset, n, opts, dbg);
|
||||||
}
|
}
|
||||||
|
|
||||||
FSRandomAccessFile* file() { return file_.get(); }
|
FSRandomAccessFile* file() { return file_.get(); }
|
||||||
@@ -184,12 +187,13 @@ class RandomAccessFileReader {
|
|||||||
|
|
||||||
bool use_direct_io() const { return file_->use_direct_io(); }
|
bool use_direct_io() const { return file_->use_direct_io(); }
|
||||||
|
|
||||||
IOStatus PrepareIOOptions(const ReadOptions& ro, IOOptions& opts) const;
|
IOStatus PrepareIOOptions(const ReadOptions& ro, IOOptions& opts,
|
||||||
|
IODebugContext* dbg = nullptr) const;
|
||||||
|
|
||||||
IOStatus ReadAsync(FSReadRequest& req, const IOOptions& opts,
|
IOStatus ReadAsync(FSReadRequest& req, const IOOptions& opts,
|
||||||
std::function<void(FSReadRequest&, void*)> cb,
|
std::function<void(FSReadRequest&, void*)> cb,
|
||||||
void* cb_arg, void** io_handle, IOHandleDeleter* del_fn,
|
void* cb_arg, void** io_handle, IOHandleDeleter* del_fn,
|
||||||
AlignedBuf* aligned_buf);
|
AlignedBuf* aligned_buf, IODebugContext* dbg = nullptr);
|
||||||
|
|
||||||
void ReadAsyncCallback(FSReadRequest& req, void* cb_arg);
|
void ReadAsyncCallback(FSReadRequest& req, void* cb_arg);
|
||||||
};
|
};
|
||||||
|
|||||||
@@ -147,8 +147,9 @@ TEST_F(RandomAccessFileReaderTest, MultiReadDirectIO) {
|
|||||||
reqs.push_back(std::move(r0));
|
reqs.push_back(std::move(r0));
|
||||||
reqs.push_back(std::move(r1));
|
reqs.push_back(std::move(r1));
|
||||||
AlignedBuf aligned_buf;
|
AlignedBuf aligned_buf;
|
||||||
ASSERT_OK(
|
IODebugContext dbg;
|
||||||
r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf));
|
ASSERT_OK(r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf,
|
||||||
|
&dbg));
|
||||||
|
|
||||||
AssertResult(content, reqs);
|
AssertResult(content, reqs);
|
||||||
|
|
||||||
@@ -192,8 +193,9 @@ TEST_F(RandomAccessFileReaderTest, MultiReadDirectIO) {
|
|||||||
reqs.push_back(std::move(r1));
|
reqs.push_back(std::move(r1));
|
||||||
reqs.push_back(std::move(r2));
|
reqs.push_back(std::move(r2));
|
||||||
AlignedBuf aligned_buf;
|
AlignedBuf aligned_buf;
|
||||||
ASSERT_OK(
|
IODebugContext dbg;
|
||||||
r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf));
|
ASSERT_OK(r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf,
|
||||||
|
&dbg));
|
||||||
|
|
||||||
AssertResult(content, reqs);
|
AssertResult(content, reqs);
|
||||||
|
|
||||||
@@ -237,8 +239,9 @@ TEST_F(RandomAccessFileReaderTest, MultiReadDirectIO) {
|
|||||||
reqs.push_back(std::move(r1));
|
reqs.push_back(std::move(r1));
|
||||||
reqs.push_back(std::move(r2));
|
reqs.push_back(std::move(r2));
|
||||||
AlignedBuf aligned_buf;
|
AlignedBuf aligned_buf;
|
||||||
ASSERT_OK(
|
IODebugContext dbg;
|
||||||
r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf));
|
ASSERT_OK(r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf,
|
||||||
|
&dbg));
|
||||||
|
|
||||||
AssertResult(content, reqs);
|
AssertResult(content, reqs);
|
||||||
|
|
||||||
@@ -274,8 +277,9 @@ TEST_F(RandomAccessFileReaderTest, MultiReadDirectIO) {
|
|||||||
reqs.push_back(std::move(r0));
|
reqs.push_back(std::move(r0));
|
||||||
reqs.push_back(std::move(r1));
|
reqs.push_back(std::move(r1));
|
||||||
AlignedBuf aligned_buf;
|
AlignedBuf aligned_buf;
|
||||||
ASSERT_OK(
|
IODebugContext dbg;
|
||||||
r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf));
|
ASSERT_OK(r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf,
|
||||||
|
&dbg));
|
||||||
|
|
||||||
AssertResult(content, reqs);
|
AssertResult(content, reqs);
|
||||||
|
|
||||||
|
|||||||
@@ -220,6 +220,8 @@ struct FileOptions : EnvOptions {
|
|||||||
|
|
||||||
// A structure to pass back some debugging information from the FileSystem
|
// A structure to pass back some debugging information from the FileSystem
|
||||||
// implementation to RocksDB in case of an IO error
|
// implementation to RocksDB in case of an IO error
|
||||||
|
// TODO(virajthakur): Update all calls to FS APIs for writes to pass in
|
||||||
|
// IODebugContext
|
||||||
struct IODebugContext {
|
struct IODebugContext {
|
||||||
// file_path to be filled in by RocksDB in case of an error
|
// file_path to be filled in by RocksDB in case of an error
|
||||||
std::string file_path;
|
std::string file_path;
|
||||||
@@ -230,8 +232,9 @@ struct IODebugContext {
|
|||||||
// To be set by the FileSystem implementation
|
// To be set by the FileSystem implementation
|
||||||
std::string msg;
|
std::string msg;
|
||||||
|
|
||||||
// To be set by the underlying FileSystem implementation.
|
// To be set by the application, to allow tracing logs/metrics from user ->
|
||||||
std::string request_id;
|
// RocksDB -> FS.
|
||||||
|
const std::string* request_id = nullptr;
|
||||||
|
|
||||||
// In order to log required information in IO tracing for different
|
// In order to log required information in IO tracing for different
|
||||||
// operations, Each bit in trace_data stores which corresponding info from
|
// operations, Each bit in trace_data stores which corresponding info from
|
||||||
@@ -255,7 +258,7 @@ struct IODebugContext {
|
|||||||
|
|
||||||
// Called by underlying file system to set request_id and log request_id in
|
// Called by underlying file system to set request_id and log request_id in
|
||||||
// IOTracing.
|
// IOTracing.
|
||||||
void SetRequestId(const std::string& _request_id) {
|
void SetRequestId(const std::string* _request_id) {
|
||||||
request_id = _request_id;
|
request_id = _request_id;
|
||||||
trace_data |= (1 << TraceData::kRequestID);
|
trace_data |= (1 << TraceData::kRequestID);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2062,6 +2062,19 @@ struct ReadOptions {
|
|||||||
|
|
||||||
// *** END options for RocksDB internal use only ***
|
// *** END options for RocksDB internal use only ***
|
||||||
|
|
||||||
|
// *** BEGIN per-request settings for internal team use only ***
|
||||||
|
|
||||||
|
// TODO: create a new struct for per-request options, potentially including
|
||||||
|
// timestamps in point lookups/scans
|
||||||
|
|
||||||
|
// request_id is a unique id assigned by the application. It is used to allow
|
||||||
|
// us to link file system metrics/logs to rocksDB and application logs. This
|
||||||
|
// request_id may not be unique to each RocksDB api call - it could refer to
|
||||||
|
// an application level request that results in multiple RocksDB api calls
|
||||||
|
const std::string* request_id = nullptr;
|
||||||
|
|
||||||
|
// *** END per-request settings for internal team use only ***
|
||||||
|
|
||||||
ReadOptions() {}
|
ReadOptions() {}
|
||||||
ReadOptions(bool _verify_checksums, bool _fill_cache);
|
ReadOptions(bool _verify_checksums, bool _fill_cache);
|
||||||
explicit ReadOptions(Env::IOActivity _io_activity);
|
explicit ReadOptions(Env::IOActivity _io_activity);
|
||||||
|
|||||||
@@ -683,7 +683,8 @@ Status BlockBasedTable::Open(
|
|||||||
// 6. [meta block: index]
|
// 6. [meta block: index]
|
||||||
// 7. [meta block: filter]
|
// 7. [meta block: filter]
|
||||||
IOOptions opts;
|
IOOptions opts;
|
||||||
s = file->PrepareIOOptions(ro, opts);
|
IODebugContext dbg;
|
||||||
|
s = file->PrepareIOOptions(ro, opts, &dbg);
|
||||||
if (s.ok()) {
|
if (s.ok()) {
|
||||||
s = ReadFooterFromFile(opts, file.get(), *ioptions.fs,
|
s = ReadFooterFromFile(opts, file.get(), *ioptions.fs,
|
||||||
prefetch_buffer.get(), file_size, &footer,
|
prefetch_buffer.get(), file_size, &footer,
|
||||||
@@ -941,7 +942,8 @@ Status BlockBasedTable::PrefetchTail(
|
|||||||
#endif // NDEBUG
|
#endif // NDEBUG
|
||||||
|
|
||||||
IOOptions opts;
|
IOOptions opts;
|
||||||
Status s = file->PrepareIOOptions(ro, opts);
|
IODebugContext dbg;
|
||||||
|
Status s = file->PrepareIOOptions(ro, opts, &dbg);
|
||||||
// Try file system prefetch
|
// Try file system prefetch
|
||||||
if (s.ok() && !file->use_direct_io() && !force_direct_prefetch) {
|
if (s.ok() && !file->use_direct_io() && !force_direct_prefetch) {
|
||||||
if (!file->Prefetch(opts, prefetch_off, prefetch_len).IsNotSupported()) {
|
if (!file->Prefetch(opts, prefetch_off, prefetch_len).IsNotSupported()) {
|
||||||
|
|||||||
@@ -138,17 +138,18 @@ DEFINE_SYNC_AND_ASYNC(void, BlockBasedTable::RetrieveMultipleBlocks)
|
|||||||
AlignedBuf direct_io_buf;
|
AlignedBuf direct_io_buf;
|
||||||
{
|
{
|
||||||
IOOptions opts;
|
IOOptions opts;
|
||||||
IOStatus s = file->PrepareIOOptions(options, opts);
|
IODebugContext dbg;
|
||||||
|
IOStatus s = file->PrepareIOOptions(options, opts, &dbg);
|
||||||
if (s.ok()) {
|
if (s.ok()) {
|
||||||
#if defined(WITH_COROUTINES)
|
#if defined(WITH_COROUTINES)
|
||||||
if (file->use_direct_io()) {
|
if (file->use_direct_io()) {
|
||||||
#endif // WITH_COROUTINES
|
#endif // WITH_COROUTINES
|
||||||
s = file->MultiRead(opts, &read_reqs[0], read_reqs.size(),
|
s = file->MultiRead(opts, &read_reqs[0], read_reqs.size(),
|
||||||
&direct_io_buf);
|
&direct_io_buf, &dbg);
|
||||||
#if defined(WITH_COROUTINES)
|
#if defined(WITH_COROUTINES)
|
||||||
} else {
|
} else {
|
||||||
co_await batch->context()->reader().MultiReadAsync(
|
co_await batch->context()->reader().MultiReadAsync(
|
||||||
file, opts, &read_reqs[0], read_reqs.size(), &direct_io_buf);
|
file, opts, &read_reqs[0], read_reqs.size(), &direct_io_buf, &dbg);
|
||||||
}
|
}
|
||||||
#endif // WITH_COROUTINES
|
#endif // WITH_COROUTINES
|
||||||
}
|
}
|
||||||
@@ -240,10 +241,11 @@ DEFINE_SYNC_AND_ASYNC(void, BlockBasedTable::RetrieveMultipleBlocks)
|
|||||||
// its not a memory mapped file
|
// its not a memory mapped file
|
||||||
Slice result;
|
Slice result;
|
||||||
IOOptions opts;
|
IOOptions opts;
|
||||||
IOStatus io_s = file->PrepareIOOptions(options, opts);
|
IODebugContext dbg;
|
||||||
|
IOStatus io_s = file->PrepareIOOptions(options, opts, &dbg);
|
||||||
opts.verify_and_reconstruct_read = true;
|
opts.verify_and_reconstruct_read = true;
|
||||||
io_s = file->Read(opts, handle.offset(), BlockSizeWithTrailer(handle),
|
io_s = file->Read(opts, handle.offset(), BlockSizeWithTrailer(handle),
|
||||||
&result, const_cast<char*>(data), nullptr);
|
&result, const_cast<char*>(data), nullptr, &dbg);
|
||||||
if (io_s.ok()) {
|
if (io_s.ok()) {
|
||||||
assert(result.data() == data);
|
assert(result.data() == data);
|
||||||
assert(result.size() == BlockSizeWithTrailer(handle));
|
assert(result.size() == BlockSizeWithTrailer(handle));
|
||||||
|
|||||||
@@ -39,7 +39,8 @@ void BlockPrefetcher::PrefetchIfNeeded(
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
IOOptions opts;
|
IOOptions opts;
|
||||||
Status s = rep->file->PrepareIOOptions(read_options, opts);
|
IODebugContext dbg;
|
||||||
|
Status s = rep->file->PrepareIOOptions(read_options, opts, &dbg);
|
||||||
if (!s.ok()) {
|
if (!s.ok()) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -591,7 +591,8 @@ Status PartitionedFilterBlockReader::CacheDependencies(
|
|||||||
/*usage=*/FilePrefetchBufferUsage::kUnknown);
|
/*usage=*/FilePrefetchBufferUsage::kUnknown);
|
||||||
|
|
||||||
IOOptions opts;
|
IOOptions opts;
|
||||||
s = rep->file->PrepareIOOptions(ro, opts);
|
IODebugContext dbg;
|
||||||
|
s = rep->file->PrepareIOOptions(ro, opts, &dbg);
|
||||||
if (s.ok()) {
|
if (s.ok()) {
|
||||||
s = prefetch_buffer->Prefetch(opts, rep->file.get(), prefetch_off,
|
s = prefetch_buffer->Prefetch(opts, rep->file.get(), prefetch_off,
|
||||||
static_cast<size_t>(prefetch_len));
|
static_cast<size_t>(prefetch_len));
|
||||||
|
|||||||
+11
-7
@@ -74,7 +74,8 @@ inline bool BlockFetcher::TryGetUncompressBlockFromPersistentCache() {
|
|||||||
inline bool BlockFetcher::TryGetFromPrefetchBuffer() {
|
inline bool BlockFetcher::TryGetFromPrefetchBuffer() {
|
||||||
if (prefetch_buffer_ != nullptr) {
|
if (prefetch_buffer_ != nullptr) {
|
||||||
IOOptions opts;
|
IOOptions opts;
|
||||||
IOStatus io_s = file_->PrepareIOOptions(read_options_, opts);
|
IODebugContext dbg;
|
||||||
|
IOStatus io_s = file_->PrepareIOOptions(read_options_, opts, &dbg);
|
||||||
if (io_s.ok()) {
|
if (io_s.ok()) {
|
||||||
bool read_from_prefetch_buffer = prefetch_buffer_->TryReadFromCache(
|
bool read_from_prefetch_buffer = prefetch_buffer_->TryReadFromCache(
|
||||||
opts, file_, handle_.offset(), block_size_with_trailer_, &slice_,
|
opts, file_, handle_.offset(), block_size_with_trailer_, &slice_,
|
||||||
@@ -246,7 +247,8 @@ inline void BlockFetcher::GetBlockContents() {
|
|||||||
void BlockFetcher::ReadBlock(bool retry) {
|
void BlockFetcher::ReadBlock(bool retry) {
|
||||||
FSReadRequest read_req;
|
FSReadRequest read_req;
|
||||||
IOOptions opts;
|
IOOptions opts;
|
||||||
io_status_ = file_->PrepareIOOptions(read_options_, opts);
|
IODebugContext dbg;
|
||||||
|
io_status_ = file_->PrepareIOOptions(read_options_, opts, &dbg);
|
||||||
opts.verify_and_reconstruct_read = retry;
|
opts.verify_and_reconstruct_read = retry;
|
||||||
read_req.status.PermitUncheckedError();
|
read_req.status.PermitUncheckedError();
|
||||||
// Actual file read
|
// Actual file read
|
||||||
@@ -256,8 +258,9 @@ void BlockFetcher::ReadBlock(bool retry) {
|
|||||||
PERF_CPU_TIMER_GUARD(
|
PERF_CPU_TIMER_GUARD(
|
||||||
block_read_cpu_time,
|
block_read_cpu_time,
|
||||||
ioptions_.env ? ioptions_.env->GetSystemClock().get() : nullptr);
|
ioptions_.env ? ioptions_.env->GetSystemClock().get() : nullptr);
|
||||||
io_status_ = file_->Read(opts, handle_.offset(), block_size_with_trailer_,
|
io_status_ =
|
||||||
&slice_, /*scratch=*/nullptr, &direct_io_buf_);
|
file_->Read(opts, handle_.offset(), block_size_with_trailer_, &slice_,
|
||||||
|
/*scratch=*/nullptr, &direct_io_buf_, &dbg);
|
||||||
PERF_COUNTER_ADD(block_read_count, 1);
|
PERF_COUNTER_ADD(block_read_count, 1);
|
||||||
used_buf_ = const_cast<char*>(slice_.data());
|
used_buf_ = const_cast<char*>(slice_.data());
|
||||||
} else if (use_fs_scratch_) {
|
} else if (use_fs_scratch_) {
|
||||||
@@ -269,7 +272,7 @@ void BlockFetcher::ReadBlock(bool retry) {
|
|||||||
read_req.len = block_size_with_trailer_;
|
read_req.len = block_size_with_trailer_;
|
||||||
read_req.scratch = nullptr;
|
read_req.scratch = nullptr;
|
||||||
io_status_ = file_->MultiRead(opts, &read_req, /*num_reqs=*/1,
|
io_status_ = file_->MultiRead(opts, &read_req, /*num_reqs=*/1,
|
||||||
/*AlignedBuf* =*/nullptr);
|
/*AlignedBuf* =*/nullptr, &dbg);
|
||||||
PERF_COUNTER_ADD(block_read_count, 1);
|
PERF_COUNTER_ADD(block_read_count, 1);
|
||||||
|
|
||||||
slice_ = Slice(read_req.result.data(), read_req.result.size());
|
slice_ = Slice(read_req.result.data(), read_req.result.size());
|
||||||
@@ -286,7 +289,7 @@ void BlockFetcher::ReadBlock(bool retry) {
|
|||||||
io_status_ =
|
io_status_ =
|
||||||
file_->Read(opts, handle_.offset(), /*size*/ block_size_with_trailer_,
|
file_->Read(opts, handle_.offset(), /*size*/ block_size_with_trailer_,
|
||||||
/*result*/ &slice_, /*scratch*/ used_buf_,
|
/*result*/ &slice_, /*scratch*/ used_buf_,
|
||||||
/*aligned_buf=*/nullptr);
|
/*aligned_buf=*/nullptr, &dbg);
|
||||||
PERF_COUNTER_ADD(block_read_count, 1);
|
PERF_COUNTER_ADD(block_read_count, 1);
|
||||||
#ifndef NDEBUG
|
#ifndef NDEBUG
|
||||||
if (slice_.data() == &stack_buf_[0]) {
|
if (slice_.data() == &stack_buf_[0]) {
|
||||||
@@ -417,7 +420,8 @@ IOStatus BlockFetcher::ReadAsyncBlockContents() {
|
|||||||
assert(prefetch_buffer_ != nullptr);
|
assert(prefetch_buffer_ != nullptr);
|
||||||
if (!for_compaction_) {
|
if (!for_compaction_) {
|
||||||
IOOptions opts;
|
IOOptions opts;
|
||||||
IOStatus io_s = file_->PrepareIOOptions(read_options_, opts);
|
IODebugContext dbg;
|
||||||
|
IOStatus io_s = file_->PrepareIOOptions(read_options_, opts, &dbg);
|
||||||
if (!io_s.ok()) {
|
if (!io_s.ok()) {
|
||||||
return io_s;
|
return io_s;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -434,15 +434,16 @@ Status ReadTablePropertiesHelper(
|
|||||||
// If retrying, use a stronger file system read to check and correct
|
// If retrying, use a stronger file system read to check and correct
|
||||||
// data corruption
|
// data corruption
|
||||||
IOOptions opts;
|
IOOptions opts;
|
||||||
if (PrepareIOFromReadOptions(ro, ioptions.clock, opts) !=
|
IODebugContext dbg;
|
||||||
|
if (PrepareIOFromReadOptions(ro, ioptions.clock, opts, &dbg) !=
|
||||||
IOStatus::OK()) {
|
IOStatus::OK()) {
|
||||||
return s;
|
return s;
|
||||||
}
|
}
|
||||||
opts.verify_and_reconstruct_read = true;
|
opts.verify_and_reconstruct_read = true;
|
||||||
std::unique_ptr<char[]> data(new char[len]);
|
std::unique_ptr<char[]> data(new char[len]);
|
||||||
Slice result;
|
Slice result;
|
||||||
IOStatus io_s =
|
IOStatus io_s = file->Read(opts, handle.offset(), len, &result,
|
||||||
file->Read(opts, handle.offset(), len, &result, data.get(), nullptr);
|
data.get(), nullptr, &dbg);
|
||||||
RecordTick(ioptions.stats, FILE_READ_CORRUPTION_RETRY_COUNT);
|
RecordTick(ioptions.stats, FILE_READ_CORRUPTION_RETRY_COUNT);
|
||||||
if (!io_s.ok()) {
|
if (!io_s.ok()) {
|
||||||
ROCKS_LOG_INFO(ioptions.info_log,
|
ROCKS_LOG_INFO(ioptions.info_log,
|
||||||
@@ -574,8 +575,9 @@ Status ReadMetaIndexBlockInFile(RandomAccessFileReader* file,
|
|||||||
Footer* footer_out) {
|
Footer* footer_out) {
|
||||||
Footer footer;
|
Footer footer;
|
||||||
IOOptions opts;
|
IOOptions opts;
|
||||||
|
IODebugContext dbg;
|
||||||
Status s;
|
Status s;
|
||||||
s = file->PrepareIOOptions(read_options, opts);
|
s = file->PrepareIOOptions(read_options, opts, &dbg);
|
||||||
if (!s.ok()) {
|
if (!s.ok()) {
|
||||||
return s;
|
return s;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -82,7 +82,7 @@ Status IOTraceWriter::WriteIOOp(const IOTraceRecord& record,
|
|||||||
uint32_t set_pos = static_cast<uint32_t>(log2(trace_data & -trace_data));
|
uint32_t set_pos = static_cast<uint32_t>(log2(trace_data & -trace_data));
|
||||||
switch (set_pos) {
|
switch (set_pos) {
|
||||||
case IODebugContext::TraceData::kRequestID: {
|
case IODebugContext::TraceData::kRequestID: {
|
||||||
Slice request_id(dbg->request_id);
|
Slice request_id(*dbg->request_id);
|
||||||
PutLengthPrefixedSlice(&trace.payload, request_id);
|
PutLengthPrefixedSlice(&trace.payload, request_id);
|
||||||
} break;
|
} break;
|
||||||
default:
|
default:
|
||||||
|
|||||||
@@ -145,7 +145,8 @@ TEST_F(IOTracerTest, MultipleRecordsWithDifferentIOOpOptions) {
|
|||||||
// Write record with IODebugContext.
|
// Write record with IODebugContext.
|
||||||
io_op_data = 0;
|
io_op_data = 0;
|
||||||
IODebugContext dbg;
|
IODebugContext dbg;
|
||||||
dbg.SetRequestId("request_id_1");
|
const std::string test_request_id = "request_id_1";
|
||||||
|
dbg.SetRequestId(&test_request_id);
|
||||||
IOTraceRecord record5(0, TraceType::kIOTracer, io_op_data,
|
IOTraceRecord record5(0, TraceType::kIOTracer, io_op_data,
|
||||||
GetFileOperation(5), 10 /*latency*/,
|
GetFileOperation(5), 10 /*latency*/,
|
||||||
IOStatus::OK().ToString(), file_name);
|
IOStatus::OK().ToString(), file_name);
|
||||||
|
|||||||
@@ -0,0 +1,2 @@
|
|||||||
|
[internal team use only]
|
||||||
|
allow an application-defined request_id to be passed to RocksDB and propagated to the filesystem via IODebugContext
|
||||||
@@ -31,7 +31,7 @@ bool AsyncFileReader::MultiReadAsyncImpl(ReadAwaiter* awaiter) {
|
|||||||
}
|
}
|
||||||
},
|
},
|
||||||
&awaiter->read_reqs_[i], &awaiter->io_handle_[i], &awaiter->del_fn_[i],
|
&awaiter->read_reqs_[i], &awaiter->io_handle_[i], &awaiter->del_fn_[i],
|
||||||
/*aligned_buf=*/nullptr);
|
/*aligned_buf=*/nullptr, awaiter->dbg_);
|
||||||
if (!s.ok()) {
|
if (!s.ok()) {
|
||||||
// For any non-ok status, the FileSystem will not call the callback
|
// For any non-ok status, the FileSystem will not call the callback
|
||||||
// So let's update the status ourselves
|
// So let's update the status ourselves
|
||||||
|
|||||||
@@ -36,9 +36,10 @@ class AsyncFileReader {
|
|||||||
const IOOptions& opts,
|
const IOOptions& opts,
|
||||||
FSReadRequest* read_reqs,
|
FSReadRequest* read_reqs,
|
||||||
size_t num_reqs,
|
size_t num_reqs,
|
||||||
AlignedBuf* aligned_buf) noexcept {
|
AlignedBuf* aligned_buf,
|
||||||
return ReadOperation<ReadAwaiter>{*this, file, opts,
|
IODebugContext* dbg) noexcept {
|
||||||
read_reqs, num_reqs, aligned_buf};
|
return ReadOperation<ReadAwaiter>{*this, file, opts, read_reqs,
|
||||||
|
num_reqs, aligned_buf, dbg};
|
||||||
}
|
}
|
||||||
|
|
||||||
private:
|
private:
|
||||||
@@ -49,12 +50,14 @@ class AsyncFileReader {
|
|||||||
public:
|
public:
|
||||||
explicit ReadAwaiter(AsyncFileReader& reader, RandomAccessFileReader* file,
|
explicit ReadAwaiter(AsyncFileReader& reader, RandomAccessFileReader* file,
|
||||||
const IOOptions& opts, FSReadRequest* read_reqs,
|
const IOOptions& opts, FSReadRequest* read_reqs,
|
||||||
size_t num_reqs, AlignedBuf* /*aligned_buf*/) noexcept
|
size_t num_reqs, AlignedBuf* /*aligned_buf*/,
|
||||||
|
IODebugContext* dbg) noexcept
|
||||||
: reader_(reader),
|
: reader_(reader),
|
||||||
file_(file),
|
file_(file),
|
||||||
opts_(opts),
|
opts_(opts),
|
||||||
read_reqs_(read_reqs),
|
read_reqs_(read_reqs),
|
||||||
num_reqs_(num_reqs),
|
num_reqs_(num_reqs),
|
||||||
|
dbg_(dbg),
|
||||||
next_(nullptr) {}
|
next_(nullptr) {}
|
||||||
|
|
||||||
bool await_ready() noexcept { return false; }
|
bool await_ready() noexcept { return false; }
|
||||||
@@ -82,6 +85,7 @@ class AsyncFileReader {
|
|||||||
const IOOptions& opts_;
|
const IOOptions& opts_;
|
||||||
FSReadRequest* read_reqs_;
|
FSReadRequest* read_reqs_;
|
||||||
size_t num_reqs_;
|
size_t num_reqs_;
|
||||||
|
IODebugContext* dbg_;
|
||||||
autovector<void*, 32> io_handle_;
|
autovector<void*, 32> io_handle_;
|
||||||
autovector<IOHandleDeleter, 32> del_fn_;
|
autovector<IOHandleDeleter, 32> del_fn_;
|
||||||
folly::coro::impl::coroutine_handle<> awaiting_coro_;
|
folly::coro::impl::coroutine_handle<> awaiting_coro_;
|
||||||
@@ -101,18 +105,20 @@ class AsyncFileReader {
|
|||||||
explicit ReadOperation(AsyncFileReader& reader,
|
explicit ReadOperation(AsyncFileReader& reader,
|
||||||
RandomAccessFileReader* file, const IOOptions& opts,
|
RandomAccessFileReader* file, const IOOptions& opts,
|
||||||
FSReadRequest* read_reqs, size_t num_reqs,
|
FSReadRequest* read_reqs, size_t num_reqs,
|
||||||
AlignedBuf* aligned_buf) noexcept
|
AlignedBuf* aligned_buf,
|
||||||
|
IODebugContext* dbg) noexcept
|
||||||
: reader_(reader),
|
: reader_(reader),
|
||||||
file_(file),
|
file_(file),
|
||||||
opts_(opts),
|
opts_(opts),
|
||||||
read_reqs_(read_reqs),
|
read_reqs_(read_reqs),
|
||||||
num_reqs_(num_reqs),
|
num_reqs_(num_reqs),
|
||||||
aligned_buf_(aligned_buf) {}
|
aligned_buf_(aligned_buf),
|
||||||
|
dbg_(dbg) {}
|
||||||
|
|
||||||
auto viaIfAsync(folly::Executor::KeepAlive<> executor) const {
|
auto viaIfAsync(folly::Executor::KeepAlive<> executor) const {
|
||||||
return folly::coro::co_viaIfAsync(
|
return folly::coro::co_viaIfAsync(
|
||||||
std::move(executor),
|
std::move(executor), Awaiter{reader_, file_, opts_, read_reqs_,
|
||||||
Awaiter{reader_, file_, opts_, read_reqs_, num_reqs_, aligned_buf_});
|
num_reqs_, aligned_buf_, dbg_});
|
||||||
}
|
}
|
||||||
|
|
||||||
private:
|
private:
|
||||||
@@ -122,6 +128,7 @@ class AsyncFileReader {
|
|||||||
FSReadRequest* read_reqs_;
|
FSReadRequest* read_reqs_;
|
||||||
size_t num_reqs_;
|
size_t num_reqs_;
|
||||||
AlignedBuf* aligned_buf_;
|
AlignedBuf* aligned_buf_;
|
||||||
|
IODebugContext* dbg_;
|
||||||
};
|
};
|
||||||
|
|
||||||
// This function does the actual work when this awaitable starts execution
|
// This function does the actual work when this awaitable starts execution
|
||||||
|
|||||||
Reference in New Issue
Block a user