Skip to content

Commit 0c2de65

Browse files
Wine93Wine93
committed
[feat][cache]: Refactor cache benchmarker.
1 parent b1d7682 commit 0c2de65

26 files changed

+760
-669
lines changed

‎src/cache/benchmark/README.md

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
cache benchmark
2+
===
3+
4+
Quick Start
5+
---
6+
7+
```bash
8+
cache-bench --flagfile bench.conf
9+
```
10+
11+
`bench.conf`:
12+
13+
```
14+
--threads=3
15+
--op=put
16+
--fsid=1
17+
--ino=1
18+
--blksize=4194304
19+
--blocks=100
20+
--writeback=false
21+
--retrive=true
22+
--async_batch=128
23+
--runtime=300
24+
--time_based=true
25+
26+
# Any other client flags here
27+
```
28+
29+
Output
30+
---
31+
32+
```
33+
libaio_read: (g=0): rw=read, bs=(R) 4096KiB-4096KiB, (W) 4096KiB-4096KiB, (T) 4096KiB-4096KiB, ioengine=libaio, iodepth=1
34+
...
35+
fio-3.28
36+
Starting 4 processes
37+
38+
...
39+
[10%] put: 584 op/s 2336 MB/s lat(0.013706 0.042489 0.002988)
40+
[11%] put: 563 op/s 2253 MB/s lat(0.014187 0.044417 0.003051)
41+
...
42+
```
43+
44+
The output shows the performance of the cache benchmark, including operations per second (op/s), throughput in megabytes per second (MB/s), and latency statistics which include average, maximum and minimum latency in seconds.

‎src/cache/benchmark/benchmarker.cpp

Lines changed: 85 additions & 83 deletions
Original file line numberDiff line numberDiff line change
@@ -24,139 +24,141 @@
2424

2525
#include <memory>
2626

27-
#include "base/time/time.h"
28-
#include "blockaccess/block_access_log.h"
27+
#include "cache/common/type.h"
2928
#include "cache/config/config.h"
29+
#include "cache/storage/storage_pool.h"
3030
#include "cache/tiercache/tier_block_cache.h"
31-
#include "cache/utils/access_log.h"
3231

