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