101

从零开始使用 ToplingDB

示例代码(省略了错误处理,完整版在这里

从 RocksDB 迁移到 ToplingDB,只需要修改 Open/Close 相关的代码,其它代码不用任何改动。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
#include <topling/side_plugin_repo.h>
#include <rocksdb/db.h>
int main(int argc, char* argv[]) {
using namespace rocksdb;
SidePluginRepo repo;
repo.ImportAutoFile(argv[1]); // argv[1] is json/yaml conf file
DB* db = nullptr;
repo.OpenDB(&db);
repo.StartHttpServer();
if (strcasecmp(argv[2], "set") == 0) {
db->Put(WriteOptions(), argv[3], argv[4]);
}
else if (strcasecmp(argv[2], "get") == 0) {
std::string val;
db->Get(ReadOptions(), argv[3], &val);
fprintf(stdout, "%s\n", val.c_str());
}
fprintf(stderr, "now visit the web(defined in json/yaml conf file)\n");
fprintf(stderr, "press enter to exit\n");
getchar(); // wait for enter
repo.CloseAllDB();
return 0;
}

1. Open


SidePluginRepo 是 ToplingDB SidePlugin 的核心 class,所有组件都被包含在其中,要使用 SidePlugin,就要先定一个 SidePluginRepo 对象:

1
SidePluginRepo repo;

有了 SidePluginRepo 对象,接下来就是导入配置参数,最便捷的方式是调用 ImportAutoFile,该函数根据文件后缀名自动识别 json 和 yaml 文件:

1
repo.ImportAutoFile(argv[1]); // argv[1] is json/yaml conf file

现在,我们可以打开 DB 了:

1
2
DB* db = nullptr;
repo.OpenDB(&db);

在 RocksDB 中,我们一般使用 DB::Open,少数情况下使用其它 Open 函数,但是在 SidePlugin 中,我们统一用 OpenDB,具体使用的是哪个 Open 函数,是在 json/yaml 中进行配置的:例如 Todis从结点

在这个例子中我们使用的是默认的,只有一个 ColumnFamily 的 DB,但其实 OpenDB 是个重载的函数,也可以用来 Open 有多个 ColumnFamily 的 DB(后面修改的 db_bench 就是这样的例子):

1
2
DB_MultiCF* dbmcf = nullptr;
repo.OpenDB(&dbmcf); // overload function name

2. 启动内嵌的 Http Web Server

理论上,内嵌的 Http Web Server 可以在 Open 中自动启动,但是,我们仍然决定让用户来显式启动:

1
repo.StartHttpServer();

主要有三个原因:

  • 用户可以显式打开同一个 json/yaml 中定义的多个 DB,但内嵌的 Http 只有一个
  • 以 RocksDB 的接口风格,错误处理使用返回值 Status,难以使用同一个 Status 来区分,DB Open 与 Http 的错误信息
  • 即使 json/yaml 中定义了 http,用户也可能并不想启用内嵌的 Http Web Server

3. Close

需要显式 Close,就一行代码:

1
repo.CloseAllDB();

进一步参考:在 ToplingDB 中使用 CompactionFilter


复杂一点的:db_bench 针对 ToplingDB 的修改

db_bench_tool.cc 是一个 8000 多行的文件,我们让它支持 ToplingDB,只新增了 47 行代码,
无任何其它修改,就拥有了 ToplingDB 的全部能力!以下是 diff 内容:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
diff --git a/tools/db_bench_tool.cc b/tools/db_bench_tool.cc
index fcd3c157a..3a1f632a5 100644
--- a/tools/db_bench_tool.cc
+++ b/tools/db_bench_tool.cc
@@ -97,6 +97,8 @@
#include <io.h> // open/close
#endif

+#include "sideplugin/rockside/src/topling/side_plugin_repo.h"
+
using GFLAGS_NAMESPACE::ParseCommandLineFlags;
using GFLAGS_NAMESPACE::RegisterFlagValidator;
using GFLAGS_NAMESPACE::SetUsageMessage;
@@ -1052,6 +1054,7 @@ DEFINE_int32(trace_replay_threads, 1,
DEFINE_bool(io_uring_enabled, true,
"If true, enable the use of IO uring if the platform supports it");
extern "C" bool RocksDbIOUringEnable() { return FLAGS_io_uring_enabled; }
+DEFINE_string(json, "", "json config file.");
#endif // ROCKSDB_LITE

DEFINE_bool(adaptive_readahead, false,
@@ -3087,6 +3090,7 @@ class Benchmark {
}

void DeleteDBs() {
+ repo_.CloseAllDB(false);
db_.DeleteDBs();
for (const DBWithColumnFamilies& dbwcf : multi_dbs_) {
delete dbwcf.db;
@@ -3104,6 +3108,11 @@ class Benchmark {
}
}

+ void exit(int code) {
+ this->~Benchmark();
+ ::exit(code);
+ }
+
Slice AllocateKey(std::unique_ptr<const char[]>* key_guard) {
char* data = new char[key_size_];
const char* const_data = data;
@@ -3236,6 +3245,7 @@ class Benchmark {
ErrorExit();
}
Open(&open_options_);
+ open_options_ = db_.db->GetOptions();
PrintHeader(open_options_);
std::stringstream benchmark_stream(FLAGS_benchmarks);
std::string name;
@@ -4516,9 +4526,45 @@ class Benchmark {
InitializeOptionsGeneral(opts);
}

+ SidePluginRepo repo_;
void OpenDb(Options options, const std::string& db_name,
DBWithColumnFamilies* db) {
uint64_t open_start = FLAGS_report_open_timing ? FLAGS_env->NowNanos() : 0;
+ if (!FLAGS_json.empty()) {
+ repo_.CloseAllDB(false);
+ repo_.CleanResetRepo();
+ DB_MultiCF* dbmcf = nullptr;
+ Status s = repo_.ImportAutoFile(FLAGS_json);
+ if (!s.ok()) {
+ fprintf(stderr, "ERROR: ImportAutoFile(%s): %s\n",
+ FLAGS_json.c_str(), s.ToString().c_str());
+ exit(1);
+ }
+ s = repo_.OpenDB(&dbmcf);
+ if (!s.ok()) {
+ fprintf(stderr, "ERROR: OpenDB(): Config File=%s: %s\n",
+ FLAGS_json.c_str(), s.ToString().c_str());
+ exit(1);
+ }
+ s = repo_.StartHttpServer();
+ if (!s.ok()) {
+ fprintf(stderr, "ERROR: StartHttpServer(): JsonFile=%s: %s\n",
+ FLAGS_json.c_str(), s.ToString().c_str());
+ exit(1);
+ }
+ db->cfh = dbmcf->cf_handles;
+ db->db = dbmcf->db;
+ if (auto tdb = dynamic_cast<OptimisticTransactionDB*>(dbmcf->db)) {
+ db->opt_txn_db = tdb;
+ db->db = tdb->GetBaseDB();
+ }
+ db->num_created = FLAGS_num_column_families;
+ db->num_hot = FLAGS_num_column_families;
+ DBOptions dbo = db->db->GetDBOptions();
+ dbstats = dbo.statistics;
+ FLAGS_db = db->db->GetName();
+ return;
+ }
Status s;
// Open with column families if necessary.
if (FLAGS_num_column_families > 1) {