?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
KEYS: add missing permission check for request_key() destination
[linux/fpc-iii.git] / net / ceph / osd_client.c
blobb30776ab1e6e5e3443b86eda907527d5b7181a9d
1
2 #include <linux/ceph/ceph_debug.h>
3
4 #include <linux/module.h>
5 #include <linux/err.h>
6 #include <linux/highmem.h>
7 #include <linux/mm.h>
8 #include <linux/pagemap.h>
9 #include <linux/slab.h>
10 #include <linux/uaccess.h>
11 #ifdef CONFIG_BLOCK
12 #include <linux/bio.h>
13 #endif
14
15 #include <linux/ceph/libceph.h>
16 #include <linux/ceph/osd_client.h>
17 #include <linux/ceph/messenger.h>
18 #include <linux/ceph/decode.h>
19 #include <linux/ceph/auth.h>
20 #include <linux/ceph/pagelist.h>
21
22 #define OSD_OP_FRONT_LEN 4096
23 #define OSD_OPREPLY_FRONT_LEN 512
24
25 static struct kmem_cache *ceph_osd_request_cache;
26
27 static const struct ceph_connection_operations osd_con_ops;
28
29 static void __send_queued(struct ceph_osd_client *osdc);
30 static int __reset_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd);
31 static void __register_request(struct ceph_osd_client *osdc,
32 struct ceph_osd_request *req);
33 static void __unregister_linger_request(struct ceph_osd_client *osdc,
34 struct ceph_osd_request *req);
35 static void __send_request(struct ceph_osd_client *osdc,
36 struct ceph_osd_request *req);
37
38 /*
39 * Implement client access to distributed object storage cluster.
40 *
41 * All data objects are stored within a cluster/cloud of OSDs, or
42 * "object storage devices." (Note that Ceph OSDs have _nothing_ to
43 * do with the T10 OSD extensions to SCSI.) Ceph OSDs are simply
44 * remote daemons serving up and coordinating consistent and safe
45 * access to storage.
46 *
47 * Cluster membership and the mapping of data objects onto storage devices
48 * are described by the osd map.
49 *
50 * We keep track of pending OSD requests (read, write), resubmit
51 * requests to different OSDs when the cluster topology/data layout
52 * change, or retry the affected requests when the communications
53 * channel with an OSD is reset.
54 */
55
56 /*
57 * calculate the mapping of a file extent onto an object, and fill out the
58 * request accordingly. shorten extent as necessary if it crosses an
59 * object boundary.
60 *
61 * fill osd op in request message.
62 */
63 static int calc_layout(struct ceph_file_layout *layout, u64 off, u64 *plen,
64 u64 *objnum, u64 *objoff, u64 *objlen)
65 {
66 u64 orig_len = *plen;
67 int r;
68
69 /* object extent? */
70 r = ceph_calc_file_object_mapping(layout, off, orig_len, objnum,
71 objoff, objlen);
72 if (r < 0)
73 return r;
74 if (*objlen < orig_len) {
75 *plen = *objlen;
76 dout(" skipping last %llu, final file extent %llu~%llu\n",
77 orig_len - *plen, off, *plen);
78 }
79
80 dout("calc_layout objnum=%llx %llu~%llu\n", *objnum, *objoff, *objlen);
81
82 return 0;
83 }
84
85 static void ceph_osd_data_init(struct ceph_osd_data *osd_data)
86 {
87 memset(osd_data, 0, sizeof (*osd_data));
88 osd_data->type = CEPH_OSD_DATA_TYPE_NONE;
89 }
90
91 static void ceph_osd_data_pages_init(struct ceph_osd_data *osd_data,
92 struct page **pages, u64 length, u32 alignment,
93 bool pages_from_pool, bool own_pages)
94 {
95 osd_data->type = CEPH_OSD_DATA_TYPE_PAGES;
96 osd_data->pages = pages;
97 osd_data->length = length;
98 osd_data->alignment = alignment;
99 osd_data->pages_from_pool = pages_from_pool;
100 osd_data->own_pages = own_pages;
101 }
102
103 static void ceph_osd_data_pagelist_init(struct ceph_osd_data *osd_data,
104 struct ceph_pagelist *pagelist)
105 {
106 osd_data->type = CEPH_OSD_DATA_TYPE_PAGELIST;
107 osd_data->pagelist = pagelist;
108 }
109
110 #ifdef CONFIG_BLOCK
111 static void ceph_osd_data_bio_init(struct ceph_osd_data *osd_data,
112 struct bio *bio, size_t bio_length)
113 {
114 osd_data->type = CEPH_OSD_DATA_TYPE_BIO;
115 osd_data->bio = bio;
116 osd_data->bio_length = bio_length;
117 }
118 #endif /* CONFIG_BLOCK */
119
120 #define osd_req_op_data(oreq, whch, typ, fld) \
121 ({ \
122 BUG_ON(whch >= (oreq)->r_num_ops); \
123 &(oreq)->r_ops[whch].typ.fld; \
124 })
125
126 static struct ceph_osd_data *
127 osd_req_op_raw_data_in(struct ceph_osd_request *osd_req, unsigned int which)
128 {
129 BUG_ON(which >= osd_req->r_num_ops);
130
131 return &osd_req->r_ops[which].raw_data_in;
132 }
133
134 struct ceph_osd_data *
135 osd_req_op_extent_osd_data(struct ceph_osd_request *osd_req,
136 unsigned int which)
137 {
138 return osd_req_op_data(osd_req, which, extent, osd_data);
139 }
140 EXPORT_SYMBOL(osd_req_op_extent_osd_data);
141
142 struct ceph_osd_data *
143 osd_req_op_cls_response_data(struct ceph_osd_request *osd_req,
144 unsigned int which)
145 {
146 return osd_req_op_data(osd_req, which, cls, response_data);
147 }
148 EXPORT_SYMBOL(osd_req_op_cls_response_data); /* ??? */
149
150 void osd_req_op_raw_data_in_pages(struct ceph_osd_request *osd_req,
151 unsigned int which, struct page **pages,
152 u64 length, u32 alignment,
153 bool pages_from_pool, bool own_pages)
154 {
155 struct ceph_osd_data *osd_data;
156
157 osd_data = osd_req_op_raw_data_in(osd_req, which);
158 ceph_osd_data_pages_init(osd_data, pages, length, alignment,
159 pages_from_pool, own_pages);
160 }
161 EXPORT_SYMBOL(osd_req_op_raw_data_in_pages);
162
163 void osd_req_op_extent_osd_data_pages(struct ceph_osd_request *osd_req,
164 unsigned int which, struct page **pages,
165 u64 length, u32 alignment,
166 bool pages_from_pool, bool own_pages)
167 {
168 struct ceph_osd_data *osd_data;
169
170 osd_data = osd_req_op_data(osd_req, which, extent, osd_data);
171 ceph_osd_data_pages_init(osd_data, pages, length, alignment,
172 pages_from_pool, own_pages);
173 }
174 EXPORT_SYMBOL(osd_req_op_extent_osd_data_pages);
175
176 void osd_req_op_extent_osd_data_pagelist(struct ceph_osd_request *osd_req,
177 unsigned int which, struct ceph_pagelist *pagelist)
178 {
179 struct ceph_osd_data *osd_data;
180
181 osd_data = osd_req_op_data(osd_req, which, extent, osd_data);
182 ceph_osd_data_pagelist_init(osd_data, pagelist);
183 }
184 EXPORT_SYMBOL(osd_req_op_extent_osd_data_pagelist);
185
186 #ifdef CONFIG_BLOCK
187 void osd_req_op_extent_osd_data_bio(struct ceph_osd_request *osd_req,
188 unsigned int which, struct bio *bio, size_t bio_length)
189 {
190 struct ceph_osd_data *osd_data;
191
192 osd_data = osd_req_op_data(osd_req, which, extent, osd_data);
193 ceph_osd_data_bio_init(osd_data, bio, bio_length);
194 }
195 EXPORT_SYMBOL(osd_req_op_extent_osd_data_bio);
196 #endif /* CONFIG_BLOCK */
197
198 static void osd_req_op_cls_request_info_pagelist(
199 struct ceph_osd_request *osd_req,
200 unsigned int which, struct ceph_pagelist *pagelist)
201 {
202 struct ceph_osd_data *osd_data;
203
204 osd_data = osd_req_op_data(osd_req, which, cls, request_info);
205 ceph_osd_data_pagelist_init(osd_data, pagelist);
206 }
207
208 void osd_req_op_cls_request_data_pagelist(
209 struct ceph_osd_request *osd_req,
210 unsigned int which, struct ceph_pagelist *pagelist)
211 {
212 struct ceph_osd_data *osd_data;
213
214 osd_data = osd_req_op_data(osd_req, which, cls, request_data);
215 ceph_osd_data_pagelist_init(osd_data, pagelist);
216 }
217 EXPORT_SYMBOL(osd_req_op_cls_request_data_pagelist);
218
219 void osd_req_op_cls_request_data_pages(struct ceph_osd_request *osd_req,
220 unsigned int which, struct page **pages, u64 length,
221 u32 alignment, bool pages_from_pool, bool own_pages)
222 {
223 struct ceph_osd_data *osd_data;
224
225 osd_data = osd_req_op_data(osd_req, which, cls, request_data);
226 ceph_osd_data_pages_init(osd_data, pages, length, alignment,
227 pages_from_pool, own_pages);
228 }
229 EXPORT_SYMBOL(osd_req_op_cls_request_data_pages);
230
231 void osd_req_op_cls_response_data_pages(struct ceph_osd_request *osd_req,
232 unsigned int which, struct page **pages, u64 length,
233 u32 alignment, bool pages_from_pool, bool own_pages)
234 {
235 struct ceph_osd_data *osd_data;
236
237 osd_data = osd_req_op_data(osd_req, which, cls, response_data);
238 ceph_osd_data_pages_init(osd_data, pages, length, alignment,
239 pages_from_pool, own_pages);
240 }
241 EXPORT_SYMBOL(osd_req_op_cls_response_data_pages);
242
243 static u64 ceph_osd_data_length(struct ceph_osd_data *osd_data)
244 {
245 switch (osd_data->type) {
246 case CEPH_OSD_DATA_TYPE_NONE:
247 return 0;
248 case CEPH_OSD_DATA_TYPE_PAGES:
249 return osd_data->length;
250 case CEPH_OSD_DATA_TYPE_PAGELIST:
251 return (u64)osd_data->pagelist->length;
252 #ifdef CONFIG_BLOCK
253 case CEPH_OSD_DATA_TYPE_BIO:
254 return (u64)osd_data->bio_length;
255 #endif /* CONFIG_BLOCK */
256 default:
257 WARN(true, "unrecognized data type %d\n", (int)osd_data->type);
258 return 0;
259 }
260 }
261
262 static void ceph_osd_data_release(struct ceph_osd_data *osd_data)
263 {
264 if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGES && osd_data->own_pages) {
265 int num_pages;
266
267 num_pages = calc_pages_for((u64)osd_data->alignment,
268 (u64)osd_data->length);
269 ceph_release_page_vector(osd_data->pages, num_pages);
270 }
271 ceph_osd_data_init(osd_data);
272 }
273
274 static void osd_req_op_data_release(struct ceph_osd_request *osd_req,
275 unsigned int which)
276 {
277 struct ceph_osd_req_op *op;
278
279 BUG_ON(which >= osd_req->r_num_ops);
280 op = &osd_req->r_ops[which];
281
282 switch (op->op) {
283 case CEPH_OSD_OP_READ:
284 case CEPH_OSD_OP_WRITE:
285 ceph_osd_data_release(&op->extent.osd_data);
286 break;
287 case CEPH_OSD_OP_CALL:
288 ceph_osd_data_release(&op->cls.request_info);
289 ceph_osd_data_release(&op->cls.request_data);
290 ceph_osd_data_release(&op->cls.response_data);
291 break;
292 default:
293 break;
294 }
295 }
296
297 /*
298 * requests
299 */
300 void ceph_osdc_release_request(struct kref *kref)
301 {
302 struct ceph_osd_request *req;
303 unsigned int which;
304
305 req = container_of(kref, struct ceph_osd_request, r_kref);
306 if (req->r_request)
307 ceph_msg_put(req->r_request);
308 if (req->r_reply) {
309 ceph_msg_revoke_incoming(req->r_reply);
310 ceph_msg_put(req->r_reply);
311 }
312
313 for (which = 0; which < req->r_num_ops; which++)
314 osd_req_op_data_release(req, which);
315
316 ceph_put_snap_context(req->r_snapc);
317 if (req->r_mempool)
318 mempool_free(req, req->r_osdc->req_mempool);
319 else
320 kmem_cache_free(ceph_osd_request_cache, req);
321
322 }
323 EXPORT_SYMBOL(ceph_osdc_release_request);
324
325 struct ceph_osd_request *ceph_osdc_alloc_request(struct ceph_osd_client *osdc,
326 struct ceph_snap_context *snapc,
327 unsigned int num_ops,
328 bool use_mempool,
329 gfp_t gfp_flags)
330 {
331 struct ceph_osd_request *req;
332 struct ceph_msg *msg;
333 size_t msg_size;
334
335 BUILD_BUG_ON(CEPH_OSD_MAX_OP > U16_MAX);
336 BUG_ON(num_ops > CEPH_OSD_MAX_OP);
337
338 msg_size = 4 + 4 + 8 + 8 + 4+8;
339 msg_size += 2 + 4 + 8 + 4 + 4; /* oloc */
340 msg_size += 1 + 8 + 4 + 4; /* pg_t */
341 msg_size += 4 + CEPH_MAX_OID_NAME_LEN; /* oid */
342 msg_size += 2 + num_ops*sizeof(struct ceph_osd_op);
343 msg_size += 8; /* snapid */
344 msg_size += 8; /* snap_seq */
345 msg_size += 8 * (snapc ? snapc->num_snaps : 0); /* snaps */
346 msg_size += 4;
347
348 if (use_mempool) {
349 req = mempool_alloc(osdc->req_mempool, gfp_flags);
350 memset(req, 0, sizeof(*req));
351 } else {
352 req = kmem_cache_zalloc(ceph_osd_request_cache, gfp_flags);
353 }
354 if (req == NULL)
355 return NULL;
356
357 req->r_osdc = osdc;
358 req->r_mempool = use_mempool;
359 req->r_num_ops = num_ops;
360
361 kref_init(&req->r_kref);
362 init_completion(&req->r_completion);
363 init_completion(&req->r_safe_completion);
364 RB_CLEAR_NODE(&req->r_node);
365 INIT_LIST_HEAD(&req->r_unsafe_item);
366 INIT_LIST_HEAD(&req->r_linger_item);
367 INIT_LIST_HEAD(&req->r_linger_osd);
368 INIT_LIST_HEAD(&req->r_req_lru_item);
369 INIT_LIST_HEAD(&req->r_osd_item);
370
371 req->r_base_oloc.pool = -1;
372 req->r_target_oloc.pool = -1;
373
374 /* create reply message */
375 if (use_mempool)
376 msg = ceph_msgpool_get(&osdc->msgpool_op_reply, 0);
377 else
378 msg = ceph_msg_new(CEPH_MSG_OSD_OPREPLY,
379 OSD_OPREPLY_FRONT_LEN, gfp_flags, true);
380 if (!msg) {
381 ceph_osdc_put_request(req);
382 return NULL;
383 }
384 req->r_reply = msg;
385
386 /* create request message; allow space for oid */
387 if (use_mempool)
388 msg = ceph_msgpool_get(&osdc->msgpool_op, 0);
389 else
390 msg = ceph_msg_new(CEPH_MSG_OSD_OP, msg_size, gfp_flags, true);
391 if (!msg) {
392 ceph_osdc_put_request(req);
393 return NULL;
394 }
395
396 memset(msg->front.iov_base, 0, msg->front.iov_len);
397
398 req->r_request = msg;
399
400 return req;
401 }
402 EXPORT_SYMBOL(ceph_osdc_alloc_request);
403
404 static bool osd_req_opcode_valid(u16 opcode)
405 {
406 switch (opcode) {
407 case CEPH_OSD_OP_READ:
408 case CEPH_OSD_OP_STAT:
409 case CEPH_OSD_OP_MAPEXT:
410 case CEPH_OSD_OP_MASKTRUNC:
411 case CEPH_OSD_OP_SPARSE_READ:
412 case CEPH_OSD_OP_NOTIFY:
413 case CEPH_OSD_OP_NOTIFY_ACK:
414 case CEPH_OSD_OP_ASSERT_VER:
415 case CEPH_OSD_OP_WRITE:
416 case CEPH_OSD_OP_WRITEFULL:
417 case CEPH_OSD_OP_TRUNCATE:
418 case CEPH_OSD_OP_ZERO:
419 case CEPH_OSD_OP_DELETE:
420 case CEPH_OSD_OP_APPEND:
421 case CEPH_OSD_OP_STARTSYNC:
422 case CEPH_OSD_OP_SETTRUNC:
423 case CEPH_OSD_OP_TRIMTRUNC:
424 case CEPH_OSD_OP_TMAPUP:
425 case CEPH_OSD_OP_TMAPPUT:
426 case CEPH_OSD_OP_TMAPGET:
427 case CEPH_OSD_OP_CREATE:
428 case CEPH_OSD_OP_ROLLBACK:
429 case CEPH_OSD_OP_WATCH:
430 case CEPH_OSD_OP_OMAPGETKEYS:
431 case CEPH_OSD_OP_OMAPGETVALS:
432 case CEPH_OSD_OP_OMAPGETHEADER:
433 case CEPH_OSD_OP_OMAPGETVALSBYKEYS:
434 case CEPH_OSD_OP_OMAPSETVALS:
435 case CEPH_OSD_OP_OMAPSETHEADER:
436 case CEPH_OSD_OP_OMAPCLEAR:
437 case CEPH_OSD_OP_OMAPRMKEYS:
438 case CEPH_OSD_OP_OMAP_CMP:
439 case CEPH_OSD_OP_SETALLOCHINT:
440 case CEPH_OSD_OP_CLONERANGE:
441 case CEPH_OSD_OP_ASSERT_SRC_VERSION:
442 case CEPH_OSD_OP_SRC_CMPXATTR:
443 case CEPH_OSD_OP_GETXATTR:
444 case CEPH_OSD_OP_GETXATTRS:
445 case CEPH_OSD_OP_CMPXATTR:
446 case CEPH_OSD_OP_SETXATTR:
447 case CEPH_OSD_OP_SETXATTRS:
448 case CEPH_OSD_OP_RESETXATTRS:
449 case CEPH_OSD_OP_RMXATTR:
450 case CEPH_OSD_OP_PULL:
451 case CEPH_OSD_OP_PUSH:
452 case CEPH_OSD_OP_BALANCEREADS:
453 case CEPH_OSD_OP_UNBALANCEREADS:
454 case CEPH_OSD_OP_SCRUB:
455 case CEPH_OSD_OP_SCRUB_RESERVE:
456 case CEPH_OSD_OP_SCRUB_UNRESERVE:
457 case CEPH_OSD_OP_SCRUB_STOP:
458 case CEPH_OSD_OP_SCRUB_MAP:
459 case CEPH_OSD_OP_WRLOCK:
460 case CEPH_OSD_OP_WRUNLOCK:
461 case CEPH_OSD_OP_RDLOCK:
462 case CEPH_OSD_OP_RDUNLOCK:
463 case CEPH_OSD_OP_UPLOCK:
464 case CEPH_OSD_OP_DNLOCK:
465 case CEPH_OSD_OP_CALL:
466 case CEPH_OSD_OP_PGLS:
467 case CEPH_OSD_OP_PGLS_FILTER:
468 return true;
469 default:
470 return false;
471 }
472 }
473
474 /*
475 * This is an osd op init function for opcodes that have no data or
476 * other information associated with them. It also serves as a
477 * common init routine for all the other init functions, below.
478 */
479 static struct ceph_osd_req_op *
480 _osd_req_op_init(struct ceph_osd_request *osd_req, unsigned int which,
481 u16 opcode)
482 {
483 struct ceph_osd_req_op *op;
484
485 BUG_ON(which >= osd_req->r_num_ops);
486 BUG_ON(!osd_req_opcode_valid(opcode));
487
488 op = &osd_req->r_ops[which];
489 memset(op, 0, sizeof (*op));
490 op->op = opcode;
491
492 return op;
493 }
494
495 void osd_req_op_init(struct ceph_osd_request *osd_req,
496 unsigned int which, u16 opcode)
497 {
498 (void)_osd_req_op_init(osd_req, which, opcode);
499 }
500 EXPORT_SYMBOL(osd_req_op_init);
501
502 void osd_req_op_extent_init(struct ceph_osd_request *osd_req,
503 unsigned int which, u16 opcode,
504 u64 offset, u64 length,
505 u64 truncate_size, u32 truncate_seq)
506 {
507 struct ceph_osd_req_op *op = _osd_req_op_init(osd_req, which, opcode);
508 size_t payload_len = 0;
509
510 BUG_ON(opcode != CEPH_OSD_OP_READ && opcode != CEPH_OSD_OP_WRITE &&
511 opcode != CEPH_OSD_OP_DELETE && opcode != CEPH_OSD_OP_ZERO &&
512 opcode != CEPH_OSD_OP_TRUNCATE);
513
514 op->extent.offset = offset;
515 op->extent.length = length;
516 op->extent.truncate_size = truncate_size;
517 op->extent.truncate_seq = truncate_seq;
518 if (opcode == CEPH_OSD_OP_WRITE)
519 payload_len += length;
520
521 op->payload_len = payload_len;
522 }
523 EXPORT_SYMBOL(osd_req_op_extent_init);
524
525 void osd_req_op_extent_update(struct ceph_osd_request *osd_req,
526 unsigned int which, u64 length)
527 {
528 struct ceph_osd_req_op *op;
529 u64 previous;
530
531 BUG_ON(which >= osd_req->r_num_ops);
532 op = &osd_req->r_ops[which];
533 previous = op->extent.length;
534
535 if (length == previous)
536 return; /* Nothing to do */
537 BUG_ON(length > previous);
538
539 op->extent.length = length;
540 op->payload_len -= previous - length;
541 }
542 EXPORT_SYMBOL(osd_req_op_extent_update);
543
544 void osd_req_op_cls_init(struct ceph_osd_request *osd_req, unsigned int which,
545 u16 opcode, const char *class, const char *method)
546 {
547 struct ceph_osd_req_op *op = _osd_req_op_init(osd_req, which, opcode);
548 struct ceph_pagelist *pagelist;
549 size_t payload_len = 0;
550 size_t size;
551
552 BUG_ON(opcode != CEPH_OSD_OP_CALL);
553
554 pagelist = kmalloc(sizeof (*pagelist), GFP_NOFS);
555 BUG_ON(!pagelist);
556 ceph_pagelist_init(pagelist);
557
558 op->cls.class_name = class;
559 size = strlen(class);
560 BUG_ON(size > (size_t) U8_MAX);
561 op->cls.class_len = size;
562 ceph_pagelist_append(pagelist, class, size);
563 payload_len += size;
564
565 op->cls.method_name = method;
566 size = strlen(method);
567 BUG_ON(size > (size_t) U8_MAX);
568 op->cls.method_len = size;
569 ceph_pagelist_append(pagelist, method, size);
570 payload_len += size;
571
572 osd_req_op_cls_request_info_pagelist(osd_req, which, pagelist);
573
574 op->cls.argc = 0; /* currently unused */
575
576 op->payload_len = payload_len;
577 }
578 EXPORT_SYMBOL(osd_req_op_cls_init);
579
580 void osd_req_op_watch_init(struct ceph_osd_request *osd_req,
581 unsigned int which, u16 opcode,
582 u64 cookie, u64 version, int flag)
583 {
584 struct ceph_osd_req_op *op = _osd_req_op_init(osd_req, which, opcode);
585
586 BUG_ON(opcode != CEPH_OSD_OP_NOTIFY_ACK && opcode != CEPH_OSD_OP_WATCH);
587
588 op->watch.cookie = cookie;
589 op->watch.ver = version;
590 if (opcode == CEPH_OSD_OP_WATCH && flag)
591 op->watch.flag = (u8)1;
592 }
593 EXPORT_SYMBOL(osd_req_op_watch_init);
594
595 void osd_req_op_alloc_hint_init(struct ceph_osd_request *osd_req,
596 unsigned int which,
597 u64 expected_object_size,
598 u64 expected_write_size)
599 {
600 struct ceph_osd_req_op *op = _osd_req_op_init(osd_req, which,
601 CEPH_OSD_OP_SETALLOCHINT);
602
603 op->alloc_hint.expected_object_size = expected_object_size;
604 op->alloc_hint.expected_write_size = expected_write_size;
605
606 /*
607 * CEPH_OSD_OP_SETALLOCHINT op is advisory and therefore deemed
608 * not worth a feature bit. Set FAILOK per-op flag to make
609 * sure older osds don't trip over an unsupported opcode.
610 */
611 op->flags |= CEPH_OSD_OP_FLAG_FAILOK;
612 }
613 EXPORT_SYMBOL(osd_req_op_alloc_hint_init);
614
615 static void ceph_osdc_msg_data_add(struct ceph_msg *msg,
616 struct ceph_osd_data *osd_data)
617 {
618 u64 length = ceph_osd_data_length(osd_data);
619
620 if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGES) {
621 BUG_ON(length > (u64) SIZE_MAX);
622 if (length)
623 ceph_msg_data_add_pages(msg, osd_data->pages,
624 length, osd_data->alignment);
625 } else if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGELIST) {
626 BUG_ON(!length);
627 ceph_msg_data_add_pagelist(msg, osd_data->pagelist);
628 #ifdef CONFIG_BLOCK
629 } else if (osd_data->type == CEPH_OSD_DATA_TYPE_BIO) {
630 ceph_msg_data_add_bio(msg, osd_data->bio, length);
631 #endif
632 } else {
633 BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_NONE);
634 }
635 }
636
637 static u64 osd_req_encode_op(struct ceph_osd_request *req,
638 struct ceph_osd_op *dst, unsigned int which)
639 {
640 struct ceph_osd_req_op *src;
641 struct ceph_osd_data *osd_data;
642 u64 request_data_len = 0;
643 u64 data_length;
644
645 BUG_ON(which >= req->r_num_ops);
646 src = &req->r_ops[which];
647 if (WARN_ON(!osd_req_opcode_valid(src->op))) {
648 pr_err("unrecognized osd opcode %d\n", src->op);
649
650 return 0;
651 }
652
653 switch (src->op) {
654 case CEPH_OSD_OP_STAT:
655 osd_data = &src->raw_data_in;
656 ceph_osdc_msg_data_add(req->r_reply, osd_data);
657 break;
658 case CEPH_OSD_OP_READ:
659 case CEPH_OSD_OP_WRITE:
660 case CEPH_OSD_OP_ZERO:
661 case CEPH_OSD_OP_DELETE:
662 case CEPH_OSD_OP_TRUNCATE:
663 if (src->op == CEPH_OSD_OP_WRITE)
664 request_data_len = src->extent.length;
665 dst->extent.offset = cpu_to_le64(src->extent.offset);
666 dst->extent.length = cpu_to_le64(src->extent.length);
667 dst->extent.truncate_size =
668 cpu_to_le64(src->extent.truncate_size);
669 dst->extent.truncate_seq =
670 cpu_to_le32(src->extent.truncate_seq);
671 osd_data = &src->extent.osd_data;
672 if (src->op == CEPH_OSD_OP_WRITE)
673 ceph_osdc_msg_data_add(req->r_request, osd_data);
674 else
675 ceph_osdc_msg_data_add(req->r_reply, osd_data);
676 break;
677 case CEPH_OSD_OP_CALL:
678 dst->cls.class_len = src->cls.class_len;
679 dst->cls.method_len = src->cls.method_len;
680 osd_data = &src->cls.request_info;
681 ceph_osdc_msg_data_add(req->r_request, osd_data);
682 BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_PAGELIST);
683 request_data_len = osd_data->pagelist->length;
684
685 osd_data = &src->cls.request_data;
686 data_length = ceph_osd_data_length(osd_data);
687 if (data_length) {
688 BUG_ON(osd_data->type == CEPH_OSD_DATA_TYPE_NONE);
689 dst->cls.indata_len = cpu_to_le32(data_length);
690 ceph_osdc_msg_data_add(req->r_request, osd_data);
691 src->payload_len += data_length;
692 request_data_len += data_length;
693 }
694 osd_data = &src->cls.response_data;
695 ceph_osdc_msg_data_add(req->r_reply, osd_data);
696 break;
697 case CEPH_OSD_OP_STARTSYNC:
698 break;
699 case CEPH_OSD_OP_NOTIFY_ACK:
700 case CEPH_OSD_OP_WATCH:
701 dst->watch.cookie = cpu_to_le64(src->watch.cookie);
702 dst->watch.ver = cpu_to_le64(src->watch.ver);
703 dst->watch.flag = src->watch.flag;
704 break;
705 case CEPH_OSD_OP_SETALLOCHINT:
706 dst->alloc_hint.expected_object_size =
707 cpu_to_le64(src->alloc_hint.expected_object_size);
708 dst->alloc_hint.expected_write_size =
709 cpu_to_le64(src->alloc_hint.expected_write_size);
710 break;
711 default:
712 pr_err("unsupported osd opcode %s\n",
713 ceph_osd_op_name(src->op));
714 WARN_ON(1);
715
716 return 0;
717 }
718
719 dst->op = cpu_to_le16(src->op);
720 dst->flags = cpu_to_le32(src->flags);
721 dst->payload_len = cpu_to_le32(src->payload_len);
722
723 return request_data_len;
724 }
725
726 /*
727 * build new request AND message, calculate layout, and adjust file
728 * extent as needed.
729 *
730 * if the file was recently truncated, we include information about its
731 * old and new size so that the object can be updated appropriately. (we
732 * avoid synchronously deleting truncated objects because it's slow.)
733 *
734 * if @do_sync, include a 'startsync' command so that the osd will flush
735 * data quickly.
736 */
737 struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *osdc,
738 struct ceph_file_layout *layout,
739 struct ceph_vino vino,
740 u64 off, u64 *plen, int num_ops,
741 int opcode, int flags,
742 struct ceph_snap_context *snapc,
743 u32 truncate_seq,
744 u64 truncate_size,
745 bool use_mempool)
746 {
747 struct ceph_osd_request *req;
748 u64 objnum = 0;
749 u64 objoff = 0;
750 u64 objlen = 0;
751 u32 object_size;
752 u64 object_base;
753 int r;
754
755 BUG_ON(opcode != CEPH_OSD_OP_READ && opcode != CEPH_OSD_OP_WRITE &&
756 opcode != CEPH_OSD_OP_DELETE && opcode != CEPH_OSD_OP_ZERO &&
757 opcode != CEPH_OSD_OP_TRUNCATE);
758
759 req = ceph_osdc_alloc_request(osdc, snapc, num_ops, use_mempool,
760 GFP_NOFS);
761 if (!req)
762 return ERR_PTR(-ENOMEM);
763
764 req->r_flags = flags;
765
766 /* calculate max write size */
767 r = calc_layout(layout, off, plen, &objnum, &objoff, &objlen);
768 if (r < 0) {
769 ceph_osdc_put_request(req);
770 return ERR_PTR(r);
771 }
772
773 object_size = le32_to_cpu(layout->fl_object_size);
774 object_base = off - objoff;
775 if (!(truncate_seq == 1 && truncate_size == -1ULL)) {
776 if (truncate_size <= object_base) {
777 truncate_size = 0;
778 } else {
779 truncate_size -= object_base;
780 if (truncate_size > object_size)
781 truncate_size = object_size;
782 }
783 }
784
785 osd_req_op_extent_init(req, 0, opcode, objoff, objlen,
786 truncate_size, truncate_seq);
787
788 /*
789 * A second op in the ops array means the caller wants to
790 * also issue a include a 'startsync' command so that the
791 * osd will flush data quickly.
792 */
793 if (num_ops > 1)
794 osd_req_op_init(req, 1, CEPH_OSD_OP_STARTSYNC);
795
796 req->r_base_oloc.pool = ceph_file_layout_pg_pool(*layout);
797
798 snprintf(req->r_base_oid.name, sizeof(req->r_base_oid.name),
799 "%llx.%08llx", vino.ino, objnum);
800 req->r_base_oid.name_len = strlen(req->r_base_oid.name);
801
802 return req;
803 }
804 EXPORT_SYMBOL(ceph_osdc_new_request);
805
806 /*
807 * We keep osd requests in an rbtree, sorted by ->r_tid.
808 */
809 static void __insert_request(struct ceph_osd_client *osdc,
810 struct ceph_osd_request *new)
811 {
812 struct rb_node **p = &osdc->requests.rb_node;
813 struct rb_node *parent = NULL;
814 struct ceph_osd_request *req = NULL;
815
816 while (*p) {
817 parent = *p;
818 req = rb_entry(parent, struct ceph_osd_request, r_node);
819 if (new->r_tid < req->r_tid)
820 p = &(*p)->rb_left;
821 else if (new->r_tid > req->r_tid)
822 p = &(*p)->rb_right;
823 else
824 BUG();
825 }
826
827 rb_link_node(&new->r_node, parent, p);
828 rb_insert_color(&new->r_node, &osdc->requests);
829 }
830
831 static struct ceph_osd_request *__lookup_request(struct ceph_osd_client *osdc,
832 u64 tid)
833 {
834 struct ceph_osd_request *req;
835 struct rb_node *n = osdc->requests.rb_node;
836
837 while (n) {
838 req = rb_entry(n, struct ceph_osd_request, r_node);
839 if (tid < req->r_tid)
840 n = n->rb_left;
841 else if (tid > req->r_tid)
842 n = n->rb_right;
843 else
844 return req;
845 }
846 return NULL;
847 }
848
849 static struct ceph_osd_request *
850 __lookup_request_ge(struct ceph_osd_client *osdc,
851 u64 tid)
852 {
853 struct ceph_osd_request *req;
854 struct rb_node *n = osdc->requests.rb_node;
855
856 while (n) {
857 req = rb_entry(n, struct ceph_osd_request, r_node);
858 if (tid < req->r_tid) {
859 if (!n->rb_left)
860 return req;
861 n = n->rb_left;
862 } else if (tid > req->r_tid) {
863 n = n->rb_right;
864 } else {
865 return req;
866 }
867 }
868 return NULL;
869 }
870
871 /*
872 * Resubmit requests pending on the given osd.
873 */
874 static void __kick_osd_requests(struct ceph_osd_client *osdc,
875 struct ceph_osd *osd)
876 {
877 struct ceph_osd_request *req, *nreq;
878 LIST_HEAD(resend);
879 int err;
880
881 dout("__kick_osd_requests osd%d\n", osd->o_osd);
882 err = __reset_osd(osdc, osd);
883 if (err)
884 return;
885 /*
886 * Build up a list of requests to resend by traversing the
887 * osd's list of requests. Requests for a given object are
888 * sent in tid order, and that is also the order they're
889 * kept on this list. Therefore all requests that are in
890 * flight will be found first, followed by all requests that
891 * have not yet been sent. And to resend requests while
892 * preserving this order we will want to put any sent
893 * requests back on the front of the osd client's unsent
894 * list.
895 *
896 * So we build a separate ordered list of already-sent
897 * requests for the affected osd and splice it onto the
898 * front of the osd client's unsent list. Once we've seen a
899 * request that has not yet been sent we're done. Those
900 * requests are already sitting right where they belong.
901 */
902 list_for_each_entry(req, &osd->o_requests, r_osd_item) {
903 if (!req->r_sent)
904 break;
905 list_move_tail(&req->r_req_lru_item, &resend);
906 dout("requeueing %p tid %llu osd%d\n", req, req->r_tid,
907 osd->o_osd);
908 if (!req->r_linger)
909 req->r_flags |= CEPH_OSD_FLAG_RETRY;
910 }
911 list_splice(&resend, &osdc->req_unsent);
912
913 /*
914 * Linger requests are re-registered before sending, which
915 * sets up a new tid for each. We add them to the unsent
916 * list at the end to keep things in tid order.
917 */
918 list_for_each_entry_safe(req, nreq, &osd->o_linger_requests,
919 r_linger_osd) {
920 /*
921 * reregister request prior to unregistering linger so
922 * that r_osd is preserved.
923 */
924 BUG_ON(!list_empty(&req->r_req_lru_item));
925 __register_request(osdc, req);
926 list_add_tail(&req->r_req_lru_item, &osdc->req_unsent);
927 list_add_tail(&req->r_osd_item, &req->r_osd->o_requests);
928 __unregister_linger_request(osdc, req);
929 dout("requeued lingering %p tid %llu osd%d\n", req, req->r_tid,
930 osd->o_osd);
931 }
932 }
933
934 /*
935 * If the osd connection drops, we need to resubmit all requests.
936 */
937 static void osd_reset(struct ceph_connection *con)
938 {
939 struct ceph_osd *osd = con->private;
940 struct ceph_osd_client *osdc;
941
942 if (!osd)
943 return;
944 dout("osd_reset osd%d\n", osd->o_osd);
945 osdc = osd->o_osdc;
946 down_read(&osdc->map_sem);
947 mutex_lock(&osdc->request_mutex);
948 __kick_osd_requests(osdc, osd);
949 __send_queued(osdc);
950 mutex_unlock(&osdc->request_mutex);
951 up_read(&osdc->map_sem);
952 }
953
954 /*
955 * Track open sessions with osds.
956 */
957 static struct ceph_osd *create_osd(struct ceph_osd_client *osdc, int onum)
958 {
959 struct ceph_osd *osd;
960
961 osd = kzalloc(sizeof(*osd), GFP_NOFS);
962 if (!osd)
963 return NULL;
964
965 atomic_set(&osd->o_ref, 1);
966 osd->o_osdc = osdc;
967 osd->o_osd = onum;
968 RB_CLEAR_NODE(&osd->o_node);
969 INIT_LIST_HEAD(&osd->o_requests);
970 INIT_LIST_HEAD(&osd->o_linger_requests);
971 INIT_LIST_HEAD(&osd->o_osd_lru);
972 osd->o_incarnation = 1;
973
974 ceph_con_init(&osd->o_con, osd, &osd_con_ops, &osdc->client->msgr);
975
976 INIT_LIST_HEAD(&osd->o_keepalive_item);
977 return osd;
978 }
979
980 static struct ceph_osd *get_osd(struct ceph_osd *osd)
981 {
982 if (atomic_inc_not_zero(&osd->o_ref)) {
983 dout("get_osd %p %d -> %d\n", osd, atomic_read(&osd->o_ref)-1,
984 atomic_read(&osd->o_ref));
985 return osd;
986 } else {
987 dout("get_osd %p FAIL\n", osd);
988 return NULL;
989 }
990 }
991
992 static void put_osd(struct ceph_osd *osd)
993 {
994 dout("put_osd %p %d -> %d\n", osd, atomic_read(&osd->o_ref),
995 atomic_read(&osd->o_ref) - 1);
996 if (atomic_dec_and_test(&osd->o_ref)) {
997 if (osd->o_auth.authorizer)
998 ceph_auth_destroy_authorizer(osd->o_auth.authorizer);
999 kfree(osd);
1000 }
1001 }
1002
1003 /*
1004 * remove an osd from our map
1005 */
1006 static void __remove_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd)
1007 {
1008 dout("%s %p osd%d\n", __func__, osd, osd->o_osd);
1009 WARN_ON(!list_empty(&osd->o_requests));
1010 WARN_ON(!list_empty(&osd->o_linger_requests));
1011
1012 list_del_init(&osd->o_osd_lru);
1013 rb_erase(&osd->o_node, &osdc->osds);
1014 RB_CLEAR_NODE(&osd->o_node);
1015 }
1016
1017 static void remove_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd)
1018 {
1019 dout("%s %p osd%d\n", __func__, osd, osd->o_osd);
1020
1021 if (!RB_EMPTY_NODE(&osd->o_node)) {
1022 ceph_con_close(&osd->o_con);
1023 __remove_osd(osdc, osd);
1024 put_osd(osd);
1025 }
1026 }
1027
1028 static void remove_all_osds(struct ceph_osd_client *osdc)
1029 {
1030 dout("%s %p\n", __func__, osdc);
1031 mutex_lock(&osdc->request_mutex);
1032 while (!RB_EMPTY_ROOT(&osdc->osds)) {
1033 struct ceph_osd *osd = rb_entry(rb_first(&osdc->osds),
1034 struct ceph_osd, o_node);
1035 remove_osd(osdc, osd);
1036 }
1037 mutex_unlock(&osdc->request_mutex);
1038 }
1039
1040 static void __move_osd_to_lru(struct ceph_osd_client *osdc,
1041 struct ceph_osd *osd)
1042 {
1043 dout("__move_osd_to_lru %p\n", osd);
1044 BUG_ON(!list_empty(&osd->o_osd_lru));
1045 list_add_tail(&osd->o_osd_lru, &osdc->osd_lru);
1046 osd->lru_ttl = jiffies + osdc->client->options->osd_idle_ttl * HZ;
1047 }
1048
1049 static void __remove_osd_from_lru(struct ceph_osd *osd)
1050 {
1051 dout("__remove_osd_from_lru %p\n", osd);
1052 if (!list_empty(&osd->o_osd_lru))
1053 list_del_init(&osd->o_osd_lru);
1054 }
1055
1056 static void remove_old_osds(struct ceph_osd_client *osdc)
1057 {
1058 struct ceph_osd *osd, *nosd;
1059
1060 dout("__remove_old_osds %p\n", osdc);
1061 mutex_lock(&osdc->request_mutex);
1062 list_for_each_entry_safe(osd, nosd, &osdc->osd_lru, o_osd_lru) {
1063 if (time_before(jiffies, osd->lru_ttl))
1064 break;
1065 remove_osd(osdc, osd);
1066 }
1067 mutex_unlock(&osdc->request_mutex);
1068 }
1069
1070 /*
1071 * reset osd connect
1072 */
1073 static int __reset_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd)
1074 {
1075 struct ceph_entity_addr *peer_addr;
1076
1077 dout("__reset_osd %p osd%d\n", osd, osd->o_osd);
1078 if (list_empty(&osd->o_requests) &&
1079 list_empty(&osd->o_linger_requests)) {
1080 remove_osd(osdc, osd);
1081 return -ENODEV;
1082 }
1083
1084 peer_addr = &osdc->osdmap->osd_addr[osd->o_osd];
1085 if (!memcmp(peer_addr, &osd->o_con.peer_addr, sizeof (*peer_addr)) &&
1086 !ceph_con_opened(&osd->o_con)) {
1087 struct ceph_osd_request *req;
1088
1089 dout("osd addr hasn't changed and connection never opened, "
1090 "letting msgr retry\n");
1091 /* touch each r_stamp for handle_timeout()'s benfit */
1092 list_for_each_entry(req, &osd->o_requests, r_osd_item)
1093 req->r_stamp = jiffies;
1094
1095 return -EAGAIN;
1096 }
1097
1098 ceph_con_close(&osd->o_con);
1099 ceph_con_open(&osd->o_con, CEPH_ENTITY_TYPE_OSD, osd->o_osd, peer_addr);
1100 osd->o_incarnation++;
1101
1102 return 0;
1103 }
1104
1105 static void __insert_osd(struct ceph_osd_client *osdc, struct ceph_osd *new)
1106 {
1107 struct rb_node **p = &osdc->osds.rb_node;
1108 struct rb_node *parent = NULL;
1109 struct ceph_osd *osd = NULL;
1110
1111 dout("__insert_osd %p osd%d\n", new, new->o_osd);
1112 while (*p) {
1113 parent = *p;
1114 osd = rb_entry(parent, struct ceph_osd, o_node);
1115 if (new->o_osd < osd->o_osd)
1116 p = &(*p)->rb_left;
1117 else if (new->o_osd > osd->o_osd)
1118 p = &(*p)->rb_right;
1119 else
1120 BUG();
1121 }
1122
1123 rb_link_node(&new->o_node, parent, p);
1124 rb_insert_color(&new->o_node, &osdc->osds);
1125 }
1126
1127 static struct ceph_osd *__lookup_osd(struct ceph_osd_client *osdc, int o)
1128 {
1129 struct ceph_osd *osd;
1130 struct rb_node *n = osdc->osds.rb_node;
1131
1132 while (n) {
1133 osd = rb_entry(n, struct ceph_osd, o_node);
1134 if (o < osd->o_osd)
1135 n = n->rb_left;
1136 else if (o > osd->o_osd)
1137 n = n->rb_right;
1138 else
1139 return osd;
1140 }
1141 return NULL;
1142 }
1143
1144 static void __schedule_osd_timeout(struct ceph_osd_client *osdc)
1145 {
1146 schedule_delayed_work(&osdc->timeout_work,
1147 osdc->client->options->osd_keepalive_timeout * HZ);
1148 }
1149
1150 static void __cancel_osd_timeout(struct ceph_osd_client *osdc)
1151 {
1152 cancel_delayed_work(&osdc->timeout_work);
1153 }
1154
1155 /*
1156 * Register request, assign tid. If this is the first request, set up
1157 * the timeout event.
1158 */
1159 static void __register_request(struct ceph_osd_client *osdc,
1160 struct ceph_osd_request *req)
1161 {
1162 req->r_tid = ++osdc->last_tid;
1163 req->r_request->hdr.tid = cpu_to_le64(req->r_tid);
1164 dout("__register_request %p tid %lld\n", req, req->r_tid);
1165 __insert_request(osdc, req);
1166 ceph_osdc_get_request(req);
1167 osdc->num_requests++;
1168 if (osdc->num_requests == 1) {
1169 dout(" first request, scheduling timeout\n");
1170 __schedule_osd_timeout(osdc);
1171 }
1172 }
1173
1174 /*
1175 * called under osdc->request_mutex
1176 */
1177 static void __unregister_request(struct ceph_osd_client *osdc,
1178 struct ceph_osd_request *req)
1179 {
1180 if (RB_EMPTY_NODE(&req->r_node)) {
1181 dout("__unregister_request %p tid %lld not registered\n",
1182 req, req->r_tid);
1183 return;
1184 }
1185
1186 dout("__unregister_request %p tid %lld\n", req, req->r_tid);
1187 rb_erase(&req->r_node, &osdc->requests);
1188 osdc->num_requests--;
1189
1190 if (req->r_osd) {
1191 /* make sure the original request isn't in flight. */
1192 ceph_msg_revoke(req->r_request);
1193
1194 list_del_init(&req->r_osd_item);
1195 if (list_empty(&req->r_osd->o_requests) &&
1196 list_empty(&req->r_osd->o_linger_requests)) {
1197 dout("moving osd to %p lru\n", req->r_osd);
1198 __move_osd_to_lru(osdc, req->r_osd);
1199 }
1200 if (list_empty(&req->r_linger_item))
1201 req->r_osd = NULL;
1202 }
1203
1204 list_del_init(&req->r_req_lru_item);
1205 ceph_osdc_put_request(req);
1206
1207 if (osdc->num_requests == 0) {
1208 dout(" no requests, canceling timeout\n");
1209 __cancel_osd_timeout(osdc);
1210 }
1211 }
1212
1213 /*
1214 * Cancel a previously queued request message
1215 */
1216 static void __cancel_request(struct ceph_osd_request *req)
1217 {
1218 if (req->r_sent && req->r_osd) {
1219 ceph_msg_revoke(req->r_request);
1220 req->r_sent = 0;
1221 }
1222 }
1223
1224 static void __register_linger_request(struct ceph_osd_client *osdc,
1225 struct ceph_osd_request *req)
1226 {
1227 dout("__register_linger_request %p\n", req);
1228 ceph_osdc_get_request(req);
1229 list_add_tail(&req->r_linger_item, &osdc->req_linger);
1230 if (req->r_osd)
1231 list_add_tail(&req->r_linger_osd,
1232 &req->r_osd->o_linger_requests);
1233 }
1234
1235 static void __unregister_linger_request(struct ceph_osd_client *osdc,
1236 struct ceph_osd_request *req)
1237 {
1238 dout("__unregister_linger_request %p\n", req);
1239 list_del_init(&req->r_linger_item);
1240 if (req->r_osd) {
1241 list_del_init(&req->r_linger_osd);
1242
1243 if (list_empty(&req->r_osd->o_requests) &&
1244 list_empty(&req->r_osd->o_linger_requests)) {
1245 dout("moving osd to %p lru\n", req->r_osd);
1246 __move_osd_to_lru(osdc, req->r_osd);
1247 }
1248 if (list_empty(&req->r_osd_item))
1249 req->r_osd = NULL;
1250 }
1251 ceph_osdc_put_request(req);
1252 }
1253
1254 void ceph_osdc_unregister_linger_request(struct ceph_osd_client *osdc,
1255 struct ceph_osd_request *req)
1256 {
1257 mutex_lock(&osdc->request_mutex);
1258 if (req->r_linger) {
1259 req->r_linger = 0;
1260 __unregister_linger_request(osdc, req);
1261 }
1262 mutex_unlock(&osdc->request_mutex);
1263 }
1264 EXPORT_SYMBOL(ceph_osdc_unregister_linger_request);
1265
1266 void ceph_osdc_set_request_linger(struct ceph_osd_client *osdc,
1267 struct ceph_osd_request *req)
1268 {
1269 if (!req->r_linger) {
1270 dout("set_request_linger %p\n", req);
1271 req->r_linger = 1;
1272 }
1273 }
1274 EXPORT_SYMBOL(ceph_osdc_set_request_linger);
1275
1276 /*
1277 * Returns whether a request should be blocked from being sent
1278 * based on the current osdmap and osd_client settings.
1279 *
1280 * Caller should hold map_sem for read.
1281 */
1282 static bool __req_should_be_paused(struct ceph_osd_client *osdc,
1283 struct ceph_osd_request *req)
1284 {
1285 bool pauserd = ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_PAUSERD);
1286 bool pausewr = ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_PAUSEWR) ||
1287 ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_FULL);
1288 return (req->r_flags & CEPH_OSD_FLAG_READ && pauserd) ||
1289 (req->r_flags & CEPH_OSD_FLAG_WRITE && pausewr);
1290 }
1291
1292 /*
1293 * Calculate mapping of a request to a PG. Takes tiering into account.
1294 */
1295 static int __calc_request_pg(struct ceph_osdmap *osdmap,
1296 struct ceph_osd_request *req,
1297 struct ceph_pg *pg_out)
1298 {
1299 bool need_check_tiering;
1300
1301 need_check_tiering = false;
1302 if (req->r_target_oloc.pool == -1) {
1303 req->r_target_oloc = req->r_base_oloc; /* struct */
1304 need_check_tiering = true;
1305 }
1306 if (req->r_target_oid.name_len == 0) {
1307 ceph_oid_copy(&req->r_target_oid, &req->r_base_oid);
1308 need_check_tiering = true;
1309 }
1310
1311 if (need_check_tiering &&
1312 (req->r_flags & CEPH_OSD_FLAG_IGNORE_OVERLAY) == 0) {
1313 struct ceph_pg_pool_info *pi;
1314
1315 pi = ceph_pg_pool_by_id(osdmap, req->r_target_oloc.pool);
1316 if (pi) {
1317 if ((req->r_flags & CEPH_OSD_FLAG_READ) &&
1318 pi->read_tier >= 0)
1319 req->r_target_oloc.pool = pi->read_tier;
1320 if ((req->r_flags & CEPH_OSD_FLAG_WRITE) &&
1321 pi->write_tier >= 0)
1322 req->r_target_oloc.pool = pi->write_tier;
1323 }
1324 /* !pi is caught in ceph_oloc_oid_to_pg() */
1325 }
1326
1327 return ceph_oloc_oid_to_pg(osdmap, &req->r_target_oloc,
1328 &req->r_target_oid, pg_out);
1329 }
1330
1331 /*
1332 * Pick an osd (the first 'up' osd in the pg), allocate the osd struct
1333 * (as needed), and set the request r_osd appropriately. If there is
1334 * no up osd, set r_osd to NULL. Move the request to the appropriate list
1335 * (unsent, homeless) or leave on in-flight lru.
1336 *
1337 * Return 0 if unchanged, 1 if changed, or negative on error.
1338 *
1339 * Caller should hold map_sem for read and request_mutex.
1340 */
1341 static int __map_request(struct ceph_osd_client *osdc,
1342 struct ceph_osd_request *req, int force_resend)
1343 {
1344 struct ceph_pg pgid;
1345 int acting[CEPH_PG_MAX_SIZE];
1346 int num, o;
1347 int err;
1348 bool was_paused;
1349
1350 dout("map_request %p tid %lld\n", req, req->r_tid);
1351
1352 err = __calc_request_pg(osdc->osdmap, req, &pgid);
1353 if (err) {
1354 list_move(&req->r_req_lru_item, &osdc->req_notarget);
1355 return err;
1356 }
1357 req->r_pgid = pgid;
1358
1359 num = ceph_calc_pg_acting(osdc->osdmap, pgid, acting, &o);
1360 if (num < 0)
1361 num = 0;
1362
1363 was_paused = req->r_paused;
1364 req->r_paused = __req_should_be_paused(osdc, req);
1365 if (was_paused && !req->r_paused)
1366 force_resend = 1;
1367
1368 if ((!force_resend &&
1369 req->r_osd && req->r_osd->o_osd == o &&
1370 req->r_sent >= req->r_osd->o_incarnation &&
1371 req->r_num_pg_osds == num &&
1372 memcmp(req->r_pg_osds, acting, sizeof(acting[0])*num) == 0) ||
1373 (req->r_osd == NULL && o == -1) ||
1374 req->r_paused)
1375 return 0; /* no change */
1376
1377 dout("map_request tid %llu pgid %lld.%x osd%d (was osd%d)\n",
1378 req->r_tid, pgid.pool, pgid.seed, o,
1379 req->r_osd ? req->r_osd->o_osd : -1);
1380
1381 /* record full pg acting set */
1382 memcpy(req->r_pg_osds, acting, sizeof(acting[0]) * num);
1383 req->r_num_pg_osds = num;
1384
1385 if (req->r_osd) {
1386 __cancel_request(req);
1387 list_del_init(&req->r_osd_item);
1388 req->r_osd = NULL;
1389 }
1390
1391 req->r_osd = __lookup_osd(osdc, o);
1392 if (!req->r_osd && o >= 0) {
1393 err = -ENOMEM;
1394 req->r_osd = create_osd(osdc, o);
1395 if (!req->r_osd) {
1396 list_move(&req->r_req_lru_item, &osdc->req_notarget);
1397 goto out;
1398 }
1399
1400 dout("map_request osd %p is osd%d\n", req->r_osd, o);
1401 __insert_osd(osdc, req->r_osd);
1402
1403 ceph_con_open(&req->r_osd->o_con,
1404 CEPH_ENTITY_TYPE_OSD, o,
1405 &osdc->osdmap->osd_addr[o]);
1406 }
1407
1408 if (req->r_osd) {
1409 __remove_osd_from_lru(req->r_osd);
1410 list_add_tail(&req->r_osd_item, &req->r_osd->o_requests);
1411 list_move_tail(&req->r_req_lru_item, &osdc->req_unsent);
1412 } else {
1413 list_move_tail(&req->r_req_lru_item, &osdc->req_notarget);
1414 }
1415 err = 1; /* osd or pg changed */
1416
1417 out:
1418 return err;
1419 }
1420
1421 /*
1422 * caller should hold map_sem (for read) and request_mutex
1423 */
1424 static void __send_request(struct ceph_osd_client *osdc,
1425 struct ceph_osd_request *req)
1426 {
1427 void *p;
1428
1429 dout("send_request %p tid %llu to osd%d flags %d pg %lld.%x\n",
1430 req, req->r_tid, req->r_osd->o_osd, req->r_flags,
1431 (unsigned long long)req->r_pgid.pool, req->r_pgid.seed);
1432
1433 /* fill in message content that changes each time we send it */
1434 put_unaligned_le32(osdc->osdmap->epoch, req->r_request_osdmap_epoch);
1435 put_unaligned_le32(req->r_flags, req->r_request_flags);
1436 put_unaligned_le64(req->r_target_oloc.pool, req->r_request_pool);
1437 p = req->r_request_pgid;
1438 ceph_encode_64(&p, req->r_pgid.pool);
1439 ceph_encode_32(&p, req->r_pgid.seed);
1440 put_unaligned_le64(1, req->r_request_attempts); /* FIXME */
1441 memcpy(req->r_request_reassert_version, &req->r_reassert_version,
1442 sizeof(req->r_reassert_version));
1443
1444 req->r_stamp = jiffies;
1445 list_move_tail(&req->r_req_lru_item, &osdc->req_lru);
1446
1447 ceph_msg_get(req->r_request); /* send consumes a ref */
1448
1449 req->r_sent = req->r_osd->o_incarnation;
1450
1451 ceph_con_send(&req->r_osd->o_con, req->r_request);
1452 }
1453
1454 /*
1455 * Send any requests in the queue (req_unsent).
1456 */
1457 static void __send_queued(struct ceph_osd_client *osdc)
1458 {
1459 struct ceph_osd_request *req, *tmp;
1460
1461 dout("__send_queued\n");
1462 list_for_each_entry_safe(req, tmp, &osdc->req_unsent, r_req_lru_item)
1463 __send_request(osdc, req);
1464 }
1465
1466 /*
1467 * Caller should hold map_sem for read and request_mutex.
1468 */
1469 static int __ceph_osdc_start_request(struct ceph_osd_client *osdc,
1470 struct ceph_osd_request *req,
1471 bool nofail)
1472 {
1473 int rc;
1474
1475 __register_request(osdc, req);
1476 req->r_sent = 0;
1477 req->r_got_reply = 0;
1478 rc = __map_request(osdc, req, 0);
1479 if (rc < 0) {
1480 if (nofail) {
1481 dout("osdc_start_request failed map, "
1482 " will retry %lld\n", req->r_tid);
1483 rc = 0;
1484 } else {
1485 __unregister_request(osdc, req);
1486 }
1487 return rc;
1488 }
1489
1490 if (req->r_osd == NULL) {
1491 dout("send_request %p no up osds in pg\n", req);
1492 ceph_monc_request_next_osdmap(&osdc->client->monc);
1493 } else {
1494 __send_queued(osdc);
1495 }
1496
1497 return 0;
1498 }
1499
1500 /*
1501 * Timeout callback, called every N seconds when 1 or more osd
1502 * requests has been active for more than N seconds. When this
1503 * happens, we ping all OSDs with requests who have timed out to
1504 * ensure any communications channel reset is detected. Reset the
1505 * request timeouts another N seconds in the future as we go.
1506 * Reschedule the timeout event another N seconds in future (unless
1507 * there are no open requests).
1508 */
1509 static void handle_timeout(struct work_struct *work)
1510 {
1511 struct ceph_osd_client *osdc =
1512 container_of(work, struct ceph_osd_client, timeout_work.work);
1513 struct ceph_osd_request *req;
1514 struct ceph_osd *osd;
1515 unsigned long keepalive =
1516 osdc->client->options->osd_keepalive_timeout * HZ;
1517 struct list_head slow_osds;
1518 dout("timeout\n");
1519 down_read(&osdc->map_sem);
1520
1521 ceph_monc_request_next_osdmap(&osdc->client->monc);
1522
1523 mutex_lock(&osdc->request_mutex);
1524
1525 /*
1526 * ping osds that are a bit slow. this ensures that if there
1527 * is a break in the TCP connection we will notice, and reopen
1528 * a connection with that osd (from the fault callback).
1529 */
1530 INIT_LIST_HEAD(&slow_osds);
1531 list_for_each_entry(req, &osdc->req_lru, r_req_lru_item) {
1532 if (time_before(jiffies, req->r_stamp + keepalive))
1533 break;
1534
1535 osd = req->r_osd;
1536 BUG_ON(!osd);
1537 dout(" tid %llu is slow, will send keepalive on osd%d\n",
1538 req->r_tid, osd->o_osd);
1539 list_move_tail(&osd->o_keepalive_item, &slow_osds);
1540 }
1541 while (!list_empty(&slow_osds)) {
1542 osd = list_entry(slow_osds.next, struct ceph_osd,
1543 o_keepalive_item);
1544 list_del_init(&osd->o_keepalive_item);
1545 ceph_con_keepalive(&osd->o_con);
1546 }
1547
1548 __schedule_osd_timeout(osdc);
1549 __send_queued(osdc);
1550 mutex_unlock(&osdc->request_mutex);
1551 up_read(&osdc->map_sem);
1552 }
1553
1554 static void handle_osds_timeout(struct work_struct *work)
1555 {
1556 struct ceph_osd_client *osdc =
1557 container_of(work, struct ceph_osd_client,
1558 osds_timeout_work.work);
1559 unsigned long delay =
1560 osdc->client->options->osd_idle_ttl * HZ >> 2;
1561
1562 dout("osds timeout\n");
1563 down_read(&osdc->map_sem);
1564 remove_old_osds(osdc);
1565 up_read(&osdc->map_sem);
1566
1567 schedule_delayed_work(&osdc->osds_timeout_work,
1568 round_jiffies_relative(delay));
1569 }
1570
1571 static int ceph_oloc_decode(void **p, void *end,
1572 struct ceph_object_locator *oloc)
1573 {
1574 u8 struct_v, struct_cv;
1575 u32 len;
1576 void *struct_end;
1577 int ret = 0;
1578
1579 ceph_decode_need(p, end, 1 + 1 + 4, e_inval);
1580 struct_v = ceph_decode_8(p);
1581 struct_cv = ceph_decode_8(p);
1582 if (struct_v < 3) {
1583 pr_warn("got v %d < 3 cv %d of ceph_object_locator\n",
1584 struct_v, struct_cv);
1585 goto e_inval;
1586 }
1587 if (struct_cv > 6) {
1588 pr_warn("got v %d cv %d > 6 of ceph_object_locator\n",
1589 struct_v, struct_cv);
1590 goto e_inval;
1591 }
1592 len = ceph_decode_32(p);
1593 ceph_decode_need(p, end, len, e_inval);
1594 struct_end = *p + len;
1595
1596 oloc->pool = ceph_decode_64(p);
1597 *p += 4; /* skip preferred */
1598
1599 len = ceph_decode_32(p);
1600 if (len > 0) {
1601 pr_warn("ceph_object_locator::key is set\n");
1602 goto e_inval;
1603 }
1604
1605 if (struct_v >= 5) {
1606 len = ceph_decode_32(p);
1607 if (len > 0) {
1608 pr_warn("ceph_object_locator::nspace is set\n");
1609 goto e_inval;
1610 }
1611 }
1612
1613 if (struct_v >= 6) {
1614 s64 hash = ceph_decode_64(p);
1615 if (hash != -1) {
1616 pr_warn("ceph_object_locator::hash is set\n");
1617 goto e_inval;
1618 }
1619 }
1620
1621 /* skip the rest */
1622 *p = struct_end;
1623 out:
1624 return ret;
1625
1626 e_inval:
1627 ret = -EINVAL;
1628 goto out;
1629 }
1630
1631 static int ceph_redirect_decode(void **p, void *end,
1632 struct ceph_request_redirect *redir)
1633 {
1634 u8 struct_v, struct_cv;
1635 u32 len;
1636 void *struct_end;
1637 int ret;
1638
1639 ceph_decode_need(p, end, 1 + 1 + 4, e_inval);
1640 struct_v = ceph_decode_8(p);
1641 struct_cv = ceph_decode_8(p);
1642 if (struct_cv > 1) {
1643 pr_warn("got v %d cv %d > 1 of ceph_request_redirect\n",
1644 struct_v, struct_cv);
1645 goto e_inval;
1646 }
1647 len = ceph_decode_32(p);
1648 ceph_decode_need(p, end, len, e_inval);
1649 struct_end = *p + len;
1650
1651 ret = ceph_oloc_decode(p, end, &redir->oloc);
1652 if (ret)
1653 goto out;
1654
1655 len = ceph_decode_32(p);
1656 if (len > 0) {
1657 pr_warn("ceph_request_redirect::object_name is set\n");
1658 goto e_inval;
1659 }
1660
1661 len = ceph_decode_32(p);
1662 *p += len; /* skip osd_instructions */
1663
1664 /* skip the rest */
1665 *p = struct_end;
1666 out:
1667 return ret;
1668
1669 e_inval:
1670 ret = -EINVAL;
1671 goto out;
1672 }
1673
1674 static void complete_request(struct ceph_osd_request *req)
1675 {
1676 complete_all(&req->r_safe_completion); /* fsync waiter */
1677 }
1678
1679 /*
1680 * handle osd op reply. either call the callback if it is specified,
1681 * or do the completion to wake up the waiting thread.
1682 */
1683 static void handle_reply(struct ceph_osd_client *osdc, struct ceph_msg *msg,
1684 struct ceph_connection *con)
1685 {
1686 void *p, *end;
1687 struct ceph_osd_request *req;
1688 struct ceph_request_redirect redir;
1689 u64 tid;
1690 int object_len;
1691 unsigned int numops;
1692 int payload_len, flags;
1693 s32 result;
1694 s32 retry_attempt;
1695 struct ceph_pg pg;
1696 int err;
1697 u32 reassert_epoch;
1698 u64 reassert_version;
1699 u32 osdmap_epoch;
1700 int already_completed;
1701 u32 bytes;
1702 unsigned int i;
1703
1704 tid = le64_to_cpu(msg->hdr.tid);
1705 dout("handle_reply %p tid %llu\n", msg, tid);
1706
1707 p = msg->front.iov_base;
1708 end = p + msg->front.iov_len;
1709
1710 ceph_decode_need(&p, end, 4, bad);
1711 object_len = ceph_decode_32(&p);
1712 ceph_decode_need(&p, end, object_len, bad);
1713 p += object_len;
1714
1715 err = ceph_decode_pgid(&p, end, &pg);
1716 if (err)
1717 goto bad;
1718
1719 ceph_decode_need(&p, end, 8 + 4 + 4 + 8 + 4, bad);
1720 flags = ceph_decode_64(&p);
1721 result = ceph_decode_32(&p);
1722 reassert_epoch = ceph_decode_32(&p);
1723 reassert_version = ceph_decode_64(&p);
1724 osdmap_epoch = ceph_decode_32(&p);
1725
1726 /* lookup */
1727 down_read(&osdc->map_sem);
1728 mutex_lock(&osdc->request_mutex);
1729 req = __lookup_request(osdc, tid);
1730 if (req == NULL) {
1731 dout("handle_reply tid %llu dne\n", tid);
1732 goto bad_mutex;
1733 }
1734 ceph_osdc_get_request(req);
1735
1736 dout("handle_reply %p tid %llu req %p result %d\n", msg, tid,
1737 req, result);
1738
1739 ceph_decode_need(&p, end, 4, bad_put);
1740 numops = ceph_decode_32(&p);
1741 if (numops > CEPH_OSD_MAX_OP)
1742 goto bad_put;
1743 if (numops != req->r_num_ops)
1744 goto bad_put;
1745 payload_len = 0;
1746 ceph_decode_need(&p, end, numops * sizeof(struct ceph_osd_op), bad_put);
1747 for (i = 0; i < numops; i++) {
1748 struct ceph_osd_op *op = p;
1749 int len;
1750
1751 len = le32_to_cpu(op->payload_len);
1752 req->r_reply_op_len[i] = len;
1753 dout(" op %d has %d bytes\n", i, len);
1754 payload_len += len;
1755 p += sizeof(*op);
1756 }
1757 bytes = le32_to_cpu(msg->hdr.data_len);
1758 if (payload_len != bytes) {
1759 pr_warning("sum of op payload lens %d != data_len %d",
1760 payload_len, bytes);
1761 goto bad_put;
1762 }
1763
1764 ceph_decode_need(&p, end, 4 + numops * 4, bad_put);
1765 retry_attempt = ceph_decode_32(&p);
1766 for (i = 0; i < numops; i++)
1767 req->r_reply_op_result[i] = ceph_decode_32(&p);
1768
1769 if (le16_to_cpu(msg->hdr.version) >= 6) {
1770 p += 8 + 4; /* skip replay_version */
1771 p += 8; /* skip user_version */
1772
1773 err = ceph_redirect_decode(&p, end, &redir);
1774 if (err)
1775 goto bad_put;
1776 } else {
1777 redir.oloc.pool = -1;
1778 }
1779
1780 if (redir.oloc.pool != -1) {
1781 dout("redirect pool %lld\n", redir.oloc.pool);
1782
1783 __unregister_request(osdc, req);
1784
1785 req->r_target_oloc = redir.oloc; /* struct */
1786
1787 /*
1788 * Start redirect requests with nofail=true. If
1789 * mapping fails, request will end up on the notarget
1790 * list, waiting for the new osdmap (which can take
1791 * a while), even though the original request mapped
1792 * successfully. In the future we might want to follow
1793 * original request's nofail setting here.
1794 */
1795 err = __ceph_osdc_start_request(osdc, req, true);
1796 BUG_ON(err);
1797
1798 goto out_unlock;
1799 }
1800
1801 already_completed = req->r_got_reply;
1802 if (!req->r_got_reply) {
1803 req->r_result = result;
1804 dout("handle_reply result %d bytes %d\n", req->r_result,
1805 bytes);
1806 if (req->r_result == 0)
1807 req->r_result = bytes;
1808
1809 /* in case this is a write and we need to replay, */
1810 req->r_reassert_version.epoch = cpu_to_le32(reassert_epoch);
1811 req->r_reassert_version.version = cpu_to_le64(reassert_version);
1812
1813 req->r_got_reply = 1;
1814 } else if ((flags & CEPH_OSD_FLAG_ONDISK) == 0) {
1815 dout("handle_reply tid %llu dup ack\n", tid);
1816 goto out_unlock;
1817 }
1818
1819 dout("handle_reply tid %llu flags %d\n", tid, flags);
1820
1821 if (req->r_linger && (flags & CEPH_OSD_FLAG_ONDISK))
1822 __register_linger_request(osdc, req);
1823
1824 /* either this is a read, or we got the safe response */
1825 if (result < 0 ||
1826 (flags & CEPH_OSD_FLAG_ONDISK) ||
1827 ((flags & CEPH_OSD_FLAG_WRITE) == 0))
1828 __unregister_request(osdc, req);
1829
1830 mutex_unlock(&osdc->request_mutex);
1831 up_read(&osdc->map_sem);
1832
1833 if (!already_completed) {
1834 if (req->r_unsafe_callback &&
1835 result >= 0 && !(flags & CEPH_OSD_FLAG_ONDISK))
1836 req->r_unsafe_callback(req, true);
1837 if (req->r_callback)
1838 req->r_callback(req, msg);
1839 else
1840 complete_all(&req->r_completion);
1841 }
1842
1843 if (flags & CEPH_OSD_FLAG_ONDISK) {
1844 if (req->r_unsafe_callback && already_completed)
1845 req->r_unsafe_callback(req, false);
1846 complete_request(req);
1847 }
1848
1849 out:
1850 dout("req=%p req->r_linger=%d\n", req, req->r_linger);
1851 ceph_osdc_put_request(req);
1852 return;
1853 out_unlock:
1854 mutex_unlock(&osdc->request_mutex);
1855 up_read(&osdc->map_sem);
1856 goto out;
1857
1858 bad_put:
1859 req->r_result = -EIO;
1860 __unregister_request(osdc, req);
1861 if (req->r_callback)
1862 req->r_callback(req, msg);
1863 else
1864 complete_all(&req->r_completion);
1865 complete_request(req);
1866 ceph_osdc_put_request(req);
1867 bad_mutex:
1868 mutex_unlock(&osdc->request_mutex);
1869 up_read(&osdc->map_sem);
1870 bad:
1871 pr_err("corrupt osd_op_reply got %d %d\n",
1872 (int)msg->front.iov_len, le32_to_cpu(msg->hdr.front_len));
1873 ceph_msg_dump(msg);
1874 }
1875
1876 static void reset_changed_osds(struct ceph_osd_client *osdc)
1877 {
1878 struct rb_node *p, *n;
1879
1880 dout("%s %p\n", __func__, osdc);
1881 for (p = rb_first(&osdc->osds); p; p = n) {
1882 struct ceph_osd *osd = rb_entry(p, struct ceph_osd, o_node);
1883
1884 n = rb_next(p);
1885 if (!ceph_osd_is_up(osdc->osdmap, osd->o_osd) ||
1886 memcmp(&osd->o_con.peer_addr,
1887 ceph_osd_addr(osdc->osdmap,
1888 osd->o_osd),
1889 sizeof(struct ceph_entity_addr)) != 0)
1890 __reset_osd(osdc, osd);
1891 }
1892 }
1893
1894 /*
1895 * Requeue requests whose mapping to an OSD has changed. If requests map to
1896 * no osd, request a new map.
1897 *
1898 * Caller should hold map_sem for read.
1899 */
1900 static void kick_requests(struct ceph_osd_client *osdc, bool force_resend,
1901 bool force_resend_writes)
1902 {
1903 struct ceph_osd_request *req, *nreq;
1904 struct rb_node *p;
1905 int needmap = 0;
1906 int err;
1907 bool force_resend_req;
1908
1909 dout("kick_requests %s %s\n", force_resend ? " (force resend)" : "",
1910 force_resend_writes ? " (force resend writes)" : "");
1911 mutex_lock(&osdc->request_mutex);
1912 for (p = rb_first(&osdc->requests); p; ) {
1913 req = rb_entry(p, struct ceph_osd_request, r_node);
1914 p = rb_next(p);
1915
1916 /*
1917 * For linger requests that have not yet been
1918 * registered, move them to the linger list; they'll
1919 * be sent to the osd in the loop below. Unregister
1920 * the request before re-registering it as a linger
1921 * request to ensure the __map_request() below
1922 * will decide it needs to be sent.
1923 */
1924 if (req->r_linger && list_empty(&req->r_linger_item)) {
1925 dout("%p tid %llu restart on osd%d\n",
1926 req, req->r_tid,
1927 req->r_osd ? req->r_osd->o_osd : -1);
1928 ceph_osdc_get_request(req);
1929 __unregister_request(osdc, req);
1930 __register_linger_request(osdc, req);
1931 ceph_osdc_put_request(req);
1932 continue;
1933 }
1934
1935 force_resend_req = force_resend ||
1936 (force_resend_writes &&
1937 req->r_flags & CEPH_OSD_FLAG_WRITE);
1938 err = __map_request(osdc, req, force_resend_req);
1939 if (err < 0)
1940 continue; /* error */
1941 if (req->r_osd == NULL) {
1942 dout("%p tid %llu maps to no osd\n", req, req->r_tid);
1943 needmap++; /* request a newer map */
1944 } else if (err > 0) {
1945 if (!req->r_linger) {
1946 dout("%p tid %llu requeued on osd%d\n", req,
1947 req->r_tid,
1948 req->r_osd ? req->r_osd->o_osd : -1);
1949 req->r_flags |= CEPH_OSD_FLAG_RETRY;
1950 }
1951 }
1952 }
1953
1954 list_for_each_entry_safe(req, nreq, &osdc->req_linger,
1955 r_linger_item) {
1956 dout("linger req=%p req->r_osd=%p\n", req, req->r_osd);
1957
1958 err = __map_request(osdc, req,
1959 force_resend || force_resend_writes);
1960 dout("__map_request returned %d\n", err);
1961 if (err < 0)
1962 continue; /* hrm! */
1963 if (req->r_osd == NULL || err > 0) {
1964 if (req->r_osd == NULL) {
1965 dout("lingering %p tid %llu maps to no osd\n",
1966 req, req->r_tid);
1967 /*
1968 * A homeless lingering request makes
1969 * no sense, as it's job is to keep
1970 * a particular OSD connection open.
1971 * Request a newer map and kick the
1972 * request, knowing that it won't be
1973 * resent until we actually get a map
1974 * that can tell us where to send it.
1975 */
1976 needmap++;
1977 }
1978
1979 dout("kicking lingering %p tid %llu osd%d\n", req,
1980 req->r_tid, req->r_osd ? req->r_osd->o_osd : -1);
1981 __register_request(osdc, req);
1982 __unregister_linger_request(osdc, req);
1983 }
1984 }
1985 reset_changed_osds(osdc);
1986 mutex_unlock(&osdc->request_mutex);
1987
1988 if (needmap) {
1989 dout("%d requests for down osds, need new map\n", needmap);
1990 ceph_monc_request_next_osdmap(&osdc->client->monc);
1991 }
1992 }
1993
1994
1995 /*
1996 * Process updated osd map.
1997 *
1998 * The message contains any number of incremental and full maps, normally
1999 * indicating some sort of topology change in the cluster. Kick requests
2000 * off to different OSDs as needed.
2001 */
2002 void ceph_osdc_handle_map(struct ceph_osd_client *osdc, struct ceph_msg *msg)
2003 {
2004 void *p, *end, *next;
2005 u32 nr_maps, maplen;
2006 u32 epoch;
2007 struct ceph_osdmap *newmap = NULL, *oldmap;
2008 int err;
2009 struct ceph_fsid fsid;
2010 bool was_full;
2011
2012 dout("handle_map have %u\n", osdc->osdmap ? osdc->osdmap->epoch : 0);
2013 p = msg->front.iov_base;
2014 end = p + msg->front.iov_len;
2015
2016 /* verify fsid */
2017 ceph_decode_need(&p, end, sizeof(fsid), bad);
2018 ceph_decode_copy(&p, &fsid, sizeof(fsid));
2019 if (ceph_check_fsid(osdc->client, &fsid) < 0)
2020 return;
2021
2022 down_write(&osdc->map_sem);
2023
2024 was_full = ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_FULL);
2025
2026 /* incremental maps */
2027 ceph_decode_32_safe(&p, end, nr_maps, bad);
2028 dout(" %d inc maps\n", nr_maps);
2029 while (nr_maps > 0) {
2030 ceph_decode_need(&p, end, 2*sizeof(u32), bad);
2031 epoch = ceph_decode_32(&p);
2032 maplen = ceph_decode_32(&p);
2033 ceph_decode_need(&p, end, maplen, bad);
2034 next = p + maplen;
2035 if (osdc->osdmap && osdc->osdmap->epoch+1 == epoch) {
2036 dout("applying incremental map %u len %d\n",
2037 epoch, maplen);
2038 newmap = osdmap_apply_incremental(&p, next,
2039 osdc->osdmap,
2040 &osdc->client->msgr);
2041 if (IS_ERR(newmap)) {
2042 err = PTR_ERR(newmap);
2043 goto bad;
2044 }
2045 BUG_ON(!newmap);
2046 if (newmap != osdc->osdmap) {
2047 ceph_osdmap_destroy(osdc->osdmap);
2048 osdc->osdmap = newmap;
2049 }
2050 was_full = was_full ||
2051 ceph_osdmap_flag(osdc->osdmap,
2052 CEPH_OSDMAP_FULL);
2053 kick_requests(osdc, 0, was_full);
2054 } else {
2055 dout("ignoring incremental map %u len %d\n",
2056 epoch, maplen);
2057 }
2058 p = next;
2059 nr_maps--;
2060 }
2061 if (newmap)
2062 goto done;
2063
2064 /* full maps */
2065 ceph_decode_32_safe(&p, end, nr_maps, bad);
2066 dout(" %d full maps\n", nr_maps);
2067 while (nr_maps) {
2068 ceph_decode_need(&p, end, 2*sizeof(u32), bad);
2069 epoch = ceph_decode_32(&p);
2070 maplen = ceph_decode_32(&p);
2071 ceph_decode_need(&p, end, maplen, bad);
2072 if (nr_maps > 1) {
2073 dout("skipping non-latest full map %u len %d\n",
2074 epoch, maplen);
2075 } else if (osdc->osdmap && osdc->osdmap->epoch >= epoch) {
2076 dout("skipping full map %u len %d, "
2077 "older than our %u\n", epoch, maplen,
2078 osdc->osdmap->epoch);
2079 } else {
2080 int skipped_map = 0;
2081
2082 dout("taking full map %u len %d\n", epoch, maplen);
2083 newmap = ceph_osdmap_decode(&p, p+maplen);
2084 if (IS_ERR(newmap)) {
2085 err = PTR_ERR(newmap);
2086 goto bad;
2087 }
2088 BUG_ON(!newmap);
2089 oldmap = osdc->osdmap;
2090 osdc->osdmap = newmap;
2091 if (oldmap) {
2092 if (oldmap->epoch + 1 < newmap->epoch)
2093 skipped_map = 1;
2094 ceph_osdmap_destroy(oldmap);
2095 }
2096 was_full = was_full ||
2097 ceph_osdmap_flag(osdc->osdmap,
2098 CEPH_OSDMAP_FULL);
2099 kick_requests(osdc, skipped_map, was_full);
2100 }
2101 p += maplen;
2102 nr_maps--;
2103 }
2104
2105 if (!osdc->osdmap)
2106 goto bad;
2107 done:
2108 downgrade_write(&osdc->map_sem);
2109 ceph_monc_got_osdmap(&osdc->client->monc, osdc->osdmap->epoch);
2110
2111 /*
2112 * subscribe to subsequent osdmap updates if full to ensure
2113 * we find out when we are no longer full and stop returning
2114 * ENOSPC.
2115 */
2116 if (ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_FULL) ||
2117 ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_PAUSERD) ||
2118 ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_PAUSEWR))
2119 ceph_monc_request_next_osdmap(&osdc->client->monc);
2120
2121 mutex_lock(&osdc->request_mutex);
2122 __send_queued(osdc);
2123 mutex_unlock(&osdc->request_mutex);
2124 up_read(&osdc->map_sem);
2125 wake_up_all(&osdc->client->auth_wq);
2126 return;
2127
2128 bad:
2129 pr_err("osdc handle_map corrupt msg\n");
2130 ceph_msg_dump(msg);
2131 up_write(&osdc->map_sem);
2132 }
2133
2134 /*
2135 * watch/notify callback event infrastructure
2136 *
2137 * These callbacks are used both for watch and notify operations.
2138 */
2139 static void __release_event(struct kref *kref)
2140 {
2141 struct ceph_osd_event *event =
2142 container_of(kref, struct ceph_osd_event, kref);
2143
2144 dout("__release_event %p\n", event);
2145 kfree(event);
2146 }
2147
2148 static void get_event(struct ceph_osd_event *event)
2149 {
2150 kref_get(&event->kref);
2151 }
2152
2153 void ceph_osdc_put_event(struct ceph_osd_event *event)
2154 {
2155 kref_put(&event->kref, __release_event);
2156 }
2157 EXPORT_SYMBOL(ceph_osdc_put_event);
2158
2159 static void __insert_event(struct ceph_osd_client *osdc,
2160 struct ceph_osd_event *new)
2161 {
2162 struct rb_node **p = &osdc->event_tree.rb_node;
2163 struct rb_node *parent = NULL;
2164 struct ceph_osd_event *event = NULL;
2165
2166 while (*p) {
2167 parent = *p;
2168 event = rb_entry(parent, struct ceph_osd_event, node);
2169 if (new->cookie < event->cookie)
2170 p = &(*p)->rb_left;
2171 else if (new->cookie > event->cookie)
2172 p = &(*p)->rb_right;
2173 else
2174 BUG();
2175 }
2176
2177 rb_link_node(&new->node, parent, p);
2178 rb_insert_color(&new->node, &osdc->event_tree);
2179 }
2180
2181 static struct ceph_osd_event *__find_event(struct ceph_osd_client *osdc,
2182 u64 cookie)
2183 {
2184 struct rb_node **p = &osdc->event_tree.rb_node;
2185 struct rb_node *parent = NULL;
2186 struct ceph_osd_event *event = NULL;
2187
2188 while (*p) {
2189 parent = *p;
2190 event = rb_entry(parent, struct ceph_osd_event, node);
2191 if (cookie < event->cookie)
2192 p = &(*p)->rb_left;
2193 else if (cookie > event->cookie)
2194 p = &(*p)->rb_right;
2195 else
2196 return event;
2197 }
2198 return NULL;
2199 }
2200
2201 static void __remove_event(struct ceph_osd_event *event)
2202 {
2203 struct ceph_osd_client *osdc = event->osdc;
2204
2205 if (!RB_EMPTY_NODE(&event->node)) {
2206 dout("__remove_event removed %p\n", event);
2207 rb_erase(&event->node, &osdc->event_tree);
2208 ceph_osdc_put_event(event);
2209 } else {
2210 dout("__remove_event didn't remove %p\n", event);
2211 }
2212 }
2213
2214 int ceph_osdc_create_event(struct ceph_osd_client *osdc,
2215 void (*event_cb)(u64, u64, u8, void *),
2216 void *data, struct ceph_osd_event **pevent)
2217 {
2218 struct ceph_osd_event *event;
2219
2220 event = kmalloc(sizeof(*event), GFP_NOIO);
2221 if (!event)
2222 return -ENOMEM;
2223
2224 dout("create_event %p\n", event);
2225 event->cb = event_cb;
2226 event->one_shot = 0;
2227 event->data = data;
2228 event->osdc = osdc;
2229 INIT_LIST_HEAD(&event->osd_node);
2230 RB_CLEAR_NODE(&event->node);
2231 kref_init(&event->kref); /* one ref for us */
2232 kref_get(&event->kref); /* one ref for the caller */
2233
2234 spin_lock(&osdc->event_lock);
2235 event->cookie = ++osdc->event_count;
2236 __insert_event(osdc, event);
2237 spin_unlock(&osdc->event_lock);
2238
2239 *pevent = event;
2240 return 0;
2241 }
2242 EXPORT_SYMBOL(ceph_osdc_create_event);
2243
2244 void ceph_osdc_cancel_event(struct ceph_osd_event *event)
2245 {
2246 struct ceph_osd_client *osdc = event->osdc;
2247
2248 dout("cancel_event %p\n", event);
2249 spin_lock(&osdc->event_lock);
2250 __remove_event(event);
2251 spin_unlock(&osdc->event_lock);
2252 ceph_osdc_put_event(event); /* caller's */
2253 }
2254 EXPORT_SYMBOL(ceph_osdc_cancel_event);
2255
2256
2257 static void do_event_work(struct work_struct *work)
2258 {
2259 struct ceph_osd_event_work *event_work =
2260 container_of(work, struct ceph_osd_event_work, work);
2261 struct ceph_osd_event *event = event_work->event;
2262 u64 ver = event_work->ver;
2263 u64 notify_id = event_work->notify_id;
2264 u8 opcode = event_work->opcode;
2265
2266 dout("do_event_work completing %p\n", event);
2267 event->cb(ver, notify_id, opcode, event->data);
2268 dout("do_event_work completed %p\n", event);
2269 ceph_osdc_put_event(event);
2270 kfree(event_work);
2271 }
2272
2273
2274 /*
2275 * Process osd watch notifications
2276 */
2277 static void handle_watch_notify(struct ceph_osd_client *osdc,
2278 struct ceph_msg *msg)
2279 {
2280 void *p, *end;
2281 u8 proto_ver;
2282 u64 cookie, ver, notify_id;
2283 u8 opcode;
2284 struct ceph_osd_event *event;
2285 struct ceph_osd_event_work *event_work;
2286
2287 p = msg->front.iov_base;
2288 end = p + msg->front.iov_len;
2289
2290 ceph_decode_8_safe(&p, end, proto_ver, bad);
2291 ceph_decode_8_safe(&p, end, opcode, bad);
2292 ceph_decode_64_safe(&p, end, cookie, bad);
2293 ceph_decode_64_safe(&p, end, ver, bad);
2294 ceph_decode_64_safe(&p, end, notify_id, bad);
2295
2296 spin_lock(&osdc->event_lock);
2297 event = __find_event(osdc, cookie);
2298 if (event) {
2299 BUG_ON(event->one_shot);
2300 get_event(event);
2301 }
2302 spin_unlock(&osdc->event_lock);
2303 dout("handle_watch_notify cookie %lld ver %lld event %p\n",
2304 cookie, ver, event);
2305 if (event) {
2306 event_work = kmalloc(sizeof(*event_work), GFP_NOIO);
2307 if (!event_work) {
2308 dout("ERROR: could not allocate event_work\n");
2309 goto done_err;
2310 }
2311 INIT_WORK(&event_work->work, do_event_work);
2312 event_work->event = event;
2313 event_work->ver = ver;
2314 event_work->notify_id = notify_id;
2315 event_work->opcode = opcode;
2316 if (!queue_work(osdc->notify_wq, &event_work->work)) {
2317 dout("WARNING: failed to queue notify event work\n");
2318 goto done_err;
2319 }
2320 }
2321
2322 return;
2323
2324 done_err:
2325 ceph_osdc_put_event(event);
2326 return;
2327
2328 bad:
2329 pr_err("osdc handle_watch_notify corrupt msg\n");
2330 }
2331
2332 /*
2333 * build new request AND message
2334 *
2335 */
2336 void ceph_osdc_build_request(struct ceph_osd_request *req, u64 off,
2337 struct ceph_snap_context *snapc, u64 snap_id,
2338 struct timespec *mtime)
2339 {
2340 struct ceph_msg *msg = req->r_request;
2341 void *p;
2342 size_t msg_size;
2343 int flags = req->r_flags;
2344 u64 data_len;
2345 unsigned int i;
2346
2347 req->r_snapid = snap_id;
2348 req->r_snapc = ceph_get_snap_context(snapc);
2349
2350 /* encode request */
2351 msg->hdr.version = cpu_to_le16(4);
2352
2353 p = msg->front.iov_base;
2354 ceph_encode_32(&p, 1); /* client_inc is always 1 */
2355 req->r_request_osdmap_epoch = p;
2356 p += 4;
2357 req->r_request_flags = p;
2358 p += 4;
2359 if (req->r_flags & CEPH_OSD_FLAG_WRITE)
2360 ceph_encode_timespec(p, mtime);
2361 p += sizeof(struct ceph_timespec);
2362 req->r_request_reassert_version = p;
2363 p += sizeof(struct ceph_eversion); /* will get filled in */
2364
2365 /* oloc */
2366 ceph_encode_8(&p, 4);
2367 ceph_encode_8(&p, 4);
2368 ceph_encode_32(&p, 8 + 4 + 4);
2369 req->r_request_pool = p;
2370 p += 8;
2371 ceph_encode_32(&p, -1); /* preferred */
2372 ceph_encode_32(&p, 0); /* key len */
2373
2374 ceph_encode_8(&p, 1);
2375 req->r_request_pgid = p;
2376 p += 8 + 4;
2377 ceph_encode_32(&p, -1); /* preferred */
2378
2379 /* oid */
2380 ceph_encode_32(&p, req->r_base_oid.name_len);
2381 memcpy(p, req->r_base_oid.name, req->r_base_oid.name_len);
2382 dout("oid '%.*s' len %d\n", req->r_base_oid.name_len,
2383 req->r_base_oid.name, req->r_base_oid.name_len);
2384 p += req->r_base_oid.name_len;
2385
2386 /* ops--can imply data */
2387 ceph_encode_16(&p, (u16)req->r_num_ops);
2388 data_len = 0;
2389 for (i = 0; i < req->r_num_ops; i++) {
2390 data_len += osd_req_encode_op(req, p, i);
2391 p += sizeof(struct ceph_osd_op);
2392 }
2393
2394 /* snaps */
2395 ceph_encode_64(&p, req->r_snapid);
2396 ceph_encode_64(&p, req->r_snapc ? req->r_snapc->seq : 0);
2397 ceph_encode_32(&p, req->r_snapc ? req->r_snapc->num_snaps : 0);
2398 if (req->r_snapc) {
2399 for (i = 0; i < snapc->num_snaps; i++) {
2400 ceph_encode_64(&p, req->r_snapc->snaps[i]);
2401 }
2402 }
2403
2404 req->r_request_attempts = p;
2405 p += 4;
2406
2407 /* data */
2408 if (flags & CEPH_OSD_FLAG_WRITE) {
2409 u16 data_off;
2410
2411 /*
2412 * The header "data_off" is a hint to the receiver
2413 * allowing it to align received data into its
2414 * buffers such that there's no need to re-copy
2415 * it before writing it to disk (direct I/O).
2416 */
2417 data_off = (u16) (off & 0xffff);
2418 req->r_request->hdr.data_off = cpu_to_le16(data_off);
2419 }
2420 req->r_request->hdr.data_len = cpu_to_le32(data_len);
2421
2422 BUG_ON(p > msg->front.iov_base + msg->front.iov_len);
2423 msg_size = p - msg->front.iov_base;
2424 msg->front.iov_len = msg_size;
2425 msg->hdr.front_len = cpu_to_le32(msg_size);
2426
2427 dout("build_request msg_size was %d\n", (int)msg_size);
2428 }
2429 EXPORT_SYMBOL(ceph_osdc_build_request);
2430
2431 /*
2432 * Register request, send initial attempt.
2433 */
2434 int ceph_osdc_start_request(struct ceph_osd_client *osdc,
2435 struct ceph_osd_request *req,
2436 bool nofail)
2437 {
2438 int rc;
2439
2440 down_read(&osdc->map_sem);
2441 mutex_lock(&osdc->request_mutex);
2442
2443 rc = __ceph_osdc_start_request(osdc, req, nofail);
2444
2445 mutex_unlock(&osdc->request_mutex);
2446 up_read(&osdc->map_sem);
2447
2448 return rc;
2449 }
2450 EXPORT_SYMBOL(ceph_osdc_start_request);
2451
2452 /*
2453 * wait for a request to complete
2454 */
2455 int ceph_osdc_wait_request(struct ceph_osd_client *osdc,
2456 struct ceph_osd_request *req)
2457 {
2458 int rc;
2459
2460 rc = wait_for_completion_interruptible(&req->r_completion);
2461 if (rc < 0) {
2462 mutex_lock(&osdc->request_mutex);
2463 __cancel_request(req);
2464 __unregister_request(osdc, req);
2465 mutex_unlock(&osdc->request_mutex);
2466 complete_request(req);
2467 dout("wait_request tid %llu canceled/timed out\n", req->r_tid);
2468 return rc;
2469 }
2470
2471 dout("wait_request tid %llu result %d\n", req->r_tid, req->r_result);
2472 return req->r_result;
2473 }
2474 EXPORT_SYMBOL(ceph_osdc_wait_request);
2475
2476 /*
2477 * sync - wait for all in-flight requests to flush. avoid starvation.
2478 */
2479 void ceph_osdc_sync(struct ceph_osd_client *osdc)
2480 {
2481 struct ceph_osd_request *req;
2482 u64 last_tid, next_tid = 0;
2483
2484 mutex_lock(&osdc->request_mutex);
2485 last_tid = osdc->last_tid;
2486 while (1) {
2487 req = __lookup_request_ge(osdc, next_tid);
2488 if (!req)
2489 break;
2490 if (req->r_tid > last_tid)
2491 break;
2492
2493 next_tid = req->r_tid + 1;
2494 if ((req->r_flags & CEPH_OSD_FLAG_WRITE) == 0)
2495 continue;
2496
2497 ceph_osdc_get_request(req);
2498 mutex_unlock(&osdc->request_mutex);
2499 dout("sync waiting on tid %llu (last is %llu)\n",
2500 req->r_tid, last_tid);
2501 wait_for_completion(&req->r_safe_completion);
2502 mutex_lock(&osdc->request_mutex);
2503 ceph_osdc_put_request(req);
2504 }
2505 mutex_unlock(&osdc->request_mutex);
2506 dout("sync done (thru tid %llu)\n", last_tid);
2507 }
2508 EXPORT_SYMBOL(ceph_osdc_sync);
2509
2510 /*
2511 * Call all pending notify callbacks - for use after a watch is
2512 * unregistered, to make sure no more callbacks for it will be invoked
2513 */
2514 void ceph_osdc_flush_notifies(struct ceph_osd_client *osdc)
2515 {
2516 flush_workqueue(osdc->notify_wq);
2517 }
2518 EXPORT_SYMBOL(ceph_osdc_flush_notifies);
2519
2520
2521 /*
2522 * init, shutdown
2523 */
2524 int ceph_osdc_init(struct ceph_osd_client *osdc, struct ceph_client *client)
2525 {
2526 int err;
2527
2528 dout("init\n");
2529 osdc->client = client;
2530 osdc->osdmap = NULL;
2531 init_rwsem(&osdc->map_sem);
2532 init_completion(&osdc->map_waiters);
2533 osdc->last_requested_map = 0;
2534 mutex_init(&osdc->request_mutex);
2535 osdc->last_tid = 0;
2536 osdc->osds = RB_ROOT;
2537 INIT_LIST_HEAD(&osdc->osd_lru);
2538 osdc->requests = RB_ROOT;
2539 INIT_LIST_HEAD(&osdc->req_lru);
2540 INIT_LIST_HEAD(&osdc->req_unsent);
2541 INIT_LIST_HEAD(&osdc->req_notarget);
2542 INIT_LIST_HEAD(&osdc->req_linger);
2543 osdc->num_requests = 0;
2544 INIT_DELAYED_WORK(&osdc->timeout_work, handle_timeout);
2545 INIT_DELAYED_WORK(&osdc->osds_timeout_work, handle_osds_timeout);
2546 spin_lock_init(&osdc->event_lock);
2547 osdc->event_tree = RB_ROOT;
2548 osdc->event_count = 0;
2549
2550 schedule_delayed_work(&osdc->osds_timeout_work,
2551 round_jiffies_relative(osdc->client->options->osd_idle_ttl * HZ));
2552
2553 err = -ENOMEM;
2554 osdc->req_mempool = mempool_create_kmalloc_pool(10,
2555 sizeof(struct ceph_osd_request));
2556 if (!osdc->req_mempool)
2557 goto out;
2558
2559 err = ceph_msgpool_init(&osdc->msgpool_op, CEPH_MSG_OSD_OP,
2560 OSD_OP_FRONT_LEN, 10, true,
2561 "osd_op");
2562 if (err < 0)
2563 goto out_mempool;
2564 err = ceph_msgpool_init(&osdc->msgpool_op_reply, CEPH_MSG_OSD_OPREPLY,
2565 OSD_OPREPLY_FRONT_LEN, 10, true,
2566 "osd_op_reply");
2567 if (err < 0)
2568 goto out_msgpool;
2569
2570 err = -ENOMEM;
2571 osdc->notify_wq = create_singlethread_workqueue("ceph-watch-notify");
2572 if (!osdc->notify_wq)
2573 goto out_msgpool_reply;
2574
2575 return 0;
2576
2577 out_msgpool_reply:
2578 ceph_msgpool_destroy(&osdc->msgpool_op_reply);
2579 out_msgpool:
2580 ceph_msgpool_destroy(&osdc->msgpool_op);
2581 out_mempool:
2582 mempool_destroy(osdc->req_mempool);
2583 out:
2584 return err;
2585 }
2586
2587 void ceph_osdc_stop(struct ceph_osd_client *osdc)
2588 {
2589 flush_workqueue(osdc->notify_wq);
2590 destroy_workqueue(osdc->notify_wq);
2591 cancel_delayed_work_sync(&osdc->timeout_work);
2592 cancel_delayed_work_sync(&osdc->osds_timeout_work);
2593 if (osdc->osdmap) {
2594 ceph_osdmap_destroy(osdc->osdmap);
2595 osdc->osdmap = NULL;
2596 }
2597 remove_all_osds(osdc);
2598 mempool_destroy(osdc->req_mempool);
2599 ceph_msgpool_destroy(&osdc->msgpool_op);
2600 ceph_msgpool_destroy(&osdc->msgpool_op_reply);
2601 }
2602
2603 /*
2604 * Read some contiguous pages. If we cross a stripe boundary, shorten
2605 * *plen. Return number of bytes read, or error.
2606 */
2607 int ceph_osdc_readpages(struct ceph_osd_client *osdc,
2608 struct ceph_vino vino, struct ceph_file_layout *layout,
2609 u64 off, u64 *plen,
2610 u32 truncate_seq, u64 truncate_size,
2611 struct page **pages, int num_pages, int page_align)
2612 {
2613 struct ceph_osd_request *req;
2614 int rc = 0;
2615
2616 dout("readpages on ino %llx.%llx on %llu~%llu\n", vino.ino,
2617 vino.snap, off, *plen);
2618 req = ceph_osdc_new_request(osdc, layout, vino, off, plen, 1,
2619 CEPH_OSD_OP_READ, CEPH_OSD_FLAG_READ,
2620 NULL, truncate_seq, truncate_size,
2621 false);
2622 if (IS_ERR(req))
2623 return PTR_ERR(req);
2624
2625 /* it may be a short read due to an object boundary */
2626
2627 osd_req_op_extent_osd_data_pages(req, 0,
2628 pages, *plen, page_align, false, false);
2629
2630 dout("readpages final extent is %llu~%llu (%llu bytes align %d)\n",
2631 off, *plen, *plen, page_align);
2632
2633 ceph_osdc_build_request(req, off, NULL, vino.snap, NULL);
2634
2635 rc = ceph_osdc_start_request(osdc, req, false);
2636 if (!rc)
2637 rc = ceph_osdc_wait_request(osdc, req);
2638
2639 ceph_osdc_put_request(req);
2640 dout("readpages result %d\n", rc);
2641 return rc;
2642 }
2643 EXPORT_SYMBOL(ceph_osdc_readpages);
2644
2645 /*
2646 * do a synchronous write on N pages
2647 */
2648 int ceph_osdc_writepages(struct ceph_osd_client *osdc, struct ceph_vino vino,
2649 struct ceph_file_layout *layout,
2650 struct ceph_snap_context *snapc,
2651 u64 off, u64 len,
2652 u32 truncate_seq, u64 truncate_size,
2653 struct timespec *mtime,
2654 struct page **pages, int num_pages)
2655 {
2656 struct ceph_osd_request *req;
2657 int rc = 0;
2658 int page_align = off & ~PAGE_MASK;
2659
2660 BUG_ON(vino.snap != CEPH_NOSNAP); /* snapshots aren't writeable */
2661 req = ceph_osdc_new_request(osdc, layout, vino, off, &len, 1,
2662 CEPH_OSD_OP_WRITE,
2663 CEPH_OSD_FLAG_ONDISK | CEPH_OSD_FLAG_WRITE,
2664 snapc, truncate_seq, truncate_size,
2665 true);
2666 if (IS_ERR(req))
2667 return PTR_ERR(req);
2668
2669 /* it may be a short write due to an object boundary */
2670 osd_req_op_extent_osd_data_pages(req, 0, pages, len, page_align,
2671 false, false);
2672 dout("writepages %llu~%llu (%llu bytes)\n", off, len, len);
2673
2674 ceph_osdc_build_request(req, off, snapc, CEPH_NOSNAP, mtime);
2675
2676 rc = ceph_osdc_start_request(osdc, req, true);
2677 if (!rc)
2678 rc = ceph_osdc_wait_request(osdc, req);
2679
2680 ceph_osdc_put_request(req);
2681 if (rc == 0)
2682 rc = len;
2683 dout("writepages result %d\n", rc);
2684 return rc;
2685 }
2686 EXPORT_SYMBOL(ceph_osdc_writepages);
2687
2688 int ceph_osdc_setup(void)
2689 {
2690 BUG_ON(ceph_osd_request_cache);
2691 ceph_osd_request_cache = kmem_cache_create("ceph_osd_request",
2692 sizeof (struct ceph_osd_request),
2693 __alignof__(struct ceph_osd_request),
2694 0, NULL);
2695
2696 return ceph_osd_request_cache ? 0 : -ENOMEM;
2697 }
2698 EXPORT_SYMBOL(ceph_osdc_setup);
2699
2700 void ceph_osdc_cleanup(void)
2701 {
2702 BUG_ON(!ceph_osd_request_cache);
2703 kmem_cache_destroy(ceph_osd_request_cache);
2704 ceph_osd_request_cache = NULL;
2705 }
2706 EXPORT_SYMBOL(ceph_osdc_cleanup);
2707
2708 /*
2709 * handle incoming message
2710 */
2711 static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
2712 {
2713 struct ceph_osd *osd = con->private;
2714 struct ceph_osd_client *osdc;
2715 int type = le16_to_cpu(msg->hdr.type);
2716
2717 if (!osd)
2718 goto out;
2719 osdc = osd->o_osdc;
2720
2721 switch (type) {
2722 case CEPH_MSG_OSD_MAP:
2723 ceph_osdc_handle_map(osdc, msg);
2724 break;
2725 case CEPH_MSG_OSD_OPREPLY:
2726 handle_reply(osdc, msg, con);
2727 break;
2728 case CEPH_MSG_WATCH_NOTIFY:
2729 handle_watch_notify(osdc, msg);
2730 break;
2731
2732 default:
2733 pr_err("received unknown message type %d %s\n", type,
2734 ceph_msg_type_name(type));
2735 }
2736 out:
2737 ceph_msg_put(msg);
2738 }
2739
2740 /*
2741 * lookup and return message for incoming reply. set up reply message
2742 * pages.
2743 */
2744 static struct ceph_msg *get_reply(struct ceph_connection *con,
2745 struct ceph_msg_header *hdr,
2746 int *skip)
2747 {
2748 struct ceph_osd *osd = con->private;
2749 struct ceph_osd_client *osdc = osd->o_osdc;
2750 struct ceph_msg *m;
2751 struct ceph_osd_request *req;
2752 int front_len = le32_to_cpu(hdr->front_len);
2753 int data_len = le32_to_cpu(hdr->data_len);
2754 u64 tid;
2755
2756 tid = le64_to_cpu(hdr->tid);
2757 mutex_lock(&osdc->request_mutex);
2758 req = __lookup_request(osdc, tid);
2759 if (!req) {
2760 *skip = 1;
2761 m = NULL;
2762 dout("get_reply unknown tid %llu from osd%d\n", tid,
2763 osd->o_osd);
2764 goto out;
2765 }
2766
2767 if (req->r_reply->con)
2768 dout("%s revoking msg %p from old con %p\n", __func__,
2769 req->r_reply, req->r_reply->con);
2770 ceph_msg_revoke_incoming(req->r_reply);
2771
2772 if (front_len > req->r_reply->front_alloc_len) {
2773 pr_warning("get_reply front %d > preallocated %d (%u#%llu)\n",
2774 front_len, req->r_reply->front_alloc_len,
2775 (unsigned int)con->peer_name.type,
2776 le64_to_cpu(con->peer_name.num));
2777 m = ceph_msg_new(CEPH_MSG_OSD_OPREPLY, front_len, GFP_NOFS,
2778 false);
2779 if (!m)
2780 goto out;
2781 ceph_msg_put(req->r_reply);
2782 req->r_reply = m;
2783 }
2784 m = ceph_msg_get(req->r_reply);
2785
2786 if (data_len > 0) {
2787 struct ceph_osd_data *osd_data;
2788
2789 /*
2790 * XXX This is assuming there is only one op containing
2791 * XXX page data. Probably OK for reads, but this
2792 * XXX ought to be done more generally.
2793 */
2794 osd_data = osd_req_op_extent_osd_data(req, 0);
2795 if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGES) {
2796 if (osd_data->pages &&
2797 unlikely(osd_data->length < data_len)) {
2798
2799 pr_warning("tid %lld reply has %d bytes "
2800 "we had only %llu bytes ready\n",
2801 tid, data_len, osd_data->length);
2802 *skip = 1;
2803 ceph_msg_put(m);
2804 m = NULL;
2805 goto out;
2806 }
2807 }
2808 }
2809 *skip = 0;
2810 dout("get_reply tid %lld %p\n", tid, m);
2811
2812 out:
2813 mutex_unlock(&osdc->request_mutex);
2814 return m;
2815
2816 }
2817
2818 static struct ceph_msg *alloc_msg(struct ceph_connection *con,
2819 struct ceph_msg_header *hdr,
2820 int *skip)
2821 {
2822 struct ceph_osd *osd = con->private;
2823 int type = le16_to_cpu(hdr->type);
2824 int front = le32_to_cpu(hdr->front_len);
2825
2826 *skip = 0;
2827 switch (type) {
2828 case CEPH_MSG_OSD_MAP:
2829 case CEPH_MSG_WATCH_NOTIFY:
2830 return ceph_msg_new(type, front, GFP_NOFS, false);
2831 case CEPH_MSG_OSD_OPREPLY:
2832 return get_reply(con, hdr, skip);
2833 default:
2834 pr_info("alloc_msg unexpected msg type %d from osd%d\n", type,
2835 osd->o_osd);
2836 *skip = 1;
2837 return NULL;
2838 }
2839 }
2840
2841 /*
2842 * Wrappers to refcount containing ceph_osd struct
2843 */
2844 static struct ceph_connection *get_osd_con(struct ceph_connection *con)
2845 {
2846 struct ceph_osd *osd = con->private;
2847 if (get_osd(osd))
2848 return con;
2849 return NULL;
2850 }
2851
2852 static void put_osd_con(struct ceph_connection *con)
2853 {
2854 struct ceph_osd *osd = con->private;
2855 put_osd(osd);
2856 }
2857
2858 /*
2859 * authentication
2860 */
2861 /*
2862 * Note: returned pointer is the address of a structure that's
2863 * managed separately. Caller must *not* attempt to free it.
2864 */
2865 static struct ceph_auth_handshake *get_authorizer(struct ceph_connection *con,
2866 int *proto, int force_new)
2867 {
2868 struct ceph_osd *o = con->private;
2869 struct ceph_osd_client *osdc = o->o_osdc;
2870 struct ceph_auth_client *ac = osdc->client->monc.auth;
2871 struct ceph_auth_handshake *auth = &o->o_auth;
2872
2873 if (force_new && auth->authorizer) {
2874 ceph_auth_destroy_authorizer(auth->authorizer);
2875 auth->authorizer = NULL;
2876 }
2877 if (!auth->authorizer) {
2878 int ret = ceph_auth_create_authorizer(ac, CEPH_ENTITY_TYPE_OSD,
2879 auth);
2880 if (ret)
2881 return ERR_PTR(ret);
2882 } else {
2883 int ret = ceph_auth_update_authorizer(ac, CEPH_ENTITY_TYPE_OSD,
2884 auth);
2885 if (ret)
2886 return ERR_PTR(ret);
2887 }
2888 *proto = ac->protocol;
2889
2890 return auth;
2891 }
2892
2893
2894 static int verify_authorizer_reply(struct ceph_connection *con, int len)
2895 {
2896 struct ceph_osd *o = con->private;
2897 struct ceph_osd_client *osdc = o->o_osdc;
2898 struct ceph_auth_client *ac = osdc->client->monc.auth;
2899
2900 return ceph_auth_verify_authorizer_reply(ac, o->o_auth.authorizer, len);
2901 }
2902
2903 static int invalidate_authorizer(struct ceph_connection *con)
2904 {
2905 struct ceph_osd *o = con->private;
2906 struct ceph_osd_client *osdc = o->o_osdc;
2907 struct ceph_auth_client *ac = osdc->client->monc.auth;
2908
2909 ceph_auth_invalidate_authorizer(ac, CEPH_ENTITY_TYPE_OSD);
2910 return ceph_monc_validate_auth(&osdc->client->monc);
2911 }
2912
2913 static const struct ceph_connection_operations osd_con_ops = {
2914 .get = get_osd_con,
2915 .put = put_osd_con,
2916 .dispatch = dispatch,
2917 .get_authorizer = get_authorizer,
2918 .verify_authorizer_reply = verify_authorizer_reply,
2919 .invalidate_authorizer = invalidate_authorizer,
2920 .alloc_msg = alloc_msg,
2921 .fault = osd_reset,
2922 };
Linux with added FPC-III support