projects / deliverable / linux.git / blob
? search:


37d43d5b828c3cd2712ca373425c24527d6001d9
[deliverable/linux.git] / net / ceph / osd_client.c
1 #include <linux/ceph/ceph_debug.h>
2
3 #include <linux/module.h>
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>
10 #ifdef CONFIG_BLOCK
11 #include <linux/bio.h>
12 #endif
13
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>
20
21 #define OSD_OP_FRONT_LEN 4096
22 #define OSD_OPREPLY_FRONT_LEN 512
23
24 static const struct ceph_connection_operations osd_con_ops;
25
26 static void send_queued(struct ceph_osd_client *osdc);
27 static int __reset_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd);
28 static void __register_request(struct ceph_osd_client *osdc,
29 struct ceph_osd_request *req);
30 static void __unregister_linger_request(struct ceph_osd_client *osdc,
31 struct ceph_osd_request *req);
32 static void __send_request(struct ceph_osd_client *osdc,
33 struct ceph_osd_request *req);
34
35 static int op_has_extent(int op)
36 {
37 return (op == CEPH_OSD_OP_READ ||
38 op == CEPH_OSD_OP_WRITE);
39 }
40
41 int ceph_calc_raw_layout(struct ceph_osd_client *osdc,
42 struct ceph_file_layout *layout,
43 u64 snapid,
44 u64 off, u64 *plen, u64 *bno,
45 struct ceph_osd_request *req,
46 struct ceph_osd_req_op *op)
47 {
48 struct ceph_osd_request_head *reqhead = req->r_request->front.iov_base;
49 u64 orig_len = *plen;
50 u64 objoff, objlen; /* extent in object */
51 int r;
52
53 reqhead->snapid = cpu_to_le64(snapid);
54
55 /* object extent? */
56 r = ceph_calc_file_object_mapping(layout, off, plen, bno,
57 &objoff, &objlen);
58 if (r < 0)
59 return r;
60 if (*plen < orig_len)
61 dout(" skipping last %llu, final file extent %llu~%llu\n",
62 orig_len - *plen, off, *plen);
63
64 if (op_has_extent(op->op)) {
65 u32 osize = le32_to_cpu(layout->fl_object_size);
66 op->extent.offset = objoff;
67 op->extent.length = objlen;
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 }
75 }
76 req->r_num_pages = calc_pages_for(off, *plen);
77 req->r_page_alignment = off & ~PAGE_MASK;
78 if (op->op == CEPH_OSD_OP_WRITE)
79 op->payload_len = *plen;
80
81 dout("calc_layout bno=%llx %llu~%llu (%d pages)\n",
82 *bno, objoff, objlen, req->r_num_pages);
83 return 0;
84 }
85 EXPORT_SYMBOL(ceph_calc_raw_layout);
86
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 */
112 static 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)
118 {
119 u64 bno;
120 int r;
121
122 r = ceph_calc_raw_layout(osdc, layout, vino.snap, off,
123 plen, &bno, req, op);
124 if (r < 0)
125 return r;
126
127 snprintf(req->r_oid, sizeof(req->r_oid), "%llx.%08llx", vino.ino, bno);
128 req->r_oid_len = strlen(req->r_oid);
129
130 return r;
131 }
132
133 /*
134 * requests
135 */
136 void ceph_osdc_release_request(struct kref *kref)
137 {
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);
144 if (req->r_con_filling_msg) {
145 dout("%s revoking pages %p from con %p\n", __func__,
146 req->r_pages, req->r_con_filling_msg);
147 ceph_msg_revoke_incoming(req->r_reply);
148 req->r_con_filling_msg->ops->put(req->r_con_filling_msg);
149 }
150 if (req->r_reply)
151 ceph_msg_put(req->r_reply);
152 if (req->r_own_pages)
153 ceph_release_page_vector(req->r_pages,
154 req->r_num_pages);
155 #ifdef CONFIG_BLOCK
156 if (req->r_bio)
157 bio_put(req->r_bio);
158 #endif
159 ceph_put_snap_context(req->r_snapc);
160 ceph_pagelist_release(&req->r_trail);
161 if (req->r_mempool)
162 mempool_free(req, req->r_osdc->req_mempool);
163 else
164 kfree(req);
165 }
166 EXPORT_SYMBOL(ceph_osdc_release_request);
167
168 static int get_num_ops(struct ceph_osd_req_op *ops)
169 {
170 int i = 0;
171
172 while (ops[i].op)
173 i++;
174
175 return i;
176 }
177
178 struct ceph_osd_request *ceph_osdc_alloc_request(struct ceph_osd_client *osdc,
179 int flags,
180 struct ceph_snap_context *snapc,
181 struct ceph_osd_req_op *ops,
182 bool use_mempool,
183 gfp_t gfp_flags,
184 struct page **pages,
185 struct bio *bio)
186 {
187 struct ceph_osd_request *req;
188 struct ceph_msg *msg;
189 int num_op = get_num_ops(ops);
190 size_t msg_size = sizeof(struct ceph_osd_request_head);
191
192 msg_size += num_op*sizeof(struct ceph_osd_op);
193
194 if (use_mempool) {
195 req = mempool_alloc(osdc->req_mempool, gfp_flags);
196 memset(req, 0, sizeof(*req));
197 } else {
198 req = kzalloc(sizeof(*req), gfp_flags);
199 }
200 if (req == NULL)
201 return NULL;
202
203 req->r_osdc = osdc;
204 req->r_mempool = use_mempool;
205
206 kref_init(&req->r_kref);
207 init_completion(&req->r_completion);
208 init_completion(&req->r_safe_completion);
209 RB_CLEAR_NODE(&req->r_node);
210 INIT_LIST_HEAD(&req->r_unsafe_item);
211 INIT_LIST_HEAD(&req->r_linger_item);
212 INIT_LIST_HEAD(&req->r_linger_osd);
213 INIT_LIST_HEAD(&req->r_req_lru_item);
214 INIT_LIST_HEAD(&req->r_osd_item);
215
216 req->r_flags = flags;
217
218 WARN_ON((flags & (CEPH_OSD_FLAG_READ|CEPH_OSD_FLAG_WRITE)) == 0);
219
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,
225 OSD_OPREPLY_FRONT_LEN, gfp_flags, true);
226 if (!msg) {
227 ceph_osdc_put_request(req);
228 return NULL;
229 }
230 req->r_reply = msg;
231
232 ceph_pagelist_init(&req->r_trail);
233
234 /* create request message; allow space for oid */
235 msg_size += MAX_OBJ_NAME_SIZE;
236 if (snapc)
237 msg_size += sizeof(u64) * snapc->num_snaps;
238 if (use_mempool)
239 msg = ceph_msgpool_get(&osdc->msgpool_op, 0);
240 else
241 msg = ceph_msg_new(CEPH_MSG_OSD_OP, msg_size, gfp_flags, true);
242 if (!msg) {
243 ceph_osdc_put_request(req);
244 return NULL;
245 }
246
247 memset(msg->front.iov_base, 0, msg->front.iov_len);
248
249 req->r_request = msg;
250 req->r_pages = pages;
251 #ifdef CONFIG_BLOCK
252 if (bio) {
253 req->r_bio = bio;
254 bio_get(req->r_bio);
255 }
256 #endif
257
258 return req;
259 }
260 EXPORT_SYMBOL(ceph_osdc_alloc_request);
261
262 static 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
268 switch (src->op) {
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:
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;
288 ceph_pagelist_append(&req->r_trail, src->xattr.name,
289 src->xattr.name_len);
290 ceph_pagelist_append(&req->r_trail, src->xattr.val,
291 src->xattr.value_len);
292 break;
293 case CEPH_OSD_OP_CALL:
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
298 ceph_pagelist_append(&req->r_trail, src->cls.class_name,
299 src->cls.class_len);
300 ceph_pagelist_append(&req->r_trail, src->cls.method_name,
301 src->cls.method_len);
302 ceph_pagelist_append(&req->r_trail, src->cls.indata,
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;
308 case CEPH_OSD_OP_STARTSYNC:
309 break;
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
315 ceph_pagelist_append(&req->r_trail,
316 &prot_ver, sizeof(prot_ver));
317 ceph_pagelist_append(&req->r_trail,
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;
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
334 /*
335 * build new request AND message
336 *
337 */
338 void ceph_osdc_build_request(struct ceph_osd_request *req,
339 u64 off, u64 *plen,
340 struct ceph_osd_req_op *src_ops,
341 struct ceph_snap_context *snapc,
342 struct timespec *mtime)
343 {
344 struct ceph_msg *msg = req->r_request;
345 struct ceph_osd_request_head *head;
346 struct ceph_osd_req_op *src_op;
347 struct ceph_osd_op *op;
348 void *p;
349 int num_op = get_num_ops(src_ops);
350 size_t msg_size = sizeof(*head) + num_op*sizeof(*op);
351 int flags = req->r_flags;
352 u64 data_len = 0;
353 int i;
354
355 head = msg->front.iov_base;
356 op = (void *)(head + 1);
357 p = (void *)(op + num_op);
358
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);
366
367
368 /* fill in oid */
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;
372
373 src_op = src_ops;
374 while (src_op->op) {
375 osd_req_encode_op(req, op, src_op);
376 src_op++;
377 op++;
378 }
379
380 data_len += req->r_trail.length;
381
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
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
399 req->r_request->page_alignment = req->r_page_alignment;
400
401 BUG_ON(p > msg->front.iov_base + msg->front.iov_len);
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);
405 return;
406 }
407 EXPORT_SYMBOL(ceph_osdc_build_request);
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 */
420 struct 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,
430 bool use_mempool, int num_reply,
431 int page_align)
432 {
433 struct ceph_osd_req_op ops[3];
434 struct ceph_osd_request *req;
435 int r;
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,
451 use_mempool,
452 GFP_NOFS, NULL, NULL);
453 if (!req)
454 return ERR_PTR(-ENOMEM);
455
456 /* calculate max write size */
457 r = calc_layout(osdc, vino, layout, off, plen, req, ops);
458 if (r < 0)
459 return ERR_PTR(r);
460 req->r_file_layout = *layout; /* keep a copy */
461
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);
465 req->r_page_alignment = page_align;
466
467 ceph_osdc_build_request(req, off, plen, ops,
468 snapc,
469 mtime);
470
471 return req;
472 }
473 EXPORT_SYMBOL(ceph_osdc_new_request);
474
475 /*
476 * We keep osd requests in an rbtree, sorted by ->r_tid.
477 */
478 static 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
500 static 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
518 static 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
540 /*
541 * Resubmit requests pending on the given osd.
542 */
543 static void __kick_osd_requests(struct ceph_osd_client *osdc,
544 struct ceph_osd *osd)
545 {
546 struct ceph_osd_request *req, *nreq;
547 int err;
548
549 dout("__kick_osd_requests osd%d\n", osd->o_osd);
550 err = __reset_osd(osdc, osd);
551 if (err)
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);
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) {
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));
569 __register_request(osdc, req);
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);
573 dout("requeued lingering %p tid %llu osd%d\n", req, req->r_tid,
574 osd->o_osd);
575 }
576 }
577
578 /*
579 * If the osd connection drops, we need to resubmit all requests.
580 */
581 static 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;
590 down_read(&osdc->map_sem);
591 mutex_lock(&osdc->request_mutex);
592 __kick_osd_requests(osdc, osd);
593 mutex_unlock(&osdc->request_mutex);
594 send_queued(osdc);
595 up_read(&osdc->map_sem);
596 }
597
598 /*
599 * Track open sessions with osds.
600 */
601 static struct ceph_osd *create_osd(struct ceph_osd_client *osdc, int onum)
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;
611 osd->o_osd = onum;
612 RB_CLEAR_NODE(&osd->o_node);
613 INIT_LIST_HEAD(&osd->o_requests);
614 INIT_LIST_HEAD(&osd->o_linger_requests);
615 INIT_LIST_HEAD(&osd->o_osd_lru);
616 osd->o_incarnation = 1;
617
618 ceph_con_init(&osd->o_con, osd, &osd_con_ops, &osdc->client->msgr);
619
620 INIT_LIST_HEAD(&osd->o_keepalive_item);
621 return osd;
622 }
623
624 static 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
636 static 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);
640 if (atomic_dec_and_test(&osd->o_ref) && osd->o_auth.authorizer) {
641 struct ceph_auth_client *ac = osd->o_osdc->client->monc.auth;
642
643 if (ac->ops && ac->ops->destroy_authorizer)
644 ac->ops->destroy_authorizer(ac, osd->o_auth.authorizer);
645 kfree(osd);
646 }
647 }
648
649 /*
650 * remove an osd from our map
651 */
652 static void __remove_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd)
653 {
654 dout("__remove_osd %p\n", osd);
655 BUG_ON(!list_empty(&osd->o_requests));
656 rb_erase(&osd->o_node, &osdc->osds);
657 list_del_init(&osd->o_osd_lru);
658 ceph_con_close(&osd->o_con);
659 put_osd(osd);
660 }
661
662 static void remove_all_osds(struct ceph_osd_client *osdc)
663 {
664 dout("%s %p\n", __func__, osdc);
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
674 static 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);
680 osd->lru_ttl = jiffies + osdc->client->options->osd_idle_ttl * HZ;
681 }
682
683 static 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
690 static void remove_old_osds(struct ceph_osd_client *osdc)
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) {
697 if (time_before(jiffies, osd->lru_ttl))
698 break;
699 __remove_osd(osdc, osd);
700 }
701 mutex_unlock(&osdc->request_mutex);
702 }
703
704 /*
705 * reset osd connect
706 */
707 static int __reset_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd)
708 {
709 struct ceph_entity_addr *peer_addr;
710
711 dout("__reset_osd %p osd%d\n", osd, osd->o_osd);
712 if (list_empty(&osd->o_requests) &&
713 list_empty(&osd->o_linger_requests)) {
714 __remove_osd(osdc, osd);
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
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;
729
730 return -EAGAIN;
731 }
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;
738 }
739
740 static 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
746 dout("__insert_osd %p osd%d\n", new, new->o_osd);
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
762 static 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
779 static void __schedule_osd_timeout(struct ceph_osd_client *osdc)
780 {
781 schedule_delayed_work(&osdc->timeout_work,
782 osdc->client->options->osd_keepalive_timeout * HZ);
783 }
784
785 static void __cancel_osd_timeout(struct ceph_osd_client *osdc)
786 {
787 cancel_delayed_work(&osdc->timeout_work);
788 }
789
790 /*
791 * Register request, assign tid. If this is the first request, set up
792 * the timeout event.
793 */
794 static void __register_request(struct ceph_osd_client *osdc,
795 struct ceph_osd_request *req)
796 {
797 req->r_tid = ++osdc->last_tid;
798 req->r_request->hdr.tid = cpu_to_le64(req->r_tid);
799 dout("__register_request %p tid %lld\n", req, req->r_tid);
800 __insert_request(osdc, req);
801 ceph_osdc_get_request(req);
802 osdc->num_requests++;
803 if (osdc->num_requests == 1) {
804 dout(" first request, scheduling timeout\n");
805 __schedule_osd_timeout(osdc);
806 }
807 }
808
809 static 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);
814 mutex_unlock(&osdc->request_mutex);
815 }
816
817 /*
818 * called under osdc->request_mutex
819 */
820 static void __unregister_request(struct ceph_osd_client *osdc,
821 struct ceph_osd_request *req)
822 {
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
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
833 if (req->r_osd) {
834 /* make sure the original request isn't in flight. */
835 ceph_msg_revoke(req->r_request);
836
837 list_del_init(&req->r_osd_item);
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);
841 __move_osd_to_lru(osdc, req->r_osd);
842 }
843 if (list_empty(&req->r_linger_item))
844 req->r_osd = NULL;
845 }
846
847 list_del_init(&req->r_req_lru_item);
848 ceph_osdc_put_request(req);
849
850 if (osdc->num_requests == 0) {
851 dout(" no requests, canceling timeout\n");
852 __cancel_osd_timeout(osdc);
853 }
854 }
855
856 /*
857 * Cancel a previously queued request message
858 */
859 static void __cancel_request(struct ceph_osd_request *req)
860 {
861 if (req->r_sent && req->r_osd) {
862 ceph_msg_revoke(req->r_request);
863 req->r_sent = 0;
864 }
865 }
866
867 static 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);
872 if (req->r_osd)
873 list_add_tail(&req->r_linger_osd,
874 &req->r_osd->o_linger_requests);
875 }
876
877 static void __unregister_linger_request(struct ceph_osd_client *osdc,
878 struct ceph_osd_request *req)
879 {
880 dout("__unregister_linger_request %p\n", req);
881 list_del_init(&req->r_linger_item);
882 if (req->r_osd) {
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 }
890 if (list_empty(&req->r_osd_item))
891 req->r_osd = NULL;
892 }
893 }
894
895 void 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 }
905 EXPORT_SYMBOL(ceph_osdc_unregister_linger_request);
906
907 void 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 }
920 EXPORT_SYMBOL(ceph_osdc_set_request_linger);
921
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
925 * no up osd, set r_osd to NULL. Move the request to the appropriate list
926 * (unsent, homeless) or leave on in-flight lru.
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 */
932 static int __map_request(struct ceph_osd_client *osdc,
933 struct ceph_osd_request *req, int force_resend)
934 {
935 struct ceph_osd_request_head *reqhead = req->r_request->front.iov_base;
936 struct ceph_pg pgid;
937 int acting[CEPH_PG_MAX_SIZE];
938 int o = -1, num = 0;
939 int err;
940
941 dout("map_request %p tid %lld\n", req, req->r_tid);
942 err = ceph_calc_object_layout(&reqhead->layout, req->r_oid,
943 &req->r_file_layout, osdc->osdmap);
944 if (err) {
945 list_move(&req->r_req_lru_item, &osdc->req_notarget);
946 return err;
947 }
948 pgid = reqhead->layout.ol_pgid;
949 req->r_pgid = pgid;
950
951 err = ceph_calc_pg_acting(osdc->osdmap, pgid, acting);
952 if (err > 0) {
953 o = acting[0];
954 num = err;
955 }
956
957 if ((!force_resend &&
958 req->r_osd && req->r_osd->o_osd == o &&
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) ||
962 (req->r_osd == NULL && o == -1))
963 return 0; /* no change */
964
965 dout("map_request tid %llu pgid %d.%x osd%d (was osd%d)\n",
966 req->r_tid, le32_to_cpu(pgid.pool), le16_to_cpu(pgid.ps), o,
967 req->r_osd ? req->r_osd->o_osd : -1);
968
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
973 if (req->r_osd) {
974 __cancel_request(req);
975 list_del_init(&req->r_osd_item);
976 req->r_osd = NULL;
977 }
978
979 req->r_osd = __lookup_osd(osdc, o);
980 if (!req->r_osd && o >= 0) {
981 err = -ENOMEM;
982 req->r_osd = create_osd(osdc, o);
983 if (!req->r_osd) {
984 list_move(&req->r_req_lru_item, &osdc->req_notarget);
985 goto out;
986 }
987
988 dout("map_request osd %p is osd%d\n", req->r_osd, o);
989 __insert_osd(osdc, req->r_osd);
990
991 ceph_con_open(&req->r_osd->o_con,
992 CEPH_ENTITY_TYPE_OSD, o,
993 &osdc->osdmap->osd_addr[o]);
994 }
995
996 if (req->r_osd) {
997 __remove_osd_from_lru(req->r_osd);
998 list_add(&req->r_osd_item, &req->r_osd->o_requests);
999 list_move(&req->r_req_lru_item, &osdc->req_unsent);
1000 } else {
1001 list_move(&req->r_req_lru_item, &osdc->req_notarget);
1002 }
1003 err = 1; /* osd or pg changed */
1004
1005 out:
1006 return err;
1007 }
1008
1009 /*
1010 * caller should hold map_sem (for read) and request_mutex
1011 */
1012 static void __send_request(struct ceph_osd_client *osdc,
1013 struct ceph_osd_request *req)
1014 {
1015 struct ceph_osd_request_head *reqhead;
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
1025 req->r_stamp = jiffies;
1026 list_move_tail(&req->r_req_lru_item, &osdc->req_lru);
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;
1031 }
1032
1033 /*
1034 * Send any requests in the queue (req_unsent).
1035 */
1036 static 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
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 */
1057 static 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);
1061 struct ceph_osd_request *req;
1062 struct ceph_osd *osd;
1063 unsigned long keepalive =
1064 osdc->client->options->osd_keepalive_timeout * HZ;
1065 struct list_head slow_osds;
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);
1072
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) {
1080 if (time_before(jiffies, req->r_stamp + keepalive))
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",
1086 req->r_tid, osd->o_osd);
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);
1093 ceph_con_keepalive(&osd->o_con);
1094 }
1095
1096 __schedule_osd_timeout(osdc);
1097 mutex_unlock(&osdc->request_mutex);
1098 send_queued(osdc);
1099 up_read(&osdc->map_sem);
1100 }
1101
1102 static 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 =
1108 osdc->client->options->osd_idle_ttl * HZ >> 2;
1109
1110 dout("osds timeout\n");
1111 down_read(&osdc->map_sem);
1112 remove_old_osds(osdc);
1113 up_read(&osdc->map_sem);
1114
1115 schedule_delayed_work(&osdc->osds_timeout_work,
1116 round_jiffies_relative(delay));
1117 }
1118
1119 static 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
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 */
1130 static void handle_reply(struct ceph_osd_client *osdc, struct ceph_msg *msg,
1131 struct ceph_connection *con)
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;
1137 s32 result;
1138
1139 tid = le64_to_cpu(msg->hdr.tid);
1140 if (msg->front.iov_len < sizeof(*rhead))
1141 goto bad;
1142 numops = le32_to_cpu(rhead->num_ops);
1143 object_len = le32_to_cpu(rhead->object_len);
1144 result = le32_to_cpu(rhead->result);
1145 if (msg->front.iov_len != sizeof(*rhead) + object_len +
1146 numops * sizeof(struct ceph_osd_op))
1147 goto bad;
1148 dout("handle_reply %p tid %llu result %d\n", msg, tid, (int)result);
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
1160 /*
1161 * if this connection filled our message, drop our reference now, to
1162 * avoid a (safe but slower) revoke later.
1163 */
1164 if (req->r_con_filling_msg == con && req->r_reply == msg) {
1165 dout(" dropping con_filling_msg ref %p\n", con);
1166 req->r_con_filling_msg = NULL;
1167 con->ops->put(con);
1168 }
1169
1170 if (!req->r_got_reply) {
1171 unsigned int bytes;
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);
1186 mutex_unlock(&osdc->request_mutex);
1187 goto done;
1188 }
1189
1190 dout("handle_reply tid %llu flags %d\n", tid, flags);
1191
1192 if (req->r_linger && (flags & CEPH_OSD_FLAG_ONDISK))
1193 __register_linger_request(osdc, req);
1194
1195 /* either this is a read, or we got the safe response */
1196 if (result < 0 ||
1197 (flags & CEPH_OSD_FLAG_ONDISK) ||
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
1206 complete_all(&req->r_completion);
1207
1208 if (flags & CEPH_OSD_FLAG_ONDISK)
1209 complete_request(req);
1210
1211 done:
1212 dout("req=%p req->r_linger=%d\n", req, req->r_linger);
1213 ceph_osdc_put_request(req);
1214 return;
1215
1216 bad:
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));
1220 ceph_msg_dump(msg);
1221 }
1222
1223 static void reset_changed_osds(struct ceph_osd_client *osdc)
1224 {
1225 struct rb_node *p, *n;
1226
1227 for (p = rb_first(&osdc->osds); p; p = n) {
1228 struct ceph_osd *osd = rb_entry(p, struct ceph_osd, o_node);
1229
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);
1237 }
1238 }
1239
1240 /*
1241 * Requeue requests whose mapping to an OSD has changed. If requests map to
1242 * no osd, request a new map.
1243 *
1244 * Caller should hold map_sem for read.
1245 */
1246 static void kick_requests(struct ceph_osd_client *osdc, int force_resend)
1247 {
1248 struct ceph_osd_request *req, *nreq;
1249 struct rb_node *p;
1250 int needmap = 0;
1251 int err;
1252
1253 dout("kick_requests %s\n", force_resend ? " (force resend)" : "");
1254 mutex_lock(&osdc->request_mutex);
1255 for (p = rb_first(&osdc->requests); p; ) {
1256 req = rb_entry(p, struct ceph_osd_request, r_node);
1257 p = rb_next(p);
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
1276 err = __map_request(osdc, req, force_resend);
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) {
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);
1287 req->r_flags |= CEPH_OSD_FLAG_RETRY;
1288 }
1289 }
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
1296 err = __map_request(osdc, req, force_resend);
1297 dout("__map_request returned %d\n", err);
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;
1306 }
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);
1310 __register_request(osdc, req);
1311 __unregister_linger_request(osdc, req);
1312 }
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 }
1319 reset_changed_osds(osdc);
1320 }
1321
1322
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 */
1330 void 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));
1346 if (ceph_check_fsid(osdc->client, &fsid) < 0)
1347 return;
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);
1356 epoch = ceph_decode_32(&p);
1357 maplen = ceph_decode_32(&p);
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,
1365 &osdc->client->msgr);
1366 if (IS_ERR(newmap)) {
1367 err = PTR_ERR(newmap);
1368 goto bad;
1369 }
1370 BUG_ON(!newmap);
1371 if (newmap != osdc->osdmap) {
1372 ceph_osdmap_destroy(osdc->osdmap);
1373 osdc->osdmap = newmap;
1374 }
1375 kick_requests(osdc, 0);
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);
1391 epoch = ceph_decode_32(&p);
1392 maplen = ceph_decode_32(&p);
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 {
1402 int skipped_map = 0;
1403
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 }
1410 BUG_ON(!newmap);
1411 oldmap = osdc->osdmap;
1412 osdc->osdmap = newmap;
1413 if (oldmap) {
1414 if (oldmap->epoch + 1 < newmap->epoch)
1415 skipped_map = 1;
1416 ceph_osdmap_destroy(oldmap);
1417 }
1418 kick_requests(osdc, skipped_map);
1419 }
1420 p += maplen;
1421 nr_maps--;
1422 }
1423
1424 done:
1425 downgrade_write(&osdc->map_sem);
1426 ceph_monc_got_osdmap(&osdc->client->monc, osdc->osdmap->epoch);
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
1436 send_queued(osdc);
1437 up_read(&osdc->map_sem);
1438 wake_up_all(&osdc->client->auth_wq);
1439 return;
1440
1441 bad:
1442 pr_err("osdc handle_map corrupt msg\n");
1443 ceph_msg_dump(msg);
1444 up_write(&osdc->map_sem);
1445 return;
1446 }
1447
1448 /*
1449 * watch/notify callback event infrastructure
1450 *
1451 * These callbacks are used both for watch and notify operations.
1452 */
1453 static 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
1462 static void get_event(struct ceph_osd_event *event)
1463 {
1464 kref_get(&event->kref);
1465 }
1466
1467 void ceph_osdc_put_event(struct ceph_osd_event *event)
1468 {
1469 kref_put(&event->kref, __release_event);
1470 }
1471 EXPORT_SYMBOL(ceph_osdc_put_event);
1472
1473 static 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
1495 static 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
1515 static 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
1528 int 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);
1545 RB_CLEAR_NODE(&event->node);
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 }
1558 EXPORT_SYMBOL(ceph_osdc_create_event);
1559
1560 void 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 }
1570 EXPORT_SYMBOL(ceph_osdc_cancel_event);
1571
1572
1573 static 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 */
1594 void 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);
1624 if (!event_work) {
1625 dout("ERROR: could not allocate event_work\n");
1626 goto done_err;
1627 }
1628 INIT_WORK(&event_work->work, do_event_work);
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
1641 done_err:
1642 complete(&event->completion);
1643 ceph_osdc_put_event(event);
1644 return;
1645
1646 bad:
1647 pr_err("osdc handle_watch_notify corrupt msg\n");
1648 return;
1649 }
1650
1651 int 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 }
1664 EXPORT_SYMBOL(ceph_osdc_wait_event);
1665
1666 /*
1667 * Register request, send initial attempt.
1668 */
1669 int ceph_osdc_start_request(struct ceph_osd_client *osdc,
1670 struct ceph_osd_request *req,
1671 bool nofail)
1672 {
1673 int rc = 0;
1674
1675 req->r_request->pages = req->r_pages;
1676 req->r_request->nr_pages = req->r_num_pages;
1677 #ifdef CONFIG_BLOCK
1678 req->r_request->bio = req->r_bio;
1679 #endif
1680 req->r_request->trail = &req->r_trail;
1681
1682 register_request(osdc, req);
1683
1684 down_read(&osdc->map_sem);
1685 mutex_lock(&osdc->request_mutex);
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) {
1692 rc = __map_request(osdc, req, 0);
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 }
1699 goto out_unlock;
1700 }
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 {
1705 __send_request(osdc, req);
1706 }
1707 rc = 0;
1708 }
1709
1710 out_unlock:
1711 mutex_unlock(&osdc->request_mutex);
1712 up_read(&osdc->map_sem);
1713 return rc;
1714 }
1715 EXPORT_SYMBOL(ceph_osdc_start_request);
1716
1717 /*
1718 * wait for a request to complete
1719 */
1720 int 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);
1729 __unregister_request(osdc, req);
1730 mutex_unlock(&osdc->request_mutex);
1731 complete_request(req);
1732 dout("wait_request tid %llu canceled/timed out\n", req->r_tid);
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 }
1739 EXPORT_SYMBOL(ceph_osdc_wait_request);
1740
1741 /*
1742 * sync - wait for all in-flight requests to flush. avoid starvation.
1743 */
1744 void 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 }
1773 EXPORT_SYMBOL(ceph_osdc_sync);
1774
1775 /*
1776 * init, shutdown
1777 */
1778 int 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);
1789 osdc->last_tid = 0;
1790 osdc->osds = RB_ROOT;
1791 INIT_LIST_HEAD(&osdc->osd_lru);
1792 osdc->requests = RB_ROOT;
1793 INIT_LIST_HEAD(&osdc->req_lru);
1794 INIT_LIST_HEAD(&osdc->req_unsent);
1795 INIT_LIST_HEAD(&osdc->req_notarget);
1796 INIT_LIST_HEAD(&osdc->req_linger);
1797 osdc->num_requests = 0;
1798 INIT_DELAYED_WORK(&osdc->timeout_work, handle_timeout);
1799 INIT_DELAYED_WORK(&osdc->osds_timeout_work, handle_osds_timeout);
1800 spin_lock_init(&osdc->event_lock);
1801 osdc->event_tree = RB_ROOT;
1802 osdc->event_count = 0;
1803
1804 schedule_delayed_work(&osdc->osds_timeout_work,
1805 round_jiffies_relative(osdc->client->options->osd_idle_ttl * HZ));
1806
1807 err = -ENOMEM;
1808 osdc->req_mempool = mempool_create_kmalloc_pool(10,
1809 sizeof(struct ceph_osd_request));
1810 if (!osdc->req_mempool)
1811 goto out;
1812
1813 err = ceph_msgpool_init(&osdc->msgpool_op, CEPH_MSG_OSD_OP,
1814 OSD_OP_FRONT_LEN, 10, true,
1815 "osd_op");
1816 if (err < 0)
1817 goto out_mempool;
1818 err = ceph_msgpool_init(&osdc->msgpool_op_reply, CEPH_MSG_OSD_OPREPLY,
1819 OSD_OPREPLY_FRONT_LEN, 10, true,
1820 "osd_op_reply");
1821 if (err < 0)
1822 goto out_msgpool;
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 }
1830 return 0;
1831
1832 out_msgpool:
1833 ceph_msgpool_destroy(&osdc->msgpool_op);
1834 out_mempool:
1835 mempool_destroy(osdc->req_mempool);
1836 out:
1837 return err;
1838 }
1839 EXPORT_SYMBOL(ceph_osdc_init);
1840
1841 void ceph_osdc_stop(struct ceph_osd_client *osdc)
1842 {
1843 flush_workqueue(osdc->notify_wq);
1844 destroy_workqueue(osdc->notify_wq);
1845 cancel_delayed_work_sync(&osdc->timeout_work);
1846 cancel_delayed_work_sync(&osdc->osds_timeout_work);
1847 if (osdc->osdmap) {
1848 ceph_osdmap_destroy(osdc->osdmap);
1849 osdc->osdmap = NULL;
1850 }
1851 remove_all_osds(osdc);
1852 mempool_destroy(osdc->req_mempool);
1853 ceph_msgpool_destroy(&osdc->msgpool_op);
1854 ceph_msgpool_destroy(&osdc->msgpool_op_reply);
1855 }
1856 EXPORT_SYMBOL(ceph_osdc_stop);
1857
1858 /*
1859 * Read some contiguous pages. If we cross a stripe boundary, shorten
1860 * *plen. Return number of bytes read, or error.
1861 */
1862 int 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,
1866 struct page **pages, int num_pages, int page_align)
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,
1876 false, 1, page_align);
1877 if (IS_ERR(req))
1878 return PTR_ERR(req);
1879
1880 /* it may be a short read due to an object boundary */
1881 req->r_pages = pages;
1882
1883 dout("readpages final extent is %llu~%llu (%d pages align %d)\n",
1884 off, *plen, req->r_num_pages, page_align);
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 }
1894 EXPORT_SYMBOL(ceph_osdc_readpages);
1895
1896 /*
1897 * do a synchronous write on N pages
1898 */
1899 int 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;
1910 int page_align = off & ~PAGE_MASK;
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,
1919 nofail, 1, page_align);
1920 if (IS_ERR(req))
1921 return PTR_ERR(req);
1922
1923 /* it may be a short write due to an object boundary */
1924 req->r_pages = pages;
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 }
1938 EXPORT_SYMBOL(ceph_osdc_writepages);
1939
1940 /*
1941 * handle incoming message
1942 */
1943 static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
1944 {
1945 struct ceph_osd *osd = con->private;
1946 struct ceph_osd_client *osdc;
1947 int type = le16_to_cpu(msg->hdr.type);
1948
1949 if (!osd)
1950 goto out;
1951 osdc = osd->o_osdc;
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:
1958 handle_reply(osdc, msg, con);
1959 break;
1960 case CEPH_MSG_WATCH_NOTIFY:
1961 handle_watch_notify(osdc, msg);
1962 break;
1963
1964 default:
1965 pr_err("received unknown message type %d %s\n", type,
1966 ceph_msg_type_name(type));
1967 }
1968 out:
1969 ceph_msg_put(msg);
1970 }
1971
1972 /*
1973 * lookup and return message for incoming reply. set up reply message
1974 * pages.
1975 */
1976 static struct ceph_msg *get_reply(struct ceph_connection *con,
1977 struct ceph_msg_header *hdr,
1978 int *skip)
1979 {
1980 struct ceph_osd *osd = con->private;
1981 struct ceph_osd_client *osdc = osd->o_osdc;
1982 struct ceph_msg *m;
1983 struct ceph_osd_request *req;
1984 int front = le32_to_cpu(hdr->front_len);
1985 int data_len = le32_to_cpu(hdr->data_len);
1986 u64 tid;
1987
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;
1994 dout("get_reply unknown tid %llu from osd%d\n", tid,
1995 osd->o_osd);
1996 goto out;
1997 }
1998
1999 if (req->r_con_filling_msg) {
2000 dout("%s revoking msg %p from old con %p\n", __func__,
2001 req->r_reply, req->r_con_filling_msg);
2002 ceph_msg_revoke_incoming(req->r_reply);
2003 req->r_con_filling_msg->ops->put(req->r_con_filling_msg);
2004 req->r_con_filling_msg = NULL;
2005 }
2006
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);
2010 m = ceph_msg_new(CEPH_MSG_OSD_OPREPLY, front, GFP_NOFS, false);
2011 if (!m)
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
2018 if (data_len > 0) {
2019 int want = calc_pages_for(req->r_page_alignment, data_len);
2020
2021 if (unlikely(req->r_num_pages < want)) {
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);
2025 *skip = 1;
2026 ceph_msg_put(m);
2027 m = NULL;
2028 goto out;
2029 }
2030 m->pages = req->r_pages;
2031 m->nr_pages = req->r_num_pages;
2032 m->page_alignment = req->r_page_alignment;
2033 #ifdef CONFIG_BLOCK
2034 m->bio = req->r_bio;
2035 #endif
2036 }
2037 *skip = 0;
2038 req->r_con_filling_msg = con->ops->get(con);
2039 dout("get_reply tid %lld %p\n", tid, m);
2040
2041 out:
2042 mutex_unlock(&osdc->request_mutex);
2043 return m;
2044
2045 }
2046
2047 static 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
2055 *skip = 0;
2056 switch (type) {
2057 case CEPH_MSG_OSD_MAP:
2058 case CEPH_MSG_WATCH_NOTIFY:
2059 return ceph_msg_new(type, front, GFP_NOFS, false);
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 }
2068 }
2069
2070 /*
2071 * Wrappers to refcount containing ceph_osd struct
2072 */
2073 static 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
2081 static void put_osd_con(struct ceph_connection *con)
2082 {
2083 struct ceph_osd *osd = con->private;
2084 put_osd(osd);
2085 }
2086
2087 /*
2088 * authentication
2089 */
2090 /*
2091 * Note: returned pointer is the address of a structure that's
2092 * managed separately. Caller must *not* attempt to free it.
2093 */
2094 static struct ceph_auth_handshake *get_authorizer(struct ceph_connection *con,
2095 int *proto, int force_new)
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;
2100 struct ceph_auth_handshake *auth = &o->o_auth;
2101
2102 if (force_new && auth->authorizer) {
2103 if (ac->ops && ac->ops->destroy_authorizer)
2104 ac->ops->destroy_authorizer(ac, auth->authorizer);
2105 auth->authorizer = NULL;
2106 }
2107 if (!auth->authorizer && ac->ops && ac->ops->create_authorizer) {
2108 int ret = ac->ops->create_authorizer(ac, CEPH_ENTITY_TYPE_OSD,
2109 auth);
2110 if (ret)
2111 return ERR_PTR(ret);
2112 }
2113 *proto = ac->protocol;
2114
2115 return auth;
2116 }
2117
2118
2119 static 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
2125 /*
2126 * XXX If ac->ops or ac->ops->verify_authorizer_reply is null,
2127 * XXX which do we do: succeed or fail?
2128 */
2129 return ac->ops->verify_authorizer_reply(ac, o->o_auth.authorizer, len);
2130 }
2131
2132 static 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
2138 if (ac->ops && ac->ops->invalidate_authorizer)
2139 ac->ops->invalidate_authorizer(ac, CEPH_ENTITY_TYPE_OSD);
2140
2141 return ceph_monc_validate_auth(&osdc->client->monc);
2142 }
2143
2144 static const struct ceph_connection_operations osd_con_ops = {
2145 .get = get_osd_con,
2146 .put = put_osd_con,
2147 .dispatch = dispatch,
2148 .get_authorizer = get_authorizer,
2149 .verify_authorizer_reply = verify_authorizer_reply,
2150 .invalidate_authorizer = invalidate_authorizer,
2151 .alloc_msg = alloc_msg,
2152 .fault = osd_reset,
2153 };
Linux kernel - Deliverables
This page took 0.144144 seconds and 5 git commands to generate.