projects / deliverable / linux.git / blame
? search:
summary | shortlog | log | commit | commitdiff | tree
blob | blame (incremental) | history | HEAD
libceph: clean up ceph_osd_new_request()
[deliverable/linux.git] / net / ceph / osd_client.c
CommitLineData
3d14c5d2 1#include <linux/ceph/ceph_debug.h>
f24e9980 2
3d14c5d2 3#include <linux/module.h>
f24e9980
SW		 4#include <linux/err.h>
5#include <linux/highmem.h>
6#include <linux/mm.h>
7#include <linux/pagemap.h>
8#include <linux/slab.h>
9#include <linux/uaccess.h>
68b4476b
YS		 10#ifdef CONFIG_BLOCK
11#include <linux/bio.h>
12#endif
f24e9980 13
3d14c5d2
YS		 14#include <linux/ceph/libceph.h>
15#include <linux/ceph/osd_client.h>
16#include <linux/ceph/messenger.h>
17#include <linux/ceph/decode.h>
18#include <linux/ceph/auth.h>
19#include <linux/ceph/pagelist.h>
f24e9980 20
c16e7869
SW		 21#define OSD_OP_FRONT_LEN 4096
22#define OSD_OPREPLY_FRONT_LEN 512
0d59ab81 23
9e32789f 24static const struct ceph_connection_operations osd_con_ops;
f24e9980 25
f9d25199 26static void __send_queued(struct ceph_osd_client *osdc);
6f6c7006 27static int __reset_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd);
a40c4f10
YS		 28static void __register_request(struct ceph_osd_client *osdc,
29 struct ceph_osd_request *req);
30static void __unregister_linger_request(struct ceph_osd_client *osdc,
31 struct ceph_osd_request *req);
56e925b6
SW		 32static void __send_request(struct ceph_osd_client *osdc,
33 struct ceph_osd_request *req);
f24e9980
SW		 34
35/*
36 * Implement client access to distributed object storage cluster.
37 *
38 * All data objects are stored within a cluster/cloud of OSDs, or
39 * "object storage devices." (Note that Ceph OSDs have _nothing_ to
40 * do with the T10 OSD extensions to SCSI.) Ceph OSDs are simply
41 * remote daemons serving up and coordinating consistent and safe
42 * access to storage.
43 *
44 * Cluster membership and the mapping of data objects onto storage devices
45 * are described by the osd map.
46 *
47 * We keep track of pending OSD requests (read, write), resubmit
48 * requests to different OSDs when the cluster topology/data layout
49 * change, or retry the affected requests when the communications
50 * channel with an OSD is reset.
51 */
52
53/*
54 * calculate the mapping of a file extent onto an object, and fill out the
55 * request accordingly. shorten extent as necessary if it crosses an
56 * object boundary.
57 *
58 * fill osd op in request message.
59 */
dbe0fc41 60static int calc_layout(struct ceph_file_layout *layout, u64 off, u64 *plen,
a19dadfb 61 u64 *objnum, u64 *objoff, u64 *objlen)
f24e9980 62{
60e56f13 63 u64 orig_len = *plen;
d63b77f4 64 int r;
f24e9980 65
60e56f13 66 /* object extent? */
75d1c941
AE		 67 r = ceph_calc_file_object_mapping(layout, off, orig_len, objnum,
68 objoff, objlen);
d63b77f4
SW		 69 if (r < 0)
70 return r;
75d1c941
AE		 71 if (*objlen < orig_len) {
72 *plen = *objlen;
60e56f13
AE		 73 dout(" skipping last %llu, final file extent %llu~%llu\n",
74 orig_len - *plen, off, *plen);
75 }
76
75d1c941 77 dout("calc_layout objnum=%llx %llu~%llu\n", *objnum, *objoff, *objlen);
f24e9980 78
3ff5f385 79 return 0;
f24e9980
SW		 80}
81
f24e9980
SW		 82/*
83 * requests
84 */
415e49a9 85void ceph_osdc_release_request(struct kref *kref)
f24e9980 86{
e0c59487 87 int num_pages;
415e49a9
SW		 88 struct ceph_osd_request *req = container_of(kref,
89 struct ceph_osd_request,
90 r_kref);
91
92 if (req->r_request)
93 ceph_msg_put(req->r_request);
0d59ab81 94 if (req->r_con_filling_msg) {
9cbb1d72
AE		 95 dout("%s revoking msg %p from con %p\n", __func__,
96 req->r_reply, req->r_con_filling_msg);
8921d114 97 ceph_msg_revoke_incoming(req->r_reply);
0d47766f 98 req->r_con_filling_msg->ops->put(req->r_con_filling_msg);
9cbb1d72 99 req->r_con_filling_msg = NULL;
350b1c32 100 }
ab8cb34a
AE		 101 if (req->r_reply)
102 ceph_msg_put(req->r_reply);
0fff87ec
AE		 103
104 if (req->r_data_in.type == CEPH_OSD_DATA_TYPE_PAGES &&
e0c59487
AE		 105 req->r_data_in.own_pages) {
106 num_pages = calc_pages_for((u64)req->r_data_in.alignment,
107 (u64)req->r_data_in.length);
108 ceph_release_page_vector(req->r_data_in.pages, num_pages);
109 }
0fff87ec 110 if (req->r_data_out.type == CEPH_OSD_DATA_TYPE_PAGES &&
e0c59487
AE		 111 req->r_data_out.own_pages) {
112 num_pages = calc_pages_for((u64)req->r_data_out.alignment,
113 (u64)req->r_data_out.length);
114 ceph_release_page_vector(req->r_data_out.pages, num_pages);
115 }
0fff87ec 116
415e49a9
SW		 117 ceph_put_snap_context(req->r_snapc);
118 if (req->r_mempool)
119 mempool_free(req, req->r_osdc->req_mempool);
120 else
121 kfree(req);
f24e9980 122}
3d14c5d2 123EXPORT_SYMBOL(ceph_osdc_release_request);
68b4476b 124
3499e8a5 125struct ceph_osd_request *ceph_osdc_alloc_request(struct ceph_osd_client *osdc,
f24e9980 126 struct ceph_snap_context *snapc,
1b83bef2 127 unsigned int num_ops,
3499e8a5 128 bool use_mempool,
54a54007 129 gfp_t gfp_flags)
f24e9980
SW		 130{
131 struct ceph_osd_request *req;
132 struct ceph_msg *msg;
1b83bef2
SW		 133 size_t msg_size;
134
135 msg_size = 4 + 4 + 8 + 8 + 4+8;
136 msg_size += 2 + 4 + 8 + 4 + 4; /* oloc */
137 msg_size += 1 + 8 + 4 + 4; /* pg_t */
138 msg_size += 4 + MAX_OBJ_NAME_SIZE;
139 msg_size += 2 + num_ops*sizeof(struct ceph_osd_op);
140 msg_size += 8; /* snapid */
141 msg_size += 8; /* snap_seq */
142 msg_size += 8 * (snapc ? snapc->num_snaps : 0); /* snaps */
143 msg_size += 4;
f24e9980
SW		 144
145 if (use_mempool) {
3499e8a5 146 req = mempool_alloc(osdc->req_mempool, gfp_flags);
f24e9980
SW		 147 memset(req, 0, sizeof(*req));
148 } else {
3499e8a5 149 req = kzalloc(sizeof(*req), gfp_flags);
f24e9980
SW		 150 }
151 if (req == NULL)
a79832f2 152 return NULL;
f24e9980 153
f24e9980
SW		 154 req->r_osdc = osdc;
155 req->r_mempool = use_mempool;
68b4476b 156
415e49a9 157 kref_init(&req->r_kref);
f24e9980
SW		 158 init_completion(&req->r_completion);
159 init_completion(&req->r_safe_completion);
a978fa20 160 RB_CLEAR_NODE(&req->r_node);
f24e9980 161 INIT_LIST_HEAD(&req->r_unsafe_item);
a40c4f10
YS		 162 INIT_LIST_HEAD(&req->r_linger_item);
163 INIT_LIST_HEAD(&req->r_linger_osd);
935b639a 164 INIT_LIST_HEAD(&req->r_req_lru_item);
cd43045c
SW		 165 INIT_LIST_HEAD(&req->r_osd_item);
166
c16e7869
SW		 167 /* create reply message */
168 if (use_mempool)
169 msg = ceph_msgpool_get(&osdc->msgpool_op_reply, 0);
170 else
171 msg = ceph_msg_new(CEPH_MSG_OSD_OPREPLY,
b61c2763 172 OSD_OPREPLY_FRONT_LEN, gfp_flags, true);
a79832f2 173 if (!msg) {
c16e7869 174 ceph_osdc_put_request(req);
a79832f2 175 return NULL;
c16e7869
SW		 176 }
177 req->r_reply = msg;
178
0fff87ec
AE		 179 req->r_data_in.type = CEPH_OSD_DATA_TYPE_NONE;
180 req->r_data_out.type = CEPH_OSD_DATA_TYPE_NONE;
d50b409f 181
c16e7869 182 /* create request message; allow space for oid */
f24e9980 183 if (use_mempool)
8f3bc053 184 msg = ceph_msgpool_get(&osdc->msgpool_op, 0);
f24e9980 185 else
b61c2763 186 msg = ceph_msg_new(CEPH_MSG_OSD_OP, msg_size, gfp_flags, true);
a79832f2 187 if (!msg) {
f24e9980 188 ceph_osdc_put_request(req);
a79832f2 189 return NULL;
f24e9980 190 }
68b4476b 191
f24e9980 192 memset(msg->front.iov_base, 0, msg->front.iov_len);
3499e8a5
YS		 193
194 req->r_request = msg;
3499e8a5
YS		 195
196 return req;
197}
3d14c5d2 198EXPORT_SYMBOL(ceph_osdc_alloc_request);
3499e8a5 199
a8dd0a37 200static bool osd_req_opcode_valid(u16 opcode)
68b4476b 201{
a8dd0a37 202 switch (opcode) {
68b4476b 203 case CEPH_OSD_OP_READ:
a8dd0a37 204 case CEPH_OSD_OP_STAT:
4c46459c
AE		 205 case CEPH_OSD_OP_MAPEXT:
206 case CEPH_OSD_OP_MASKTRUNC:
207 case CEPH_OSD_OP_SPARSE_READ:
a9f36c3e 208 case CEPH_OSD_OP_NOTIFY:
a8dd0a37 209 case CEPH_OSD_OP_NOTIFY_ACK:
4c46459c 210 case CEPH_OSD_OP_ASSERT_VER:
a8dd0a37 211 case CEPH_OSD_OP_WRITE:
4c46459c
AE		 212 case CEPH_OSD_OP_WRITEFULL:
213 case CEPH_OSD_OP_TRUNCATE:
214 case CEPH_OSD_OP_ZERO:
215 case CEPH_OSD_OP_DELETE:
216 case CEPH_OSD_OP_APPEND:
a8dd0a37 217 case CEPH_OSD_OP_STARTSYNC:
4c46459c
AE		 218 case CEPH_OSD_OP_SETTRUNC:
219 case CEPH_OSD_OP_TRIMTRUNC:
220 case CEPH_OSD_OP_TMAPUP:
221 case CEPH_OSD_OP_TMAPPUT:
222 case CEPH_OSD_OP_TMAPGET:
223 case CEPH_OSD_OP_CREATE:
a9f36c3e 224 case CEPH_OSD_OP_ROLLBACK:
a8dd0a37 225 case CEPH_OSD_OP_WATCH:
4c46459c
AE		 226 case CEPH_OSD_OP_OMAPGETKEYS:
227 case CEPH_OSD_OP_OMAPGETVALS:
228 case CEPH_OSD_OP_OMAPGETHEADER:
229 case CEPH_OSD_OP_OMAPGETVALSBYKEYS:
4c46459c
AE		 230 case CEPH_OSD_OP_OMAPSETVALS:
231 case CEPH_OSD_OP_OMAPSETHEADER:
232 case CEPH_OSD_OP_OMAPCLEAR:
233 case CEPH_OSD_OP_OMAPRMKEYS:
234 case CEPH_OSD_OP_OMAP_CMP:
235 case CEPH_OSD_OP_CLONERANGE:
236 case CEPH_OSD_OP_ASSERT_SRC_VERSION:
237 case CEPH_OSD_OP_SRC_CMPXATTR:
a9f36c3e 238 case CEPH_OSD_OP_GETXATTR:
4c46459c 239 case CEPH_OSD_OP_GETXATTRS:
a9f36c3e
AE		 240 case CEPH_OSD_OP_CMPXATTR:
241 case CEPH_OSD_OP_SETXATTR:
4c46459c
AE		 242 case CEPH_OSD_OP_SETXATTRS:
243 case CEPH_OSD_OP_RESETXATTRS:
244 case CEPH_OSD_OP_RMXATTR:
245 case CEPH_OSD_OP_PULL:
246 case CEPH_OSD_OP_PUSH:
247 case CEPH_OSD_OP_BALANCEREADS:
248 case CEPH_OSD_OP_UNBALANCEREADS:
249 case CEPH_OSD_OP_SCRUB:
250 case CEPH_OSD_OP_SCRUB_RESERVE:
251 case CEPH_OSD_OP_SCRUB_UNRESERVE:
252 case CEPH_OSD_OP_SCRUB_STOP:
253 case CEPH_OSD_OP_SCRUB_MAP:
254 case CEPH_OSD_OP_WRLOCK:
255 case CEPH_OSD_OP_WRUNLOCK:
256 case CEPH_OSD_OP_RDLOCK:
257 case CEPH_OSD_OP_RDUNLOCK:
258 case CEPH_OSD_OP_UPLOCK:
259 case CEPH_OSD_OP_DNLOCK:
a8dd0a37 260 case CEPH_OSD_OP_CALL:
4c46459c
AE		 261 case CEPH_OSD_OP_PGLS:
262 case CEPH_OSD_OP_PGLS_FILTER:
a8dd0a37
AE		 263 return true;
264 default:
265 return false;
266 }
267}
268
33803f33
AE		 269/*
270 * This is an osd op init function for opcodes that have no data or
271 * other information associated with them. It also serves as a
272 * common init routine for all the other init functions, below.
273 */
274void osd_req_op_init(struct ceph_osd_req_op *op, u16 opcode)
275{
276 BUG_ON(!osd_req_opcode_valid(opcode));
277
278 memset(op, 0, sizeof (*op));
279
280 op->op = opcode;
281}
282
283void osd_req_op_extent_init(struct ceph_osd_req_op *op, u16 opcode,
284 u64 offset, u64 length,
285 u64 truncate_size, u32 truncate_seq)
286{
287 size_t payload_len = 0;
288
289 BUG_ON(opcode != CEPH_OSD_OP_READ && opcode != CEPH_OSD_OP_WRITE);
290
291 osd_req_op_init(op, opcode);
292
293 op->extent.offset = offset;
294 op->extent.length = length;
295 op->extent.truncate_size = truncate_size;
296 op->extent.truncate_seq = truncate_seq;
297 if (opcode == CEPH_OSD_OP_WRITE)
298 payload_len += length;
299
300 op->payload_len = payload_len;
301}
302EXPORT_SYMBOL(osd_req_op_extent_init);
303
304void osd_req_op_cls_init(struct ceph_osd_req_op *op, u16 opcode,
305 const char *class, const char *method,
306 const void *request_data, size_t request_data_size)
307{
308 size_t payload_len = 0;
309 size_t size;
310
311 BUG_ON(opcode != CEPH_OSD_OP_CALL);
312
313 osd_req_op_init(op, opcode);
314
315 op->cls.class_name = class;
316 size = strlen(class);
317 BUG_ON(size > (size_t) U8_MAX);
318 op->cls.class_len = size;
319 payload_len += size;
320
321 op->cls.method_name = method;
322 size = strlen(method);
323 BUG_ON(size > (size_t) U8_MAX);
324 op->cls.method_len = size;
325 payload_len += size;
326
327 op->cls.indata = request_data;
328 BUG_ON(request_data_size > (size_t) U32_MAX);
329 op->cls.indata_len = (u32) request_data_size;
330 payload_len += request_data_size;
331
332 op->cls.argc = 0; /* currently unused */
333
334 op->payload_len = payload_len;
335}
336EXPORT_SYMBOL(osd_req_op_cls_init);
337
338void osd_req_op_watch_init(struct ceph_osd_req_op *op, u16 opcode,
339 u64 cookie, u64 version, int flag)
340{
341 BUG_ON(opcode != CEPH_OSD_OP_NOTIFY_ACK && opcode != CEPH_OSD_OP_WATCH);
342
343 osd_req_op_init(op, opcode);
344
345 op->watch.cookie = cookie;
346 /* op->watch.ver = version; */ /* XXX 3847 */
347 op->watch.ver = cpu_to_le64(version);
348 if (opcode == CEPH_OSD_OP_WATCH && flag)
349 op->watch.flag = (u8) 1;
350}
351EXPORT_SYMBOL(osd_req_op_watch_init);
352
a8dd0a37
AE		 353static u64 osd_req_encode_op(struct ceph_osd_request *req,
354 struct ceph_osd_op *dst,
355 struct ceph_osd_req_op *src)
356{
357 u64 out_data_len = 0;
358 struct ceph_pagelist *pagelist;
359
360 if (WARN_ON(!osd_req_opcode_valid(src->op))) {
361 pr_err("unrecognized osd opcode %d\n", src->op);
362
363 return 0;
364 }
365
366 switch (src->op) {
367 case CEPH_OSD_OP_STAT:
368 break;
369 case CEPH_OSD_OP_READ:
370 case CEPH_OSD_OP_WRITE:
371 if (src->op == CEPH_OSD_OP_WRITE)
372 out_data_len = src->extent.length;
373 dst->extent.offset = cpu_to_le64(src->extent.offset);
374 dst->extent.length = cpu_to_le64(src->extent.length);
375 dst->extent.truncate_size =
376 cpu_to_le64(src->extent.truncate_size);
377 dst->extent.truncate_seq =
378 cpu_to_le32(src->extent.truncate_seq);
379 break;
380 case CEPH_OSD_OP_CALL:
381 pagelist = kmalloc(sizeof (*pagelist), GFP_NOFS);
382 BUG_ON(!pagelist);
383 ceph_pagelist_init(pagelist);
384
385 dst->cls.class_len = src->cls.class_len;
386 dst->cls.method_len = src->cls.method_len;
387 dst->cls.indata_len = cpu_to_le32(src->cls.indata_len);
388 ceph_pagelist_append(pagelist, src->cls.class_name,
389 src->cls.class_len);
390 ceph_pagelist_append(pagelist, src->cls.method_name,
391 src->cls.method_len);
392 ceph_pagelist_append(pagelist, src->cls.indata,
393 src->cls.indata_len);
394
395 req->r_data_out.type = CEPH_OSD_DATA_TYPE_PAGELIST;
396 req->r_data_out.pagelist = pagelist;
397 out_data_len = pagelist->length;
398 break;
399 case CEPH_OSD_OP_STARTSYNC:
400 break;
401 case CEPH_OSD_OP_NOTIFY_ACK:
402 case CEPH_OSD_OP_WATCH:
403 dst->watch.cookie = cpu_to_le64(src->watch.cookie);
404 dst->watch.ver = cpu_to_le64(src->watch.ver);
405 dst->watch.flag = src->watch.flag;
406 break;
407 default:
4c46459c 408 pr_err("unsupported osd opcode %s\n",
8f63ca2d 409 ceph_osd_op_name(src->op));
4c46459c 410 WARN_ON(1);
a8dd0a37
AE		 411
412 return 0;
68b4476b 413 }
a8dd0a37 414 dst->op = cpu_to_le16(src->op);
68b4476b 415 dst->payload_len = cpu_to_le32(src->payload_len);
175face2
AE		 416
417 return out_data_len;
68b4476b
YS		 418}
419
3499e8a5
YS		 420/*
421 * build new request AND message
422 *
423 */
424void ceph_osdc_build_request(struct ceph_osd_request *req,
175face2 425 u64 off, unsigned int num_ops,
68b4476b 426 struct ceph_osd_req_op *src_ops,
4d6b250b 427 struct ceph_snap_context *snapc, u64 snap_id,
af77f26c 428 struct timespec *mtime)
3499e8a5
YS		 429{
430 struct ceph_msg *msg = req->r_request;
68b4476b 431 struct ceph_osd_req_op *src_op;
3499e8a5 432 void *p;
1b83bef2 433 size_t msg_size;
3499e8a5 434 int flags = req->r_flags;
f44246e3 435 u64 data_len;
68b4476b 436 int i;
3499e8a5 437
1b83bef2
SW		 438 req->r_num_ops = num_ops;
439 req->r_snapid = snap_id;
f24e9980
SW		 440 req->r_snapc = ceph_get_snap_context(snapc);
441
1b83bef2
SW		 442 /* encode request */
443 msg->hdr.version = cpu_to_le16(4);
444
445 p = msg->front.iov_base;
446 ceph_encode_32(&p, 1); /* client_inc is always 1 */
447 req->r_request_osdmap_epoch = p;
448 p += 4;
449 req->r_request_flags = p;
450 p += 4;
451 if (req->r_flags & CEPH_OSD_FLAG_WRITE)
452 ceph_encode_timespec(p, mtime);
453 p += sizeof(struct ceph_timespec);
454 req->r_request_reassert_version = p;
455 p += sizeof(struct ceph_eversion); /* will get filled in */
456
457 /* oloc */
458 ceph_encode_8(&p, 4);
459 ceph_encode_8(&p, 4);
460 ceph_encode_32(&p, 8 + 4 + 4);
461 req->r_request_pool = p;
462 p += 8;
463 ceph_encode_32(&p, -1); /* preferred */
464 ceph_encode_32(&p, 0); /* key len */
465
466 ceph_encode_8(&p, 1);
467 req->r_request_pgid = p;
468 p += 8 + 4;
469 ceph_encode_32(&p, -1); /* preferred */
470
471 /* oid */
472 ceph_encode_32(&p, req->r_oid_len);
af77f26c 473 memcpy(p, req->r_oid, req->r_oid_len);
1b83bef2 474 dout("oid '%.*s' len %d\n", req->r_oid_len, req->r_oid, req->r_oid_len);
af77f26c 475 p += req->r_oid_len;
f24e9980 476
175face2 477 /* ops--can imply data */
1b83bef2 478 ceph_encode_16(&p, num_ops);
68b4476b 479 src_op = src_ops;
1b83bef2 480 req->r_request_ops = p;
175face2 481 data_len = 0;
1b83bef2 482 for (i = 0; i < num_ops; i++, src_op++) {
175face2 483 data_len += osd_req_encode_op(req, p, src_op);
1b83bef2
SW		 484 p += sizeof(struct ceph_osd_op);
485 }
68b4476b 486
1b83bef2
SW		 487 /* snaps */
488 ceph_encode_64(&p, req->r_snapid);
489 ceph_encode_64(&p, req->r_snapc ? req->r_snapc->seq : 0);
490 ceph_encode_32(&p, req->r_snapc ? req->r_snapc->num_snaps : 0);
491 if (req->r_snapc) {
f24e9980 492 for (i = 0; i < snapc->num_snaps; i++) {
1b83bef2 493 ceph_encode_64(&p, req->r_snapc->snaps[i]);
f24e9980
SW		 494 }
495 }
496
1b83bef2
SW		 497 req->r_request_attempts = p;
498 p += 4;
499
175face2 500 /* data */
0baa1bd9
AE		 501 if (flags & CEPH_OSD_FLAG_WRITE) {
502 u16 data_off;
503
504 /*
505 * The header "data_off" is a hint to the receiver
506 * allowing it to align received data into its
507 * buffers such that there's no need to re-copy
508 * it before writing it to disk (direct I/O).
509 */
510 data_off = (u16) (off & 0xffff);
511 req->r_request->hdr.data_off = cpu_to_le16(data_off);
512 }
f44246e3 513 req->r_request->hdr.data_len = cpu_to_le32(data_len);
c5c6b19d 514
f24e9980 515 BUG_ON(p > msg->front.iov_base + msg->front.iov_len);
6f863e71
SW		 516 msg_size = p - msg->front.iov_base;
517 msg->front.iov_len = msg_size;
518 msg->hdr.front_len = cpu_to_le32(msg_size);
1b83bef2
SW		 519
520 dout("build_request msg_size was %d num_ops %d\n", (int)msg_size,
521 num_ops);
3499e8a5
YS		 522 return;
523}
3d14c5d2 524EXPORT_SYMBOL(ceph_osdc_build_request);
3499e8a5
YS		 525
526/*
527 * build new request AND message, calculate layout, and adjust file
528 * extent as needed.
529 *
530 * if the file was recently truncated, we include information about its
531 * old and new size so that the object can be updated appropriately. (we
532 * avoid synchronously deleting truncated objects because it's slow.)
533 *
534 * if @do_sync, include a 'startsync' command so that the osd will flush
535 * data quickly.
536 */
537struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *osdc,
538 struct ceph_file_layout *layout,
539 struct ceph_vino vino,
540 u64 off, u64 *plen,
541 int opcode, int flags,
542 struct ceph_snap_context *snapc,
543 int do_sync,
544 u32 truncate_seq,
545 u64 truncate_size,
546 struct timespec *mtime,
153e5167 547 bool use_mempool)
3499e8a5 548{
ae7ca4a3 549 struct ceph_osd_req_op ops[2];
68b4476b 550 struct ceph_osd_request *req;
d18d1e28 551 unsigned int num_op = do_sync ? 2 : 1;
75d1c941
AE		 552 u64 objnum = 0;
553 u64 objoff = 0;
554 u64 objlen = 0;
d18d1e28
AE		 555 u32 object_size;
556 u64 object_base;
6816282d 557 int r;
68b4476b 558
d18d1e28 559 BUG_ON(opcode != CEPH_OSD_OP_READ && opcode != CEPH_OSD_OP_WRITE);
68b4476b 560
ae7ca4a3
AE		 561 req = ceph_osdc_alloc_request(osdc, snapc, num_op, use_mempool,
562 GFP_NOFS);
4ad12621 563 if (!req)
6816282d 564 return ERR_PTR(-ENOMEM);
d178a9e7 565 req->r_flags = flags;
3499e8a5
YS		 566
567 /* calculate max write size */
a19dadfb 568 r = calc_layout(layout, off, plen, &objnum, &objoff, &objlen);
3ff5f385
AE		 569 if (r < 0) {
570 ceph_osdc_put_request(req);
6816282d 571 return ERR_PTR(r);
3ff5f385 572 }
a19dadfb 573
d18d1e28
AE		 574 object_size = le32_to_cpu(layout->fl_object_size);
575 object_base = off - objoff;
576 if (truncate_size <= object_base) {
577 truncate_size = 0;
578 } else {
579 truncate_size -= object_base;
580 if (truncate_size > object_size)
581 truncate_size = object_size;
a19dadfb 582 }
d18d1e28
AE		 583
584 memset(&ops, 0, sizeof ops);
585 ops[0].op = opcode;
586 ops[0].extent.offset = objoff;
587 ops[0].extent.length = objlen;
588 ops[0].extent.truncate_size = truncate_size;
589 ops[0].extent.truncate_seq = truncate_seq;
a19dadfb
AE		 590 if (ops[0].op == CEPH_OSD_OP_WRITE)
591 ops[0].payload_len = *plen;
592
d18d1e28
AE		 593 if (do_sync)
594 ops[1].op = CEPH_OSD_OP_STARTSYNC;
595
3499e8a5
YS		 596 req->r_file_layout = *layout; /* keep a copy */
597
75d1c941
AE		 598 snprintf(req->r_oid, sizeof(req->r_oid), "%llx.%08llx",
599 vino.ino, objnum);
dbe0fc41
AE		 600 req->r_oid_len = strlen(req->r_oid);
601
175face2 602 ceph_osdc_build_request(req, off, num_op, ops,
ae7ca4a3 603 snapc, vino.snap, mtime);
3499e8a5 604
f24e9980
SW		 605 return req;
606}
3d14c5d2 607EXPORT_SYMBOL(ceph_osdc_new_request);
f24e9980
SW		 608
609/*
610 * We keep osd requests in an rbtree, sorted by ->r_tid.
611 */
612static void __insert_request(struct ceph_osd_client *osdc,
613 struct ceph_osd_request *new)
614{
615 struct rb_node **p = &osdc->requests.rb_node;
616 struct rb_node *parent = NULL;
617 struct ceph_osd_request *req = NULL;
618
619 while (*p) {
620 parent = *p;
621 req = rb_entry(parent, struct ceph_osd_request, r_node);
622 if (new->r_tid < req->r_tid)
623 p = &(*p)->rb_left;
624 else if (new->r_tid > req->r_tid)
625 p = &(*p)->rb_right;
626 else
627 BUG();
628 }
629
630 rb_link_node(&new->r_node, parent, p);
631 rb_insert_color(&new->r_node, &osdc->requests);
632}
633
634static struct ceph_osd_request *__lookup_request(struct ceph_osd_client *osdc,
635 u64 tid)
636{
637 struct ceph_osd_request *req;
638 struct rb_node *n = osdc->requests.rb_node;
639
640 while (n) {
641 req = rb_entry(n, struct ceph_osd_request, r_node);
642 if (tid < req->r_tid)
643 n = n->rb_left;
644 else if (tid > req->r_tid)
645 n = n->rb_right;
646 else
647 return req;
648 }
649 return NULL;
650}
651
652static struct ceph_osd_request *
653__lookup_request_ge(struct ceph_osd_client *osdc,
654 u64 tid)
655{
656 struct ceph_osd_request *req;
657 struct rb_node *n = osdc->requests.rb_node;
658
659 while (n) {
660 req = rb_entry(n, struct ceph_osd_request, r_node);
661 if (tid < req->r_tid) {
662 if (!n->rb_left)
663 return req;
664 n = n->rb_left;
665 } else if (tid > req->r_tid) {
666 n = n->rb_right;
667 } else {
668 return req;
669 }
670 }
671 return NULL;
672}
673
6f6c7006
SW		 674/*
675 * Resubmit requests pending on the given osd.
676 */
677static void __kick_osd_requests(struct ceph_osd_client *osdc,
678 struct ceph_osd *osd)
679{
a40c4f10 680 struct ceph_osd_request *req, *nreq;
e02493c0 681 LIST_HEAD(resend);
6f6c7006
SW		 682 int err;
683
684 dout("__kick_osd_requests osd%d\n", osd->o_osd);
685 err = __reset_osd(osdc, osd);
685a7555 686 if (err)
6f6c7006 687 return;
e02493c0
AE		 688 /*
689 * Build up a list of requests to resend by traversing the
690 * osd's list of requests. Requests for a given object are
691 * sent in tid order, and that is also the order they're
692 * kept on this list. Therefore all requests that are in
693 * flight will be found first, followed by all requests that
694 * have not yet been sent. And to resend requests while
695 * preserving this order we will want to put any sent
696 * requests back on the front of the osd client's unsent
697 * list.
698 *
699 * So we build a separate ordered list of already-sent
700 * requests for the affected osd and splice it onto the
701 * front of the osd client's unsent list. Once we've seen a
702 * request that has not yet been sent we're done. Those
703 * requests are already sitting right where they belong.
704 */
6f6c7006 705 list_for_each_entry(req, &osd->o_requests, r_osd_item) {
e02493c0
AE		 706 if (!req->r_sent)
707 break;
708 list_move_tail(&req->r_req_lru_item, &resend);
709 dout("requeueing %p tid %llu osd%d\n", req, req->r_tid,
6f6c7006 710 osd->o_osd);
a40c4f10
YS		 711 if (!req->r_linger)
712 req->r_flags |= CEPH_OSD_FLAG_RETRY;
713 }
e02493c0 714 list_splice(&resend, &osdc->req_unsent);
a40c4f10 715
e02493c0
AE		 716 /*
717 * Linger requests are re-registered before sending, which
718 * sets up a new tid for each. We add them to the unsent
719 * list at the end to keep things in tid order.
720 */
a40c4f10
YS		 721 list_for_each_entry_safe(req, nreq, &osd->o_linger_requests,
722 r_linger_osd) {
77f38e0e
SW		 723 /*
724 * reregister request prior to unregistering linger so
725 * that r_osd is preserved.
726 */
727 BUG_ON(!list_empty(&req->r_req_lru_item));
a40c4f10 728 __register_request(osdc, req);
e02493c0 729 list_add_tail(&req->r_req_lru_item, &osdc->req_unsent);
ad885927 730 list_add_tail(&req->r_osd_item, &req->r_osd->o_requests);
77f38e0e 731 __unregister_linger_request(osdc, req);
a40c4f10
YS		 732 dout("requeued lingering %p tid %llu osd%d\n", req, req->r_tid,
733 osd->o_osd);
6f6c7006
SW		 734 }
735}
736
f24e9980 737/*
81b024e7 738 * If the osd connection drops, we need to resubmit all requests.
f24e9980
SW		 739 */
740static void osd_reset(struct ceph_connection *con)
741{
742 struct ceph_osd *osd = con->private;
743 struct ceph_osd_client *osdc;
744
745 if (!osd)
746 return;
747 dout("osd_reset osd%d\n", osd->o_osd);
748 osdc = osd->o_osdc;
f24e9980 749 down_read(&osdc->map_sem);
83aff95e
SW		 750 mutex_lock(&osdc->request_mutex);
751 __kick_osd_requests(osdc, osd);
f9d25199 752 __send_queued(osdc);
83aff95e 753 mutex_unlock(&osdc->request_mutex);
f24e9980
SW		 754 up_read(&osdc->map_sem);
755}
756
757/*
758 * Track open sessions with osds.
759 */
e10006f8 760static struct ceph_osd *create_osd(struct ceph_osd_client *osdc, int onum)
f24e9980
SW		 761{
762 struct ceph_osd *osd;
763
764 osd = kzalloc(sizeof(*osd), GFP_NOFS);
765 if (!osd)
766 return NULL;
767
768 atomic_set(&osd->o_ref, 1);
769 osd->o_osdc = osdc;
e10006f8 770 osd->o_osd = onum;
f407731d 771 RB_CLEAR_NODE(&osd->o_node);
f24e9980 772 INIT_LIST_HEAD(&osd->o_requests);
a40c4f10 773 INIT_LIST_HEAD(&osd->o_linger_requests);
f5a2041b 774 INIT_LIST_HEAD(&osd->o_osd_lru);
f24e9980
SW		 775 osd->o_incarnation = 1;
776
b7a9e5dd 777 ceph_con_init(&osd->o_con, osd, &osd_con_ops, &osdc->client->msgr);
4e7a5dcd 778
422d2cb8 779 INIT_LIST_HEAD(&osd->o_keepalive_item);
f24e9980
SW		 780 return osd;
781}
782
783static struct ceph_osd *get_osd(struct ceph_osd *osd)
784{
785 if (atomic_inc_not_zero(&osd->o_ref)) {
786 dout("get_osd %p %d -> %d\n", osd, atomic_read(&osd->o_ref)-1,
787 atomic_read(&osd->o_ref));
788 return osd;
789 } else {
790 dout("get_osd %p FAIL\n", osd);
791 return NULL;
792 }
793}
794
795static void put_osd(struct ceph_osd *osd)
796{
797 dout("put_osd %p %d -> %d\n", osd, atomic_read(&osd->o_ref),
798 atomic_read(&osd->o_ref) - 1);
a255651d 799 if (atomic_dec_and_test(&osd->o_ref) && osd->o_auth.authorizer) {
79494d1b
SW		 800 struct ceph_auth_client *ac = osd->o_osdc->client->monc.auth;
801
27859f97 802 ceph_auth_destroy_authorizer(ac, osd->o_auth.authorizer);
f24e9980 803 kfree(osd);
79494d1b 804 }
f24e9980
SW		 805}
806
807/*
808 * remove an osd from our map
809 */
f5a2041b 810static void __remove_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd)
f24e9980 811{
f5a2041b 812 dout("__remove_osd %p\n", osd);
f24e9980
SW		 813 BUG_ON(!list_empty(&osd->o_requests));
814 rb_erase(&osd->o_node, &osdc->osds);
f5a2041b 815 list_del_init(&osd->o_osd_lru);
f24e9980
SW		 816 ceph_con_close(&osd->o_con);
817 put_osd(osd);
818}
819
aca420bc
SW		 820static void remove_all_osds(struct ceph_osd_client *osdc)
821{
048a9d2d 822 dout("%s %p\n", __func__, osdc);
aca420bc
SW		 823 mutex_lock(&osdc->request_mutex);
824 while (!RB_EMPTY_ROOT(&osdc->osds)) {
825 struct ceph_osd *osd = rb_entry(rb_first(&osdc->osds),
826 struct ceph_osd, o_node);
827 __remove_osd(osdc, osd);
828 }
829 mutex_unlock(&osdc->request_mutex);
830}
831
f5a2041b
YS		 832static void __move_osd_to_lru(struct ceph_osd_client *osdc,
833 struct ceph_osd *osd)
834{
835 dout("__move_osd_to_lru %p\n", osd);
836 BUG_ON(!list_empty(&osd->o_osd_lru));
837 list_add_tail(&osd->o_osd_lru, &osdc->osd_lru);
3d14c5d2 838 osd->lru_ttl = jiffies + osdc->client->options->osd_idle_ttl * HZ;
f5a2041b
YS		 839}
840
841static void __remove_osd_from_lru(struct ceph_osd *osd)
842{
843 dout("__remove_osd_from_lru %p\n", osd);
844 if (!list_empty(&osd->o_osd_lru))
845 list_del_init(&osd->o_osd_lru);
846}
847
aca420bc 848static void remove_old_osds(struct ceph_osd_client *osdc)
f5a2041b
YS		 849{
850 struct ceph_osd *osd, *nosd;
851
852 dout("__remove_old_osds %p\n", osdc);
853 mutex_lock(&osdc->request_mutex);
854 list_for_each_entry_safe(osd, nosd, &osdc->osd_lru, o_osd_lru) {
aca420bc 855 if (time_before(jiffies, osd->lru_ttl))
f5a2041b
YS		 856 break;
857 __remove_osd(osdc, osd);
858 }
859 mutex_unlock(&osdc->request_mutex);
860}
861
f24e9980
SW		 862/*
863 * reset osd connect
864 */
f5a2041b 865static int __reset_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd)
f24e9980 866{
c3acb181 867 struct ceph_entity_addr *peer_addr;
f24e9980 868
f5a2041b 869 dout("__reset_osd %p osd%d\n", osd, osd->o_osd);
a40c4f10
YS		 870 if (list_empty(&osd->o_requests) &&
871 list_empty(&osd->o_linger_requests)) {
f5a2041b 872 __remove_osd(osdc, osd);
c3acb181
AE		 873
874 return -ENODEV;
875 }
876
877 peer_addr = &osdc->osdmap->osd_addr[osd->o_osd];
878 if (!memcmp(peer_addr, &osd->o_con.peer_addr, sizeof (*peer_addr)) &&
879 !ceph_con_opened(&osd->o_con)) {
880 struct ceph_osd_request *req;
881
87b315a5
SW		 882 dout(" osd addr hasn't changed and connection never opened,"
883 " letting msgr retry");
884 /* touch each r_stamp for handle_timeout()'s benfit */
885 list_for_each_entry(req, &osd->o_requests, r_osd_item)
886 req->r_stamp = jiffies;
c3acb181
AE		 887
888 return -EAGAIN;
f24e9980 889 }
c3acb181
AE		 890
891 ceph_con_close(&osd->o_con);
892 ceph_con_open(&osd->o_con, CEPH_ENTITY_TYPE_OSD, osd->o_osd, peer_addr);
893 osd->o_incarnation++;
894
895 return 0;
f24e9980
SW		 896}
897
898static void __insert_osd(struct ceph_osd_client *osdc, struct ceph_osd *new)
899{
900 struct rb_node **p = &osdc->osds.rb_node;
901 struct rb_node *parent = NULL;
902 struct ceph_osd *osd = NULL;
903
aca420bc 904 dout("__insert_osd %p osd%d\n", new, new->o_osd);
f24e9980
SW		 905 while (*p) {
906 parent = *p;
907 osd = rb_entry(parent, struct ceph_osd, o_node);
908 if (new->o_osd < osd->o_osd)
909 p = &(*p)->rb_left;
910 else if (new->o_osd > osd->o_osd)
911 p = &(*p)->rb_right;
912 else
913 BUG();
914 }
915
916 rb_link_node(&new->o_node, parent, p);
917 rb_insert_color(&new->o_node, &osdc->osds);
918}
919
920static struct ceph_osd *__lookup_osd(struct ceph_osd_client *osdc, int o)
921{
922 struct ceph_osd *osd;
923 struct rb_node *n = osdc->osds.rb_node;
924
925 while (n) {
926 osd = rb_entry(n, struct ceph_osd, o_node);
927 if (o < osd->o_osd)
928 n = n->rb_left;
929 else if (o > osd->o_osd)
930 n = n->rb_right;
931 else
932 return osd;
933 }
934 return NULL;
935}
936
422d2cb8
YS		 937static void __schedule_osd_timeout(struct ceph_osd_client *osdc)
938{
939 schedule_delayed_work(&osdc->timeout_work,
3d14c5d2 940 osdc->client->options->osd_keepalive_timeout * HZ);
422d2cb8
YS		 941}
942
943static void __cancel_osd_timeout(struct ceph_osd_client *osdc)
944{
945 cancel_delayed_work(&osdc->timeout_work);
946}
f24e9980
SW		 947
948/*
949 * Register request, assign tid. If this is the first request, set up
950 * the timeout event.
951 */
a40c4f10
YS		 952static void __register_request(struct ceph_osd_client *osdc,
953 struct ceph_osd_request *req)
f24e9980 954{
f24e9980 955 req->r_tid = ++osdc->last_tid;
6df058c0 956 req->r_request->hdr.tid = cpu_to_le64(req->r_tid);
77f38e0e 957 dout("__register_request %p tid %lld\n", req, req->r_tid);
f24e9980
SW		 958 __insert_request(osdc, req);
959 ceph_osdc_get_request(req);
960 osdc->num_requests++;
f24e9980 961 if (osdc->num_requests == 1) {
422d2cb8
YS		 962 dout(" first request, scheduling timeout\n");
963 __schedule_osd_timeout(osdc);
f24e9980 964 }
a40c4f10
YS		 965}
966
f24e9980
SW		 967/*
968 * called under osdc->request_mutex
969 */
970static void __unregister_request(struct ceph_osd_client *osdc,
971 struct ceph_osd_request *req)
972{
35f9f8a0
SW		 973 if (RB_EMPTY_NODE(&req->r_node)) {
974 dout("__unregister_request %p tid %lld not registered\n",
975 req, req->r_tid);
976 return;
977 }
978
f24e9980
SW		 979 dout("__unregister_request %p tid %lld\n", req, req->r_tid);
980 rb_erase(&req->r_node, &osdc->requests);
981 osdc->num_requests--;
982
0ba6478d
SW		 983 if (req->r_osd) {
984 /* make sure the original request isn't in flight. */
6740a845 985 ceph_msg_revoke(req->r_request);
0ba6478d
SW		 986
987 list_del_init(&req->r_osd_item);
a40c4f10
YS		 988 if (list_empty(&req->r_osd->o_requests) &&
989 list_empty(&req->r_osd->o_linger_requests)) {
990 dout("moving osd to %p lru\n", req->r_osd);
f5a2041b 991 __move_osd_to_lru(osdc, req->r_osd);
a40c4f10 992 }
fbdb9190 993 if (list_empty(&req->r_linger_item))
a40c4f10 994 req->r_osd = NULL;
0ba6478d 995 }
f24e9980 996
7d5f2481 997 list_del_init(&req->r_req_lru_item);
f24e9980
SW		 998 ceph_osdc_put_request(req);
999
422d2cb8
YS		 1000 if (osdc->num_requests == 0) {
1001 dout(" no requests, canceling timeout\n");
1002 __cancel_osd_timeout(osdc);
f24e9980
SW		 1003 }
1004}
1005
1006/*
1007 * Cancel a previously queued request message
1008 */
1009static void __cancel_request(struct ceph_osd_request *req)
1010{
6bc18876 1011 if (req->r_sent && req->r_osd) {
6740a845 1012 ceph_msg_revoke(req->r_request);
f24e9980
SW		 1013 req->r_sent = 0;
1014 }
1015}
1016
a40c4f10
YS		 1017static void __register_linger_request(struct ceph_osd_client *osdc,
1018 struct ceph_osd_request *req)
1019{
1020 dout("__register_linger_request %p\n", req);
1021 list_add_tail(&req->r_linger_item, &osdc->req_linger);
6194ea89
SW		 1022 if (req->r_osd)
1023 list_add_tail(&req->r_linger_osd,
1024 &req->r_osd->o_linger_requests);
a40c4f10
YS		 1025}
1026
1027static void __unregister_linger_request(struct ceph_osd_client *osdc,
1028 struct ceph_osd_request *req)
1029{
1030 dout("__unregister_linger_request %p\n", req);
61c74035 1031 list_del_init(&req->r_linger_item);
a40c4f10 1032 if (req->r_osd) {
a40c4f10
YS		 1033 list_del_init(&req->r_linger_osd);
1034
1035 if (list_empty(&req->r_osd->o_requests) &&
1036 list_empty(&req->r_osd->o_linger_requests)) {
1037 dout("moving osd to %p lru\n", req->r_osd);
1038 __move_osd_to_lru(osdc, req->r_osd);
1039 }
fbdb9190
SW		 1040 if (list_empty(&req->r_osd_item))
1041 req->r_osd = NULL;
a40c4f10
YS		 1042 }
1043}
1044
1045void ceph_osdc_unregister_linger_request(struct ceph_osd_client *osdc,
1046 struct ceph_osd_request *req)
1047{
1048 mutex_lock(&osdc->request_mutex);
1049 if (req->r_linger) {
1050 __unregister_linger_request(osdc, req);
1051 ceph_osdc_put_request(req);
1052 }
1053 mutex_unlock(&osdc->request_mutex);
1054}
1055EXPORT_SYMBOL(ceph_osdc_unregister_linger_request);
1056
1057void ceph_osdc_set_request_linger(struct ceph_osd_client *osdc,
1058 struct ceph_osd_request *req)
1059{
1060 if (!req->r_linger) {
1061 dout("set_request_linger %p\n", req);
1062 req->r_linger = 1;
1063 /*
1064 * caller is now responsible for calling
1065 * unregister_linger_request
1066 */
1067 ceph_osdc_get_request(req);
1068 }
1069}
1070EXPORT_SYMBOL(ceph_osdc_set_request_linger);
1071
f24e9980
SW		 1072/*
1073 * Pick an osd (the first 'up' osd in the pg), allocate the osd struct
1074 * (as needed), and set the request r_osd appropriately. If there is
25985edc 1075 * no up osd, set r_osd to NULL. Move the request to the appropriate list
6f6c7006 1076 * (unsent, homeless) or leave on in-flight lru.
f24e9980
SW		 1077 *
1078 * Return 0 if unchanged, 1 if changed, or negative on error.
1079 *
1080 * Caller should hold map_sem for read and request_mutex.
1081 */
6f6c7006 1082static int __map_request(struct ceph_osd_client *osdc,
38d6453c 1083 struct ceph_osd_request *req, int force_resend)
f24e9980 1084{
5b191d99 1085 struct ceph_pg pgid;
d85b7056
SW		 1086 int acting[CEPH_PG_MAX_SIZE];
1087 int o = -1, num = 0;
f24e9980 1088 int err;
f24e9980 1089
6f6c7006 1090 dout("map_request %p tid %lld\n", req, req->r_tid);
41766f87
AE		 1091 err = ceph_calc_ceph_pg(&pgid, req->r_oid, osdc->osdmap,
1092 ceph_file_layout_pg_pool(req->r_file_layout));
6f6c7006
SW		 1093 if (err) {
1094 list_move(&req->r_req_lru_item, &osdc->req_notarget);
f24e9980 1095 return err;
6f6c7006 1096 }
7740a42f
SW		 1097 req->r_pgid = pgid;
1098
d85b7056
SW		 1099 err = ceph_calc_pg_acting(osdc->osdmap, pgid, acting);
1100 if (err > 0) {
1101 o = acting[0];
1102 num = err;
1103 }
f24e9980 1104
38d6453c
SW		 1105 if ((!force_resend &&
1106 req->r_osd && req->r_osd->o_osd == o &&
d85b7056
SW		 1107 req->r_sent >= req->r_osd->o_incarnation &&
1108 req->r_num_pg_osds == num &&
1109 memcmp(req->r_pg_osds, acting, sizeof(acting[0])*num) == 0) ||
f24e9980
SW		 1110 (req->r_osd == NULL && o == -1))
1111 return 0; /* no change */
1112
5b191d99
SW		 1113 dout("map_request tid %llu pgid %lld.%x osd%d (was osd%d)\n",
1114 req->r_tid, pgid.pool, pgid.seed, o,
f24e9980
SW		 1115 req->r_osd ? req->r_osd->o_osd : -1);
1116
d85b7056
SW		 1117 /* record full pg acting set */
1118 memcpy(req->r_pg_osds, acting, sizeof(acting[0]) * num);
1119 req->r_num_pg_osds = num;
1120
f24e9980
SW		 1121 if (req->r_osd) {
1122 __cancel_request(req);
1123 list_del_init(&req->r_osd_item);
f24e9980
SW		 1124 req->r_osd = NULL;
1125 }
1126
1127 req->r_osd = __lookup_osd(osdc, o);
1128 if (!req->r_osd && o >= 0) {
c99eb1c7 1129 err = -ENOMEM;
e10006f8 1130 req->r_osd = create_osd(osdc, o);
6f6c7006
SW		 1131 if (!req->r_osd) {
1132 list_move(&req->r_req_lru_item, &osdc->req_notarget);
c99eb1c7 1133 goto out;
6f6c7006 1134 }
f24e9980 1135
6f6c7006 1136 dout("map_request osd %p is osd%d\n", req->r_osd, o);
f24e9980
SW		 1137 __insert_osd(osdc, req->r_osd);
1138
b7a9e5dd
SW		 1139 ceph_con_open(&req->r_osd->o_con,
1140 CEPH_ENTITY_TYPE_OSD, o,
1141 &osdc->osdmap->osd_addr[o]);
f24e9980
SW		 1142 }
1143
f5a2041b
YS		 1144 if (req->r_osd) {
1145 __remove_osd_from_lru(req->r_osd);
ad885927
AE		 1146 list_add_tail(&req->r_osd_item, &req->r_osd->o_requests);
1147 list_move_tail(&req->r_req_lru_item, &osdc->req_unsent);
6f6c7006 1148 } else {
ad885927 1149 list_move_tail(&req->r_req_lru_item, &osdc->req_notarget);
f5a2041b 1150 }
d85b7056 1151 err = 1; /* osd or pg changed */
f24e9980
SW		 1152
1153out:
f24e9980
SW		 1154 return err;
1155}
1156
1157/*
1158 * caller should hold map_sem (for read) and request_mutex
1159 */
56e925b6
SW		 1160static void __send_request(struct ceph_osd_client *osdc,
1161 struct ceph_osd_request *req)
f24e9980 1162{
1b83bef2 1163 void *p;
f24e9980 1164
1b83bef2
SW		 1165 dout("send_request %p tid %llu to osd%d flags %d pg %lld.%x\n",
1166 req, req->r_tid, req->r_osd->o_osd, req->r_flags,
1167 (unsigned long long)req->r_pgid.pool, req->r_pgid.seed);
1168
1169 /* fill in message content that changes each time we send it */
1170 put_unaligned_le32(osdc->osdmap->epoch, req->r_request_osdmap_epoch);
1171 put_unaligned_le32(req->r_flags, req->r_request_flags);
1172 put_unaligned_le64(req->r_pgid.pool, req->r_request_pool);
1173 p = req->r_request_pgid;
1174 ceph_encode_64(&p, req->r_pgid.pool);
1175 ceph_encode_32(&p, req->r_pgid.seed);
1176 put_unaligned_le64(1, req->r_request_attempts); /* FIXME */
1177 memcpy(req->r_request_reassert_version, &req->r_reassert_version,
1178 sizeof(req->r_reassert_version));
2169aea6 1179
3dd72fc0 1180 req->r_stamp = jiffies;
07a27e22 1181 list_move_tail(&req->r_req_lru_item, &osdc->req_lru);
f24e9980
SW		 1182
1183 ceph_msg_get(req->r_request); /* send consumes a ref */
1184 ceph_con_send(&req->r_osd->o_con, req->r_request);
1185 req->r_sent = req->r_osd->o_incarnation;
f24e9980
SW		 1186}
1187
6f6c7006
SW		 1188/*
1189 * Send any requests in the queue (req_unsent).
1190 */
f9d25199 1191static void __send_queued(struct ceph_osd_client *osdc)
6f6c7006
SW		 1192{
1193 struct ceph_osd_request *req, *tmp;
1194
f9d25199
AE		 1195 dout("__send_queued\n");
1196 list_for_each_entry_safe(req, tmp, &osdc->req_unsent, r_req_lru_item)
6f6c7006 1197 __send_request(osdc, req);
6f6c7006
SW		 1198}
1199
f24e9980
SW		 1200/*
1201 * Timeout callback, called every N seconds when 1 or more osd
1202 * requests has been active for more than N seconds. When this
1203 * happens, we ping all OSDs with requests who have timed out to
1204 * ensure any communications channel reset is detected. Reset the
1205 * request timeouts another N seconds in the future as we go.
1206 * Reschedule the timeout event another N seconds in future (unless
1207 * there are no open requests).
1208 */
1209static void handle_timeout(struct work_struct *work)
1210{
1211 struct ceph_osd_client *osdc =
1212 container_of(work, struct ceph_osd_client, timeout_work.work);
83aff95e 1213 struct ceph_osd_request *req;
f24e9980 1214 struct ceph_osd *osd;
422d2cb8 1215 unsigned long keepalive =
3d14c5d2 1216 osdc->client->options->osd_keepalive_timeout * HZ;
422d2cb8 1217 struct list_head slow_osds;
f24e9980
SW		 1218 dout("timeout\n");
1219 down_read(&osdc->map_sem);
1220
1221 ceph_monc_request_next_osdmap(&osdc->client->monc);
1222
1223 mutex_lock(&osdc->request_mutex);
f24e9980 1224
422d2cb8
YS		 1225 /*
1226 * ping osds that are a bit slow. this ensures that if there
1227 * is a break in the TCP connection we will notice, and reopen
1228 * a connection with that osd (from the fault callback).
1229 */
1230 INIT_LIST_HEAD(&slow_osds);
1231 list_for_each_entry(req, &osdc->req_lru, r_req_lru_item) {
3dd72fc0 1232 if (time_before(jiffies, req->r_stamp + keepalive))
422d2cb8
YS		 1233 break;
1234
1235 osd = req->r_osd;
1236 BUG_ON(!osd);
1237 dout(" tid %llu is slow, will send keepalive on osd%d\n",
f24e9980 1238 req->r_tid, osd->o_osd);
422d2cb8
YS		 1239 list_move_tail(&osd->o_keepalive_item, &slow_osds);
1240 }
1241 while (!list_empty(&slow_osds)) {
1242 osd = list_entry(slow_osds.next, struct ceph_osd,
1243 o_keepalive_item);
1244 list_del_init(&osd->o_keepalive_item);
f24e9980
SW		 1245 ceph_con_keepalive(&osd->o_con);
1246 }
1247
422d2cb8 1248 __schedule_osd_timeout(osdc);
f9d25199 1249 __send_queued(osdc);
f24e9980 1250 mutex_unlock(&osdc->request_mutex);
f24e9980
SW		 1251 up_read(&osdc->map_sem);
1252}
1253
f5a2041b
YS		 1254static void handle_osds_timeout(struct work_struct *work)
1255{
1256 struct ceph_osd_client *osdc =
1257 container_of(work, struct ceph_osd_client,
1258 osds_timeout_work.work);
1259 unsigned long delay =
3d14c5d2 1260 osdc->client->options->osd_idle_ttl * HZ >> 2;
f5a2041b
YS		 1261
1262 dout("osds timeout\n");
1263 down_read(&osdc->map_sem);
aca420bc 1264 remove_old_osds(osdc);
f5a2041b
YS		 1265 up_read(&osdc->map_sem);
1266
1267 schedule_delayed_work(&osdc->osds_timeout_work,
1268 round_jiffies_relative(delay));
1269}
1270
25845472
SW		 1271static void complete_request(struct ceph_osd_request *req)
1272{
1273 if (req->r_safe_callback)
1274 req->r_safe_callback(req, NULL);
1275 complete_all(&req->r_safe_completion); /* fsync waiter */
1276}
1277
1b83bef2
SW		 1278static int __decode_pgid(void **p, void *end, struct ceph_pg *pgid)
1279{
1280 __u8 v;
1281
1282 ceph_decode_need(p, end, 1 + 8 + 4 + 4, bad);
1283 v = ceph_decode_8(p);
1284 if (v > 1) {
1285 pr_warning("do not understand pg encoding %d > 1", v);
1286 return -EINVAL;
1287 }
1288 pgid->pool = ceph_decode_64(p);
1289 pgid->seed = ceph_decode_32(p);
1290 *p += 4;
1291 return 0;
1292
1293bad:
1294 pr_warning("incomplete pg encoding");
1295 return -EINVAL;
1296}
1297
f24e9980
SW		 1298/*
1299 * handle osd op reply. either call the callback if it is specified,
1300 * or do the completion to wake up the waiting thread.
1301 */
350b1c32
SW		 1302static void handle_reply(struct ceph_osd_client *osdc, struct ceph_msg *msg,
1303 struct ceph_connection *con)
f24e9980 1304{
1b83bef2 1305 void *p, *end;
f24e9980
SW		 1306 struct ceph_osd_request *req;
1307 u64 tid;
1b83bef2
SW		 1308 int object_len;
1309 int numops, payload_len, flags;
0ceed5db 1310 s32 result;
1b83bef2
SW		 1311 s32 retry_attempt;
1312 struct ceph_pg pg;
1313 int err;
1314 u32 reassert_epoch;
1315 u64 reassert_version;
1316 u32 osdmap_epoch;
0d5af164 1317 int already_completed;
1b83bef2 1318 int i;
f24e9980 1319
6df058c0 1320 tid = le64_to_cpu(msg->hdr.tid);
1b83bef2
SW		 1321 dout("handle_reply %p tid %llu\n", msg, tid);
1322
1323 p = msg->front.iov_base;
1324 end = p + msg->front.iov_len;
1325
1326 ceph_decode_need(&p, end, 4, bad);
1327 object_len = ceph_decode_32(&p);
1328 ceph_decode_need(&p, end, object_len, bad);
1329 p += object_len;
1330
1331 err = __decode_pgid(&p, end, &pg);
1332 if (err)
f24e9980 1333 goto bad;
1b83bef2
SW		 1334
1335 ceph_decode_need(&p, end, 8 + 4 + 4 + 8 + 4, bad);
1336 flags = ceph_decode_64(&p);
1337 result = ceph_decode_32(&p);
1338 reassert_epoch = ceph_decode_32(&p);
1339 reassert_version = ceph_decode_64(&p);
1340 osdmap_epoch = ceph_decode_32(&p);
1341
f24e9980
SW		 1342 /* lookup */
1343 mutex_lock(&osdc->request_mutex);
1344 req = __lookup_request(osdc, tid);
1345 if (req == NULL) {
1346 dout("handle_reply tid %llu dne\n", tid);
1347 mutex_unlock(&osdc->request_mutex);
1348 return;
1349 }
1350 ceph_osdc_get_request(req);
1b83bef2
SW		 1351
1352 dout("handle_reply %p tid %llu req %p result %d\n", msg, tid,
1353 req, result);
1354
1355 ceph_decode_need(&p, end, 4, bad);
1356 numops = ceph_decode_32(&p);
1357 if (numops > CEPH_OSD_MAX_OP)
1358 goto bad_put;
1359 if (numops != req->r_num_ops)
1360 goto bad_put;
1361 payload_len = 0;
1362 ceph_decode_need(&p, end, numops * sizeof(struct ceph_osd_op), bad);
1363 for (i = 0; i < numops; i++) {
1364 struct ceph_osd_op *op = p;
1365 int len;
1366
1367 len = le32_to_cpu(op->payload_len);
1368 req->r_reply_op_len[i] = len;
1369 dout(" op %d has %d bytes\n", i, len);
1370 payload_len += len;
1371 p += sizeof(*op);
1372 }
1373 if (payload_len != le32_to_cpu(msg->hdr.data_len)) {
1374 pr_warning("sum of op payload lens %d != data_len %d",
1375 payload_len, le32_to_cpu(msg->hdr.data_len));
1376 goto bad_put;
1377 }
1378
1379 ceph_decode_need(&p, end, 4 + numops * 4, bad);
1380 retry_attempt = ceph_decode_32(&p);
1381 for (i = 0; i < numops; i++)
1382 req->r_reply_op_result[i] = ceph_decode_32(&p);
f24e9980 1383
350b1c32 1384 /*
0d59ab81 1385 * if this connection filled our message, drop our reference now, to
350b1c32
SW		 1386 * avoid a (safe but slower) revoke later.
1387 */
0d59ab81 1388 if (req->r_con_filling_msg == con && req->r_reply == msg) {
c16e7869 1389 dout(" dropping con_filling_msg ref %p\n", con);
0d59ab81 1390 req->r_con_filling_msg = NULL;
0d47766f 1391 con->ops->put(con);
350b1c32
SW		 1392 }
1393
f24e9980 1394 if (!req->r_got_reply) {
95c96174 1395 unsigned int bytes;
f24e9980 1396
1b83bef2 1397 req->r_result = result;
f24e9980
SW		 1398 bytes = le32_to_cpu(msg->hdr.data_len);
1399 dout("handle_reply result %d bytes %d\n", req->r_result,
1400 bytes);
1401 if (req->r_result == 0)
1402 req->r_result = bytes;
1403
1404 /* in case this is a write and we need to replay, */
1b83bef2
SW		 1405 req->r_reassert_version.epoch = cpu_to_le32(reassert_epoch);
1406 req->r_reassert_version.version = cpu_to_le64(reassert_version);
f24e9980
SW		 1407
1408 req->r_got_reply = 1;
1409 } else if ((flags & CEPH_OSD_FLAG_ONDISK) == 0) {
1410 dout("handle_reply tid %llu dup ack\n", tid);
34b43a56 1411 mutex_unlock(&osdc->request_mutex);
f24e9980
SW		 1412 goto done;
1413 }
1414
1415 dout("handle_reply tid %llu flags %d\n", tid, flags);
1416
a40c4f10
YS		 1417 if (req->r_linger && (flags & CEPH_OSD_FLAG_ONDISK))
1418 __register_linger_request(osdc, req);
1419
f24e9980 1420 /* either this is a read, or we got the safe response */
0ceed5db
SW		 1421 if (result < 0 ||
1422 (flags & CEPH_OSD_FLAG_ONDISK) ||
f24e9980
SW		 1423 ((flags & CEPH_OSD_FLAG_WRITE) == 0))
1424 __unregister_request(osdc, req);
1425
0d5af164
AE		 1426 already_completed = req->r_completed;
1427 req->r_completed = 1;
f24e9980 1428 mutex_unlock(&osdc->request_mutex);
0d5af164
AE		 1429 if (already_completed)
1430 goto done;
f24e9980
SW		 1431
1432 if (req->r_callback)
1433 req->r_callback(req, msg);
1434 else
03066f23 1435 complete_all(&req->r_completion);
f24e9980 1436
25845472
SW		 1437 if (flags & CEPH_OSD_FLAG_ONDISK)
1438 complete_request(req);
f24e9980
SW		 1439
1440done:
a40c4f10 1441 dout("req=%p req->r_linger=%d\n", req, req->r_linger);
f24e9980
SW		 1442 ceph_osdc_put_request(req);
1443 return;
1444
1b83bef2
SW		 1445bad_put:
1446 ceph_osdc_put_request(req);
f24e9980 1447bad:
1b83bef2
SW		 1448 pr_err("corrupt osd_op_reply got %d %d\n",
1449 (int)msg->front.iov_len, le32_to_cpu(msg->hdr.front_len));
9ec7cab1 1450 ceph_msg_dump(msg);
f24e9980
SW		 1451}
1452
6f6c7006 1453static void reset_changed_osds(struct ceph_osd_client *osdc)
f24e9980 1454{
f24e9980 1455 struct rb_node *p, *n;
f24e9980 1456
6f6c7006
SW		 1457 for (p = rb_first(&osdc->osds); p; p = n) {
1458 struct ceph_osd *osd = rb_entry(p, struct ceph_osd, o_node);
f24e9980 1459
6f6c7006
SW		 1460 n = rb_next(p);
1461 if (!ceph_osd_is_up(osdc->osdmap, osd->o_osd) ||
1462 memcmp(&osd->o_con.peer_addr,
1463 ceph_osd_addr(osdc->osdmap,
1464 osd->o_osd),
1465 sizeof(struct ceph_entity_addr)) != 0)
1466 __reset_osd(osdc, osd);
f24e9980 1467 }
422d2cb8
YS		 1468}
1469
1470/*
6f6c7006
SW		 1471 * Requeue requests whose mapping to an OSD has changed. If requests map to
1472 * no osd, request a new map.
422d2cb8 1473 *
e6d50f67 1474 * Caller should hold map_sem for read.
422d2cb8 1475 */
38d6453c 1476static void kick_requests(struct ceph_osd_client *osdc, int force_resend)
422d2cb8 1477{
a40c4f10 1478 struct ceph_osd_request *req, *nreq;
6f6c7006
SW		 1479 struct rb_node *p;
1480 int needmap = 0;
1481 int err;
422d2cb8 1482
38d6453c 1483 dout("kick_requests %s\n", force_resend ? " (force resend)" : "");
422d2cb8 1484 mutex_lock(&osdc->request_mutex);
6194ea89 1485 for (p = rb_first(&osdc->requests); p; ) {
6f6c7006 1486 req = rb_entry(p, struct ceph_osd_request, r_node);
6194ea89 1487 p = rb_next(p);
ab60b16d
AE		 1488
1489 /*
1490 * For linger requests that have not yet been
1491 * registered, move them to the linger list; they'll
1492 * be sent to the osd in the loop below. Unregister
1493 * the request before re-registering it as a linger
1494 * request to ensure the __map_request() below
1495 * will decide it needs to be sent.
1496 */
1497 if (req->r_linger && list_empty(&req->r_linger_item)) {
1498 dout("%p tid %llu restart on osd%d\n",
1499 req, req->r_tid,
1500 req->r_osd ? req->r_osd->o_osd : -1);
1501 __unregister_request(osdc, req);
1502 __register_linger_request(osdc, req);
1503 continue;
1504 }
1505
38d6453c 1506 err = __map_request(osdc, req, force_resend);
6f6c7006
SW		 1507 if (err < 0)
1508 continue; /* error */
1509 if (req->r_osd == NULL) {
1510 dout("%p tid %llu maps to no osd\n", req, req->r_tid);
1511 needmap++; /* request a newer map */
1512 } else if (err > 0) {
6194ea89
SW		 1513 if (!req->r_linger) {
1514 dout("%p tid %llu requeued on osd%d\n", req,
1515 req->r_tid,
1516 req->r_osd ? req->r_osd->o_osd : -1);
a40c4f10 1517 req->r_flags |= CEPH_OSD_FLAG_RETRY;
6194ea89
SW		 1518 }
1519 }
a40c4f10
YS		 1520 }
1521
1522 list_for_each_entry_safe(req, nreq, &osdc->req_linger,
1523 r_linger_item) {
1524 dout("linger req=%p req->r_osd=%p\n", req, req->r_osd);
1525
38d6453c 1526 err = __map_request(osdc, req, force_resend);
ab60b16d 1527 dout("__map_request returned %d\n", err);
a40c4f10
YS		 1528 if (err == 0)
1529 continue; /* no change and no osd was specified */
1530 if (err < 0)
1531 continue; /* hrm! */
1532 if (req->r_osd == NULL) {
1533 dout("tid %llu maps to no valid osd\n", req->r_tid);
1534 needmap++; /* request a newer map */
1535 continue;
6f6c7006 1536 }
a40c4f10
YS		 1537
1538 dout("kicking lingering %p tid %llu osd%d\n", req, req->r_tid,
1539 req->r_osd ? req->r_osd->o_osd : -1);
a40c4f10 1540 __register_request(osdc, req);
c89ce05e 1541 __unregister_linger_request(osdc, req);
6f6c7006 1542 }
f24e9980
SW		 1543 mutex_unlock(&osdc->request_mutex);
1544
1545 if (needmap) {
1546 dout("%d requests for down osds, need new map\n", needmap);
1547 ceph_monc_request_next_osdmap(&osdc->client->monc);
1548 }
e6d50f67 1549 reset_changed_osds(osdc);
422d2cb8 1550}
6f6c7006
SW		 1551
1552
f24e9980
SW		 1553/*
1554 * Process updated osd map.
1555 *
1556 * The message contains any number of incremental and full maps, normally
1557 * indicating some sort of topology change in the cluster. Kick requests
1558 * off to different OSDs as needed.
1559 */
1560void ceph_osdc_handle_map(struct ceph_osd_client *osdc, struct ceph_msg *msg)
1561{
1562 void *p, *end, *next;
1563 u32 nr_maps, maplen;
1564 u32 epoch;
1565 struct ceph_osdmap *newmap = NULL, *oldmap;
1566 int err;
1567 struct ceph_fsid fsid;
1568
1569 dout("handle_map have %u\n", osdc->osdmap ? osdc->osdmap->epoch : 0);
1570 p = msg->front.iov_base;
1571 end = p + msg->front.iov_len;
1572
1573 /* verify fsid */
1574 ceph_decode_need(&p, end, sizeof(fsid), bad);
1575 ceph_decode_copy(&p, &fsid, sizeof(fsid));
0743304d
SW		 1576 if (ceph_check_fsid(osdc->client, &fsid) < 0)
1577 return;
f24e9980
SW		 1578
1579 down_write(&osdc->map_sem);
1580
1581 /* incremental maps */
1582 ceph_decode_32_safe(&p, end, nr_maps, bad);
1583 dout(" %d inc maps\n", nr_maps);
1584 while (nr_maps > 0) {
1585 ceph_decode_need(&p, end, 2*sizeof(u32), bad);
c89136ea
SW		 1586 epoch = ceph_decode_32(&p);
1587 maplen = ceph_decode_32(&p);
f24e9980
SW		 1588 ceph_decode_need(&p, end, maplen, bad);
1589 next = p + maplen;
1590 if (osdc->osdmap && osdc->osdmap->epoch+1 == epoch) {
1591 dout("applying incremental map %u len %d\n",
1592 epoch, maplen);
1593 newmap = osdmap_apply_incremental(&p, next,
1594 osdc->osdmap,
15d9882c 1595 &osdc->client->msgr);
f24e9980
SW		 1596 if (IS_ERR(newmap)) {
1597 err = PTR_ERR(newmap);
1598 goto bad;
1599 }
30dc6381 1600 BUG_ON(!newmap);
f24e9980
SW		 1601 if (newmap != osdc->osdmap) {
1602 ceph_osdmap_destroy(osdc->osdmap);
1603 osdc->osdmap = newmap;
1604 }
38d6453c 1605 kick_requests(osdc, 0);
f24e9980
SW		 1606 } else {
1607 dout("ignoring incremental map %u len %d\n",
1608 epoch, maplen);
1609 }
1610 p = next;
1611 nr_maps--;
1612 }
1613 if (newmap)
1614 goto done;
1615
1616 /* full maps */
1617 ceph_decode_32_safe(&p, end, nr_maps, bad);
1618 dout(" %d full maps\n", nr_maps);
1619 while (nr_maps) {
1620 ceph_decode_need(&p, end, 2*sizeof(u32), bad);
c89136ea
SW		 1621 epoch = ceph_decode_32(&p);
1622 maplen = ceph_decode_32(&p);
f24e9980
SW		 1623 ceph_decode_need(&p, end, maplen, bad);
1624 if (nr_maps > 1) {
1625 dout("skipping non-latest full map %u len %d\n",
1626 epoch, maplen);
1627 } else if (osdc->osdmap && osdc->osdmap->epoch >= epoch) {
1628 dout("skipping full map %u len %d, "
1629 "older than our %u\n", epoch, maplen,
1630 osdc->osdmap->epoch);
1631 } else {
38d6453c
SW		 1632 int skipped_map = 0;
1633
f24e9980
SW		 1634 dout("taking full map %u len %d\n", epoch, maplen);
1635 newmap = osdmap_decode(&p, p+maplen);
1636 if (IS_ERR(newmap)) {
1637 err = PTR_ERR(newmap);
1638 goto bad;
1639 }
30dc6381 1640 BUG_ON(!newmap);
f24e9980
SW		 1641 oldmap = osdc->osdmap;
1642 osdc->osdmap = newmap;
38d6453c
SW		 1643 if (oldmap) {
1644 if (oldmap->epoch + 1 < newmap->epoch)
1645 skipped_map = 1;
f24e9980 1646 ceph_osdmap_destroy(oldmap);
38d6453c
SW		 1647 }
1648 kick_requests(osdc, skipped_map);
f24e9980
SW		 1649 }
1650 p += maplen;
1651 nr_maps--;
1652 }
1653
1654done:
1655 downgrade_write(&osdc->map_sem);
1656 ceph_monc_got_osdmap(&osdc->client->monc, osdc->osdmap->epoch);
cd634fb6
SW		 1657
1658 /*
1659 * subscribe to subsequent osdmap updates if full to ensure
1660 * we find out when we are no longer full and stop returning
1661 * ENOSPC.
1662 */
1663 if (ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_FULL))
1664 ceph_monc_request_next_osdmap(&osdc->client->monc);
1665
f9d25199
AE		 1666 mutex_lock(&osdc->request_mutex);
1667 __send_queued(osdc);
1668 mutex_unlock(&osdc->request_mutex);
f24e9980 1669 up_read(&osdc->map_sem);
03066f23 1670 wake_up_all(&osdc->client->auth_wq);
f24e9980
SW		 1671 return;
1672
1673bad:
1674 pr_err("osdc handle_map corrupt msg\n");
9ec7cab1 1675 ceph_msg_dump(msg);
f24e9980
SW		 1676 up_write(&osdc->map_sem);
1677 return;
1678}
1679
a40c4f10
YS		 1680/*
1681 * watch/notify callback event infrastructure
1682 *
1683 * These callbacks are used both for watch and notify operations.
1684 */
1685static void __release_event(struct kref *kref)
1686{
1687 struct ceph_osd_event *event =
1688 container_of(kref, struct ceph_osd_event, kref);
1689
1690 dout("__release_event %p\n", event);
1691 kfree(event);
1692}
1693
1694static void get_event(struct ceph_osd_event *event)
1695{
1696 kref_get(&event->kref);
1697}
1698
1699void ceph_osdc_put_event(struct ceph_osd_event *event)
1700{
1701 kref_put(&event->kref, __release_event);
1702}
1703EXPORT_SYMBOL(ceph_osdc_put_event);
1704
1705static void __insert_event(struct ceph_osd_client *osdc,
1706 struct ceph_osd_event *new)
1707{
1708 struct rb_node **p = &osdc->event_tree.rb_node;
1709 struct rb_node *parent = NULL;
1710 struct ceph_osd_event *event = NULL;
1711
1712 while (*p) {
1713 parent = *p;
1714 event = rb_entry(parent, struct ceph_osd_event, node);
1715 if (new->cookie < event->cookie)
1716 p = &(*p)->rb_left;
1717 else if (new->cookie > event->cookie)
1718 p = &(*p)->rb_right;
1719 else
1720 BUG();
1721 }
1722
1723 rb_link_node(&new->node, parent, p);
1724 rb_insert_color(&new->node, &osdc->event_tree);
1725}
1726
1727static struct ceph_osd_event *__find_event(struct ceph_osd_client *osdc,
1728 u64 cookie)
1729{
1730 struct rb_node **p = &osdc->event_tree.rb_node;
1731 struct rb_node *parent = NULL;
1732 struct ceph_osd_event *event = NULL;
1733
1734 while (*p) {
1735 parent = *p;
1736 event = rb_entry(parent, struct ceph_osd_event, node);
1737 if (cookie < event->cookie)
1738 p = &(*p)->rb_left;
1739 else if (cookie > event->cookie)
1740 p = &(*p)->rb_right;
1741 else
1742 return event;
1743 }
1744 return NULL;
1745}
1746
1747static void __remove_event(struct ceph_osd_event *event)
1748{
1749 struct ceph_osd_client *osdc = event->osdc;
1750
1751 if (!RB_EMPTY_NODE(&event->node)) {
1752 dout("__remove_event removed %p\n", event);
1753 rb_erase(&event->node, &osdc->event_tree);
1754 ceph_osdc_put_event(event);
1755 } else {
1756 dout("__remove_event didn't remove %p\n", event);
1757 }
1758}
1759
1760int ceph_osdc_create_event(struct ceph_osd_client *osdc,
1761 void (*event_cb)(u64, u64, u8, void *),
3c663bbd 1762 void *data, struct ceph_osd_event **pevent)
a40c4f10
YS		 1763{
1764 struct ceph_osd_event *event;
1765
1766 event = kmalloc(sizeof(*event), GFP_NOIO);
1767 if (!event)
1768 return -ENOMEM;
1769
1770 dout("create_event %p\n", event);
1771 event->cb = event_cb;
3c663bbd 1772 event->one_shot = 0;
a40c4f10
YS		 1773 event->data = data;
1774 event->osdc = osdc;
1775 INIT_LIST_HEAD(&event->osd_node);
3ee5234d 1776 RB_CLEAR_NODE(&event->node);
a40c4f10
YS		 1777 kref_init(&event->kref); /* one ref for us */
1778 kref_get(&event->kref); /* one ref for the caller */
a40c4f10
YS		 1779
1780 spin_lock(&osdc->event_lock);
1781 event->cookie = ++osdc->event_count;
1782 __insert_event(osdc, event);
1783 spin_unlock(&osdc->event_lock);
1784
1785 *pevent = event;
1786 return 0;
1787}
1788EXPORT_SYMBOL(ceph_osdc_create_event);
1789
1790void ceph_osdc_cancel_event(struct ceph_osd_event *event)
1791{
1792 struct ceph_osd_client *osdc = event->osdc;
1793
1794 dout("cancel_event %p\n", event);
1795 spin_lock(&osdc->event_lock);
1796 __remove_event(event);
1797 spin_unlock(&osdc->event_lock);
1798 ceph_osdc_put_event(event); /* caller's */
1799}
1800EXPORT_SYMBOL(ceph_osdc_cancel_event);
1801
1802
1803static void do_event_work(struct work_struct *work)
1804{
1805 struct ceph_osd_event_work *event_work =
1806 container_of(work, struct ceph_osd_event_work, work);
1807 struct ceph_osd_event *event = event_work->event;
1808 u64 ver = event_work->ver;
1809 u64 notify_id = event_work->notify_id;
1810 u8 opcode = event_work->opcode;
1811
1812 dout("do_event_work completing %p\n", event);
1813 event->cb(ver, notify_id, opcode, event->data);
a40c4f10
YS		 1814 dout("do_event_work completed %p\n", event);
1815 ceph_osdc_put_event(event);
1816 kfree(event_work);
1817}
1818
1819
1820/*
1821 * Process osd watch notifications
1822 */
3c663bbd
AE		 1823static void handle_watch_notify(struct ceph_osd_client *osdc,
1824 struct ceph_msg *msg)
a40c4f10
YS		 1825{
1826 void *p, *end;
1827 u8 proto_ver;
1828 u64 cookie, ver, notify_id;
1829 u8 opcode;
1830 struct ceph_osd_event *event;
1831 struct ceph_osd_event_work *event_work;
1832
1833 p = msg->front.iov_base;
1834 end = p + msg->front.iov_len;
1835
1836 ceph_decode_8_safe(&p, end, proto_ver, bad);
1837 ceph_decode_8_safe(&p, end, opcode, bad);
1838 ceph_decode_64_safe(&p, end, cookie, bad);
1839 ceph_decode_64_safe(&p, end, ver, bad);
1840 ceph_decode_64_safe(&p, end, notify_id, bad);
1841
1842 spin_lock(&osdc->event_lock);
1843 event = __find_event(osdc, cookie);
1844 if (event) {
3c663bbd 1845 BUG_ON(event->one_shot);
a40c4f10 1846 get_event(event);
a40c4f10
YS		 1847 }
1848 spin_unlock(&osdc->event_lock);
1849 dout("handle_watch_notify cookie %lld ver %lld event %p\n",
1850 cookie, ver, event);
1851 if (event) {
1852 event_work = kmalloc(sizeof(*event_work), GFP_NOIO);
a40c4f10
YS		 1853 if (!event_work) {
1854 dout("ERROR: could not allocate event_work\n");
1855 goto done_err;
1856 }
6b0ae409 1857 INIT_WORK(&event_work->work, do_event_work);
a40c4f10
YS		 1858 event_work->event = event;
1859 event_work->ver = ver;
1860 event_work->notify_id = notify_id;
1861 event_work->opcode = opcode;
1862 if (!queue_work(osdc->notify_wq, &event_work->work)) {
1863 dout("WARNING: failed to queue notify event work\n");
1864 goto done_err;
1865 }
1866 }
1867
1868 return;
1869
1870done_err:
a40c4f10
YS		 1871 ceph_osdc_put_event(event);
1872 return;
1873
1874bad:
1875 pr_err("osdc handle_watch_notify corrupt msg\n");
1876 return;
1877}
1878
70636773
AE		 1879static void ceph_osdc_msg_data_set(struct ceph_msg *msg,
1880 struct ceph_osd_data *osd_data)
f24e9980 1881{
0fff87ec 1882 if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGES) {
f1baeb2b 1883 BUG_ON(osd_data->length > (u64) SIZE_MAX);
ebf18f47 1884 if (osd_data->length)
70636773
AE		 1885 ceph_msg_data_set_pages(msg, osd_data->pages,
1886 osd_data->length, osd_data->alignment);
9a5e6d09
AE		 1887 } else if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGELIST) {
1888 BUG_ON(!osd_data->pagelist->length);
1889 ceph_msg_data_set_pagelist(msg, osd_data->pagelist);
68b4476b 1890#ifdef CONFIG_BLOCK
0fff87ec 1891 } else if (osd_data->type == CEPH_OSD_DATA_TYPE_BIO) {
70636773 1892 ceph_msg_data_set_bio(msg, osd_data->bio);
68b4476b 1893#endif
2ac2b7a6 1894 } else {
0fff87ec 1895 BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_NONE);
2ac2b7a6 1896 }
70636773
AE		 1897}
1898
1899/*
1900 * Register request, send initial attempt.
1901 */
1902int ceph_osdc_start_request(struct ceph_osd_client *osdc,
1903 struct ceph_osd_request *req,
1904 bool nofail)
1905{
1906 int rc = 0;
1907
1908 /* Set up response incoming data and request outgoing data fields */
1909
1910 ceph_osdc_msg_data_set(req->r_reply, &req->r_data_in);
1911 ceph_osdc_msg_data_set(req->r_request, &req->r_data_out);
f24e9980 1912
f24e9980
SW		 1913 down_read(&osdc->map_sem);
1914 mutex_lock(&osdc->request_mutex);
dc4b870c 1915 __register_request(osdc, req);
92451b49
AE		 1916 WARN_ON(req->r_sent);
1917 rc = __map_request(osdc, req, 0);
1918 if (rc < 0) {
1919 if (nofail) {
1920 dout("osdc_start_request failed map, "
1921 " will retry %lld\n", req->r_tid);
1922 rc = 0;
f24e9980 1923 }
92451b49 1924 goto out_unlock;
f24e9980 1925 }
92451b49
AE		 1926 if (req->r_osd == NULL) {
1927 dout("send_request %p no up osds in pg\n", req);
1928 ceph_monc_request_next_osdmap(&osdc->client->monc);
1929 } else {
7e2766a1 1930 __send_queued(osdc);
92451b49
AE		 1931 }
1932 rc = 0;
234af26f 1933out_unlock:
f24e9980
SW		 1934 mutex_unlock(&osdc->request_mutex);
1935 up_read(&osdc->map_sem);
1936 return rc;
1937}
3d14c5d2 1938EXPORT_SYMBOL(ceph_osdc_start_request);
f24e9980
SW		 1939
1940/*
1941 * wait for a request to complete
1942 */
1943int ceph_osdc_wait_request(struct ceph_osd_client *osdc,
1944 struct ceph_osd_request *req)
1945{
1946 int rc;
1947
1948 rc = wait_for_completion_interruptible(&req->r_completion);
1949 if (rc < 0) {
1950 mutex_lock(&osdc->request_mutex);
1951 __cancel_request(req);
529cfcc4 1952 __unregister_request(osdc, req);
f24e9980 1953 mutex_unlock(&osdc->request_mutex);
25845472 1954 complete_request(req);
529cfcc4 1955 dout("wait_request tid %llu canceled/timed out\n", req->r_tid);
f24e9980
SW		 1956 return rc;
1957 }
1958
1959 dout("wait_request tid %llu result %d\n", req->r_tid, req->r_result);
1960 return req->r_result;
1961}
3d14c5d2 1962EXPORT_SYMBOL(ceph_osdc_wait_request);
f24e9980
SW		 1963
1964/*
1965 * sync - wait for all in-flight requests to flush. avoid starvation.
1966 */
1967void ceph_osdc_sync(struct ceph_osd_client *osdc)
1968{
1969 struct ceph_osd_request *req;
1970 u64 last_tid, next_tid = 0;
1971
1972 mutex_lock(&osdc->request_mutex);
1973 last_tid = osdc->last_tid;
1974 while (1) {
1975 req = __lookup_request_ge(osdc, next_tid);
1976 if (!req)
1977 break;
1978 if (req->r_tid > last_tid)
1979 break;
1980
1981 next_tid = req->r_tid + 1;
1982 if ((req->r_flags & CEPH_OSD_FLAG_WRITE) == 0)
1983 continue;
1984
1985 ceph_osdc_get_request(req);
1986 mutex_unlock(&osdc->request_mutex);
1987 dout("sync waiting on tid %llu (last is %llu)\n",
1988 req->r_tid, last_tid);
1989 wait_for_completion(&req->r_safe_completion);
1990 mutex_lock(&osdc->request_mutex);
1991 ceph_osdc_put_request(req);
1992 }
1993 mutex_unlock(&osdc->request_mutex);
1994 dout("sync done (thru tid %llu)\n", last_tid);
1995}
3d14c5d2 1996EXPORT_SYMBOL(ceph_osdc_sync);
f24e9980
SW		 1997
1998/*
1999 * init, shutdown
2000 */
2001int ceph_osdc_init(struct ceph_osd_client *osdc, struct ceph_client *client)
2002{
2003 int err;
2004
2005 dout("init\n");
2006 osdc->client = client;
2007 osdc->osdmap = NULL;
2008 init_rwsem(&osdc->map_sem);
2009 init_completion(&osdc->map_waiters);
2010 osdc->last_requested_map = 0;
2011 mutex_init(&osdc->request_mutex);
f24e9980
SW		 2012 osdc->last_tid = 0;
2013 osdc->osds = RB_ROOT;
f5a2041b 2014 INIT_LIST_HEAD(&osdc->osd_lru);
f24e9980 2015 osdc->requests = RB_ROOT;
422d2cb8 2016 INIT_LIST_HEAD(&osdc->req_lru);
6f6c7006
SW		 2017 INIT_LIST_HEAD(&osdc->req_unsent);
2018 INIT_LIST_HEAD(&osdc->req_notarget);
a40c4f10 2019 INIT_LIST_HEAD(&osdc->req_linger);
f24e9980
SW		 2020 osdc->num_requests = 0;
2021 INIT_DELAYED_WORK(&osdc->timeout_work, handle_timeout);
f5a2041b 2022 INIT_DELAYED_WORK(&osdc->osds_timeout_work, handle_osds_timeout);
a40c4f10
YS		 2023 spin_lock_init(&osdc->event_lock);
2024 osdc->event_tree = RB_ROOT;
2025 osdc->event_count = 0;
f5a2041b
YS		 2026
2027 schedule_delayed_work(&osdc->osds_timeout_work,
3d14c5d2 2028 round_jiffies_relative(osdc->client->options->osd_idle_ttl * HZ));
f24e9980 2029
5f44f142 2030 err = -ENOMEM;
f24e9980
SW		 2031 osdc->req_mempool = mempool_create_kmalloc_pool(10,
2032 sizeof(struct ceph_osd_request));
2033 if (!osdc->req_mempool)
5f44f142 2034 goto out;
f24e9980 2035
d50b409f
SW		 2036 err = ceph_msgpool_init(&osdc->msgpool_op, CEPH_MSG_OSD_OP,
2037 OSD_OP_FRONT_LEN, 10, true,
4f48280e 2038 "osd_op");
f24e9980 2039 if (err < 0)
5f44f142 2040 goto out_mempool;
d50b409f 2041 err = ceph_msgpool_init(&osdc->msgpool_op_reply, CEPH_MSG_OSD_OPREPLY,
4f48280e
SW		 2042 OSD_OPREPLY_FRONT_LEN, 10, true,
2043 "osd_op_reply");
c16e7869
SW		 2044 if (err < 0)
2045 goto out_msgpool;
a40c4f10
YS		 2046
2047 osdc->notify_wq = create_singlethread_workqueue("ceph-watch-notify");
2048 if (IS_ERR(osdc->notify_wq)) {
2049 err = PTR_ERR(osdc->notify_wq);
2050 osdc->notify_wq = NULL;
2051 goto out_msgpool;
2052 }
f24e9980 2053 return 0;
5f44f142 2054
c16e7869
SW		 2055out_msgpool:
2056 ceph_msgpool_destroy(&osdc->msgpool_op);
5f44f142
SW		 2057out_mempool:
2058 mempool_destroy(osdc->req_mempool);
2059out:
2060 return err;
f24e9980
SW		 2061}
2062
2063void ceph_osdc_stop(struct ceph_osd_client *osdc)
2064{
a40c4f10
YS		 2065 flush_workqueue(osdc->notify_wq);
2066 destroy_workqueue(osdc->notify_wq);
f24e9980 2067 cancel_delayed_work_sync(&osdc->timeout_work);
f5a2041b 2068 cancel_delayed_work_sync(&osdc->osds_timeout_work);
f24e9980
SW		 2069 if (osdc->osdmap) {
2070 ceph_osdmap_destroy(osdc->osdmap);
2071 osdc->osdmap = NULL;
2072 }
aca420bc 2073 remove_all_osds(osdc);
f24e9980
SW		 2074 mempool_destroy(osdc->req_mempool);
2075 ceph_msgpool_destroy(&osdc->msgpool_op);
c16e7869 2076 ceph_msgpool_destroy(&osdc->msgpool_op_reply);
f24e9980
SW		 2077}
2078
2079/*
2080 * Read some contiguous pages. If we cross a stripe boundary, shorten
2081 * *plen. Return number of bytes read, or error.
2082 */
2083int ceph_osdc_readpages(struct ceph_osd_client *osdc,
2084 struct ceph_vino vino, struct ceph_file_layout *layout,
2085 u64 off, u64 *plen,
2086 u32 truncate_seq, u64 truncate_size,
b7495fc2 2087 struct page **pages, int num_pages, int page_align)
f24e9980
SW		 2088{
2089 struct ceph_osd_request *req;
0fff87ec 2090 struct ceph_osd_data *osd_data;
f24e9980
SW		 2091 int rc = 0;
2092
2093 dout("readpages on ino %llx.%llx on %llu~%llu\n", vino.ino,
2094 vino.snap, off, *plen);
2095 req = ceph_osdc_new_request(osdc, layout, vino, off, plen,
2096 CEPH_OSD_OP_READ, CEPH_OSD_FLAG_READ,
2097 NULL, 0, truncate_seq, truncate_size, NULL,
153e5167 2098 false);
6816282d
SW		 2099 if (IS_ERR(req))
2100 return PTR_ERR(req);
f24e9980
SW		 2101
2102 /* it may be a short read due to an object boundary */
0fff87ec
AE		 2103
2104 osd_data = &req->r_data_in;
2105 osd_data->type = CEPH_OSD_DATA_TYPE_PAGES;
2106 osd_data->pages = pages;
e0c59487 2107 osd_data->length = *plen;
0fff87ec 2108 osd_data->alignment = page_align;
f24e9980 2109
e0c59487
AE		 2110 dout("readpages final extent is %llu~%llu (%llu bytes align %d)\n",
2111 off, *plen, osd_data->length, page_align);
f24e9980
SW		 2112
2113 rc = ceph_osdc_start_request(osdc, req, false);
2114 if (!rc)
2115 rc = ceph_osdc_wait_request(osdc, req);
2116
2117 ceph_osdc_put_request(req);
2118 dout("readpages result %d\n", rc);
2119 return rc;
2120}
3d14c5d2 2121EXPORT_SYMBOL(ceph_osdc_readpages);
f24e9980
SW		 2122
2123/*
2124 * do a synchronous write on N pages
2125 */
2126int ceph_osdc_writepages(struct ceph_osd_client *osdc, struct ceph_vino vino,
2127 struct ceph_file_layout *layout,
2128 struct ceph_snap_context *snapc,
2129 u64 off, u64 len,
2130 u32 truncate_seq, u64 truncate_size,
2131 struct timespec *mtime,
24808826 2132 struct page **pages, int num_pages)
f24e9980
SW		 2133{
2134 struct ceph_osd_request *req;
0fff87ec 2135 struct ceph_osd_data *osd_data;
f24e9980 2136 int rc = 0;
b7495fc2 2137 int page_align = off & ~PAGE_MASK;
f24e9980
SW		 2138
2139 BUG_ON(vino.snap != CEPH_NOSNAP);
2140 req = ceph_osdc_new_request(osdc, layout, vino, off, &len,
2141 CEPH_OSD_OP_WRITE,
24808826 2142 CEPH_OSD_FLAG_ONDISK | CEPH_OSD_FLAG_WRITE,
fbf8685f 2143 snapc, 0,
f24e9980 2144 truncate_seq, truncate_size, mtime,
153e5167 2145 true);
6816282d
SW		 2146 if (IS_ERR(req))
2147 return PTR_ERR(req);
f24e9980
SW		 2148
2149 /* it may be a short write due to an object boundary */
0fff87ec
AE		 2150 osd_data = &req->r_data_out;
2151 osd_data->type = CEPH_OSD_DATA_TYPE_PAGES;
2152 osd_data->pages = pages;
e0c59487 2153 osd_data->length = len;
0fff87ec 2154 osd_data->alignment = page_align;
e0c59487 2155 dout("writepages %llu~%llu (%llu bytes)\n", off, len, osd_data->length);
f24e9980 2156
87f979d3 2157 rc = ceph_osdc_start_request(osdc, req, true);
f24e9980
SW		 2158 if (!rc)
2159 rc = ceph_osdc_wait_request(osdc, req);
2160
2161 ceph_osdc_put_request(req);
2162 if (rc == 0)
2163 rc = len;
2164 dout("writepages result %d\n", rc);
2165 return rc;
2166}
3d14c5d2 2167EXPORT_SYMBOL(ceph_osdc_writepages);
f24e9980
SW		 2168
2169/*
2170 * handle incoming message
2171 */
2172static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
2173{
2174 struct ceph_osd *osd = con->private;
32c895e7 2175 struct ceph_osd_client *osdc;
f24e9980
SW		 2176 int type = le16_to_cpu(msg->hdr.type);
2177
2178 if (!osd)
4a32f93d 2179 goto out;
32c895e7 2180 osdc = osd->o_osdc;
f24e9980
SW		 2181
2182 switch (type) {
2183 case CEPH_MSG_OSD_MAP:
2184 ceph_osdc_handle_map(osdc, msg);
2185 break;
2186 case CEPH_MSG_OSD_OPREPLY:
350b1c32 2187 handle_reply(osdc, msg, con);
f24e9980 2188 break;
a40c4f10
YS		 2189 case CEPH_MSG_WATCH_NOTIFY:
2190 handle_watch_notify(osdc, msg);
2191 break;
f24e9980
SW		 2192
2193 default:
2194 pr_err("received unknown message type %d %s\n", type,
2195 ceph_msg_type_name(type));
2196 }
4a32f93d 2197out:
f24e9980
SW		 2198 ceph_msg_put(msg);
2199}
2200
5b3a4db3 2201/*
21b667f6
SW		 2202 * lookup and return message for incoming reply. set up reply message
2203 * pages.
5b3a4db3
SW		 2204 */
2205static struct ceph_msg *get_reply(struct ceph_connection *con,
2450418c
YS		 2206 struct ceph_msg_header *hdr,
2207 int *skip)
f24e9980
SW		 2208{
2209 struct ceph_osd *osd = con->private;
2210 struct ceph_osd_client *osdc = osd->o_osdc;
2450418c 2211 struct ceph_msg *m;
0547a9b3 2212 struct ceph_osd_request *req;
5b3a4db3
SW		 2213 int front = le32_to_cpu(hdr->front_len);
2214 int data_len = le32_to_cpu(hdr->data_len);
0547a9b3 2215 u64 tid;
f24e9980 2216
0547a9b3
YS		 2217 tid = le64_to_cpu(hdr->tid);
2218 mutex_lock(&osdc->request_mutex);
2219 req = __lookup_request(osdc, tid);
2220 if (!req) {
2221 *skip = 1;
2222 m = NULL;
756a16a5
SW		 2223 dout("get_reply unknown tid %llu from osd%d\n", tid,
2224 osd->o_osd);
0547a9b3
YS		 2225 goto out;
2226 }
c16e7869
SW		 2227
2228 if (req->r_con_filling_msg) {
8921d114 2229 dout("%s revoking msg %p from old con %p\n", __func__,
c16e7869 2230 req->r_reply, req->r_con_filling_msg);
8921d114 2231 ceph_msg_revoke_incoming(req->r_reply);
0d47766f 2232 req->r_con_filling_msg->ops->put(req->r_con_filling_msg);
6f46cb29 2233 req->r_con_filling_msg = NULL;
0547a9b3
YS		 2234 }
2235
c16e7869
SW		 2236 if (front > req->r_reply->front.iov_len) {
2237 pr_warning("get_reply front %d > preallocated %d\n",
2238 front, (int)req->r_reply->front.iov_len);
b61c2763 2239 m = ceph_msg_new(CEPH_MSG_OSD_OPREPLY, front, GFP_NOFS, false);
a79832f2 2240 if (!m)
c16e7869
SW		 2241 goto out;
2242 ceph_msg_put(req->r_reply);
2243 req->r_reply = m;
2244 }
2245 m = ceph_msg_get(req->r_reply);
2246
0547a9b3 2247 if (data_len > 0) {
0fff87ec
AE		 2248 struct ceph_osd_data *osd_data = &req->r_data_in;
2249
2250 if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGES) {
0fff87ec 2251 if (osd_data->pages &&
e0c59487 2252 unlikely(osd_data->length < data_len)) {
2ac2b7a6 2253
e0c59487
AE		 2254 pr_warning("tid %lld reply has %d bytes "
2255 "we had only %llu bytes ready\n",
2256 tid, data_len, osd_data->length);
2ac2b7a6
AE		 2257 *skip = 1;
2258 ceph_msg_put(m);
2259 m = NULL;
2260 goto out;
2261 }
2ac2b7a6 2262 }
0547a9b3 2263 }
5b3a4db3 2264 *skip = 0;
0d47766f 2265 req->r_con_filling_msg = con->ops->get(con);
c16e7869 2266 dout("get_reply tid %lld %p\n", tid, m);
0547a9b3
YS		 2267
2268out:
2269 mutex_unlock(&osdc->request_mutex);
2450418c 2270 return m;
5b3a4db3
SW		 2271
2272}
2273
2274static struct ceph_msg *alloc_msg(struct ceph_connection *con,
2275 struct ceph_msg_header *hdr,
2276 int *skip)
2277{
2278 struct ceph_osd *osd = con->private;
2279 int type = le16_to_cpu(hdr->type);
2280 int front = le32_to_cpu(hdr->front_len);
2281
1c20f2d2 2282 *skip = 0;
5b3a4db3
SW		 2283 switch (type) {
2284 case CEPH_MSG_OSD_MAP:
a40c4f10 2285 case CEPH_MSG_WATCH_NOTIFY:
b61c2763 2286 return ceph_msg_new(type, front, GFP_NOFS, false);
5b3a4db3
SW		 2287 case CEPH_MSG_OSD_OPREPLY:
2288 return get_reply(con, hdr, skip);
2289 default:
2290 pr_info("alloc_msg unexpected msg type %d from osd%d\n", type,
2291 osd->o_osd);
2292 *skip = 1;
2293 return NULL;
2294 }
f24e9980
SW		 2295}
2296
2297/*
2298 * Wrappers to refcount containing ceph_osd struct
2299 */
2300static struct ceph_connection *get_osd_con(struct ceph_connection *con)
2301{
2302 struct ceph_osd *osd = con->private;
2303 if (get_osd(osd))
2304 return con;
2305 return NULL;
2306}
2307
2308static void put_osd_con(struct ceph_connection *con)
2309{
2310 struct ceph_osd *osd = con->private;
2311 put_osd(osd);
2312}
2313
4e7a5dcd
SW		 2314/*
2315 * authentication
2316 */
a3530df3
AE		 2317/*
2318 * Note: returned pointer is the address of a structure that's
2319 * managed separately. Caller must *not* attempt to free it.
2320 */
2321static struct ceph_auth_handshake *get_authorizer(struct ceph_connection *con,
8f43fb53 2322 int *proto, int force_new)
4e7a5dcd
SW		 2323{
2324 struct ceph_osd *o = con->private;
2325 struct ceph_osd_client *osdc = o->o_osdc;
2326 struct ceph_auth_client *ac = osdc->client->monc.auth;
74f1869f 2327 struct ceph_auth_handshake *auth = &o->o_auth;
4e7a5dcd 2328
74f1869f 2329 if (force_new && auth->authorizer) {
27859f97 2330 ceph_auth_destroy_authorizer(ac, auth->authorizer);
74f1869f
AE		 2331 auth->authorizer = NULL;
2332 }
27859f97
SW		 2333 if (!auth->authorizer) {
2334 int ret = ceph_auth_create_authorizer(ac, CEPH_ENTITY_TYPE_OSD,
2335 auth);
4e7a5dcd 2336 if (ret)
a3530df3 2337 return ERR_PTR(ret);
27859f97
SW		 2338 } else {
2339 int ret = ceph_auth_update_authorizer(ac, CEPH_ENTITY_TYPE_OSD,
0bed9b5c
SW		 2340 auth);
2341 if (ret)
2342 return ERR_PTR(ret);
4e7a5dcd 2343 }
4e7a5dcd 2344 *proto = ac->protocol;
74f1869f 2345
a3530df3 2346 return auth;
4e7a5dcd
SW		 2347}
2348
2349
2350static int verify_authorizer_reply(struct ceph_connection *con, int len)
2351{
2352 struct ceph_osd *o = con->private;
2353 struct ceph_osd_client *osdc = o->o_osdc;
2354 struct ceph_auth_client *ac = osdc->client->monc.auth;
2355
27859f97 2356 return ceph_auth_verify_authorizer_reply(ac, o->o_auth.authorizer, len);
4e7a5dcd
SW		 2357}
2358
9bd2e6f8
SW		 2359static int invalidate_authorizer(struct ceph_connection *con)
2360{
2361 struct ceph_osd *o = con->private;
2362 struct ceph_osd_client *osdc = o->o_osdc;
2363 struct ceph_auth_client *ac = osdc->client->monc.auth;
2364
27859f97 2365 ceph_auth_invalidate_authorizer(ac, CEPH_ENTITY_TYPE_OSD);
9bd2e6f8
SW		 2366 return ceph_monc_validate_auth(&osdc->client->monc);
2367}
4e7a5dcd 2368
9e32789f 2369static const struct ceph_connection_operations osd_con_ops = {
f24e9980
SW		 2370 .get = get_osd_con,
2371 .put = put_osd_con,
2372 .dispatch = dispatch,
4e7a5dcd
SW		 2373 .get_authorizer = get_authorizer,
2374 .verify_authorizer_reply = verify_authorizer_reply,
9bd2e6f8 2375 .invalidate_authorizer = invalidate_authorizer,
f24e9980 2376 .alloc_msg = alloc_msg,
81b024e7 2377 .fault = osd_reset,
f24e9980 2378};
Linux kernel - Deliverables
This page took 0.34064 seconds and 5 git commands to generate.