mirror of
https://github.com/facebook/rocksdb.git
synced 2024-11-25 14:31:35 +00:00
8515bd50c9
Summary: Added rate limiter and read rate-limiting support to SequentialFileReader. I've updated call sites to SequentialFileReader::Read with appropriate IO priority (or left a TODO and specified IO_TOTAL for now). The PR is separated into four commits: the first one added the rate-limiting support, but with some fixes in the unit test since the number of request bytes from rate limiter in SequentialFileReader are not accurate (there is overcharge at EOF). The second commit fixed this by allowing SequentialFileReader to check file size and determine how many bytes are left in the file to read. The third commit added benchmark related code. The fourth commit moved the logic of using file size to avoid overcharging the rate limiter into backup engine (the main user of SequentialFileReader). Pull Request resolved: https://github.com/facebook/rocksdb/pull/9973 Test Plan: - `make check`, backup_engine_test covers usage of SequentialFileReader with rate limiter. - Run db_bench to check if rate limiting is throttling as expected: Verified that reads and writes are together throttled at 2MB/s, and at 0.2MB chunks that are 100ms apart. - Set up: `./db_bench --benchmarks=fillrandom -db=/dev/shm/test_rocksdb` - Benchmark: ``` strace -ttfe read,write ./db_bench --benchmarks=backup -db=/dev/shm/test_rocksdb --backup_rate_limit=2097152 --use_existing_db strace -ttfe read,write ./db_bench --benchmarks=restore -db=/dev/shm/test_rocksdb --restore_rate_limit=2097152 --use_existing_db ``` - db bench on backup and restore to ensure no performance regression. - backup (avg over 50 runs): pre-change: 1.90443e+06 micros/op; post-change: 1.8993e+06 micros/op (improve by 0.2%) - restore (avg over 50 runs): pre-change: 1.79105e+06 micros/op; post-change: 1.78192e+06 micros/op (improve by 0.5%) ``` # Set up ./db_bench --benchmarks=fillrandom -db=/tmp/test_rocksdb -num=10000000 # benchmark TEST_TMPDIR=/tmp/test_rocksdb NUM_RUN=50 for ((j=0;j<$NUM_RUN;j++)) do ./db_bench -db=$TEST_TMPDIR -num=10000000 -benchmarks=backup -use_existing_db | egrep 'backup' # Restore #./db_bench -db=$TEST_TMPDIR -num=10000000 -benchmarks=restore -use_existing_db done > rate_limit.txt && awk -v NUM_RUN=$NUM_RUN '{sum+=$3;sum_sqrt+=$3^2}END{print sum/NUM_RUN, sqrt(sum_sqrt/NUM_RUN-(sum/NUM_RUN)^2)}' rate_limit.txt >> rate_limit_2.txt ``` Reviewed By: hx235 Differential Revision: D36327418 Pulled By: cbi42 fbshipit-source-id: e75d4307cff815945482df5ba630c1e88d064691
74 lines
2.3 KiB
C++
74 lines
2.3 KiB
C++
// Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved.
|
|
// This source code is licensed under both the GPLv2 (found in the
|
|
// COPYING file in the root directory) and Apache 2.0 License
|
|
// (found in the LICENSE.Apache file in the root directory).
|
|
|
|
#include "file/line_file_reader.h"
|
|
|
|
#include <cstring>
|
|
|
|
#include "monitoring/iostats_context_imp.h"
|
|
|
|
namespace ROCKSDB_NAMESPACE {
|
|
|
|
IOStatus LineFileReader::Create(const std::shared_ptr<FileSystem>& fs,
|
|
const std::string& fname,
|
|
const FileOptions& file_opts,
|
|
std::unique_ptr<LineFileReader>* reader,
|
|
IODebugContext* dbg,
|
|
RateLimiter* rate_limiter) {
|
|
std::unique_ptr<FSSequentialFile> file;
|
|
IOStatus io_s = fs->NewSequentialFile(fname, file_opts, &file, dbg);
|
|
if (io_s.ok()) {
|
|
reader->reset(new LineFileReader(
|
|
std::move(file), fname, nullptr,
|
|
std::vector<std::shared_ptr<EventListener>>{}, rate_limiter));
|
|
}
|
|
return io_s;
|
|
}
|
|
|
|
bool LineFileReader::ReadLine(std::string* out,
|
|
Env::IOPriority rate_limiter_priority) {
|
|
assert(out);
|
|
if (!io_status_.ok()) {
|
|
// Status should be checked (or permit unchecked) any time we return false.
|
|
io_status_.MustCheck();
|
|
return false;
|
|
}
|
|
out->clear();
|
|
for (;;) {
|
|
// Look for line delimiter
|
|
const char* found = static_cast<const char*>(
|
|
std::memchr(buf_begin_, '\n', buf_end_ - buf_begin_));
|
|
if (found) {
|
|
size_t len = found - buf_begin_;
|
|
out->append(buf_begin_, len);
|
|
buf_begin_ += len + /*delim*/ 1;
|
|
++line_number_;
|
|
return true;
|
|
}
|
|
if (at_eof_) {
|
|
io_status_.MustCheck();
|
|
return false;
|
|
}
|
|
// else flush and reload buffer
|
|
out->append(buf_begin_, buf_end_ - buf_begin_);
|
|
Slice result;
|
|
io_status_ =
|
|
sfr_.Read(buf_.size(), &result, buf_.data(), rate_limiter_priority);
|
|
IOSTATS_ADD(bytes_read, result.size());
|
|
if (!io_status_.ok()) {
|
|
io_status_.MustCheck();
|
|
return false;
|
|
}
|
|
if (result.size() != buf_.size()) {
|
|
// The obscure way of indicating EOF
|
|
at_eof_ = true;
|
|
}
|
|
buf_begin_ = result.data();
|
|
buf_end_ = result.data() + result.size();
|
|
}
|
|
}
|
|
|
|
} // namespace ROCKSDB_NAMESPACE
|