Skip to content

Commit

Permalink
use the cpplint.py specification code format (OpenAtomFoundation#114)
Browse files Browse the repository at this point in the history
  • Loading branch information
@Axlgrep
Axlgrep authored Oct 23, 2018
1 parent 803550f commit 3715974
Show file tree
Hide file tree
Showing 29 changed files with 1,411 additions and 1,117 deletions.
62 changes: 37 additions & 25 deletions src/backupable.cc
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,8 @@
#include <sys/types.h>
// Copyright (c) 2017-present The blackwidow Authors. All rights reserved.
// This source code is licensed under the BSD-style license found in the
// LICENSE file in the root directory of this source tree. An additional grant
// of patent rights can be found in the PATENTS file in the same directory.
//
#include <dirent.h>
#include <utility>

Expand All @@ -17,21 +21,22 @@ BackupEngine::~BackupEngine() {
engines_.clear();
}

Status BackupEngine::NewCheckpoint(rocksdb::DB* rocksdb_db, const std::string& type) {
Status BackupEngine::NewCheckpoint(rocksdb::DB* rocksdb_db,
const std::string& type) {
rocksdb::DBCheckpoint* checkpoint;
Status s = rocksdb::DBCheckpoint::Create(rocksdb_db, &checkpoint);
if (!s.ok()) {
//log_warn("create checkpoint failed, error %s", s.ToString().c_str());
// log_warn("create checkpoint failed, error %s", s.ToString().c_str());
return s;
}
engines_.insert(std::make_pair(type, checkpoint));
return s;
}

Status BackupEngine::Open(blackwidow::BlackWidow *blackwidow,
BackupEngine** backup_engine_ptr) {
BackupEngine** backup_engine_ptr) {
*backup_engine_ptr = new BackupEngine();
if (!*backup_engine_ptr){
if (!*backup_engine_ptr) {
return Status::Corruption("New BackupEngine failed!");
}

Expand Down Expand Up @@ -59,41 +64,45 @@ Status BackupEngine::Open(blackwidow::BlackWidow *blackwidow,
Status BackupEngine::SetBackupContent() {
Status s;
for (const auto& engine : engines_) {
//Get backup content
// Get backup content
BackupContent bcontent;
s = engine.second->GetCheckpointFiles(bcontent.live_files,
bcontent.live_wal_files,
bcontent.manifest_file_size, bcontent.sequence_number);
if (!s.ok()) {
//log_warn("get backup files faild for type: %s", engine.first.c_str());
// log_warn("get backup files faild for type: %s", engine.first.c_str());
return s;
}
backup_content_[engine.first] = std::move(bcontent);
}
return s;
}

Status BackupEngine::CreateNewBackupSpecify(const std::string &backup_dir, const std::string &type) {
std::map<std::string, rocksdb::DBCheckpoint*>::iterator it_engine = engines_.find(type);
std::map<std::string, BackupContent>::iterator it_content = backup_content_.find(type);
Status BackupEngine::CreateNewBackupSpecify(const std::string &backup_dir,
const std::string &type) {
std::map<std::string, rocksdb::DBCheckpoint*>::iterator it_engine =
engines_.find(type);
std::map<std::string, BackupContent>::iterator it_content =
backup_content_.find(type);
std::string dir = GetSaveDirByType(backup_dir, type);
delete_dir(dir.c_str());

if (it_content != backup_content_.end() &&
if (it_content != backup_content_.end() &&
it_engine != engines_.end()) {
Status s = it_engine->second->CreateCheckpointWithFiles(
dir,
it_content->second.live_files,
it_content->second.live_files,
it_content->second.live_wal_files,
it_content->second.manifest_file_size,
it_content->second.sequence_number);
if (!s.ok()) {
//log_warn("backup engine create new failed, type: %s, error %s", type.c_str(), s.ToString().c_str());
// log_warn("backup engine create new failed, type: %s, error %s",
// type.c_str(), s.ToString().c_str());
return s;
}

} else {
//log_warn("invalid db type: %s", type.c_str());
// log_warn("invalid db type: %s", type.c_str());
return Status::Corruption("invalid db type");
}
return Status::OK();
Expand All @@ -102,24 +111,25 @@ Status BackupEngine::CreateNewBackupSpecify(const std::string &backup_dir, const
void* ThreadFuncSaveSpecify(void *arg) {
BackupSaveArgs* arg_ptr = static_cast<BackupSaveArgs*>(arg);
BackupEngine* p = static_cast<BackupEngine*>(arg_ptr->p_engine);

arg_ptr->res = p->CreateNewBackupSpecify(arg_ptr->backup_dir, arg_ptr->key_type);

arg_ptr->res = p->CreateNewBackupSpecify(arg_ptr->backup_dir,
arg_ptr->key_type);
pthread_exit(&(arg_ptr->res));
}

Status BackupEngine::WaitBackupPthread() {
Status BackupEngine::WaitBackupPthread() {
int ret;
Status s = Status::OK();
for (auto& pthread : backup_pthread_ts_) {
void *res;
if ((ret = pthread_join(pthread.second, &res)) != 0) {
//log_warn("pthread_join failed with backup thread for key_type: %s, error %d", pthread.first.c_str(), ret);
// log_warn("pthread_join failed with backup thread for key_type:
// %s, error %d", pthread.first.c_str(), ret);
}
Status cur_s = *(static_cast<Status*>(res));
if (!cur_s.ok()) {
//log_warn("pthread executed failed with key_type: %s, error %s", pthread.first.c_str(), cur_s.ToString().c_str());
StopBackup(); //stop others when someone failed
// log_warn("pthread executed failed with key_type: %s, error %s",
// pthread.first.c_str(), cur_s.ToString().c_str());
StopBackup(); // stop others when someone failed
s = cur_s;
}
}
Expand All @@ -133,14 +143,16 @@ Status BackupEngine::CreateNewBackup(const std::string &dir) {
std::vector<BackupSaveArgs*> args;
for (const auto& engine : engines_) {
pthread_t tid;
BackupSaveArgs *arg = new BackupSaveArgs((void*)this, dir, engine.first);
BackupSaveArgs *arg = new BackupSaveArgs(
reinterpret_cast<void*>(this), dir, engine.first);
args.push_back(arg);
if (pthread_create(&tid, NULL, &ThreadFuncSaveSpecify, arg) != 0) {
s = Status::Corruption("pthead_create failed.");
break;
}
if (!(backup_pthread_ts_.insert(std::make_pair(engine.first, tid)).second)) {
//log_warn("thread open dupilicated, type: %s", engine.first.c_str());
if (!(backup_pthread_ts_.insert(
std::make_pair(engine.first, tid)).second)) {
// log_warn("thread open dupilicated, type: %s", engine.first.c_str());
backup_pthread_ts_[engine.first] = tid;
}
}
Expand All @@ -161,4 +173,4 @@ void BackupEngine::StopBackup() {
// DEPRECATED
}

}
} // namespace blackwidow
133 changes: 66 additions & 67 deletions src/base_data_key_format.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,16 +8,15 @@

namespace blackwidow {
class BaseDataKey {
public:
BaseDataKey(const Slice& key, int32_t version, const Slice& data) :
start_(nullptr), key_(key), version_(version), data_(data) {
}
public:
BaseDataKey(const Slice& key, int32_t version, const Slice& data) :
start_(nullptr), key_(key), version_(version), data_(data) {}

~BaseDataKey() {
if (start_ != space_) {
delete[] start_;
}
~BaseDataKey() {
if (start_ != space_) {
delete[] start_;
}
}

const Slice Encode() {
size_t usize = key_.size() + data_.size();
Expand Down Expand Up @@ -45,89 +44,89 @@ class BaseDataKey {
return Slice(start_, needed);
}

private:
char space_[200];
char* start_;
Slice key_;
int32_t version_;
Slice data_;
private:
char space_[200];
char* start_;
Slice key_;
int32_t version_;
Slice data_;
};

class ParsedBaseDataKey {
public:
explicit ParsedBaseDataKey(const std::string* key) {
const char* ptr = key->data();
int32_t key_len = DecodeFixed32(ptr);
ptr += sizeof(int32_t);
key_ = Slice(ptr, key_len);
ptr += key_len;
version_ = DecodeFixed32(ptr);
ptr += sizeof(int32_t);
data_ = Slice(ptr, key->size() - key_len - sizeof(int32_t) * 2);
}
public:
explicit ParsedBaseDataKey(const std::string* key) {
const char* ptr = key->data();
int32_t key_len = DecodeFixed32(ptr);
ptr += sizeof(int32_t);
key_ = Slice(ptr, key_len);
ptr += key_len;
version_ = DecodeFixed32(ptr);
ptr += sizeof(int32_t);
data_ = Slice(ptr, key->size() - key_len - sizeof(int32_t) * 2);
}

explicit ParsedBaseDataKey(const Slice& key) {
const char* ptr = key.data();
int32_t key_len = DecodeFixed32(ptr);
ptr += sizeof(int32_t);
key_ = Slice(ptr, key_len);
ptr += key_len;
version_ = DecodeFixed32(ptr);
ptr += sizeof(int32_t);
data_ = Slice(ptr, key.size() - key_len - sizeof(int32_t) * 2);
}
explicit ParsedBaseDataKey(const Slice& key) {
const char* ptr = key.data();
int32_t key_len = DecodeFixed32(ptr);
ptr += sizeof(int32_t);
key_ = Slice(ptr, key_len);
ptr += key_len;
version_ = DecodeFixed32(ptr);
ptr += sizeof(int32_t);
data_ = Slice(ptr, key.size() - key_len - sizeof(int32_t) * 2);
}

virtual ~ParsedBaseDataKey() = default;
virtual ~ParsedBaseDataKey() = default;

Slice key() {
return key_;
}
Slice key() {
return key_;
}

int32_t version() {
return version_;
}
int32_t version() {
return version_;
}

Slice data() {
return data_;
}
Slice data() {
return data_;
}

protected:
Slice key_;
int32_t version_;
Slice data_;
protected:
Slice key_;
int32_t version_;
Slice data_;
};

class ParsedHashesDataKey : public ParsedBaseDataKey {
public:
explicit ParsedHashesDataKey(const std::string* key)
public:
explicit ParsedHashesDataKey(const std::string* key)
: ParsedBaseDataKey(key) {}
explicit ParsedHashesDataKey(const Slice& key)
explicit ParsedHashesDataKey(const Slice& key)
: ParsedBaseDataKey(key) {}
Slice field() {
return data_;
}
Slice field() {
return data_;
}
};

class ParsedSetsMemberKey : public ParsedBaseDataKey {
public:
explicit ParsedSetsMemberKey(const std::string* key)
public:
explicit ParsedSetsMemberKey(const std::string* key)
: ParsedBaseDataKey(key) {}
explicit ParsedSetsMemberKey(const Slice& key)
explicit ParsedSetsMemberKey(const Slice& key)
: ParsedBaseDataKey(key) {}
Slice member() {
return data_;
}
Slice member() {
return data_;
}
};

class ParsedZSetsMemberKey : public ParsedBaseDataKey {
public:
explicit ParsedZSetsMemberKey(const std::string* key)
public:
explicit ParsedZSetsMemberKey(const std::string* key)
: ParsedBaseDataKey(key) {}
explicit ParsedZSetsMemberKey(const Slice& key)
explicit ParsedZSetsMemberKey(const Slice& key)
: ParsedBaseDataKey(key) {}
Slice member() {
return data_;
}
Slice member() {
return data_;
}
};

typedef BaseDataKey HashesDataKey;
Expand Down
Loading

0 comments on commit 3715974

Please sign in to comment.