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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
|
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
#ifndef BENCHERH
#define BENCHERH
#include <utility>
#include "distribution.h"
#include "stat_collector.h"
#include "backend.h"
#include <boost/scoped_ptr.hpp>
#include "common/Mutex.h"
#include "common/Cond.h"
#include "common/Thread.h"
class OnWriteApplied;
class OnWriteCommit;
class OnReadComplete;
class Clenaup;
class Bencher : public Thread {
public:
enum OpType {
WRITE,
READ
};
private:
boost::scoped_ptr<
Distribution<boost::tuple<std::string,uint64_t,uint64_t, OpType> > > op_dist;
std::tr1::shared_ptr<StatCollector> stat_collector;
boost::scoped_ptr<Backend> backend;
const uint64_t max_in_flight;
const uint64_t max_duration;
const uint64_t max_ops;
Mutex lock;
Cond open_ops_cond;
uint64_t open_ops;
void start_op();
void drain_ops();
void complete_op();
public:
Bencher(
Distribution<boost::tuple<std::string, uint64_t, uint64_t, OpType> > *op_gen,
std::tr1::shared_ptr<StatCollector> stat_collector,
Backend *backend,
uint64_t max_in_flight,
uint64_t max_duration,
uint64_t max_ops) :
op_dist(op_gen),
stat_collector(stat_collector),
backend(backend),
max_in_flight(max_in_flight),
max_duration(max_duration),
max_ops(max_ops),
lock("Bencher::lock"),
open_ops(0)
{}
Bencher(
Distribution<boost::tuple<std::string, uint64_t, uint64_t, OpType> > *op_gen,
StatCollector *stat_collector,
Backend *backend,
uint64_t max_in_flight,
uint64_t max_duration,
uint64_t max_ops) :
op_dist(op_gen),
stat_collector(stat_collector),
backend(backend),
max_in_flight(max_in_flight),
max_duration(max_duration),
max_ops(max_ops),
lock("Bencher::lock"),
open_ops(0)
{}
Bencher(
Distribution<std::string> *object_gen,
Distribution<uint64_t> *offset_gen,
Distribution<uint64_t> *length_gen,
Distribution<OpType> *op_type_gen,
StatCollector *stat_collector,
Backend *backend,
uint64_t max_in_flight,
uint64_t max_duration,
uint64_t max_ops) :
op_dist(
new FourTupleDist<std::string, uint64_t, uint64_t, OpType>(
object_gen, offset_gen, length_gen, op_type_gen)),
stat_collector(stat_collector),
backend(backend),
max_in_flight(max_in_flight),
max_duration(max_duration),
max_ops(max_ops),
lock("Bencher::lock"),
open_ops(0)
{}
void init(
const set<std::string> &objects,
uint64_t size,
std::ostream *out
);
void run_bench();
void *entry() {
run_bench();
return 0;
}
friend class OnWriteApplied;
friend class OnWriteCommit;
friend class OnReadComplete;
friend class Cleanup;
};
class SequentialLoad :
public Distribution<
boost::tuple<string, uint64_t, uint64_t, Bencher::OpType> > {
set<string> objects;
uint64_t size;
uint64_t length;
set<string>::iterator object_pos;
uint64_t cur_pos;
boost::scoped_ptr<Distribution<Bencher::OpType> > op_dist;
SequentialLoad(const SequentialLoad &other);
public:
SequentialLoad(
const set<string> &_objects, uint64_t size,
uint64_t length,
Distribution<Bencher::OpType> *op_dist)
: objects(_objects), size(size), length(length),
object_pos(objects.begin()), cur_pos(0),
op_dist(op_dist) {}
boost::tuple<string, uint64_t, uint64_t, Bencher::OpType>
operator()() {
boost::tuple<string, uint64_t, uint64_t, Bencher::OpType> ret =
boost::make_tuple(*object_pos, cur_pos, length, (*op_dist)());
cur_pos += length;
if (cur_pos > size) {
cur_pos = 0;
object_pos++;
}
if (object_pos == objects.end())
object_pos = objects.begin();
return ret;
}
};
#endif
|
