andreacavalli/rocksdb
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fix hang with large write batches and column families.

Browse Source 
Summary:
This diff fixes a hang reported by a Github user.
https://www.facebook.com/l.php?u=https%3A%2F%2Fgithub.com%2Ffacebook%2Frocksdb%2Fissues%2F595%23issuecomment-96983273&h=9AQFYOWlo
Multiple large write batches with column families cause a hang.
The issue was caused by not doing flushes/compaction when the
write controller was stopped.

Test Plan: Create a DBTest from the user's test case

Reviewers: igor

Reviewed By: igor

Subscribers: dhruba, leveldb

Differential Revision: https://reviews.facebook.net/D37929
This commit is contained in:
Venkatesh Radhakrishnan 2015-05-01 15:41:50 -07:00 committed by Igor Canadi
parent a11863cb76
commit c7c2e1968f
2 changed files with 37 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
db/db_impl.cc
View File

@ -3131,6 +3131,11 @@ Status DBImpl::Write(const WriteOptions& write_options, WriteBatch* my_batch) {
if (UNLIKELY(status.ok()) &&
(write_controller_.IsStopped() || write_controller_.GetDelay() > 0)) {
// If writer is stopped, we need to get it going,
// so schedule flushes/compactions
if (context.schedule_bg_work_) {
MaybeScheduleFlushOrCompaction();
}
status = DelayWrite(expiration_time);
}

32
db/db_test.cc
View File

@ -11982,6 +11982,38 @@ TEST_F(DBTest, HugeNumberOfLevels) {
ASSERT_OK(db_->CompactRange(nullptr, nullptr));
}
// Github issue #595
// Large write batch with column families
TEST_F(DBTest, LargeBatchWithColumnFamilies) {
Options options;
options.env = env_;
options = CurrentOptions(options);
options.write_buffer_size = 100000; // Small write buffer
CreateAndReopenWithCF({"pikachu"}, options);
int64_t j = 0;
for (int i = 0; i < 5; i++) {
for (int pass = 1; pass <= 3; pass++) {
WriteBatch batch;
size_t write_size = 1024 * 1024 * (5 + i);
fprintf(stderr, "prepare: %ld MB, pass:%d\n", (write_size / 1024 / 1024),
pass);
for (;;) {
std::string data(3000, j++ % 127 + 20);
data += std::to_string(j);
batch.Put(handles_[0], Slice(data), Slice(data));
if (batch.GetDataSize() > write_size) {
break;
}
}
fprintf(stderr, "write: %ld MB\n", (batch.GetDataSize() / 1024 / 1024));
ASSERT_OK(dbfull()->Write(WriteOptions(), &batch));
fprintf(stderr, "done\n");
}
}
// make sure we can re-open it.
ASSERT_OK(TryReopenWithColumnFamilies({"default", "pikachu"}, options));
}
} // namespace rocksdb
int main(int argc, char** argv) {