|
|
@ -6,8 +6,10 @@ |
|
|
|
|
|
|
|
#include <algorithm>
|
|
|
|
#include <atomic>
|
|
|
|
#include <cassert>
|
|
|
|
#include <cstdint>
|
|
|
|
#include <cstdio>
|
|
|
|
#include <cstring>
|
|
|
|
#include <set>
|
|
|
|
#include <string>
|
|
|
|
#include <vector>
|
|
|
@ -24,9 +26,12 @@ |
|
|
|
#include "db/write_batch_internal.h"
|
|
|
|
#include "leveldb/db.h"
|
|
|
|
#include "leveldb/env.h"
|
|
|
|
#include "leveldb/filter_policy.h"
|
|
|
|
#include "leveldb/slice.h"
|
|
|
|
#include "leveldb/status.h"
|
|
|
|
#include "leveldb/table.h"
|
|
|
|
#include "leveldb/table_builder.h"
|
|
|
|
#include "leveldb/write_batch.h"
|
|
|
|
#include "port/port.h"
|
|
|
|
#include "table/block.h"
|
|
|
|
#include "table/merger.h"
|
|
|
@ -35,6 +40,7 @@ |
|
|
|
#include "util/logging.h"
|
|
|
|
#include "util/mutexlock.h"
|
|
|
|
#include "util/serialize_value.h"
|
|
|
|
#include "kv_sep/kvlog.h"
|
|
|
|
|
|
|
|
namespace leveldb { |
|
|
|
|
|
|
@ -145,6 +151,7 @@ DBImpl::DBImpl(const Options& raw_options, const std::string& dbname) |
|
|
|
log_(nullptr), |
|
|
|
seed_(0), |
|
|
|
tmp_batch_(new WriteBatch), |
|
|
|
tmp_kp_batch_(new WriteBatch), |
|
|
|
background_compaction_scheduled_(false), |
|
|
|
manual_compaction_(nullptr), |
|
|
|
versions_(new VersionSet(dbname_, &options_, table_cache_, |
|
|
@ -235,6 +242,8 @@ void DBImpl::RemoveObsoleteFiles() { |
|
|
|
// Make a set of all of the live files
|
|
|
|
std::set<uint64_t> live = pending_outputs_; |
|
|
|
versions_->AddLiveFiles(&live); |
|
|
|
//将所有的live的kvlog加入集合,防止被删除
|
|
|
|
versions_->AddLiveKVLogs(&live); |
|
|
|
|
|
|
|
std::vector<std::string> filenames; |
|
|
|
env_->GetChildren(dbname_, &filenames); // Ignoring errors on purpose
|
|
|
@ -540,6 +549,10 @@ Status DBImpl::WriteLevel0Table(MemTable* mem, VersionEdit* edit, |
|
|
|
meta.largest); |
|
|
|
} |
|
|
|
|
|
|
|
edit->AddKVLogs(imm_kvlogfile_number); |
|
|
|
imm_kvlogfile_number = 0; |
|
|
|
delete imm_kvlogfile_; |
|
|
|
|
|
|
|
CompactionStats stats; |
|
|
|
stats.micros = env_->NowMicros() - start_micros; |
|
|
|
stats.bytes_written = meta.file_size; |
|
|
@ -1118,6 +1131,18 @@ int64_t DBImpl::TEST_MaxNextLevelOverlappingBytes() { |
|
|
|
return versions_->MaxNextLevelOverlappingBytes(); |
|
|
|
} |
|
|
|
|
|
|
|
Slice DBImpl::GetValueFromFP(const FilePointer &fp,std::string *value) { |
|
|
|
RandomAccessFile *file; |
|
|
|
Status s = env_->NewRandomAccessFile(KVLogFileName(dbname_, fp.FileNumber), &file); |
|
|
|
Slice slice; |
|
|
|
char *buf = new char[fp.Size]; |
|
|
|
s = file->Read(fp.FileOffset, fp.Size, &slice, buf); |
|
|
|
*value = slice.ToString(); |
|
|
|
delete[] buf; |
|
|
|
delete file; |
|
|
|
return slice; |
|
|
|
} |
|
|
|
|
|
|
|
Status DBImpl::Get(const ReadOptions& options, const Slice& key, |
|
|
|
std::string* value) { |
|
|
|
Status s; |
|
|
@ -1159,6 +1184,15 @@ Status DBImpl::Get(const ReadOptions& options, const Slice& key, |
|
|
|
if (have_stat_update && current->UpdateStats(stats)) { |
|
|
|
MaybeScheduleCompaction(); |
|
|
|
} |
|
|
|
if(!s.ok()) { |
|
|
|
// printf( "not found : %s\n",key.ToString().c_str());
|
|
|
|
} else { |
|
|
|
// printf("found key:%s value:%s\n",key.ToString().c_str(),value->c_str());
|
|
|
|
FilePointer fp; |
|
|
|
DecodeFp(fp, value->data()); |
|
|
|
GetValueFromFP(fp, value); |
|
|
|
} |
|
|
|
|
|
|
|
mem->Unref(); |
|
|
|
if (imm != nullptr) imm->Unref(); |
|
|
|
current->Unref(); |
|
|
@ -1252,7 +1286,7 @@ Status DBImpl::Write(const WriteOptions& options, WriteBatch* updates) { |
|
|
|
|
|
|
|
// May temporarily unlock and wait.
|
|
|
|
Status status = MakeRoomForWrite(updates == nullptr); |
|
|
|
uint64_t last_sequence = versions_->LastSequence(); |
|
|
|
uint64_t last_sequence = versions_->LastSequence(), temp_seq = last_sequence; |
|
|
|
Writer* last_writer = &w; |
|
|
|
if (status.ok() && updates != nullptr) { // nullptr batch is for compactions
|
|
|
|
WriteBatch* write_batch = BuildBatchGroup(&last_writer); |
|
|
@ -1266,7 +1300,14 @@ Status DBImpl::Write(const WriteOptions& options, WriteBatch* updates) { |
|
|
|
{ |
|
|
|
mutex_.Unlock(); |
|
|
|
// uint64_t start_write = env_->NowMicros();
|
|
|
|
status = log_->AddRecord(WriteBatchInternal::Contents(write_batch)); |
|
|
|
FilePointer fp; |
|
|
|
//1. 将WriteBatch写入到kvlog中
|
|
|
|
status = kvlog_->AddRecord(WriteBatchInternal::Contents(write_batch), fp); |
|
|
|
//2. 将writebatch的filepointer写入到log中
|
|
|
|
// status = log_->AddRecord(WriteBatchInternal::Contents(write_batch));
|
|
|
|
char rep[8 * 3]; |
|
|
|
EncodeFP(fp, rep); |
|
|
|
status = log_->AddRecord(Slice(rep, 3 * 8)); |
|
|
|
bool sync_error = false; |
|
|
|
if (status.ok() && options.sync) { |
|
|
|
status = logfile_->Sync(); |
|
|
@ -1274,8 +1315,11 @@ Status DBImpl::Write(const WriteOptions& options, WriteBatch* updates) { |
|
|
|
sync_error = true; |
|
|
|
} |
|
|
|
} |
|
|
|
//3. 根据write_batch里面的内容,构建kp_batch
|
|
|
|
WriteBatchInternal::ConstructKPBatch(tmp_kp_batch_, write_batch, fp); |
|
|
|
WriteBatchInternal::SetSequence(tmp_kp_batch_,temp_seq + 1); |
|
|
|
if (status.ok()) { |
|
|
|
status = WriteBatchInternal::InsertInto(write_batch, mem_); |
|
|
|
status = WriteBatchInternal::InsertInto(tmp_kp_batch_, mem_); |
|
|
|
} |
|
|
|
// BatchSize += write_batch->ApproximateSize();
|
|
|
|
// write_elapsed += env_->NowMicros() - start_write;
|
|
|
@ -1412,6 +1456,22 @@ Status DBImpl::MakeRoomForWrite(bool force) { |
|
|
|
versions_->ReuseFileNumber(new_log_number); |
|
|
|
break; |
|
|
|
} |
|
|
|
/*更换新的kvlog*/ |
|
|
|
uint64_t new_kvlog_number = versions_->NewFileNumber(); |
|
|
|
WritableFile* kvlogfile = nullptr; |
|
|
|
s = env_->NewWritableFile(KVLogFileName(dbname_,new_kvlog_number),&kvlogfile); |
|
|
|
if(!s.ok()) { |
|
|
|
versions_->ReuseFileNumber(new_kvlog_number); |
|
|
|
break; |
|
|
|
} |
|
|
|
pending_outputs_.insert(new_kvlog_number); |
|
|
|
kvlogfile_->Close(); |
|
|
|
imm_kvlogfile_ = kvlogfile_; |
|
|
|
kvlogfile_ = kvlogfile; |
|
|
|
imm_kvlogfile_number = kvlogfile_number_; |
|
|
|
kvlogfile_number_ = new_kvlog_number; |
|
|
|
delete kvlog_; |
|
|
|
kvlog_ = new KVLog(kvlogfile,new_kvlog_number); |
|
|
|
|
|
|
|
delete log_; |
|
|
|
|
|
|
@ -1566,6 +1626,14 @@ Status DB::Open(const Options& options, const std::string& dbname, DB** dbptr) { |
|
|
|
impl->mem_ = new MemTable(impl->internal_comparator_); |
|
|
|
impl->mem_->Ref(); |
|
|
|
} |
|
|
|
|
|
|
|
uint64_t new_kvlog_number = impl->versions_->NewFileNumber(); |
|
|
|
WritableFile* kvlogfile; |
|
|
|
s = options.env->NewWritableFile(KVLogFileName(dbname, new_kvlog_number), &kvlogfile); |
|
|
|
impl->pending_outputs_.insert(new_kvlog_number); |
|
|
|
impl->kvlogfile_number_ = new_kvlog_number; |
|
|
|
impl->kvlogfile_ = kvlogfile; |
|
|
|
impl->kvlog_ = new KVLog(kvlogfile,new_kvlog_number); |
|
|
|
} |
|
|
|
if (s.ok() && save_manifest) { |
|
|
|
edit.SetPrevLogNumber(0); // No older logs needed after recovery.
|
|
|
|