3332
namespace dingofs {
3433
namespace cache {
3534

3635
Benchmarker::Benchmarker()
37-
: block_accesser_(std::make_unique<blockaccess::BlockAccesserImpl>(
38-
NewBlockAccessOptions())),
39-
block_cache_(std::make_shared<TierBlockCache>(
40-
BlockCacheOption(), RemoteBlockCacheOption(), block_accesser_.get())),
41-
task_pool_(std::make_unique<TaskThreadPool>("benchmarker")),
42-
countdown_event_(std::make_shared<BthreadCountdownEvent>(FLAGS_threads)),
43-
reporter_(std::make_shared<Reporter>()) {
44-
if (FLAGS_ino == 0) {
45-
FLAGS_ino = base::time::TimeNow().seconds;
46-
}
47-
}
48-
49-
Status Benchmarker::Run() {
50-
// Init logger, block cache, workers
51-
auto status = Init();
52-
if (!status.ok()) {
53-
return status;
54-
}
55-
56-
// Start reporter, workers
57-
status = Start();
58-
if (!status.ok()) {
59-
return status;
36+
: mds_base_(std::make_unique<stub::rpcclient::MDSBaseClient>()),
37+
mds_client_(std::make_shared<stub::rpcclient::MdsClientImpl>()),
38+
collector_(std::make_unique<Collector>()),
39+
reporter_(std::make_shared<Reporter>(collector_)),
40+
task_pool_(std::make_unique<TaskThreadPool>("benchmarker_worker")) {}
41+
42+
Status Benchmarker::Init() { return InitAll(); }
43+
44+
Status Benchmarker::InitAll() {
45+
auto initers = std::vector<std::function<Status()>>{
46+
[this]() { return InitMdsClient(); },
47+
[this]() { return InitStorage(); },
48+
[this]() { return InitBlockCache(); },
49+
[this]() { return InitCollector(); },
50+
[this]() { return InitReporter(); },
51+
[this]() {
52+
InitFactory();
53+
return Status::OK();
54+
},
55+
[this]() {
56+
InitWorkers();
57+
return Status::OK();
58+
}};
59+
60+
for (const auto& initer : initers) {
61+
auto status = initer();
62+
if (!status.ok()) {
63+
return status;
64+
}
6065
}
6166

6267
return Status::OK();
6368
}
6469

65-
void Benchmarker::Shutdown() {
66-
// Wait for all workers to complete
67-
countdown_event_->wait();
70+
void Benchmarker::RunUntilFinish() {
71+
StartAll();
72+
StopAll();
73+
}
6874

69-
LOG(INFO) << "All workers completed, shutting down...";
75+
void Benchmarker::StartAll() {
76+
StartReporter();
77+
StartWorkers();
78+
}
7079

71-
// stop worker, reporter
72-
Stop();
80+
void Benchmarker::StopAll() {
81+
StopWorkers();
82+
StopReporter();
83+
StopCollector();
7384
}
7485

75-
Status Benchmarker::Init() {
76-
auto status = InitBlockCache();
77-
if (!status.ok()) {
78-
LOG(ERROR) << "Init block cache failed: " << status.ToString();
79-
return status;
86+
// init
87+
Status Benchmarker::InitMdsClient() {
88+
auto rc = mds_client_->Init(NewMdsOption(), mds_base_.get());
89+
if (rc != PBFSStatusCode::OK) {
90+
return Status::Internal("init mds client failed");
8091
}
92+
return Status::OK();
93+
}
8194

82-
status = InitWrokers();
95+
Status Benchmarker::InitStorage() {
96+
auto storage_pool = StoragePoolImpl(mds_client_);
97+
auto status = storage_pool.GetStorage(FLAGS_fsid, storage_);
8398
if (!status.ok()) {
84-
LOG(ERROR) << "Init workers failed: " << status.ToString();
99+
LOG(ERROR) << "Init storage failed: " << status.ToString();
85100
return status;
86101
}
87-
88102
return Status::OK();
89103
}
90104

91105
Status Benchmarker::InitBlockCache() {
92-
auto status = block_accesser_->Init();
106+
block_cache_ = std::make_shared<TierBlockCache>(
107+
BlockCacheOption(), RemoteBlockCacheOption(), storage_);
108+
109+
auto status = block_cache_->Init();
93110
if (!status.ok()) {
94-
return status;
111+
LOG(ERROR) << "Init block cache failed: " << status.ToString();
95112
}
96-
return block_cache_->Init();
113+
return status;
97114
}
98115

99-
Status Benchmarker::InitWrokers() {
100-
for (auto i = 0; i < FLAGS_threads; i++) {
101-
auto worker =
102-
std::make_shared<Worker>(i, block_cache_, reporter_, countdown_event_);
103-
auto status = worker->Init();
104-
if (!status.ok()) {
105-
return status;
106-
}
107-
108-
workers_.emplace_back(worker);
116+
Status Benchmarker::InitCollector() {
117+
auto status = collector_->Init();
118+
if (!status.ok()) {
119+
LOG(ERROR) << "Init collector failed: " << status.ToString();
109120
}
110-
return Status::OK();
121+
return status;
111122
}
112123

113-
Status Benchmarker::Start() {
114-
auto status = StartReporter();
124+
Status Benchmarker::InitReporter() {
125+
auto status = reporter_->Init();
115126
if (!status.ok()) {
116-
LOG(ERROR) << "Start reporter failed: " << status.ToString();
117-
return status;
127+
LOG(ERROR) << "Init reporter failed: " << status.ToString();
118128
}
129+
return status;
130+
}
119131

120-
StartWorkers();
132+
void Benchmarker::InitFactory() {
133+
factory_ = NewFactory(block_cache_, FLAGS_op);
134+
}
121135

122-
return Status::OK();
136+
void Benchmarker::InitWorkers() {
137+
CHECK_EQ(task_pool_->Start(FLAGS_threads), 0);
138+
for (auto i = 0; i < FLAGS_threads; i++) {
139+
workers_.emplace_back(std::make_unique<Worker>(i, factory_, collector_));
140+
}
123141
}
124142

125-
Status Benchmarker::StartReporter() { return reporter_->Start(); }
143+
// start
144+
void Benchmarker::StartReporter() { reporter_->Start(); }
126145

127146
void Benchmarker::StartWorkers() {
128-
CHECK_EQ(task_pool_->Start(FLAGS_threads), 0);
129-
130147
for (auto& worker : workers_) {
131-
task_pool_->Enqueue([worker]() { worker->Run(); });
148+
task_pool_->Enqueue([&worker]() { worker->Start(); });
132149
}
133150
}
134151

135-
void Benchmarker::Stop() {
136-
StopWorkers();
137-
StopReporter();
138-
StopBlockCache();
139-
}
140-
152+
// stop
141153
void Benchmarker::StopWorkers() {
142154
for (auto& worker : workers_) {
143-
worker->Shutdown();
155+
worker->Stop();
144156
}
145157
}
146158

147-
void Benchmarker::StopReporter() {
148-
auto status = reporter_->Stop();
149-
if (!status.ok()) {
150-
LOG(ERROR) << "Stop reporter failed: " << status.ToString();
151-
}
152-
}
159+
void Benchmarker::StopReporter() { reporter_->Stop(); }
153160

154-
void Benchmarker::StopBlockCache() {
155-
auto status = block_cache_->Shutdown();
156-
if (!status.ok()) {
157-
LOG(ERROR) << "Shutdown block cache failed: " << status.ToString();
158-
}
159-
}
161+
void Benchmarker::StopCollector() { collector_->Detory(); }
160162

161163
} // namespace cache
162164
} // namespace dingofs

‎src/cache/benchmark/benchmarker.h

Lines changed: 29 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -23,10 +23,13 @@
2323
#ifndef DINGOFS_SRC_CACHE_BENCHMARK_BENCHMARKER_H_
2424
#define DINGOFS_SRC_CACHE_BENCHMARK_BENCHMARKER_H_
2525

26-
#include "blockaccess/block_accesser.h"
26+
#include "cache/benchmark/factory.h"
2727
#include "cache/benchmark/reporter.h"
2828
#include "cache/benchmark/worker.h"
2929
#include "cache/blockcache/block_cache.h"
30+
#include "cache/common/common.h"
31+
#include "cache/storage/storage.h"
32+
#include "stub/rpcclient/mds_client.h"
3033

3134
namespace dingofs {
3235
namespace cache {
@@ -35,29 +38,42 @@ class Benchmarker {
3538
public:
3639
Benchmarker();
3740

38-
Status Run();
39-
void Shutdown();
41+
Status Init();
42+
43+
void RunUntilFinish();
4044

4145
private:
42-
Status Init();
46+
// Init
47+
Status InitAll();
48+
Status InitMdsClient();
49+
Status InitStorage();
4350
Status InitBlockCache();
44-
Status InitWrokers();
51+
Status InitCollector();
52+
Status InitReporter();
53+
void InitFactory();
54+
void InitWorkers();
4555

46-
Status Start();
47-
Status StartReporter();
56+
// Start
57+
void StartAll();
58+
void StartReporter();
4859
void StartWorkers();
4960

50-
void Stop();
61+
// Stop
62+
void StopAll();
5163
void StopWorkers();
5264
void StopReporter();
53-
void StopBlockCache();
65+
void StopCollector();
5466

55-
blockaccess::BlockAccesserUPtr block_accesser_;
67+
private:
68+
std::unique_ptr<stub::rpcclient::MDSBaseClient> mds_base_;
69+
std::shared_ptr<stub::rpcclient::MdsClient> mds_client_;
70+
StorageSPtr storage_;
5671
BlockCacheSPtr block_cache_;
57-
TaskThreadPoolUPtr task_pool_;
58-
std::vector<WorkerSPtr> workers_;
59-
BthreadCountdownEventSPtr countdown_event_;
72+
CollectorSPtr collector_;
6073
ReporterSPtr reporter_;
74+
TaskFactorySPtr factory_;
75+
std::vector<WorkerUPtr> workers_;
76+
TaskThreadPoolUPtr task_pool_;
6177
};
6278

6379
} // namespace cache

0 commit comments

Comments
 (0)