|
|
@ -17,8 +17,6 @@ |
|
|
|
namespace fielddb { |
|
|
|
using namespace leveldb; |
|
|
|
|
|
|
|
const char EMPTY[1] = {0}; |
|
|
|
|
|
|
|
//为虚函数提供最基本的实现
|
|
|
|
void Request::PendReq(Request *req) { |
|
|
|
assert(0); |
|
|
@ -52,15 +50,15 @@ void FieldsReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch, |
|
|
|
WriteBatch &MetaBatch,fielddb::FieldDB *DB, |
|
|
|
std::unordered_set<std::string> &batchKeySet) |
|
|
|
{ |
|
|
|
if (batchKeySet.find(*Key) != batchKeySet.end()){ |
|
|
|
if (batchKeySet.find(Key.ToString()) != batchKeySet.end()){ |
|
|
|
return;//并发的被合并的put/delete请求只处理一次
|
|
|
|
} else { |
|
|
|
batchKeySet.insert(*Key); |
|
|
|
batchKeySet.insert(Key.ToString()); |
|
|
|
} |
|
|
|
std::string val_str; |
|
|
|
Status s = Status::NotFound("test"); |
|
|
|
uint64_t start_ = DB->env_->NowMicros(); |
|
|
|
s = DB->kvDB_->Get(ReadOptions(), *Key, &val_str); |
|
|
|
s = DB->kvDB_->Get(ReadOptions(), Key.ToString(), &val_str); |
|
|
|
DB->construct_FieldsReq_Read_elapsed += DB->env_->NowMicros() - start_; |
|
|
|
FieldArray *oldFields; |
|
|
|
if (s.IsNotFound()){ |
|
|
@ -78,10 +76,10 @@ void FieldsReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch, |
|
|
|
// MutexLock L(&DB->index_mu); //互斥访问索引状态表
|
|
|
|
DB->index_mu.AssertHeld(); |
|
|
|
//1.将存在冲突的put pend到对应的请求
|
|
|
|
for(auto [field_name,field_value] : *Fields) { |
|
|
|
for(auto [field_name,field_value] : SliceFields) { |
|
|
|
if(field_name == EMPTY) break; |
|
|
|
if(DB->index_.count(field_name)) { |
|
|
|
auto [index_status,parent_req] = DB->index_[field_name]; |
|
|
|
if(DB->index_.count(field_name.ToString())) { |
|
|
|
auto [index_status,parent_req] = DB->index_[field_name.ToString()]; |
|
|
|
if(index_status == IndexStatus::Creating || index_status == IndexStatus::Deleting) { |
|
|
|
parent_req->PendReq(this->parent); |
|
|
|
return; |
|
|
@ -107,13 +105,13 @@ void FieldsReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch, |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
KVBatch.Put(Slice(*Key), Slice(SerializeValue(*Fields))); |
|
|
|
std::string scrach = SerializeValue(SliceFields); |
|
|
|
KVBatch.Put(Slice(Key), Slice(scrach)); |
|
|
|
//2.对于没有冲突但含有索引操作的put,构建metaKV,这里直接将KV对简单编码后写入metaDB
|
|
|
|
if(HasIndex || HasOldIndex) { |
|
|
|
std::string MetaKey,MetaValue; |
|
|
|
std::string serialized = SerializeValue(*Fields); |
|
|
|
MetaKV MKV = MetaKV(Key,&serialized); |
|
|
|
std::string serialized = SerializeValue(SliceFields); |
|
|
|
MetaKV MKV = MetaKV(Key,serialized); |
|
|
|
MKV.TransPut(MetaKey, MetaValue); |
|
|
|
MetaBatch.Put(MetaKey, serialized); |
|
|
|
|
|
|
@ -125,7 +123,7 @@ void FieldsReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch, |
|
|
|
if(DB->index_.count(field_name)) { |
|
|
|
std::string indexKey; |
|
|
|
AppendIndexKey(&indexKey, ParsedInternalIndexKey( |
|
|
|
*Key,field_name,field_value)); |
|
|
|
Key,field_name,field_value)); |
|
|
|
IndexBatch.Delete(indexKey); |
|
|
|
} |
|
|
|
} |
|
|
@ -133,12 +131,12 @@ void FieldsReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch, |
|
|
|
|
|
|
|
//3.2对于含有索引的field建立索引
|
|
|
|
if (HasIndex) { |
|
|
|
for(auto [field_name,field_value] : *Fields) { |
|
|
|
for(auto [field_name,field_value] : SliceFields) { |
|
|
|
if(field_name == EMPTY) continue; |
|
|
|
if(DB->index_.count(field_name)) { |
|
|
|
if(DB->index_.count(field_name.ToString())) { |
|
|
|
std::string indexKey; |
|
|
|
AppendIndexKey(&indexKey, ParsedInternalIndexKey( |
|
|
|
*Key,field_name,field_value)); |
|
|
|
Key,field_name,field_value)); |
|
|
|
IndexBatch.Put(indexKey, Slice()); |
|
|
|
} |
|
|
|
} |
|
|
@ -157,10 +155,10 @@ void DeleteReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch, |
|
|
|
WriteBatch &MetaBatch,fielddb::FieldDB *DB, |
|
|
|
std::unordered_set<std::string> &batchKeySet) |
|
|
|
{ |
|
|
|
if (batchKeySet.find(*Key) != batchKeySet.end()){ |
|
|
|
if (batchKeySet.find(Key.ToString()) != batchKeySet.end()){ |
|
|
|
return;//并发的被合并的put/delete请求只处理一次
|
|
|
|
} else { |
|
|
|
batchKeySet.insert(*Key); |
|
|
|
batchKeySet.insert(Key.ToString()); |
|
|
|
} |
|
|
|
//1. 读取当前的最新的键值对,判断是否存在含有键值对的field
|
|
|
|
//2.1 如果无,则正常构造delete
|
|
|
@ -168,11 +166,11 @@ void DeleteReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch, |
|
|
|
//在kvDB和metaDB中写入对应的delete
|
|
|
|
//2.3 如果存在field的索引状态是Creating或者Deleting,那么在那个队列上面进行等待
|
|
|
|
std::string val_str; |
|
|
|
Status s = DB->kvDB_->Get(ReadOptions(), *Key, &val_str); |
|
|
|
Status s = DB->kvDB_->Get(ReadOptions(), Key, &val_str); |
|
|
|
if (s.IsNotFound()) return; |
|
|
|
FieldArray *Fields = new FieldArray; |
|
|
|
ParseValue(val_str,Fields); |
|
|
|
KVBatch.Delete(Slice(*Key)); |
|
|
|
KVBatch.Delete(Slice(Key)); |
|
|
|
bool HasIndex = false; |
|
|
|
{ |
|
|
|
// MutexLock L(&DB->index_mu); //互斥访问索引状态表
|
|
|
@ -191,7 +189,7 @@ void DeleteReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch, |
|
|
|
//assert(0);
|
|
|
|
} |
|
|
|
} |
|
|
|
KVBatch.Delete(Slice(*Key)); |
|
|
|
KVBatch.Delete(Slice(Key)); |
|
|
|
//2.对于没有冲突但含有索引操作的delete,构建metaKV,这里直接将KV对简单编码后写入metaDB
|
|
|
|
if(HasIndex) { |
|
|
|
std::string MetaKey; |
|
|
@ -204,7 +202,7 @@ void DeleteReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch, |
|
|
|
if(DB->index_.count(field_name)) { |
|
|
|
std::string indexKey; |
|
|
|
AppendIndexKey(&indexKey, ParsedInternalIndexKey( |
|
|
|
*Key,field_name,field_value)); |
|
|
|
Key,field_name,field_value)); |
|
|
|
IndexBatch.Delete(indexKey); |
|
|
|
} |
|
|
|
} |
|
|
@ -217,8 +215,8 @@ void DeleteReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch, |
|
|
|
void iCreateReq::Prepare(FieldDB *DB) { |
|
|
|
//在index_中完成索引状态更新,在这里可以避免重复创建
|
|
|
|
DB->index_mu.AssertHeld(); |
|
|
|
if(DB->index_.count(*Field)) { |
|
|
|
auto [istatus,parent] = DB->index_[*Field]; |
|
|
|
if(DB->index_.count(Field.ToString())) { |
|
|
|
auto [istatus,parent] = DB->index_[Field.ToString()]; |
|
|
|
if(istatus == IndexStatus::Exist) { |
|
|
|
//如果已经完成建立索引,则返回成功
|
|
|
|
done = true; |
|
|
@ -232,7 +230,7 @@ void iCreateReq::Prepare(FieldDB *DB) { |
|
|
|
} |
|
|
|
//如果索引状态表中没有,则表示尚未创建,更新相应的状态
|
|
|
|
//这里将done设置为true表示在taskqueue中需要完成的部分已经完成,不需要pend
|
|
|
|
DB->index_[*Field] = {IndexStatus::Creating,this}; |
|
|
|
DB->index_[Field.ToString()] = {IndexStatus::Creating,this}; |
|
|
|
done = true; |
|
|
|
} |
|
|
|
|
|
|
@ -248,12 +246,12 @@ void iCreateReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch, |
|
|
|
//遍历数据库,构建二级索引到indexbatch,(更新metaDB中的元数据为Index类型的(Field,Creating))
|
|
|
|
//一个indexwritebatch写入,那么索引创建删除应该和metadb没有交互
|
|
|
|
std::vector<std::pair<std::string, std::string>> keysAndVal = |
|
|
|
DB->FindKeysAndValByFieldName(*Field); |
|
|
|
DB->FindKeysAndValByFieldName(Field.ToString()); |
|
|
|
Slice value = Slice(); |
|
|
|
for (auto &kvPair : keysAndVal){ |
|
|
|
std::string indexKey; |
|
|
|
AppendIndexKey(&indexKey, |
|
|
|
ParsedInternalIndexKey(kvPair.first, *Field, kvPair.second)); |
|
|
|
ParsedInternalIndexKey(kvPair.first, Field, kvPair.second)); |
|
|
|
IndexBatch.Put(indexKey, value); |
|
|
|
} |
|
|
|
} |
|
|
@ -261,7 +259,7 @@ void iCreateReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch, |
|
|
|
void iCreateReq::Finalize(FieldDB *DB) { |
|
|
|
//1. 写入完成后,更新index状态表,(并将metaDB的值改为Index类型的(Field,Existing))
|
|
|
|
MutexLock iL(&DB->index_mu); |
|
|
|
DB->index_[*Field] = {IndexStatus::Exist, nullptr}; |
|
|
|
DB->index_[Field.ToString()] = {IndexStatus::Exist, nullptr}; |
|
|
|
DB->index_mu.Unlock(); |
|
|
|
|
|
|
|
if (pending_list.empty()) return; |
|
|
@ -280,15 +278,15 @@ void iCreateReq::Finalize(FieldDB *DB) { |
|
|
|
/*******iDeleteReq*******/ |
|
|
|
void iDeleteReq::Prepare(FieldDB *DB) { |
|
|
|
DB->index_mu.AssertHeld(); |
|
|
|
if(DB->index_.count(*Field) == 0) { |
|
|
|
if(DB->index_.count(Field.ToString()) == 0) { |
|
|
|
done = true; |
|
|
|
Deleted = true; |
|
|
|
s = Status::OK(); |
|
|
|
return ; |
|
|
|
} |
|
|
|
auto [istatus,parent] = DB->index_[*Field]; |
|
|
|
auto [istatus,parent] = DB->index_[Field.ToString()]; |
|
|
|
if(istatus == IndexStatus::Exist) { |
|
|
|
DB->index_[*Field] = {IndexStatus::Deleting,this}; |
|
|
|
DB->index_[Field.ToString()] = {IndexStatus::Deleting,this}; |
|
|
|
done = true; |
|
|
|
} else { |
|
|
|
//如果正在创建或者删除,那么pend到对应的请求上
|
|
|
@ -305,19 +303,19 @@ void iDeleteReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch, |
|
|
|
WriteBatch &MetaBatch,fielddb::FieldDB *DB,std::unordered_set<std::string> &batchKeySet) |
|
|
|
{ |
|
|
|
std::vector<std::pair<std::string, std::string>> keysAndVal = |
|
|
|
DB->FindKeysAndValByFieldName(*Field); |
|
|
|
DB->FindKeysAndValByFieldName(Field); |
|
|
|
Slice value = Slice(); |
|
|
|
for (auto &kvPair : keysAndVal){ |
|
|
|
std::string indexKey; |
|
|
|
AppendIndexKey(&indexKey, |
|
|
|
ParsedInternalIndexKey(kvPair.first, *Field, kvPair.second)); |
|
|
|
ParsedInternalIndexKey(kvPair.first, Field, kvPair.second)); |
|
|
|
IndexBatch.Delete(indexKey); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
void iDeleteReq::Finalize(FieldDB *DB) { |
|
|
|
MutexLock iL(&DB->index_mu); |
|
|
|
DB->index_.erase(*Field); |
|
|
|
DB->index_.erase(Field.ToString()); |
|
|
|
DB->index_mu.Unlock(); |
|
|
|
|
|
|
|
if (pending_list.empty()) return; |
|
|
@ -339,37 +337,37 @@ BatchReq::BatchReq(WriteBatch *Batch,port::Mutex *mu): |
|
|
|
struct BatchHandler : WriteBatch::Handler { |
|
|
|
void Put(const Slice &key, const Slice &value) override { |
|
|
|
//为key和value构造存储的地方,防止由于string的析构造成可能得内存访问错误
|
|
|
|
str_buf->push_back(key.ToString()); |
|
|
|
fa_buf->push_back({{EMPTY,value.ToString()}}); |
|
|
|
// str_buf->push_back(key.ToString());
|
|
|
|
// FieldArray *field = new FieldArray;
|
|
|
|
// field = ParseValue(value.ToString(), field);
|
|
|
|
// if (field->empty()){ //batch中的value没有field
|
|
|
|
// fa_buf->push_back({{EMPTY,value.ToString()}});
|
|
|
|
// } else {
|
|
|
|
// fa_buf->push_back(*field);
|
|
|
|
// }
|
|
|
|
|
|
|
|
sub_requests->emplace_back(new FieldsReq(&str_buf->back(),&fa_buf->back(),mu)); |
|
|
|
//默认所有WriteBatch中的东西都是有Field的!!!!!
|
|
|
|
sub_requests->emplace_back(new FieldsReq(key,value,mu)); |
|
|
|
sub_requests->back()->parent = req; |
|
|
|
// delete field;
|
|
|
|
} |
|
|
|
void Delete(const Slice &key) override { |
|
|
|
str_buf->push_back(key.ToString()); |
|
|
|
sub_requests->emplace_back(new DeleteReq(&str_buf->back(),mu)); |
|
|
|
// str_buf->push_back(key.ToString());
|
|
|
|
sub_requests->emplace_back(new DeleteReq(key,mu)); |
|
|
|
sub_requests->back()->parent = req; |
|
|
|
} |
|
|
|
|
|
|
|
BatchReq *req; |
|
|
|
port::Mutex *mu; |
|
|
|
std::deque<std::string> *str_buf; |
|
|
|
std::deque<FieldArray> *fa_buf; |
|
|
|
// std::deque<std::string> *str_buf;
|
|
|
|
// std::deque<FieldArray> *fa_buf;
|
|
|
|
std::deque<Request*> *sub_requests; |
|
|
|
}; |
|
|
|
|
|
|
|
BatchHandler Handler; |
|
|
|
Handler.req = this; |
|
|
|
Handler.mu = mu; |
|
|
|
Handler.str_buf = &str_buf; |
|
|
|
Handler.fa_buf = &fa_buf; |
|
|
|
// Handler.str_buf = &str_buf;
|
|
|
|
// Handler.fa_buf = &fa_buf;
|
|
|
|
Handler.sub_requests = &sub_requests; |
|
|
|
|
|
|
|
Batch->Iterate(&Handler); |
|
|
|