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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
|
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab
#include <errno.h>
#include <time.h>
#include <thread>
#include "common/debug.h"
#include "common/Cond.h"
#include "common/Finisher.h"
#include "common/ceph_mutex.h"
#include "include/ceph_assert.h"
#include "common/ceph_time.h"
#include "MemWriteback.h"
#define dout_context g_ceph_context
#define dout_subsys ceph_subsys_objectcacher
#undef dout_prefix
#define dout_prefix *_dout << "MemWriteback(" << this << ") "
class C_DelayRead : public Context {
MemWriteback *wb;
CephContext *m_cct;
Context *m_con;
ceph::timespan m_delay;
ceph::mutex *m_lock;
object_t m_oid;
uint64_t m_off;
uint64_t m_len;
bufferlist *m_bl;
public:
C_DelayRead(MemWriteback *mwb, CephContext *cct, Context *c, ceph::mutex *lock,
const object_t& oid, uint64_t off, uint64_t len, bufferlist *pbl,
uint64_t delay_ns=0)
: wb(mwb), m_cct(cct), m_con(c),
m_delay(delay_ns * std::chrono::nanoseconds(1)),
m_lock(lock), m_oid(oid), m_off(off), m_len(len), m_bl(pbl) {}
void finish(int r) override {
std::this_thread::sleep_for(m_delay);
std::lock_guard locker{*m_lock};
r = wb->read_object_data(m_oid, m_off, m_len, m_bl);
if (m_con)
m_con->complete(r);
}
};
class C_DelayWrite : public Context {
MemWriteback *wb;
CephContext *m_cct;
Context *m_con;
ceph::timespan m_delay;
ceph::mutex *m_lock;
object_t m_oid;
uint64_t m_off;
uint64_t m_len;
const bufferlist& m_bl;
public:
C_DelayWrite(MemWriteback *mwb, CephContext *cct, Context *c, ceph::mutex *lock,
const object_t& oid, uint64_t off, uint64_t len,
const bufferlist& bl, uint64_t delay_ns=0)
: wb(mwb), m_cct(cct), m_con(c),
m_delay(delay_ns * std::chrono::nanoseconds(1)),
m_lock(lock), m_oid(oid), m_off(off), m_len(len), m_bl(bl) {}
void finish(int r) override {
std::this_thread::sleep_for(m_delay);
std::lock_guard locker{*m_lock};
wb->write_object_data(m_oid, m_off, m_len, m_bl);
if (m_con)
m_con->complete(r);
}
};
MemWriteback::MemWriteback(CephContext *cct, ceph::mutex *lock, uint64_t delay_ns)
: m_cct(cct), m_lock(lock), m_delay_ns(delay_ns)
{
m_finisher = new Finisher(cct);
m_finisher->start();
}
MemWriteback::~MemWriteback()
{
m_finisher->stop();
delete m_finisher;
}
void MemWriteback::read(const object_t& oid, uint64_t object_no,
const object_locator_t& oloc,
uint64_t off, uint64_t len, snapid_t snapid,
bufferlist *pbl, uint64_t trunc_size,
__u32 trunc_seq, int op_flags,
const ZTracer::Trace &parent_trace,
Context *onfinish)
{
ceph_assert(snapid == CEPH_NOSNAP);
C_DelayRead *wrapper = new C_DelayRead(this, m_cct, onfinish, m_lock, oid,
off, len, pbl, m_delay_ns);
m_finisher->queue(wrapper, len);
}
ceph_tid_t MemWriteback::write(const object_t& oid,
const object_locator_t& oloc,
uint64_t off, uint64_t len,
const SnapContext& snapc,
const bufferlist &bl, ceph::real_time mtime,
uint64_t trunc_size, __u32 trunc_seq,
ceph_tid_t journal_tid,
const ZTracer::Trace &parent_trace,
Context *oncommit)
{
ceph_assert(snapc.seq == 0);
C_DelayWrite *wrapper = new C_DelayWrite(this, m_cct, oncommit, m_lock, oid,
off, len, bl, m_delay_ns);
m_finisher->queue(wrapper, 0);
return ++m_tid;
}
void MemWriteback::write_object_data(const object_t& oid, uint64_t off, uint64_t len,
const bufferlist& data_bl)
{
dout(1) << "writing " << oid << " " << off << "~" << len << dendl;
ceph_assert(len == data_bl.length());
bufferlist& obj_bl = object_data[oid];
bufferlist new_obj_bl;
// ensure size, or set it if new object
if (off + len > obj_bl.length()) {
obj_bl.append_zero(off + len - obj_bl.length());
}
// beginning
new_obj_bl.substr_of(obj_bl, 0, off);
// overwritten bit
new_obj_bl.append(data_bl);
// tail bit
bufferlist tmp;
tmp.substr_of(obj_bl, off+len, obj_bl.length()-(off+len));
new_obj_bl.append(tmp);
obj_bl.swap(new_obj_bl);
dout(1) << oid << " final size " << obj_bl.length() << dendl;
}
int MemWriteback::read_object_data(const object_t& oid, uint64_t off, uint64_t len,
bufferlist *data_bl)
{
dout(1) << "reading " << oid << " " << off << "~" << len << dendl;
auto obj_i = object_data.find(oid);
if (obj_i == object_data.end()) {
dout(1) << oid << "DNE!" << dendl;
return -ENOENT;
}
const bufferlist& obj_bl = obj_i->second;
dout(1) << "reading " << oid << " from total size " << obj_bl.length() << dendl;
uint64_t read_len = std::min(len, obj_bl.length()-off);
data_bl->substr_of(obj_bl, off, read_len);
return 0;
}
bool MemWriteback::may_copy_on_write(const object_t&, uint64_t, uint64_t,
snapid_t)
{
return false;
}
|
