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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
|
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab
/*
* Ceph - scalable distributed file system
*
* Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
*
* This is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License version 2.1, as published by the Free Software
* Foundation. See file COPYING.
*
*/
#ifndef CEPH_MOSDOPREPLY_H
#define CEPH_MOSDOPREPLY_H
#include "msg/Message.h"
#include "MOSDOp.h"
#include "os/ObjectStore.h"
/*
* OSD op reply
*
* oid - object id
* op - OSD_OP_DELETE, etc.
*
*/
class MOSDOpReply : public Message {
object_t oid;
pg_t pgid;
vector<OSDOp> ops;
int64_t flags;
int32_t result;
eversion_t reassert_version;
epoch_t osdmap_epoch;
int32_t retry_attempt;
public:
object_t get_oid() const { return oid; }
pg_t get_pg() const { return pgid; }
int get_flags() const { return flags; }
bool is_ondisk() const { return get_flags() & CEPH_OSD_FLAG_ONDISK; }
bool is_onnvram() const { return get_flags() & CEPH_OSD_FLAG_ONNVRAM; }
int get_result() const { return result; }
eversion_t get_version() { return reassert_version; }
bool may_read() const { return flags & CEPH_OSD_FLAG_READ; }
bool may_write() const { return flags & CEPH_OSD_FLAG_WRITE; }
void set_result(int r) { result = r; }
void set_version(eversion_t v) { reassert_version = v; }
void add_flags(int f) { flags |= f; }
void claim_op_out_data(vector<OSDOp>& o) {
assert(ops.size() == o.size());
for (unsigned i = 0; i < o.size(); i++) {
ops[i].outdata.claim(o[i].outdata);
}
}
void claim_ops(vector<OSDOp>& o) {
o.swap(ops);
}
/**
* get retry attempt
*
* If we don't know the attempt (because the server is old), return -1.
*/
int get_retry_attempt() const {
return retry_attempt;
}
// osdmap
epoch_t get_map_epoch() { return osdmap_epoch; }
/*osd_reqid_t get_reqid() { return osd_reqid_t(get_dest(),
head.client_inc,
head.tid); }
*/
public:
MOSDOpReply(MOSDOp *req, int r, epoch_t e, int acktype) :
Message(CEPH_MSG_OSD_OPREPLY) {
set_tid(req->get_tid());
ops = req->ops;
result = r;
flags =
(req->flags & ~(CEPH_OSD_FLAG_ONDISK|CEPH_OSD_FLAG_ONNVRAM|CEPH_OSD_FLAG_ACK)) | acktype;
oid = req->oid;
pgid = req->pgid;
osdmap_epoch = e;
reassert_version = req->reassert_version;
retry_attempt = req->get_retry_attempt();
// zero out ops payload_len
for (unsigned i = 0; i < ops.size(); i++)
ops[i].op.payload_len = 0;
}
MOSDOpReply() {}
private:
~MOSDOpReply() {}
public:
virtual void encode_payload(uint64_t features) {
OSDOp::merge_osd_op_vector_out_data(ops, data);
if ((features & CEPH_FEATURE_PGID64) == 0) {
ceph_osd_reply_head head;
memset(&head, 0, sizeof(head));
head.layout.ol_pgid = pgid.get_old_pg().v;
head.flags = flags;
head.osdmap_epoch = osdmap_epoch;
head.result = result;
head.num_ops = ops.size();
head.object_len = oid.name.length();
::encode(head, payload);
for (unsigned i = 0; i < head.num_ops; i++) {
::encode(ops[i].op, payload);
}
::encode_nohead(oid.name, payload);
} else {
header.version = 4;
::encode(oid, payload);
::encode(pgid, payload);
::encode(flags, payload);
::encode(result, payload);
::encode(reassert_version, payload);
::encode(osdmap_epoch, payload);
__u32 num_ops = ops.size();
::encode(num_ops, payload);
for (unsigned i = 0; i < num_ops; i++)
::encode(ops[i].op, payload);
::encode(retry_attempt, payload);
for (unsigned i = 0; i < num_ops; i++)
::encode(ops[i].rval, payload);
}
}
virtual void decode_payload() {
bufferlist::iterator p = payload.begin();
if (header.version < 2) {
ceph_osd_reply_head head;
::decode(head, p);
ops.resize(head.num_ops);
for (unsigned i = 0; i < head.num_ops; i++) {
::decode(ops[i].op, p);
}
::decode_nohead(head.object_len, oid.name, p);
pgid = pg_t(head.layout.ol_pgid);
result = head.result;
flags = head.flags;
reassert_version = head.reassert_version;
osdmap_epoch = head.osdmap_epoch;
retry_attempt = -1;
} else {
::decode(oid, p);
::decode(pgid, p);
::decode(flags, p);
::decode(result, p);
::decode(reassert_version, p);
::decode(osdmap_epoch, p);
__u32 num_ops = ops.size();
::decode(num_ops, p);
ops.resize(num_ops);
for (unsigned i = 0; i < num_ops; i++)
::decode(ops[i].op, p);
if (header.version >= 3)
::decode(retry_attempt, p);
else
retry_attempt = -1;
if (header.version >= 4) {
for (unsigned i = 0; i < num_ops; ++i)
::decode(ops[i].rval, p);
OSDOp::split_osd_op_vector_out_data(ops, data);
}
}
}
const char *get_type_name() const { return "osd_op_reply"; }
void print(ostream& out) const {
out << "osd_op_reply(" << get_tid()
<< " " << oid << " " << ops;
if (may_write()) {
if (is_ondisk())
out << " ondisk";
else if (is_onnvram())
out << " onnvram";
else
out << " ack";
}
out << " = " << get_result();
if (get_result() < 0) {
char buf[80];
out << " (" << strerror_r(-get_result(), buf, sizeof(buf)) << ")";
}
out << ")";
}
};
#endif
|
