Posted on 2012-03-21 14:41
小明 閱讀(3460)
評(píng)論(0) 編輯 收藏 所屬分類:
分布式計(jì)算
總體來(lái)說(shuō),leveldb的寫操作有兩個(gè)步驟，首先是針對(duì)log的append操作，然后是對(duì)memtable的插入操作。
影響寫性能的因素有：
1.
write_buffer_size2.
kL0_SlowdownWritesTrigger and
kL0_StopWritesTrigger.提高這兩個(gè)值，能夠增加寫的性能，但是降低讀的性能
看看WriteOptions有哪些參數(shù)可以指定
struct WriteOptions {
//設(shè)置sync=true,leveldb會(huì)調(diào)用fsync()，這會(huì)降低插入性能
//同時(shí)會(huì)增加數(shù)據(jù)的安全性
//Default: false
bool sync;
WriteOptions()
: sync(false) {
}
};
首先把Key,value轉(zhuǎn)成WriteBatch
Status DB::Put(const WriteOptions& opt, const Slice& key, const Slice& value) {
WriteBatch batch;
batch.Put(key, value);
return Write(opt, &batch);
}
接下來(lái)就是真正的插入了
這里使用了兩把鎖，主要是想提高并發(fā)能力，減少上鎖的時(shí)間。
首先是檢查是否可寫，然后append log，最后是插入memtable
<db/dbimpl.cc>
Status DBImpl::Write(const WriteOptions& options, WriteBatch* updates) {
Status status;
//加鎖
MutexLock l(&mutex_);
LoggerId self;
//拿到寫log的權(quán)利
AcquireLoggingResponsibility(&self);
//檢查是否可寫
status = MakeRoomForWrite(false); // May temporarily release lock and wait
uint64_t last_sequence = versions_->LastSequence();
if (status.ok()) {
WriteBatchInternal::SetSequence(updates, last_sequence + 1);
last_sequence += WriteBatchInternal::Count(updates);
// Add to log and apply to memtable. We can release the lock during
// this phase since the "logger_" flag protects against concurrent
// loggers and concurrent writes into mem_.
{
assert(logger_ == &self);
mutex_.Unlock();
//IO操作：寫入LOG
status = log_->AddRecord(WriteBatchInternal::Contents(updates));
if (status.ok() && options.sync) {
status = logfile_->Sync();
}
//插入memtable
if (status.ok()) {
status = WriteBatchInternal::InsertInto(updates, mem_);
}
mutex_.Lock();
assert(logger_ == &self);
}
//設(shè)置新的seqence number
versions_->SetLastSequence(last_sequence);
}
//釋放寫LOG鎖
ReleaseLoggingResponsibility(&self);
return status;
}
寫流量控制：
<db/dbimpl.cc>
Status DBImpl::MakeRoomForWrite(bool force) {
mutex_.AssertHeld();
assert(logger_ != NULL);
bool allow_delay = !force;
Status s;
while (true) {
if (!bg_error_.ok()) {
// Yield previous error
s = bg_error_;
break;
} else if (
allow_delay &&
versions_->NumLevelFiles(0) >= config::kL0_SlowdownWritesTrigger) {
mutex_.Unlock();
//如果level0的文件大于kL0_SlowdownWritesTrigger閾值，則sleep 1s，這樣給compaction更多的CPU
env_->SleepForMicroseconds(1000);
allow_delay = false; // Do not delay a single write more than once
mutex_.Lock();
} else if (!force &&
(mem_->ApproximateMemoryUsage() <= options_.write_buffer_size)) {
//可寫
break;
} else if (imm_ != NULL) {
// imm_:之前的memtable 沒有被compaction，需要等待
bg_cv_.Wait();
} else if (versions_->NumLevelFiles(0) >= config::kL0_StopWritesTrigger) {
// level0文件個(gè)數(shù)大于kL0_StopWritesTrigger,需要等待
Log(options_.info_log, "waiting
\n");
bg_cv_.Wait();
} else {
//生成新的額memtable和logfile，把當(dāng)前memtable傳給imm_
assert(versions_->PrevLogNumber() == 0);
uint64_t new_log_number = versions_->NewFileNumber();
WritableFile* lfile = NULL;
s = env_->NewWritableFile(LogFileName(dbname_, new_log_number), &lfile);
if (!s.ok()) {
break;
}
delete log_;
delete logfile_;
logfile_ = lfile;
logfile_number_ = new_log_number;
log_ = new log::Writer(lfile);
imm_ = mem_;
has_imm_.Release_Store(imm_);
mem_ = new MemTable(internal_comparator_);
mem_->Ref();
force = false; // Do not force another compaction if have room
// 發(fā)起compaction,dump imm_
MaybeScheduleCompaction();
}
}
return s;
}