xref: /openbmc/linux/net/ceph/osd_client.c (revision 7587eb18)
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_OPREPLY_FRONT_LEN	512
23 
24 static struct kmem_cache	*ceph_osd_request_cache;
25 
26 static const struct ceph_connection_operations osd_con_ops;
27 
28 /*
29  * Implement client access to distributed object storage cluster.
30  *
31  * All data objects are stored within a cluster/cloud of OSDs, or
32  * "object storage devices."  (Note that Ceph OSDs have _nothing_ to
33  * do with the T10 OSD extensions to SCSI.)  Ceph OSDs are simply
34  * remote daemons serving up and coordinating consistent and safe
35  * access to storage.
36  *
37  * Cluster membership and the mapping of data objects onto storage devices
38  * are described by the osd map.
39  *
40  * We keep track of pending OSD requests (read, write), resubmit
41  * requests to different OSDs when the cluster topology/data layout
42  * change, or retry the affected requests when the communications
43  * channel with an OSD is reset.
44  */
45 
46 static void link_request(struct ceph_osd *osd, struct ceph_osd_request *req);
47 static void unlink_request(struct ceph_osd *osd, struct ceph_osd_request *req);
48 static void link_linger(struct ceph_osd *osd,
49 			struct ceph_osd_linger_request *lreq);
50 static void unlink_linger(struct ceph_osd *osd,
51 			  struct ceph_osd_linger_request *lreq);
52 
53 #if 1
54 static inline bool rwsem_is_wrlocked(struct rw_semaphore *sem)
55 {
56 	bool wrlocked = true;
57 
58 	if (unlikely(down_read_trylock(sem))) {
59 		wrlocked = false;
60 		up_read(sem);
61 	}
62 
63 	return wrlocked;
64 }
65 static inline void verify_osdc_locked(struct ceph_osd_client *osdc)
66 {
67 	WARN_ON(!rwsem_is_locked(&osdc->lock));
68 }
69 static inline void verify_osdc_wrlocked(struct ceph_osd_client *osdc)
70 {
71 	WARN_ON(!rwsem_is_wrlocked(&osdc->lock));
72 }
73 static inline void verify_osd_locked(struct ceph_osd *osd)
74 {
75 	struct ceph_osd_client *osdc = osd->o_osdc;
76 
77 	WARN_ON(!(mutex_is_locked(&osd->lock) &&
78 		  rwsem_is_locked(&osdc->lock)) &&
79 		!rwsem_is_wrlocked(&osdc->lock));
80 }
81 static inline void verify_lreq_locked(struct ceph_osd_linger_request *lreq)
82 {
83 	WARN_ON(!mutex_is_locked(&lreq->lock));
84 }
85 #else
86 static inline void verify_osdc_locked(struct ceph_osd_client *osdc) { }
87 static inline void verify_osdc_wrlocked(struct ceph_osd_client *osdc) { }
88 static inline void verify_osd_locked(struct ceph_osd *osd) { }
89 static inline void verify_lreq_locked(struct ceph_osd_linger_request *lreq) { }
90 #endif
91 
92 /*
93  * calculate the mapping of a file extent onto an object, and fill out the
94  * request accordingly.  shorten extent as necessary if it crosses an
95  * object boundary.
96  *
97  * fill osd op in request message.
98  */
99 static int calc_layout(struct ceph_file_layout *layout, u64 off, u64 *plen,
100 			u64 *objnum, u64 *objoff, u64 *objlen)
101 {
102 	u64 orig_len = *plen;
103 	int r;
104 
105 	/* object extent? */
106 	r = ceph_calc_file_object_mapping(layout, off, orig_len, objnum,
107 					  objoff, objlen);
108 	if (r < 0)
109 		return r;
110 	if (*objlen < orig_len) {
111 		*plen = *objlen;
112 		dout(" skipping last %llu, final file extent %llu~%llu\n",
113 		     orig_len - *plen, off, *plen);
114 	}
115 
116 	dout("calc_layout objnum=%llx %llu~%llu\n", *objnum, *objoff, *objlen);
117 
118 	return 0;
119 }
120 
121 static void ceph_osd_data_init(struct ceph_osd_data *osd_data)
122 {
123 	memset(osd_data, 0, sizeof (*osd_data));
124 	osd_data->type = CEPH_OSD_DATA_TYPE_NONE;
125 }
126 
127 static void ceph_osd_data_pages_init(struct ceph_osd_data *osd_data,
128 			struct page **pages, u64 length, u32 alignment,
129 			bool pages_from_pool, bool own_pages)
130 {
131 	osd_data->type = CEPH_OSD_DATA_TYPE_PAGES;
132 	osd_data->pages = pages;
133 	osd_data->length = length;
134 	osd_data->alignment = alignment;
135 	osd_data->pages_from_pool = pages_from_pool;
136 	osd_data->own_pages = own_pages;
137 }
138 
139 static void ceph_osd_data_pagelist_init(struct ceph_osd_data *osd_data,
140 			struct ceph_pagelist *pagelist)
141 {
142 	osd_data->type = CEPH_OSD_DATA_TYPE_PAGELIST;
143 	osd_data->pagelist = pagelist;
144 }
145 
146 #ifdef CONFIG_BLOCK
147 static void ceph_osd_data_bio_init(struct ceph_osd_data *osd_data,
148 			struct bio *bio, size_t bio_length)
149 {
150 	osd_data->type = CEPH_OSD_DATA_TYPE_BIO;
151 	osd_data->bio = bio;
152 	osd_data->bio_length = bio_length;
153 }
154 #endif /* CONFIG_BLOCK */
155 
156 #define osd_req_op_data(oreq, whch, typ, fld)				\
157 ({									\
158 	struct ceph_osd_request *__oreq = (oreq);			\
159 	unsigned int __whch = (whch);					\
160 	BUG_ON(__whch >= __oreq->r_num_ops);				\
161 	&__oreq->r_ops[__whch].typ.fld;					\
162 })
163 
164 static struct ceph_osd_data *
165 osd_req_op_raw_data_in(struct ceph_osd_request *osd_req, unsigned int which)
166 {
167 	BUG_ON(which >= osd_req->r_num_ops);
168 
169 	return &osd_req->r_ops[which].raw_data_in;
170 }
171 
172 struct ceph_osd_data *
173 osd_req_op_extent_osd_data(struct ceph_osd_request *osd_req,
174 			unsigned int which)
175 {
176 	return osd_req_op_data(osd_req, which, extent, osd_data);
177 }
178 EXPORT_SYMBOL(osd_req_op_extent_osd_data);
179 
180 void osd_req_op_raw_data_in_pages(struct ceph_osd_request *osd_req,
181 			unsigned int which, struct page **pages,
182 			u64 length, u32 alignment,
183 			bool pages_from_pool, bool own_pages)
184 {
185 	struct ceph_osd_data *osd_data;
186 
187 	osd_data = osd_req_op_raw_data_in(osd_req, which);
188 	ceph_osd_data_pages_init(osd_data, pages, length, alignment,
189 				pages_from_pool, own_pages);
190 }
191 EXPORT_SYMBOL(osd_req_op_raw_data_in_pages);
192 
193 void osd_req_op_extent_osd_data_pages(struct ceph_osd_request *osd_req,
194 			unsigned int which, struct page **pages,
195 			u64 length, u32 alignment,
196 			bool pages_from_pool, bool own_pages)
197 {
198 	struct ceph_osd_data *osd_data;
199 
200 	osd_data = osd_req_op_data(osd_req, which, extent, osd_data);
201 	ceph_osd_data_pages_init(osd_data, pages, length, alignment,
202 				pages_from_pool, own_pages);
203 }
204 EXPORT_SYMBOL(osd_req_op_extent_osd_data_pages);
205 
206 void osd_req_op_extent_osd_data_pagelist(struct ceph_osd_request *osd_req,
207 			unsigned int which, struct ceph_pagelist *pagelist)
208 {
209 	struct ceph_osd_data *osd_data;
210 
211 	osd_data = osd_req_op_data(osd_req, which, extent, osd_data);
212 	ceph_osd_data_pagelist_init(osd_data, pagelist);
213 }
214 EXPORT_SYMBOL(osd_req_op_extent_osd_data_pagelist);
215 
216 #ifdef CONFIG_BLOCK
217 void osd_req_op_extent_osd_data_bio(struct ceph_osd_request *osd_req,
218 			unsigned int which, struct bio *bio, size_t bio_length)
219 {
220 	struct ceph_osd_data *osd_data;
221 
222 	osd_data = osd_req_op_data(osd_req, which, extent, osd_data);
223 	ceph_osd_data_bio_init(osd_data, bio, bio_length);
224 }
225 EXPORT_SYMBOL(osd_req_op_extent_osd_data_bio);
226 #endif /* CONFIG_BLOCK */
227 
228 static void osd_req_op_cls_request_info_pagelist(
229 			struct ceph_osd_request *osd_req,
230 			unsigned int which, struct ceph_pagelist *pagelist)
231 {
232 	struct ceph_osd_data *osd_data;
233 
234 	osd_data = osd_req_op_data(osd_req, which, cls, request_info);
235 	ceph_osd_data_pagelist_init(osd_data, pagelist);
236 }
237 
238 void osd_req_op_cls_request_data_pagelist(
239 			struct ceph_osd_request *osd_req,
240 			unsigned int which, struct ceph_pagelist *pagelist)
241 {
242 	struct ceph_osd_data *osd_data;
243 
244 	osd_data = osd_req_op_data(osd_req, which, cls, request_data);
245 	ceph_osd_data_pagelist_init(osd_data, pagelist);
246 	osd_req->r_ops[which].cls.indata_len += pagelist->length;
247 	osd_req->r_ops[which].indata_len += pagelist->length;
248 }
249 EXPORT_SYMBOL(osd_req_op_cls_request_data_pagelist);
250 
251 void osd_req_op_cls_request_data_pages(struct ceph_osd_request *osd_req,
252 			unsigned int which, struct page **pages, u64 length,
253 			u32 alignment, bool pages_from_pool, bool own_pages)
254 {
255 	struct ceph_osd_data *osd_data;
256 
257 	osd_data = osd_req_op_data(osd_req, which, cls, request_data);
258 	ceph_osd_data_pages_init(osd_data, pages, length, alignment,
259 				pages_from_pool, own_pages);
260 	osd_req->r_ops[which].cls.indata_len += length;
261 	osd_req->r_ops[which].indata_len += length;
262 }
263 EXPORT_SYMBOL(osd_req_op_cls_request_data_pages);
264 
265 void osd_req_op_cls_response_data_pages(struct ceph_osd_request *osd_req,
266 			unsigned int which, struct page **pages, u64 length,
267 			u32 alignment, bool pages_from_pool, bool own_pages)
268 {
269 	struct ceph_osd_data *osd_data;
270 
271 	osd_data = osd_req_op_data(osd_req, which, cls, response_data);
272 	ceph_osd_data_pages_init(osd_data, pages, length, alignment,
273 				pages_from_pool, own_pages);
274 }
275 EXPORT_SYMBOL(osd_req_op_cls_response_data_pages);
276 
277 static u64 ceph_osd_data_length(struct ceph_osd_data *osd_data)
278 {
279 	switch (osd_data->type) {
280 	case CEPH_OSD_DATA_TYPE_NONE:
281 		return 0;
282 	case CEPH_OSD_DATA_TYPE_PAGES:
283 		return osd_data->length;
284 	case CEPH_OSD_DATA_TYPE_PAGELIST:
285 		return (u64)osd_data->pagelist->length;
286 #ifdef CONFIG_BLOCK
287 	case CEPH_OSD_DATA_TYPE_BIO:
288 		return (u64)osd_data->bio_length;
289 #endif /* CONFIG_BLOCK */
290 	default:
291 		WARN(true, "unrecognized data type %d\n", (int)osd_data->type);
292 		return 0;
293 	}
294 }
295 
296 static void ceph_osd_data_release(struct ceph_osd_data *osd_data)
297 {
298 	if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGES && osd_data->own_pages) {
299 		int num_pages;
300 
301 		num_pages = calc_pages_for((u64)osd_data->alignment,
302 						(u64)osd_data->length);
303 		ceph_release_page_vector(osd_data->pages, num_pages);
304 	}
305 	ceph_osd_data_init(osd_data);
306 }
307 
308 static void osd_req_op_data_release(struct ceph_osd_request *osd_req,
309 			unsigned int which)
310 {
311 	struct ceph_osd_req_op *op;
312 
313 	BUG_ON(which >= osd_req->r_num_ops);
314 	op = &osd_req->r_ops[which];
315 
316 	switch (op->op) {
317 	case CEPH_OSD_OP_READ:
318 	case CEPH_OSD_OP_WRITE:
319 	case CEPH_OSD_OP_WRITEFULL:
320 		ceph_osd_data_release(&op->extent.osd_data);
321 		break;
322 	case CEPH_OSD_OP_CALL:
323 		ceph_osd_data_release(&op->cls.request_info);
324 		ceph_osd_data_release(&op->cls.request_data);
325 		ceph_osd_data_release(&op->cls.response_data);
326 		break;
327 	case CEPH_OSD_OP_SETXATTR:
328 	case CEPH_OSD_OP_CMPXATTR:
329 		ceph_osd_data_release(&op->xattr.osd_data);
330 		break;
331 	case CEPH_OSD_OP_STAT:
332 		ceph_osd_data_release(&op->raw_data_in);
333 		break;
334 	case CEPH_OSD_OP_NOTIFY_ACK:
335 		ceph_osd_data_release(&op->notify_ack.request_data);
336 		break;
337 	case CEPH_OSD_OP_NOTIFY:
338 		ceph_osd_data_release(&op->notify.request_data);
339 		ceph_osd_data_release(&op->notify.response_data);
340 		break;
341 	default:
342 		break;
343 	}
344 }
345 
346 /*
347  * Assumes @t is zero-initialized.
348  */
349 static void target_init(struct ceph_osd_request_target *t)
350 {
351 	ceph_oid_init(&t->base_oid);
352 	ceph_oloc_init(&t->base_oloc);
353 	ceph_oid_init(&t->target_oid);
354 	ceph_oloc_init(&t->target_oloc);
355 
356 	ceph_osds_init(&t->acting);
357 	ceph_osds_init(&t->up);
358 	t->size = -1;
359 	t->min_size = -1;
360 
361 	t->osd = CEPH_HOMELESS_OSD;
362 }
363 
364 static void target_copy(struct ceph_osd_request_target *dest,
365 			const struct ceph_osd_request_target *src)
366 {
367 	ceph_oid_copy(&dest->base_oid, &src->base_oid);
368 	ceph_oloc_copy(&dest->base_oloc, &src->base_oloc);
369 	ceph_oid_copy(&dest->target_oid, &src->target_oid);
370 	ceph_oloc_copy(&dest->target_oloc, &src->target_oloc);
371 
372 	dest->pgid = src->pgid; /* struct */
373 	dest->pg_num = src->pg_num;
374 	dest->pg_num_mask = src->pg_num_mask;
375 	ceph_osds_copy(&dest->acting, &src->acting);
376 	ceph_osds_copy(&dest->up, &src->up);
377 	dest->size = src->size;
378 	dest->min_size = src->min_size;
379 	dest->sort_bitwise = src->sort_bitwise;
380 
381 	dest->flags = src->flags;
382 	dest->paused = src->paused;
383 
384 	dest->osd = src->osd;
385 }
386 
387 static void target_destroy(struct ceph_osd_request_target *t)
388 {
389 	ceph_oid_destroy(&t->base_oid);
390 	ceph_oid_destroy(&t->target_oid);
391 }
392 
393 /*
394  * requests
395  */
396 static void request_release_checks(struct ceph_osd_request *req)
397 {
398 	WARN_ON(!RB_EMPTY_NODE(&req->r_node));
399 	WARN_ON(!RB_EMPTY_NODE(&req->r_mc_node));
400 	WARN_ON(!list_empty(&req->r_unsafe_item));
401 	WARN_ON(req->r_osd);
402 }
403 
404 static void ceph_osdc_release_request(struct kref *kref)
405 {
406 	struct ceph_osd_request *req = container_of(kref,
407 					    struct ceph_osd_request, r_kref);
408 	unsigned int which;
409 
410 	dout("%s %p (r_request %p r_reply %p)\n", __func__, req,
411 	     req->r_request, req->r_reply);
412 	request_release_checks(req);
413 
414 	if (req->r_request)
415 		ceph_msg_put(req->r_request);
416 	if (req->r_reply)
417 		ceph_msg_put(req->r_reply);
418 
419 	for (which = 0; which < req->r_num_ops; which++)
420 		osd_req_op_data_release(req, which);
421 
422 	target_destroy(&req->r_t);
423 	ceph_put_snap_context(req->r_snapc);
424 
425 	if (req->r_mempool)
426 		mempool_free(req, req->r_osdc->req_mempool);
427 	else if (req->r_num_ops <= CEPH_OSD_SLAB_OPS)
428 		kmem_cache_free(ceph_osd_request_cache, req);
429 	else
430 		kfree(req);
431 }
432 
433 void ceph_osdc_get_request(struct ceph_osd_request *req)
434 {
435 	dout("%s %p (was %d)\n", __func__, req,
436 	     atomic_read(&req->r_kref.refcount));
437 	kref_get(&req->r_kref);
438 }
439 EXPORT_SYMBOL(ceph_osdc_get_request);
440 
441 void ceph_osdc_put_request(struct ceph_osd_request *req)
442 {
443 	if (req) {
444 		dout("%s %p (was %d)\n", __func__, req,
445 		     atomic_read(&req->r_kref.refcount));
446 		kref_put(&req->r_kref, ceph_osdc_release_request);
447 	}
448 }
449 EXPORT_SYMBOL(ceph_osdc_put_request);
450 
451 static void request_init(struct ceph_osd_request *req)
452 {
453 	/* req only, each op is zeroed in _osd_req_op_init() */
454 	memset(req, 0, sizeof(*req));
455 
456 	kref_init(&req->r_kref);
457 	init_completion(&req->r_completion);
458 	init_completion(&req->r_safe_completion);
459 	RB_CLEAR_NODE(&req->r_node);
460 	RB_CLEAR_NODE(&req->r_mc_node);
461 	INIT_LIST_HEAD(&req->r_unsafe_item);
462 
463 	target_init(&req->r_t);
464 }
465 
466 /*
467  * This is ugly, but it allows us to reuse linger registration and ping
468  * requests, keeping the structure of the code around send_linger{_ping}()
469  * reasonable.  Setting up a min_nr=2 mempool for each linger request
470  * and dealing with copying ops (this blasts req only, watch op remains
471  * intact) isn't any better.
472  */
473 static void request_reinit(struct ceph_osd_request *req)
474 {
475 	struct ceph_osd_client *osdc = req->r_osdc;
476 	bool mempool = req->r_mempool;
477 	unsigned int num_ops = req->r_num_ops;
478 	u64 snapid = req->r_snapid;
479 	struct ceph_snap_context *snapc = req->r_snapc;
480 	bool linger = req->r_linger;
481 	struct ceph_msg *request_msg = req->r_request;
482 	struct ceph_msg *reply_msg = req->r_reply;
483 
484 	dout("%s req %p\n", __func__, req);
485 	WARN_ON(atomic_read(&req->r_kref.refcount) != 1);
486 	request_release_checks(req);
487 
488 	WARN_ON(atomic_read(&request_msg->kref.refcount) != 1);
489 	WARN_ON(atomic_read(&reply_msg->kref.refcount) != 1);
490 	target_destroy(&req->r_t);
491 
492 	request_init(req);
493 	req->r_osdc = osdc;
494 	req->r_mempool = mempool;
495 	req->r_num_ops = num_ops;
496 	req->r_snapid = snapid;
497 	req->r_snapc = snapc;
498 	req->r_linger = linger;
499 	req->r_request = request_msg;
500 	req->r_reply = reply_msg;
501 }
502 
503 struct ceph_osd_request *ceph_osdc_alloc_request(struct ceph_osd_client *osdc,
504 					       struct ceph_snap_context *snapc,
505 					       unsigned int num_ops,
506 					       bool use_mempool,
507 					       gfp_t gfp_flags)
508 {
509 	struct ceph_osd_request *req;
510 
511 	if (use_mempool) {
512 		BUG_ON(num_ops > CEPH_OSD_SLAB_OPS);
513 		req = mempool_alloc(osdc->req_mempool, gfp_flags);
514 	} else if (num_ops <= CEPH_OSD_SLAB_OPS) {
515 		req = kmem_cache_alloc(ceph_osd_request_cache, gfp_flags);
516 	} else {
517 		BUG_ON(num_ops > CEPH_OSD_MAX_OPS);
518 		req = kmalloc(sizeof(*req) + num_ops * sizeof(req->r_ops[0]),
519 			      gfp_flags);
520 	}
521 	if (unlikely(!req))
522 		return NULL;
523 
524 	request_init(req);
525 	req->r_osdc = osdc;
526 	req->r_mempool = use_mempool;
527 	req->r_num_ops = num_ops;
528 	req->r_snapid = CEPH_NOSNAP;
529 	req->r_snapc = ceph_get_snap_context(snapc);
530 
531 	dout("%s req %p\n", __func__, req);
532 	return req;
533 }
534 EXPORT_SYMBOL(ceph_osdc_alloc_request);
535 
536 int ceph_osdc_alloc_messages(struct ceph_osd_request *req, gfp_t gfp)
537 {
538 	struct ceph_osd_client *osdc = req->r_osdc;
539 	struct ceph_msg *msg;
540 	int msg_size;
541 
542 	WARN_ON(ceph_oid_empty(&req->r_base_oid));
543 
544 	/* create request message */
545 	msg_size = 4 + 4 + 4; /* client_inc, osdmap_epoch, flags */
546 	msg_size += 4 + 4 + 4 + 8; /* mtime, reassert_version */
547 	msg_size += 2 + 4 + 8 + 4 + 4; /* oloc */
548 	msg_size += 1 + 8 + 4 + 4; /* pgid */
549 	msg_size += 4 + req->r_base_oid.name_len; /* oid */
550 	msg_size += 2 + req->r_num_ops * sizeof(struct ceph_osd_op);
551 	msg_size += 8; /* snapid */
552 	msg_size += 8; /* snap_seq */
553 	msg_size += 4 + 8 * (req->r_snapc ? req->r_snapc->num_snaps : 0);
554 	msg_size += 4; /* retry_attempt */
555 
556 	if (req->r_mempool)
557 		msg = ceph_msgpool_get(&osdc->msgpool_op, 0);
558 	else
559 		msg = ceph_msg_new(CEPH_MSG_OSD_OP, msg_size, gfp, true);
560 	if (!msg)
561 		return -ENOMEM;
562 
563 	memset(msg->front.iov_base, 0, msg->front.iov_len);
564 	req->r_request = msg;
565 
566 	/* create reply message */
567 	msg_size = OSD_OPREPLY_FRONT_LEN;
568 	msg_size += req->r_base_oid.name_len;
569 	msg_size += req->r_num_ops * sizeof(struct ceph_osd_op);
570 
571 	if (req->r_mempool)
572 		msg = ceph_msgpool_get(&osdc->msgpool_op_reply, 0);
573 	else
574 		msg = ceph_msg_new(CEPH_MSG_OSD_OPREPLY, msg_size, gfp, true);
575 	if (!msg)
576 		return -ENOMEM;
577 
578 	req->r_reply = msg;
579 
580 	return 0;
581 }
582 EXPORT_SYMBOL(ceph_osdc_alloc_messages);
583 
584 static bool osd_req_opcode_valid(u16 opcode)
585 {
586 	switch (opcode) {
587 #define GENERATE_CASE(op, opcode, str)	case CEPH_OSD_OP_##op: return true;
588 __CEPH_FORALL_OSD_OPS(GENERATE_CASE)
589 #undef GENERATE_CASE
590 	default:
591 		return false;
592 	}
593 }
594 
595 /*
596  * This is an osd op init function for opcodes that have no data or
597  * other information associated with them.  It also serves as a
598  * common init routine for all the other init functions, below.
599  */
600 static struct ceph_osd_req_op *
601 _osd_req_op_init(struct ceph_osd_request *osd_req, unsigned int which,
602 		 u16 opcode, u32 flags)
603 {
604 	struct ceph_osd_req_op *op;
605 
606 	BUG_ON(which >= osd_req->r_num_ops);
607 	BUG_ON(!osd_req_opcode_valid(opcode));
608 
609 	op = &osd_req->r_ops[which];
610 	memset(op, 0, sizeof (*op));
611 	op->op = opcode;
612 	op->flags = flags;
613 
614 	return op;
615 }
616 
617 void osd_req_op_init(struct ceph_osd_request *osd_req,
618 		     unsigned int which, u16 opcode, u32 flags)
619 {
620 	(void)_osd_req_op_init(osd_req, which, opcode, flags);
621 }
622 EXPORT_SYMBOL(osd_req_op_init);
623 
624 void osd_req_op_extent_init(struct ceph_osd_request *osd_req,
625 				unsigned int which, u16 opcode,
626 				u64 offset, u64 length,
627 				u64 truncate_size, u32 truncate_seq)
628 {
629 	struct ceph_osd_req_op *op = _osd_req_op_init(osd_req, which,
630 						      opcode, 0);
631 	size_t payload_len = 0;
632 
633 	BUG_ON(opcode != CEPH_OSD_OP_READ && opcode != CEPH_OSD_OP_WRITE &&
634 	       opcode != CEPH_OSD_OP_WRITEFULL && opcode != CEPH_OSD_OP_ZERO &&
635 	       opcode != CEPH_OSD_OP_TRUNCATE);
636 
637 	op->extent.offset = offset;
638 	op->extent.length = length;
639 	op->extent.truncate_size = truncate_size;
640 	op->extent.truncate_seq = truncate_seq;
641 	if (opcode == CEPH_OSD_OP_WRITE || opcode == CEPH_OSD_OP_WRITEFULL)
642 		payload_len += length;
643 
644 	op->indata_len = payload_len;
645 }
646 EXPORT_SYMBOL(osd_req_op_extent_init);
647 
648 void osd_req_op_extent_update(struct ceph_osd_request *osd_req,
649 				unsigned int which, u64 length)
650 {
651 	struct ceph_osd_req_op *op;
652 	u64 previous;
653 
654 	BUG_ON(which >= osd_req->r_num_ops);
655 	op = &osd_req->r_ops[which];
656 	previous = op->extent.length;
657 
658 	if (length == previous)
659 		return;		/* Nothing to do */
660 	BUG_ON(length > previous);
661 
662 	op->extent.length = length;
663 	op->indata_len -= previous - length;
664 }
665 EXPORT_SYMBOL(osd_req_op_extent_update);
666 
667 void osd_req_op_extent_dup_last(struct ceph_osd_request *osd_req,
668 				unsigned int which, u64 offset_inc)
669 {
670 	struct ceph_osd_req_op *op, *prev_op;
671 
672 	BUG_ON(which + 1 >= osd_req->r_num_ops);
673 
674 	prev_op = &osd_req->r_ops[which];
675 	op = _osd_req_op_init(osd_req, which + 1, prev_op->op, prev_op->flags);
676 	/* dup previous one */
677 	op->indata_len = prev_op->indata_len;
678 	op->outdata_len = prev_op->outdata_len;
679 	op->extent = prev_op->extent;
680 	/* adjust offset */
681 	op->extent.offset += offset_inc;
682 	op->extent.length -= offset_inc;
683 
684 	if (op->op == CEPH_OSD_OP_WRITE || op->op == CEPH_OSD_OP_WRITEFULL)
685 		op->indata_len -= offset_inc;
686 }
687 EXPORT_SYMBOL(osd_req_op_extent_dup_last);
688 
689 void osd_req_op_cls_init(struct ceph_osd_request *osd_req, unsigned int which,
690 			u16 opcode, const char *class, const char *method)
691 {
692 	struct ceph_osd_req_op *op = _osd_req_op_init(osd_req, which,
693 						      opcode, 0);
694 	struct ceph_pagelist *pagelist;
695 	size_t payload_len = 0;
696 	size_t size;
697 
698 	BUG_ON(opcode != CEPH_OSD_OP_CALL);
699 
700 	pagelist = kmalloc(sizeof (*pagelist), GFP_NOFS);
701 	BUG_ON(!pagelist);
702 	ceph_pagelist_init(pagelist);
703 
704 	op->cls.class_name = class;
705 	size = strlen(class);
706 	BUG_ON(size > (size_t) U8_MAX);
707 	op->cls.class_len = size;
708 	ceph_pagelist_append(pagelist, class, size);
709 	payload_len += size;
710 
711 	op->cls.method_name = method;
712 	size = strlen(method);
713 	BUG_ON(size > (size_t) U8_MAX);
714 	op->cls.method_len = size;
715 	ceph_pagelist_append(pagelist, method, size);
716 	payload_len += size;
717 
718 	osd_req_op_cls_request_info_pagelist(osd_req, which, pagelist);
719 
720 	op->indata_len = payload_len;
721 }
722 EXPORT_SYMBOL(osd_req_op_cls_init);
723 
724 int osd_req_op_xattr_init(struct ceph_osd_request *osd_req, unsigned int which,
725 			  u16 opcode, const char *name, const void *value,
726 			  size_t size, u8 cmp_op, u8 cmp_mode)
727 {
728 	struct ceph_osd_req_op *op = _osd_req_op_init(osd_req, which,
729 						      opcode, 0);
730 	struct ceph_pagelist *pagelist;
731 	size_t payload_len;
732 
733 	BUG_ON(opcode != CEPH_OSD_OP_SETXATTR && opcode != CEPH_OSD_OP_CMPXATTR);
734 
735 	pagelist = kmalloc(sizeof(*pagelist), GFP_NOFS);
736 	if (!pagelist)
737 		return -ENOMEM;
738 
739 	ceph_pagelist_init(pagelist);
740 
741 	payload_len = strlen(name);
742 	op->xattr.name_len = payload_len;
743 	ceph_pagelist_append(pagelist, name, payload_len);
744 
745 	op->xattr.value_len = size;
746 	ceph_pagelist_append(pagelist, value, size);
747 	payload_len += size;
748 
749 	op->xattr.cmp_op = cmp_op;
750 	op->xattr.cmp_mode = cmp_mode;
751 
752 	ceph_osd_data_pagelist_init(&op->xattr.osd_data, pagelist);
753 	op->indata_len = payload_len;
754 	return 0;
755 }
756 EXPORT_SYMBOL(osd_req_op_xattr_init);
757 
758 /*
759  * @watch_opcode: CEPH_OSD_WATCH_OP_*
760  */
761 static void osd_req_op_watch_init(struct ceph_osd_request *req, int which,
762 				  u64 cookie, u8 watch_opcode)
763 {
764 	struct ceph_osd_req_op *op;
765 
766 	op = _osd_req_op_init(req, which, CEPH_OSD_OP_WATCH, 0);
767 	op->watch.cookie = cookie;
768 	op->watch.op = watch_opcode;
769 	op->watch.gen = 0;
770 }
771 
772 void osd_req_op_alloc_hint_init(struct ceph_osd_request *osd_req,
773 				unsigned int which,
774 				u64 expected_object_size,
775 				u64 expected_write_size)
776 {
777 	struct ceph_osd_req_op *op = _osd_req_op_init(osd_req, which,
778 						      CEPH_OSD_OP_SETALLOCHINT,
779 						      0);
780 
781 	op->alloc_hint.expected_object_size = expected_object_size;
782 	op->alloc_hint.expected_write_size = expected_write_size;
783 
784 	/*
785 	 * CEPH_OSD_OP_SETALLOCHINT op is advisory and therefore deemed
786 	 * not worth a feature bit.  Set FAILOK per-op flag to make
787 	 * sure older osds don't trip over an unsupported opcode.
788 	 */
789 	op->flags |= CEPH_OSD_OP_FLAG_FAILOK;
790 }
791 EXPORT_SYMBOL(osd_req_op_alloc_hint_init);
792 
793 static void ceph_osdc_msg_data_add(struct ceph_msg *msg,
794 				struct ceph_osd_data *osd_data)
795 {
796 	u64 length = ceph_osd_data_length(osd_data);
797 
798 	if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGES) {
799 		BUG_ON(length > (u64) SIZE_MAX);
800 		if (length)
801 			ceph_msg_data_add_pages(msg, osd_data->pages,
802 					length, osd_data->alignment);
803 	} else if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGELIST) {
804 		BUG_ON(!length);
805 		ceph_msg_data_add_pagelist(msg, osd_data->pagelist);
806 #ifdef CONFIG_BLOCK
807 	} else if (osd_data->type == CEPH_OSD_DATA_TYPE_BIO) {
808 		ceph_msg_data_add_bio(msg, osd_data->bio, length);
809 #endif
810 	} else {
811 		BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_NONE);
812 	}
813 }
814 
815 static u32 osd_req_encode_op(struct ceph_osd_op *dst,
816 			     const struct ceph_osd_req_op *src)
817 {
818 	if (WARN_ON(!osd_req_opcode_valid(src->op))) {
819 		pr_err("unrecognized osd opcode %d\n", src->op);
820 
821 		return 0;
822 	}
823 
824 	switch (src->op) {
825 	case CEPH_OSD_OP_STAT:
826 		break;
827 	case CEPH_OSD_OP_READ:
828 	case CEPH_OSD_OP_WRITE:
829 	case CEPH_OSD_OP_WRITEFULL:
830 	case CEPH_OSD_OP_ZERO:
831 	case CEPH_OSD_OP_TRUNCATE:
832 		dst->extent.offset = cpu_to_le64(src->extent.offset);
833 		dst->extent.length = cpu_to_le64(src->extent.length);
834 		dst->extent.truncate_size =
835 			cpu_to_le64(src->extent.truncate_size);
836 		dst->extent.truncate_seq =
837 			cpu_to_le32(src->extent.truncate_seq);
838 		break;
839 	case CEPH_OSD_OP_CALL:
840 		dst->cls.class_len = src->cls.class_len;
841 		dst->cls.method_len = src->cls.method_len;
842 		dst->cls.indata_len = cpu_to_le32(src->cls.indata_len);
843 		break;
844 	case CEPH_OSD_OP_STARTSYNC:
845 		break;
846 	case CEPH_OSD_OP_WATCH:
847 		dst->watch.cookie = cpu_to_le64(src->watch.cookie);
848 		dst->watch.ver = cpu_to_le64(0);
849 		dst->watch.op = src->watch.op;
850 		dst->watch.gen = cpu_to_le32(src->watch.gen);
851 		break;
852 	case CEPH_OSD_OP_NOTIFY_ACK:
853 		break;
854 	case CEPH_OSD_OP_NOTIFY:
855 		dst->notify.cookie = cpu_to_le64(src->notify.cookie);
856 		break;
857 	case CEPH_OSD_OP_SETALLOCHINT:
858 		dst->alloc_hint.expected_object_size =
859 		    cpu_to_le64(src->alloc_hint.expected_object_size);
860 		dst->alloc_hint.expected_write_size =
861 		    cpu_to_le64(src->alloc_hint.expected_write_size);
862 		break;
863 	case CEPH_OSD_OP_SETXATTR:
864 	case CEPH_OSD_OP_CMPXATTR:
865 		dst->xattr.name_len = cpu_to_le32(src->xattr.name_len);
866 		dst->xattr.value_len = cpu_to_le32(src->xattr.value_len);
867 		dst->xattr.cmp_op = src->xattr.cmp_op;
868 		dst->xattr.cmp_mode = src->xattr.cmp_mode;
869 		break;
870 	case CEPH_OSD_OP_CREATE:
871 	case CEPH_OSD_OP_DELETE:
872 		break;
873 	default:
874 		pr_err("unsupported osd opcode %s\n",
875 			ceph_osd_op_name(src->op));
876 		WARN_ON(1);
877 
878 		return 0;
879 	}
880 
881 	dst->op = cpu_to_le16(src->op);
882 	dst->flags = cpu_to_le32(src->flags);
883 	dst->payload_len = cpu_to_le32(src->indata_len);
884 
885 	return src->indata_len;
886 }
887 
888 /*
889  * build new request AND message, calculate layout, and adjust file
890  * extent as needed.
891  *
892  * if the file was recently truncated, we include information about its
893  * old and new size so that the object can be updated appropriately.  (we
894  * avoid synchronously deleting truncated objects because it's slow.)
895  *
896  * if @do_sync, include a 'startsync' command so that the osd will flush
897  * data quickly.
898  */
899 struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *osdc,
900 					       struct ceph_file_layout *layout,
901 					       struct ceph_vino vino,
902 					       u64 off, u64 *plen,
903 					       unsigned int which, int num_ops,
904 					       int opcode, int flags,
905 					       struct ceph_snap_context *snapc,
906 					       u32 truncate_seq,
907 					       u64 truncate_size,
908 					       bool use_mempool)
909 {
910 	struct ceph_osd_request *req;
911 	u64 objnum = 0;
912 	u64 objoff = 0;
913 	u64 objlen = 0;
914 	int r;
915 
916 	BUG_ON(opcode != CEPH_OSD_OP_READ && opcode != CEPH_OSD_OP_WRITE &&
917 	       opcode != CEPH_OSD_OP_ZERO && opcode != CEPH_OSD_OP_TRUNCATE &&
918 	       opcode != CEPH_OSD_OP_CREATE && opcode != CEPH_OSD_OP_DELETE);
919 
920 	req = ceph_osdc_alloc_request(osdc, snapc, num_ops, use_mempool,
921 					GFP_NOFS);
922 	if (!req) {
923 		r = -ENOMEM;
924 		goto fail;
925 	}
926 
927 	/* calculate max write size */
928 	r = calc_layout(layout, off, plen, &objnum, &objoff, &objlen);
929 	if (r)
930 		goto fail;
931 
932 	if (opcode == CEPH_OSD_OP_CREATE || opcode == CEPH_OSD_OP_DELETE) {
933 		osd_req_op_init(req, which, opcode, 0);
934 	} else {
935 		u32 object_size = le32_to_cpu(layout->fl_object_size);
936 		u32 object_base = off - objoff;
937 		if (!(truncate_seq == 1 && truncate_size == -1ULL)) {
938 			if (truncate_size <= object_base) {
939 				truncate_size = 0;
940 			} else {
941 				truncate_size -= object_base;
942 				if (truncate_size > object_size)
943 					truncate_size = object_size;
944 			}
945 		}
946 		osd_req_op_extent_init(req, which, opcode, objoff, objlen,
947 				       truncate_size, truncate_seq);
948 	}
949 
950 	req->r_flags = flags;
951 	req->r_base_oloc.pool = ceph_file_layout_pg_pool(*layout);
952 	ceph_oid_printf(&req->r_base_oid, "%llx.%08llx", vino.ino, objnum);
953 
954 	req->r_snapid = vino.snap;
955 	if (flags & CEPH_OSD_FLAG_WRITE)
956 		req->r_data_offset = off;
957 
958 	r = ceph_osdc_alloc_messages(req, GFP_NOFS);
959 	if (r)
960 		goto fail;
961 
962 	return req;
963 
964 fail:
965 	ceph_osdc_put_request(req);
966 	return ERR_PTR(r);
967 }
968 EXPORT_SYMBOL(ceph_osdc_new_request);
969 
970 /*
971  * We keep osd requests in an rbtree, sorted by ->r_tid.
972  */
973 DEFINE_RB_FUNCS(request, struct ceph_osd_request, r_tid, r_node)
974 DEFINE_RB_FUNCS(request_mc, struct ceph_osd_request, r_tid, r_mc_node)
975 
976 static bool osd_homeless(struct ceph_osd *osd)
977 {
978 	return osd->o_osd == CEPH_HOMELESS_OSD;
979 }
980 
981 static bool osd_registered(struct ceph_osd *osd)
982 {
983 	verify_osdc_locked(osd->o_osdc);
984 
985 	return !RB_EMPTY_NODE(&osd->o_node);
986 }
987 
988 /*
989  * Assumes @osd is zero-initialized.
990  */
991 static void osd_init(struct ceph_osd *osd)
992 {
993 	atomic_set(&osd->o_ref, 1);
994 	RB_CLEAR_NODE(&osd->o_node);
995 	osd->o_requests = RB_ROOT;
996 	osd->o_linger_requests = RB_ROOT;
997 	INIT_LIST_HEAD(&osd->o_osd_lru);
998 	INIT_LIST_HEAD(&osd->o_keepalive_item);
999 	osd->o_incarnation = 1;
1000 	mutex_init(&osd->lock);
1001 }
1002 
1003 static void osd_cleanup(struct ceph_osd *osd)
1004 {
1005 	WARN_ON(!RB_EMPTY_NODE(&osd->o_node));
1006 	WARN_ON(!RB_EMPTY_ROOT(&osd->o_requests));
1007 	WARN_ON(!RB_EMPTY_ROOT(&osd->o_linger_requests));
1008 	WARN_ON(!list_empty(&osd->o_osd_lru));
1009 	WARN_ON(!list_empty(&osd->o_keepalive_item));
1010 
1011 	if (osd->o_auth.authorizer) {
1012 		WARN_ON(osd_homeless(osd));
1013 		ceph_auth_destroy_authorizer(osd->o_auth.authorizer);
1014 	}
1015 }
1016 
1017 /*
1018  * Track open sessions with osds.
1019  */
1020 static struct ceph_osd *create_osd(struct ceph_osd_client *osdc, int onum)
1021 {
1022 	struct ceph_osd *osd;
1023 
1024 	WARN_ON(onum == CEPH_HOMELESS_OSD);
1025 
1026 	osd = kzalloc(sizeof(*osd), GFP_NOIO | __GFP_NOFAIL);
1027 	osd_init(osd);
1028 	osd->o_osdc = osdc;
1029 	osd->o_osd = onum;
1030 
1031 	ceph_con_init(&osd->o_con, osd, &osd_con_ops, &osdc->client->msgr);
1032 
1033 	return osd;
1034 }
1035 
1036 static struct ceph_osd *get_osd(struct ceph_osd *osd)
1037 {
1038 	if (atomic_inc_not_zero(&osd->o_ref)) {
1039 		dout("get_osd %p %d -> %d\n", osd, atomic_read(&osd->o_ref)-1,
1040 		     atomic_read(&osd->o_ref));
1041 		return osd;
1042 	} else {
1043 		dout("get_osd %p FAIL\n", osd);
1044 		return NULL;
1045 	}
1046 }
1047 
1048 static void put_osd(struct ceph_osd *osd)
1049 {
1050 	dout("put_osd %p %d -> %d\n", osd, atomic_read(&osd->o_ref),
1051 	     atomic_read(&osd->o_ref) - 1);
1052 	if (atomic_dec_and_test(&osd->o_ref)) {
1053 		osd_cleanup(osd);
1054 		kfree(osd);
1055 	}
1056 }
1057 
1058 DEFINE_RB_FUNCS(osd, struct ceph_osd, o_osd, o_node)
1059 
1060 static void __move_osd_to_lru(struct ceph_osd *osd)
1061 {
1062 	struct ceph_osd_client *osdc = osd->o_osdc;
1063 
1064 	dout("%s osd %p osd%d\n", __func__, osd, osd->o_osd);
1065 	BUG_ON(!list_empty(&osd->o_osd_lru));
1066 
1067 	spin_lock(&osdc->osd_lru_lock);
1068 	list_add_tail(&osd->o_osd_lru, &osdc->osd_lru);
1069 	spin_unlock(&osdc->osd_lru_lock);
1070 
1071 	osd->lru_ttl = jiffies + osdc->client->options->osd_idle_ttl;
1072 }
1073 
1074 static void maybe_move_osd_to_lru(struct ceph_osd *osd)
1075 {
1076 	if (RB_EMPTY_ROOT(&osd->o_requests) &&
1077 	    RB_EMPTY_ROOT(&osd->o_linger_requests))
1078 		__move_osd_to_lru(osd);
1079 }
1080 
1081 static void __remove_osd_from_lru(struct ceph_osd *osd)
1082 {
1083 	struct ceph_osd_client *osdc = osd->o_osdc;
1084 
1085 	dout("%s osd %p osd%d\n", __func__, osd, osd->o_osd);
1086 
1087 	spin_lock(&osdc->osd_lru_lock);
1088 	if (!list_empty(&osd->o_osd_lru))
1089 		list_del_init(&osd->o_osd_lru);
1090 	spin_unlock(&osdc->osd_lru_lock);
1091 }
1092 
1093 /*
1094  * Close the connection and assign any leftover requests to the
1095  * homeless session.
1096  */
1097 static void close_osd(struct ceph_osd *osd)
1098 {
1099 	struct ceph_osd_client *osdc = osd->o_osdc;
1100 	struct rb_node *n;
1101 
1102 	verify_osdc_wrlocked(osdc);
1103 	dout("%s osd %p osd%d\n", __func__, osd, osd->o_osd);
1104 
1105 	ceph_con_close(&osd->o_con);
1106 
1107 	for (n = rb_first(&osd->o_requests); n; ) {
1108 		struct ceph_osd_request *req =
1109 		    rb_entry(n, struct ceph_osd_request, r_node);
1110 
1111 		n = rb_next(n); /* unlink_request() */
1112 
1113 		dout(" reassigning req %p tid %llu\n", req, req->r_tid);
1114 		unlink_request(osd, req);
1115 		link_request(&osdc->homeless_osd, req);
1116 	}
1117 	for (n = rb_first(&osd->o_linger_requests); n; ) {
1118 		struct ceph_osd_linger_request *lreq =
1119 		    rb_entry(n, struct ceph_osd_linger_request, node);
1120 
1121 		n = rb_next(n); /* unlink_linger() */
1122 
1123 		dout(" reassigning lreq %p linger_id %llu\n", lreq,
1124 		     lreq->linger_id);
1125 		unlink_linger(osd, lreq);
1126 		link_linger(&osdc->homeless_osd, lreq);
1127 	}
1128 
1129 	__remove_osd_from_lru(osd);
1130 	erase_osd(&osdc->osds, osd);
1131 	put_osd(osd);
1132 }
1133 
1134 /*
1135  * reset osd connect
1136  */
1137 static int reopen_osd(struct ceph_osd *osd)
1138 {
1139 	struct ceph_entity_addr *peer_addr;
1140 
1141 	dout("%s osd %p osd%d\n", __func__, osd, osd->o_osd);
1142 
1143 	if (RB_EMPTY_ROOT(&osd->o_requests) &&
1144 	    RB_EMPTY_ROOT(&osd->o_linger_requests)) {
1145 		close_osd(osd);
1146 		return -ENODEV;
1147 	}
1148 
1149 	peer_addr = &osd->o_osdc->osdmap->osd_addr[osd->o_osd];
1150 	if (!memcmp(peer_addr, &osd->o_con.peer_addr, sizeof (*peer_addr)) &&
1151 			!ceph_con_opened(&osd->o_con)) {
1152 		struct rb_node *n;
1153 
1154 		dout("osd addr hasn't changed and connection never opened, "
1155 		     "letting msgr retry\n");
1156 		/* touch each r_stamp for handle_timeout()'s benfit */
1157 		for (n = rb_first(&osd->o_requests); n; n = rb_next(n)) {
1158 			struct ceph_osd_request *req =
1159 			    rb_entry(n, struct ceph_osd_request, r_node);
1160 			req->r_stamp = jiffies;
1161 		}
1162 
1163 		return -EAGAIN;
1164 	}
1165 
1166 	ceph_con_close(&osd->o_con);
1167 	ceph_con_open(&osd->o_con, CEPH_ENTITY_TYPE_OSD, osd->o_osd, peer_addr);
1168 	osd->o_incarnation++;
1169 
1170 	return 0;
1171 }
1172 
1173 static struct ceph_osd *lookup_create_osd(struct ceph_osd_client *osdc, int o,
1174 					  bool wrlocked)
1175 {
1176 	struct ceph_osd *osd;
1177 
1178 	if (wrlocked)
1179 		verify_osdc_wrlocked(osdc);
1180 	else
1181 		verify_osdc_locked(osdc);
1182 
1183 	if (o != CEPH_HOMELESS_OSD)
1184 		osd = lookup_osd(&osdc->osds, o);
1185 	else
1186 		osd = &osdc->homeless_osd;
1187 	if (!osd) {
1188 		if (!wrlocked)
1189 			return ERR_PTR(-EAGAIN);
1190 
1191 		osd = create_osd(osdc, o);
1192 		insert_osd(&osdc->osds, osd);
1193 		ceph_con_open(&osd->o_con, CEPH_ENTITY_TYPE_OSD, osd->o_osd,
1194 			      &osdc->osdmap->osd_addr[osd->o_osd]);
1195 	}
1196 
1197 	dout("%s osdc %p osd%d -> osd %p\n", __func__, osdc, o, osd);
1198 	return osd;
1199 }
1200 
1201 /*
1202  * Create request <-> OSD session relation.
1203  *
1204  * @req has to be assigned a tid, @osd may be homeless.
1205  */
1206 static void link_request(struct ceph_osd *osd, struct ceph_osd_request *req)
1207 {
1208 	verify_osd_locked(osd);
1209 	WARN_ON(!req->r_tid || req->r_osd);
1210 	dout("%s osd %p osd%d req %p tid %llu\n", __func__, osd, osd->o_osd,
1211 	     req, req->r_tid);
1212 
1213 	if (!osd_homeless(osd))
1214 		__remove_osd_from_lru(osd);
1215 	else
1216 		atomic_inc(&osd->o_osdc->num_homeless);
1217 
1218 	get_osd(osd);
1219 	insert_request(&osd->o_requests, req);
1220 	req->r_osd = osd;
1221 }
1222 
1223 static void unlink_request(struct ceph_osd *osd, struct ceph_osd_request *req)
1224 {
1225 	verify_osd_locked(osd);
1226 	WARN_ON(req->r_osd != osd);
1227 	dout("%s osd %p osd%d req %p tid %llu\n", __func__, osd, osd->o_osd,
1228 	     req, req->r_tid);
1229 
1230 	req->r_osd = NULL;
1231 	erase_request(&osd->o_requests, req);
1232 	put_osd(osd);
1233 
1234 	if (!osd_homeless(osd))
1235 		maybe_move_osd_to_lru(osd);
1236 	else
1237 		atomic_dec(&osd->o_osdc->num_homeless);
1238 }
1239 
1240 static bool __pool_full(struct ceph_pg_pool_info *pi)
1241 {
1242 	return pi->flags & CEPH_POOL_FLAG_FULL;
1243 }
1244 
1245 static bool have_pool_full(struct ceph_osd_client *osdc)
1246 {
1247 	struct rb_node *n;
1248 
1249 	for (n = rb_first(&osdc->osdmap->pg_pools); n; n = rb_next(n)) {
1250 		struct ceph_pg_pool_info *pi =
1251 		    rb_entry(n, struct ceph_pg_pool_info, node);
1252 
1253 		if (__pool_full(pi))
1254 			return true;
1255 	}
1256 
1257 	return false;
1258 }
1259 
1260 static bool pool_full(struct ceph_osd_client *osdc, s64 pool_id)
1261 {
1262 	struct ceph_pg_pool_info *pi;
1263 
1264 	pi = ceph_pg_pool_by_id(osdc->osdmap, pool_id);
1265 	if (!pi)
1266 		return false;
1267 
1268 	return __pool_full(pi);
1269 }
1270 
1271 /*
1272  * Returns whether a request should be blocked from being sent
1273  * based on the current osdmap and osd_client settings.
1274  */
1275 static bool target_should_be_paused(struct ceph_osd_client *osdc,
1276 				    const struct ceph_osd_request_target *t,
1277 				    struct ceph_pg_pool_info *pi)
1278 {
1279 	bool pauserd = ceph_osdmap_flag(osdc, CEPH_OSDMAP_PAUSERD);
1280 	bool pausewr = ceph_osdmap_flag(osdc, CEPH_OSDMAP_PAUSEWR) ||
1281 		       ceph_osdmap_flag(osdc, CEPH_OSDMAP_FULL) ||
1282 		       __pool_full(pi);
1283 
1284 	WARN_ON(pi->id != t->base_oloc.pool);
1285 	return (t->flags & CEPH_OSD_FLAG_READ && pauserd) ||
1286 	       (t->flags & CEPH_OSD_FLAG_WRITE && pausewr);
1287 }
1288 
1289 enum calc_target_result {
1290 	CALC_TARGET_NO_ACTION = 0,
1291 	CALC_TARGET_NEED_RESEND,
1292 	CALC_TARGET_POOL_DNE,
1293 };
1294 
1295 static enum calc_target_result calc_target(struct ceph_osd_client *osdc,
1296 					   struct ceph_osd_request_target *t,
1297 					   u32 *last_force_resend,
1298 					   bool any_change)
1299 {
1300 	struct ceph_pg_pool_info *pi;
1301 	struct ceph_pg pgid, last_pgid;
1302 	struct ceph_osds up, acting;
1303 	bool force_resend = false;
1304 	bool need_check_tiering = false;
1305 	bool need_resend = false;
1306 	bool sort_bitwise = ceph_osdmap_flag(osdc, CEPH_OSDMAP_SORTBITWISE);
1307 	enum calc_target_result ct_res;
1308 	int ret;
1309 
1310 	pi = ceph_pg_pool_by_id(osdc->osdmap, t->base_oloc.pool);
1311 	if (!pi) {
1312 		t->osd = CEPH_HOMELESS_OSD;
1313 		ct_res = CALC_TARGET_POOL_DNE;
1314 		goto out;
1315 	}
1316 
1317 	if (osdc->osdmap->epoch == pi->last_force_request_resend) {
1318 		if (last_force_resend &&
1319 		    *last_force_resend < pi->last_force_request_resend) {
1320 			*last_force_resend = pi->last_force_request_resend;
1321 			force_resend = true;
1322 		} else if (!last_force_resend) {
1323 			force_resend = true;
1324 		}
1325 	}
1326 	if (ceph_oid_empty(&t->target_oid) || force_resend) {
1327 		ceph_oid_copy(&t->target_oid, &t->base_oid);
1328 		need_check_tiering = true;
1329 	}
1330 	if (ceph_oloc_empty(&t->target_oloc) || force_resend) {
1331 		ceph_oloc_copy(&t->target_oloc, &t->base_oloc);
1332 		need_check_tiering = true;
1333 	}
1334 
1335 	if (need_check_tiering &&
1336 	    (t->flags & CEPH_OSD_FLAG_IGNORE_OVERLAY) == 0) {
1337 		if (t->flags & CEPH_OSD_FLAG_READ && pi->read_tier >= 0)
1338 			t->target_oloc.pool = pi->read_tier;
1339 		if (t->flags & CEPH_OSD_FLAG_WRITE && pi->write_tier >= 0)
1340 			t->target_oloc.pool = pi->write_tier;
1341 	}
1342 
1343 	ret = ceph_object_locator_to_pg(osdc->osdmap, &t->target_oid,
1344 					&t->target_oloc, &pgid);
1345 	if (ret) {
1346 		WARN_ON(ret != -ENOENT);
1347 		t->osd = CEPH_HOMELESS_OSD;
1348 		ct_res = CALC_TARGET_POOL_DNE;
1349 		goto out;
1350 	}
1351 	last_pgid.pool = pgid.pool;
1352 	last_pgid.seed = ceph_stable_mod(pgid.seed, t->pg_num, t->pg_num_mask);
1353 
1354 	ceph_pg_to_up_acting_osds(osdc->osdmap, &pgid, &up, &acting);
1355 	if (any_change &&
1356 	    ceph_is_new_interval(&t->acting,
1357 				 &acting,
1358 				 &t->up,
1359 				 &up,
1360 				 t->size,
1361 				 pi->size,
1362 				 t->min_size,
1363 				 pi->min_size,
1364 				 t->pg_num,
1365 				 pi->pg_num,
1366 				 t->sort_bitwise,
1367 				 sort_bitwise,
1368 				 &last_pgid))
1369 		force_resend = true;
1370 
1371 	if (t->paused && !target_should_be_paused(osdc, t, pi)) {
1372 		t->paused = false;
1373 		need_resend = true;
1374 	}
1375 
1376 	if (ceph_pg_compare(&t->pgid, &pgid) ||
1377 	    ceph_osds_changed(&t->acting, &acting, any_change) ||
1378 	    force_resend) {
1379 		t->pgid = pgid; /* struct */
1380 		ceph_osds_copy(&t->acting, &acting);
1381 		ceph_osds_copy(&t->up, &up);
1382 		t->size = pi->size;
1383 		t->min_size = pi->min_size;
1384 		t->pg_num = pi->pg_num;
1385 		t->pg_num_mask = pi->pg_num_mask;
1386 		t->sort_bitwise = sort_bitwise;
1387 
1388 		t->osd = acting.primary;
1389 		need_resend = true;
1390 	}
1391 
1392 	ct_res = need_resend ? CALC_TARGET_NEED_RESEND : CALC_TARGET_NO_ACTION;
1393 out:
1394 	dout("%s t %p -> ct_res %d osd %d\n", __func__, t, ct_res, t->osd);
1395 	return ct_res;
1396 }
1397 
1398 static void setup_request_data(struct ceph_osd_request *req,
1399 			       struct ceph_msg *msg)
1400 {
1401 	u32 data_len = 0;
1402 	int i;
1403 
1404 	if (!list_empty(&msg->data))
1405 		return;
1406 
1407 	WARN_ON(msg->data_length);
1408 	for (i = 0; i < req->r_num_ops; i++) {
1409 		struct ceph_osd_req_op *op = &req->r_ops[i];
1410 
1411 		switch (op->op) {
1412 		/* request */
1413 		case CEPH_OSD_OP_WRITE:
1414 		case CEPH_OSD_OP_WRITEFULL:
1415 			WARN_ON(op->indata_len != op->extent.length);
1416 			ceph_osdc_msg_data_add(msg, &op->extent.osd_data);
1417 			break;
1418 		case CEPH_OSD_OP_SETXATTR:
1419 		case CEPH_OSD_OP_CMPXATTR:
1420 			WARN_ON(op->indata_len != op->xattr.name_len +
1421 						  op->xattr.value_len);
1422 			ceph_osdc_msg_data_add(msg, &op->xattr.osd_data);
1423 			break;
1424 		case CEPH_OSD_OP_NOTIFY_ACK:
1425 			ceph_osdc_msg_data_add(msg,
1426 					       &op->notify_ack.request_data);
1427 			break;
1428 
1429 		/* reply */
1430 		case CEPH_OSD_OP_STAT:
1431 			ceph_osdc_msg_data_add(req->r_reply,
1432 					       &op->raw_data_in);
1433 			break;
1434 		case CEPH_OSD_OP_READ:
1435 			ceph_osdc_msg_data_add(req->r_reply,
1436 					       &op->extent.osd_data);
1437 			break;
1438 
1439 		/* both */
1440 		case CEPH_OSD_OP_CALL:
1441 			WARN_ON(op->indata_len != op->cls.class_len +
1442 						  op->cls.method_len +
1443 						  op->cls.indata_len);
1444 			ceph_osdc_msg_data_add(msg, &op->cls.request_info);
1445 			/* optional, can be NONE */
1446 			ceph_osdc_msg_data_add(msg, &op->cls.request_data);
1447 			/* optional, can be NONE */
1448 			ceph_osdc_msg_data_add(req->r_reply,
1449 					       &op->cls.response_data);
1450 			break;
1451 		case CEPH_OSD_OP_NOTIFY:
1452 			ceph_osdc_msg_data_add(msg,
1453 					       &op->notify.request_data);
1454 			ceph_osdc_msg_data_add(req->r_reply,
1455 					       &op->notify.response_data);
1456 			break;
1457 		}
1458 
1459 		data_len += op->indata_len;
1460 	}
1461 
1462 	WARN_ON(data_len != msg->data_length);
1463 }
1464 
1465 static void encode_request(struct ceph_osd_request *req, struct ceph_msg *msg)
1466 {
1467 	void *p = msg->front.iov_base;
1468 	void *const end = p + msg->front_alloc_len;
1469 	u32 data_len = 0;
1470 	int i;
1471 
1472 	if (req->r_flags & CEPH_OSD_FLAG_WRITE) {
1473 		/* snapshots aren't writeable */
1474 		WARN_ON(req->r_snapid != CEPH_NOSNAP);
1475 	} else {
1476 		WARN_ON(req->r_mtime.tv_sec || req->r_mtime.tv_nsec ||
1477 			req->r_data_offset || req->r_snapc);
1478 	}
1479 
1480 	setup_request_data(req, msg);
1481 
1482 	ceph_encode_32(&p, 1); /* client_inc, always 1 */
1483 	ceph_encode_32(&p, req->r_osdc->osdmap->epoch);
1484 	ceph_encode_32(&p, req->r_flags);
1485 	ceph_encode_timespec(p, &req->r_mtime);
1486 	p += sizeof(struct ceph_timespec);
1487 	/* aka reassert_version */
1488 	memcpy(p, &req->r_replay_version, sizeof(req->r_replay_version));
1489 	p += sizeof(req->r_replay_version);
1490 
1491 	/* oloc */
1492 	ceph_encode_8(&p, 4);
1493 	ceph_encode_8(&p, 4);
1494 	ceph_encode_32(&p, 8 + 4 + 4);
1495 	ceph_encode_64(&p, req->r_t.target_oloc.pool);
1496 	ceph_encode_32(&p, -1); /* preferred */
1497 	ceph_encode_32(&p, 0); /* key len */
1498 
1499 	/* pgid */
1500 	ceph_encode_8(&p, 1);
1501 	ceph_encode_64(&p, req->r_t.pgid.pool);
1502 	ceph_encode_32(&p, req->r_t.pgid.seed);
1503 	ceph_encode_32(&p, -1); /* preferred */
1504 
1505 	/* oid */
1506 	ceph_encode_32(&p, req->r_t.target_oid.name_len);
1507 	memcpy(p, req->r_t.target_oid.name, req->r_t.target_oid.name_len);
1508 	p += req->r_t.target_oid.name_len;
1509 
1510 	/* ops, can imply data */
1511 	ceph_encode_16(&p, req->r_num_ops);
1512 	for (i = 0; i < req->r_num_ops; i++) {
1513 		data_len += osd_req_encode_op(p, &req->r_ops[i]);
1514 		p += sizeof(struct ceph_osd_op);
1515 	}
1516 
1517 	ceph_encode_64(&p, req->r_snapid); /* snapid */
1518 	if (req->r_snapc) {
1519 		ceph_encode_64(&p, req->r_snapc->seq);
1520 		ceph_encode_32(&p, req->r_snapc->num_snaps);
1521 		for (i = 0; i < req->r_snapc->num_snaps; i++)
1522 			ceph_encode_64(&p, req->r_snapc->snaps[i]);
1523 	} else {
1524 		ceph_encode_64(&p, 0); /* snap_seq */
1525 		ceph_encode_32(&p, 0); /* snaps len */
1526 	}
1527 
1528 	ceph_encode_32(&p, req->r_attempts); /* retry_attempt */
1529 
1530 	BUG_ON(p > end);
1531 	msg->front.iov_len = p - msg->front.iov_base;
1532 	msg->hdr.version = cpu_to_le16(4); /* MOSDOp v4 */
1533 	msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
1534 	msg->hdr.data_len = cpu_to_le32(data_len);
1535 	/*
1536 	 * The header "data_off" is a hint to the receiver allowing it
1537 	 * to align received data into its buffers such that there's no
1538 	 * need to re-copy it before writing it to disk (direct I/O).
1539 	 */
1540 	msg->hdr.data_off = cpu_to_le16(req->r_data_offset);
1541 
1542 	dout("%s req %p oid %s oid_len %d front %zu data %u\n", __func__,
1543 	     req, req->r_t.target_oid.name, req->r_t.target_oid.name_len,
1544 	     msg->front.iov_len, data_len);
1545 }
1546 
1547 /*
1548  * @req has to be assigned a tid and registered.
1549  */
1550 static void send_request(struct ceph_osd_request *req)
1551 {
1552 	struct ceph_osd *osd = req->r_osd;
1553 
1554 	verify_osd_locked(osd);
1555 	WARN_ON(osd->o_osd != req->r_t.osd);
1556 
1557 	/*
1558 	 * We may have a previously queued request message hanging
1559 	 * around.  Cancel it to avoid corrupting the msgr.
1560 	 */
1561 	if (req->r_sent)
1562 		ceph_msg_revoke(req->r_request);
1563 
1564 	req->r_flags |= CEPH_OSD_FLAG_KNOWN_REDIR;
1565 	if (req->r_attempts)
1566 		req->r_flags |= CEPH_OSD_FLAG_RETRY;
1567 	else
1568 		WARN_ON(req->r_flags & CEPH_OSD_FLAG_RETRY);
1569 
1570 	encode_request(req, req->r_request);
1571 
1572 	dout("%s req %p tid %llu to pg %llu.%x osd%d flags 0x%x attempt %d\n",
1573 	     __func__, req, req->r_tid, req->r_t.pgid.pool, req->r_t.pgid.seed,
1574 	     req->r_t.osd, req->r_flags, req->r_attempts);
1575 
1576 	req->r_t.paused = false;
1577 	req->r_stamp = jiffies;
1578 	req->r_attempts++;
1579 
1580 	req->r_sent = osd->o_incarnation;
1581 	req->r_request->hdr.tid = cpu_to_le64(req->r_tid);
1582 	ceph_con_send(&osd->o_con, ceph_msg_get(req->r_request));
1583 }
1584 
1585 static void maybe_request_map(struct ceph_osd_client *osdc)
1586 {
1587 	bool continuous = false;
1588 
1589 	verify_osdc_locked(osdc);
1590 	WARN_ON(!osdc->osdmap->epoch);
1591 
1592 	if (ceph_osdmap_flag(osdc, CEPH_OSDMAP_FULL) ||
1593 	    ceph_osdmap_flag(osdc, CEPH_OSDMAP_PAUSERD) ||
1594 	    ceph_osdmap_flag(osdc, CEPH_OSDMAP_PAUSEWR)) {
1595 		dout("%s osdc %p continuous\n", __func__, osdc);
1596 		continuous = true;
1597 	} else {
1598 		dout("%s osdc %p onetime\n", __func__, osdc);
1599 	}
1600 
1601 	if (ceph_monc_want_map(&osdc->client->monc, CEPH_SUB_OSDMAP,
1602 			       osdc->osdmap->epoch + 1, continuous))
1603 		ceph_monc_renew_subs(&osdc->client->monc);
1604 }
1605 
1606 static void send_map_check(struct ceph_osd_request *req);
1607 
1608 static void __submit_request(struct ceph_osd_request *req, bool wrlocked)
1609 {
1610 	struct ceph_osd_client *osdc = req->r_osdc;
1611 	struct ceph_osd *osd;
1612 	enum calc_target_result ct_res;
1613 	bool need_send = false;
1614 	bool promoted = false;
1615 
1616 	WARN_ON(req->r_tid || req->r_got_reply);
1617 	dout("%s req %p wrlocked %d\n", __func__, req, wrlocked);
1618 
1619 again:
1620 	ct_res = calc_target(osdc, &req->r_t, &req->r_last_force_resend, false);
1621 	if (ct_res == CALC_TARGET_POOL_DNE && !wrlocked)
1622 		goto promote;
1623 
1624 	osd = lookup_create_osd(osdc, req->r_t.osd, wrlocked);
1625 	if (IS_ERR(osd)) {
1626 		WARN_ON(PTR_ERR(osd) != -EAGAIN || wrlocked);
1627 		goto promote;
1628 	}
1629 
1630 	if ((req->r_flags & CEPH_OSD_FLAG_WRITE) &&
1631 	    ceph_osdmap_flag(osdc, CEPH_OSDMAP_PAUSEWR)) {
1632 		dout("req %p pausewr\n", req);
1633 		req->r_t.paused = true;
1634 		maybe_request_map(osdc);
1635 	} else if ((req->r_flags & CEPH_OSD_FLAG_READ) &&
1636 		   ceph_osdmap_flag(osdc, CEPH_OSDMAP_PAUSERD)) {
1637 		dout("req %p pauserd\n", req);
1638 		req->r_t.paused = true;
1639 		maybe_request_map(osdc);
1640 	} else if ((req->r_flags & CEPH_OSD_FLAG_WRITE) &&
1641 		   !(req->r_flags & (CEPH_OSD_FLAG_FULL_TRY |
1642 				     CEPH_OSD_FLAG_FULL_FORCE)) &&
1643 		   (ceph_osdmap_flag(osdc, CEPH_OSDMAP_FULL) ||
1644 		    pool_full(osdc, req->r_t.base_oloc.pool))) {
1645 		dout("req %p full/pool_full\n", req);
1646 		pr_warn_ratelimited("FULL or reached pool quota\n");
1647 		req->r_t.paused = true;
1648 		maybe_request_map(osdc);
1649 	} else if (!osd_homeless(osd)) {
1650 		need_send = true;
1651 	} else {
1652 		maybe_request_map(osdc);
1653 	}
1654 
1655 	mutex_lock(&osd->lock);
1656 	/*
1657 	 * Assign the tid atomically with send_request() to protect
1658 	 * multiple writes to the same object from racing with each
1659 	 * other, resulting in out of order ops on the OSDs.
1660 	 */
1661 	req->r_tid = atomic64_inc_return(&osdc->last_tid);
1662 	link_request(osd, req);
1663 	if (need_send)
1664 		send_request(req);
1665 	mutex_unlock(&osd->lock);
1666 
1667 	if (ct_res == CALC_TARGET_POOL_DNE)
1668 		send_map_check(req);
1669 
1670 	if (promoted)
1671 		downgrade_write(&osdc->lock);
1672 	return;
1673 
1674 promote:
1675 	up_read(&osdc->lock);
1676 	down_write(&osdc->lock);
1677 	wrlocked = true;
1678 	promoted = true;
1679 	goto again;
1680 }
1681 
1682 static void account_request(struct ceph_osd_request *req)
1683 {
1684 	unsigned int mask = CEPH_OSD_FLAG_ACK | CEPH_OSD_FLAG_ONDISK;
1685 
1686 	if (req->r_flags & CEPH_OSD_FLAG_READ) {
1687 		WARN_ON(req->r_flags & mask);
1688 		req->r_flags |= CEPH_OSD_FLAG_ACK;
1689 	} else if (req->r_flags & CEPH_OSD_FLAG_WRITE)
1690 		WARN_ON(!(req->r_flags & mask));
1691 	else
1692 		WARN_ON(1);
1693 
1694 	WARN_ON(req->r_unsafe_callback && (req->r_flags & mask) != mask);
1695 	atomic_inc(&req->r_osdc->num_requests);
1696 }
1697 
1698 static void submit_request(struct ceph_osd_request *req, bool wrlocked)
1699 {
1700 	ceph_osdc_get_request(req);
1701 	account_request(req);
1702 	__submit_request(req, wrlocked);
1703 }
1704 
1705 static void __finish_request(struct ceph_osd_request *req)
1706 {
1707 	struct ceph_osd_client *osdc = req->r_osdc;
1708 	struct ceph_osd *osd = req->r_osd;
1709 
1710 	verify_osd_locked(osd);
1711 	dout("%s req %p tid %llu\n", __func__, req, req->r_tid);
1712 
1713 	WARN_ON(lookup_request_mc(&osdc->map_checks, req->r_tid));
1714 	unlink_request(osd, req);
1715 	atomic_dec(&osdc->num_requests);
1716 
1717 	/*
1718 	 * If an OSD has failed or returned and a request has been sent
1719 	 * twice, it's possible to get a reply and end up here while the
1720 	 * request message is queued for delivery.  We will ignore the
1721 	 * reply, so not a big deal, but better to try and catch it.
1722 	 */
1723 	ceph_msg_revoke(req->r_request);
1724 	ceph_msg_revoke_incoming(req->r_reply);
1725 }
1726 
1727 static void finish_request(struct ceph_osd_request *req)
1728 {
1729 	__finish_request(req);
1730 	ceph_osdc_put_request(req);
1731 }
1732 
1733 static void __complete_request(struct ceph_osd_request *req)
1734 {
1735 	if (req->r_callback)
1736 		req->r_callback(req);
1737 	else
1738 		complete_all(&req->r_completion);
1739 }
1740 
1741 /*
1742  * Note that this is open-coded in handle_reply(), which has to deal
1743  * with ack vs commit, dup acks, etc.
1744  */
1745 static void complete_request(struct ceph_osd_request *req, int err)
1746 {
1747 	dout("%s req %p tid %llu err %d\n", __func__, req, req->r_tid, err);
1748 
1749 	req->r_result = err;
1750 	__finish_request(req);
1751 	__complete_request(req);
1752 	complete_all(&req->r_safe_completion);
1753 	ceph_osdc_put_request(req);
1754 }
1755 
1756 static void cancel_map_check(struct ceph_osd_request *req)
1757 {
1758 	struct ceph_osd_client *osdc = req->r_osdc;
1759 	struct ceph_osd_request *lookup_req;
1760 
1761 	verify_osdc_wrlocked(osdc);
1762 
1763 	lookup_req = lookup_request_mc(&osdc->map_checks, req->r_tid);
1764 	if (!lookup_req)
1765 		return;
1766 
1767 	WARN_ON(lookup_req != req);
1768 	erase_request_mc(&osdc->map_checks, req);
1769 	ceph_osdc_put_request(req);
1770 }
1771 
1772 static void cancel_request(struct ceph_osd_request *req)
1773 {
1774 	dout("%s req %p tid %llu\n", __func__, req, req->r_tid);
1775 
1776 	cancel_map_check(req);
1777 	finish_request(req);
1778 }
1779 
1780 static void check_pool_dne(struct ceph_osd_request *req)
1781 {
1782 	struct ceph_osd_client *osdc = req->r_osdc;
1783 	struct ceph_osdmap *map = osdc->osdmap;
1784 
1785 	verify_osdc_wrlocked(osdc);
1786 	WARN_ON(!map->epoch);
1787 
1788 	if (req->r_attempts) {
1789 		/*
1790 		 * We sent a request earlier, which means that
1791 		 * previously the pool existed, and now it does not
1792 		 * (i.e., it was deleted).
1793 		 */
1794 		req->r_map_dne_bound = map->epoch;
1795 		dout("%s req %p tid %llu pool disappeared\n", __func__, req,
1796 		     req->r_tid);
1797 	} else {
1798 		dout("%s req %p tid %llu map_dne_bound %u have %u\n", __func__,
1799 		     req, req->r_tid, req->r_map_dne_bound, map->epoch);
1800 	}
1801 
1802 	if (req->r_map_dne_bound) {
1803 		if (map->epoch >= req->r_map_dne_bound) {
1804 			/* we had a new enough map */
1805 			pr_info_ratelimited("tid %llu pool does not exist\n",
1806 					    req->r_tid);
1807 			complete_request(req, -ENOENT);
1808 		}
1809 	} else {
1810 		send_map_check(req);
1811 	}
1812 }
1813 
1814 static void map_check_cb(struct ceph_mon_generic_request *greq)
1815 {
1816 	struct ceph_osd_client *osdc = &greq->monc->client->osdc;
1817 	struct ceph_osd_request *req;
1818 	u64 tid = greq->private_data;
1819 
1820 	WARN_ON(greq->result || !greq->u.newest);
1821 
1822 	down_write(&osdc->lock);
1823 	req = lookup_request_mc(&osdc->map_checks, tid);
1824 	if (!req) {
1825 		dout("%s tid %llu dne\n", __func__, tid);
1826 		goto out_unlock;
1827 	}
1828 
1829 	dout("%s req %p tid %llu map_dne_bound %u newest %llu\n", __func__,
1830 	     req, req->r_tid, req->r_map_dne_bound, greq->u.newest);
1831 	if (!req->r_map_dne_bound)
1832 		req->r_map_dne_bound = greq->u.newest;
1833 	erase_request_mc(&osdc->map_checks, req);
1834 	check_pool_dne(req);
1835 
1836 	ceph_osdc_put_request(req);
1837 out_unlock:
1838 	up_write(&osdc->lock);
1839 }
1840 
1841 static void send_map_check(struct ceph_osd_request *req)
1842 {
1843 	struct ceph_osd_client *osdc = req->r_osdc;
1844 	struct ceph_osd_request *lookup_req;
1845 	int ret;
1846 
1847 	verify_osdc_wrlocked(osdc);
1848 
1849 	lookup_req = lookup_request_mc(&osdc->map_checks, req->r_tid);
1850 	if (lookup_req) {
1851 		WARN_ON(lookup_req != req);
1852 		return;
1853 	}
1854 
1855 	ceph_osdc_get_request(req);
1856 	insert_request_mc(&osdc->map_checks, req);
1857 	ret = ceph_monc_get_version_async(&osdc->client->monc, "osdmap",
1858 					  map_check_cb, req->r_tid);
1859 	WARN_ON(ret);
1860 }
1861 
1862 /*
1863  * lingering requests, watch/notify v2 infrastructure
1864  */
1865 static void linger_release(struct kref *kref)
1866 {
1867 	struct ceph_osd_linger_request *lreq =
1868 	    container_of(kref, struct ceph_osd_linger_request, kref);
1869 
1870 	dout("%s lreq %p reg_req %p ping_req %p\n", __func__, lreq,
1871 	     lreq->reg_req, lreq->ping_req);
1872 	WARN_ON(!RB_EMPTY_NODE(&lreq->node));
1873 	WARN_ON(!RB_EMPTY_NODE(&lreq->osdc_node));
1874 	WARN_ON(!RB_EMPTY_NODE(&lreq->mc_node));
1875 	WARN_ON(!list_empty(&lreq->scan_item));
1876 	WARN_ON(!list_empty(&lreq->pending_lworks));
1877 	WARN_ON(lreq->osd);
1878 
1879 	if (lreq->reg_req)
1880 		ceph_osdc_put_request(lreq->reg_req);
1881 	if (lreq->ping_req)
1882 		ceph_osdc_put_request(lreq->ping_req);
1883 	target_destroy(&lreq->t);
1884 	kfree(lreq);
1885 }
1886 
1887 static void linger_put(struct ceph_osd_linger_request *lreq)
1888 {
1889 	if (lreq)
1890 		kref_put(&lreq->kref, linger_release);
1891 }
1892 
1893 static struct ceph_osd_linger_request *
1894 linger_get(struct ceph_osd_linger_request *lreq)
1895 {
1896 	kref_get(&lreq->kref);
1897 	return lreq;
1898 }
1899 
1900 static struct ceph_osd_linger_request *
1901 linger_alloc(struct ceph_osd_client *osdc)
1902 {
1903 	struct ceph_osd_linger_request *lreq;
1904 
1905 	lreq = kzalloc(sizeof(*lreq), GFP_NOIO);
1906 	if (!lreq)
1907 		return NULL;
1908 
1909 	kref_init(&lreq->kref);
1910 	mutex_init(&lreq->lock);
1911 	RB_CLEAR_NODE(&lreq->node);
1912 	RB_CLEAR_NODE(&lreq->osdc_node);
1913 	RB_CLEAR_NODE(&lreq->mc_node);
1914 	INIT_LIST_HEAD(&lreq->scan_item);
1915 	INIT_LIST_HEAD(&lreq->pending_lworks);
1916 	init_completion(&lreq->reg_commit_wait);
1917 	init_completion(&lreq->notify_finish_wait);
1918 
1919 	lreq->osdc = osdc;
1920 	target_init(&lreq->t);
1921 
1922 	dout("%s lreq %p\n", __func__, lreq);
1923 	return lreq;
1924 }
1925 
1926 DEFINE_RB_INSDEL_FUNCS(linger, struct ceph_osd_linger_request, linger_id, node)
1927 DEFINE_RB_FUNCS(linger_osdc, struct ceph_osd_linger_request, linger_id, osdc_node)
1928 DEFINE_RB_FUNCS(linger_mc, struct ceph_osd_linger_request, linger_id, mc_node)
1929 
1930 /*
1931  * Create linger request <-> OSD session relation.
1932  *
1933  * @lreq has to be registered, @osd may be homeless.
1934  */
1935 static void link_linger(struct ceph_osd *osd,
1936 			struct ceph_osd_linger_request *lreq)
1937 {
1938 	verify_osd_locked(osd);
1939 	WARN_ON(!lreq->linger_id || lreq->osd);
1940 	dout("%s osd %p osd%d lreq %p linger_id %llu\n", __func__, osd,
1941 	     osd->o_osd, lreq, lreq->linger_id);
1942 
1943 	if (!osd_homeless(osd))
1944 		__remove_osd_from_lru(osd);
1945 	else
1946 		atomic_inc(&osd->o_osdc->num_homeless);
1947 
1948 	get_osd(osd);
1949 	insert_linger(&osd->o_linger_requests, lreq);
1950 	lreq->osd = osd;
1951 }
1952 
1953 static void unlink_linger(struct ceph_osd *osd,
1954 			  struct ceph_osd_linger_request *lreq)
1955 {
1956 	verify_osd_locked(osd);
1957 	WARN_ON(lreq->osd != osd);
1958 	dout("%s osd %p osd%d lreq %p linger_id %llu\n", __func__, osd,
1959 	     osd->o_osd, lreq, lreq->linger_id);
1960 
1961 	lreq->osd = NULL;
1962 	erase_linger(&osd->o_linger_requests, lreq);
1963 	put_osd(osd);
1964 
1965 	if (!osd_homeless(osd))
1966 		maybe_move_osd_to_lru(osd);
1967 	else
1968 		atomic_dec(&osd->o_osdc->num_homeless);
1969 }
1970 
1971 static bool __linger_registered(struct ceph_osd_linger_request *lreq)
1972 {
1973 	verify_osdc_locked(lreq->osdc);
1974 
1975 	return !RB_EMPTY_NODE(&lreq->osdc_node);
1976 }
1977 
1978 static bool linger_registered(struct ceph_osd_linger_request *lreq)
1979 {
1980 	struct ceph_osd_client *osdc = lreq->osdc;
1981 	bool registered;
1982 
1983 	down_read(&osdc->lock);
1984 	registered = __linger_registered(lreq);
1985 	up_read(&osdc->lock);
1986 
1987 	return registered;
1988 }
1989 
1990 static void linger_register(struct ceph_osd_linger_request *lreq)
1991 {
1992 	struct ceph_osd_client *osdc = lreq->osdc;
1993 
1994 	verify_osdc_wrlocked(osdc);
1995 	WARN_ON(lreq->linger_id);
1996 
1997 	linger_get(lreq);
1998 	lreq->linger_id = ++osdc->last_linger_id;
1999 	insert_linger_osdc(&osdc->linger_requests, lreq);
2000 }
2001 
2002 static void linger_unregister(struct ceph_osd_linger_request *lreq)
2003 {
2004 	struct ceph_osd_client *osdc = lreq->osdc;
2005 
2006 	verify_osdc_wrlocked(osdc);
2007 
2008 	erase_linger_osdc(&osdc->linger_requests, lreq);
2009 	linger_put(lreq);
2010 }
2011 
2012 static void cancel_linger_request(struct ceph_osd_request *req)
2013 {
2014 	struct ceph_osd_linger_request *lreq = req->r_priv;
2015 
2016 	WARN_ON(!req->r_linger);
2017 	cancel_request(req);
2018 	linger_put(lreq);
2019 }
2020 
2021 struct linger_work {
2022 	struct work_struct work;
2023 	struct ceph_osd_linger_request *lreq;
2024 	struct list_head pending_item;
2025 	unsigned long queued_stamp;
2026 
2027 	union {
2028 		struct {
2029 			u64 notify_id;
2030 			u64 notifier_id;
2031 			void *payload; /* points into @msg front */
2032 			size_t payload_len;
2033 
2034 			struct ceph_msg *msg; /* for ceph_msg_put() */
2035 		} notify;
2036 		struct {
2037 			int err;
2038 		} error;
2039 	};
2040 };
2041 
2042 static struct linger_work *lwork_alloc(struct ceph_osd_linger_request *lreq,
2043 				       work_func_t workfn)
2044 {
2045 	struct linger_work *lwork;
2046 
2047 	lwork = kzalloc(sizeof(*lwork), GFP_NOIO);
2048 	if (!lwork)
2049 		return NULL;
2050 
2051 	INIT_WORK(&lwork->work, workfn);
2052 	INIT_LIST_HEAD(&lwork->pending_item);
2053 	lwork->lreq = linger_get(lreq);
2054 
2055 	return lwork;
2056 }
2057 
2058 static void lwork_free(struct linger_work *lwork)
2059 {
2060 	struct ceph_osd_linger_request *lreq = lwork->lreq;
2061 
2062 	mutex_lock(&lreq->lock);
2063 	list_del(&lwork->pending_item);
2064 	mutex_unlock(&lreq->lock);
2065 
2066 	linger_put(lreq);
2067 	kfree(lwork);
2068 }
2069 
2070 static void lwork_queue(struct linger_work *lwork)
2071 {
2072 	struct ceph_osd_linger_request *lreq = lwork->lreq;
2073 	struct ceph_osd_client *osdc = lreq->osdc;
2074 
2075 	verify_lreq_locked(lreq);
2076 	WARN_ON(!list_empty(&lwork->pending_item));
2077 
2078 	lwork->queued_stamp = jiffies;
2079 	list_add_tail(&lwork->pending_item, &lreq->pending_lworks);
2080 	queue_work(osdc->notify_wq, &lwork->work);
2081 }
2082 
2083 static void do_watch_notify(struct work_struct *w)
2084 {
2085 	struct linger_work *lwork = container_of(w, struct linger_work, work);
2086 	struct ceph_osd_linger_request *lreq = lwork->lreq;
2087 
2088 	if (!linger_registered(lreq)) {
2089 		dout("%s lreq %p not registered\n", __func__, lreq);
2090 		goto out;
2091 	}
2092 
2093 	WARN_ON(!lreq->is_watch);
2094 	dout("%s lreq %p notify_id %llu notifier_id %llu payload_len %zu\n",
2095 	     __func__, lreq, lwork->notify.notify_id, lwork->notify.notifier_id,
2096 	     lwork->notify.payload_len);
2097 	lreq->wcb(lreq->data, lwork->notify.notify_id, lreq->linger_id,
2098 		  lwork->notify.notifier_id, lwork->notify.payload,
2099 		  lwork->notify.payload_len);
2100 
2101 out:
2102 	ceph_msg_put(lwork->notify.msg);
2103 	lwork_free(lwork);
2104 }
2105 
2106 static void do_watch_error(struct work_struct *w)
2107 {
2108 	struct linger_work *lwork = container_of(w, struct linger_work, work);
2109 	struct ceph_osd_linger_request *lreq = lwork->lreq;
2110 
2111 	if (!linger_registered(lreq)) {
2112 		dout("%s lreq %p not registered\n", __func__, lreq);
2113 		goto out;
2114 	}
2115 
2116 	dout("%s lreq %p err %d\n", __func__, lreq, lwork->error.err);
2117 	lreq->errcb(lreq->data, lreq->linger_id, lwork->error.err);
2118 
2119 out:
2120 	lwork_free(lwork);
2121 }
2122 
2123 static void queue_watch_error(struct ceph_osd_linger_request *lreq)
2124 {
2125 	struct linger_work *lwork;
2126 
2127 	lwork = lwork_alloc(lreq, do_watch_error);
2128 	if (!lwork) {
2129 		pr_err("failed to allocate error-lwork\n");
2130 		return;
2131 	}
2132 
2133 	lwork->error.err = lreq->last_error;
2134 	lwork_queue(lwork);
2135 }
2136 
2137 static void linger_reg_commit_complete(struct ceph_osd_linger_request *lreq,
2138 				       int result)
2139 {
2140 	if (!completion_done(&lreq->reg_commit_wait)) {
2141 		lreq->reg_commit_error = (result <= 0 ? result : 0);
2142 		complete_all(&lreq->reg_commit_wait);
2143 	}
2144 }
2145 
2146 static void linger_commit_cb(struct ceph_osd_request *req)
2147 {
2148 	struct ceph_osd_linger_request *lreq = req->r_priv;
2149 
2150 	mutex_lock(&lreq->lock);
2151 	dout("%s lreq %p linger_id %llu result %d\n", __func__, lreq,
2152 	     lreq->linger_id, req->r_result);
2153 	WARN_ON(!__linger_registered(lreq));
2154 	linger_reg_commit_complete(lreq, req->r_result);
2155 	lreq->committed = true;
2156 
2157 	if (!lreq->is_watch) {
2158 		struct ceph_osd_data *osd_data =
2159 		    osd_req_op_data(req, 0, notify, response_data);
2160 		void *p = page_address(osd_data->pages[0]);
2161 
2162 		WARN_ON(req->r_ops[0].op != CEPH_OSD_OP_NOTIFY ||
2163 			osd_data->type != CEPH_OSD_DATA_TYPE_PAGES);
2164 
2165 		/* make note of the notify_id */
2166 		if (req->r_ops[0].outdata_len >= sizeof(u64)) {
2167 			lreq->notify_id = ceph_decode_64(&p);
2168 			dout("lreq %p notify_id %llu\n", lreq,
2169 			     lreq->notify_id);
2170 		} else {
2171 			dout("lreq %p no notify_id\n", lreq);
2172 		}
2173 	}
2174 
2175 	mutex_unlock(&lreq->lock);
2176 	linger_put(lreq);
2177 }
2178 
2179 static int normalize_watch_error(int err)
2180 {
2181 	/*
2182 	 * Translate ENOENT -> ENOTCONN so that a delete->disconnection
2183 	 * notification and a failure to reconnect because we raced with
2184 	 * the delete appear the same to the user.
2185 	 */
2186 	if (err == -ENOENT)
2187 		err = -ENOTCONN;
2188 
2189 	return err;
2190 }
2191 
2192 static void linger_reconnect_cb(struct ceph_osd_request *req)
2193 {
2194 	struct ceph_osd_linger_request *lreq = req->r_priv;
2195 
2196 	mutex_lock(&lreq->lock);
2197 	dout("%s lreq %p linger_id %llu result %d last_error %d\n", __func__,
2198 	     lreq, lreq->linger_id, req->r_result, lreq->last_error);
2199 	if (req->r_result < 0) {
2200 		if (!lreq->last_error) {
2201 			lreq->last_error = normalize_watch_error(req->r_result);
2202 			queue_watch_error(lreq);
2203 		}
2204 	}
2205 
2206 	mutex_unlock(&lreq->lock);
2207 	linger_put(lreq);
2208 }
2209 
2210 static void send_linger(struct ceph_osd_linger_request *lreq)
2211 {
2212 	struct ceph_osd_request *req = lreq->reg_req;
2213 	struct ceph_osd_req_op *op = &req->r_ops[0];
2214 
2215 	verify_osdc_wrlocked(req->r_osdc);
2216 	dout("%s lreq %p linger_id %llu\n", __func__, lreq, lreq->linger_id);
2217 
2218 	if (req->r_osd)
2219 		cancel_linger_request(req);
2220 
2221 	request_reinit(req);
2222 	ceph_oid_copy(&req->r_base_oid, &lreq->t.base_oid);
2223 	ceph_oloc_copy(&req->r_base_oloc, &lreq->t.base_oloc);
2224 	req->r_flags = lreq->t.flags;
2225 	req->r_mtime = lreq->mtime;
2226 
2227 	mutex_lock(&lreq->lock);
2228 	if (lreq->is_watch && lreq->committed) {
2229 		WARN_ON(op->op != CEPH_OSD_OP_WATCH ||
2230 			op->watch.cookie != lreq->linger_id);
2231 		op->watch.op = CEPH_OSD_WATCH_OP_RECONNECT;
2232 		op->watch.gen = ++lreq->register_gen;
2233 		dout("lreq %p reconnect register_gen %u\n", lreq,
2234 		     op->watch.gen);
2235 		req->r_callback = linger_reconnect_cb;
2236 	} else {
2237 		if (!lreq->is_watch)
2238 			lreq->notify_id = 0;
2239 		else
2240 			WARN_ON(op->watch.op != CEPH_OSD_WATCH_OP_WATCH);
2241 		dout("lreq %p register\n", lreq);
2242 		req->r_callback = linger_commit_cb;
2243 	}
2244 	mutex_unlock(&lreq->lock);
2245 
2246 	req->r_priv = linger_get(lreq);
2247 	req->r_linger = true;
2248 
2249 	submit_request(req, true);
2250 }
2251 
2252 static void linger_ping_cb(struct ceph_osd_request *req)
2253 {
2254 	struct ceph_osd_linger_request *lreq = req->r_priv;
2255 
2256 	mutex_lock(&lreq->lock);
2257 	dout("%s lreq %p linger_id %llu result %d ping_sent %lu last_error %d\n",
2258 	     __func__, lreq, lreq->linger_id, req->r_result, lreq->ping_sent,
2259 	     lreq->last_error);
2260 	if (lreq->register_gen == req->r_ops[0].watch.gen) {
2261 		if (!req->r_result) {
2262 			lreq->watch_valid_thru = lreq->ping_sent;
2263 		} else if (!lreq->last_error) {
2264 			lreq->last_error = normalize_watch_error(req->r_result);
2265 			queue_watch_error(lreq);
2266 		}
2267 	} else {
2268 		dout("lreq %p register_gen %u ignoring old pong %u\n", lreq,
2269 		     lreq->register_gen, req->r_ops[0].watch.gen);
2270 	}
2271 
2272 	mutex_unlock(&lreq->lock);
2273 	linger_put(lreq);
2274 }
2275 
2276 static void send_linger_ping(struct ceph_osd_linger_request *lreq)
2277 {
2278 	struct ceph_osd_client *osdc = lreq->osdc;
2279 	struct ceph_osd_request *req = lreq->ping_req;
2280 	struct ceph_osd_req_op *op = &req->r_ops[0];
2281 
2282 	if (ceph_osdmap_flag(osdc, CEPH_OSDMAP_PAUSERD)) {
2283 		dout("%s PAUSERD\n", __func__);
2284 		return;
2285 	}
2286 
2287 	lreq->ping_sent = jiffies;
2288 	dout("%s lreq %p linger_id %llu ping_sent %lu register_gen %u\n",
2289 	     __func__, lreq, lreq->linger_id, lreq->ping_sent,
2290 	     lreq->register_gen);
2291 
2292 	if (req->r_osd)
2293 		cancel_linger_request(req);
2294 
2295 	request_reinit(req);
2296 	target_copy(&req->r_t, &lreq->t);
2297 
2298 	WARN_ON(op->op != CEPH_OSD_OP_WATCH ||
2299 		op->watch.cookie != lreq->linger_id ||
2300 		op->watch.op != CEPH_OSD_WATCH_OP_PING);
2301 	op->watch.gen = lreq->register_gen;
2302 	req->r_callback = linger_ping_cb;
2303 	req->r_priv = linger_get(lreq);
2304 	req->r_linger = true;
2305 
2306 	ceph_osdc_get_request(req);
2307 	account_request(req);
2308 	req->r_tid = atomic64_inc_return(&osdc->last_tid);
2309 	link_request(lreq->osd, req);
2310 	send_request(req);
2311 }
2312 
2313 static void linger_submit(struct ceph_osd_linger_request *lreq)
2314 {
2315 	struct ceph_osd_client *osdc = lreq->osdc;
2316 	struct ceph_osd *osd;
2317 
2318 	calc_target(osdc, &lreq->t, &lreq->last_force_resend, false);
2319 	osd = lookup_create_osd(osdc, lreq->t.osd, true);
2320 	link_linger(osd, lreq);
2321 
2322 	send_linger(lreq);
2323 }
2324 
2325 static void cancel_linger_map_check(struct ceph_osd_linger_request *lreq)
2326 {
2327 	struct ceph_osd_client *osdc = lreq->osdc;
2328 	struct ceph_osd_linger_request *lookup_lreq;
2329 
2330 	verify_osdc_wrlocked(osdc);
2331 
2332 	lookup_lreq = lookup_linger_mc(&osdc->linger_map_checks,
2333 				       lreq->linger_id);
2334 	if (!lookup_lreq)
2335 		return;
2336 
2337 	WARN_ON(lookup_lreq != lreq);
2338 	erase_linger_mc(&osdc->linger_map_checks, lreq);
2339 	linger_put(lreq);
2340 }
2341 
2342 /*
2343  * @lreq has to be both registered and linked.
2344  */
2345 static void __linger_cancel(struct ceph_osd_linger_request *lreq)
2346 {
2347 	if (lreq->is_watch && lreq->ping_req->r_osd)
2348 		cancel_linger_request(lreq->ping_req);
2349 	if (lreq->reg_req->r_osd)
2350 		cancel_linger_request(lreq->reg_req);
2351 	cancel_linger_map_check(lreq);
2352 	unlink_linger(lreq->osd, lreq);
2353 	linger_unregister(lreq);
2354 }
2355 
2356 static void linger_cancel(struct ceph_osd_linger_request *lreq)
2357 {
2358 	struct ceph_osd_client *osdc = lreq->osdc;
2359 
2360 	down_write(&osdc->lock);
2361 	if (__linger_registered(lreq))
2362 		__linger_cancel(lreq);
2363 	up_write(&osdc->lock);
2364 }
2365 
2366 static void send_linger_map_check(struct ceph_osd_linger_request *lreq);
2367 
2368 static void check_linger_pool_dne(struct ceph_osd_linger_request *lreq)
2369 {
2370 	struct ceph_osd_client *osdc = lreq->osdc;
2371 	struct ceph_osdmap *map = osdc->osdmap;
2372 
2373 	verify_osdc_wrlocked(osdc);
2374 	WARN_ON(!map->epoch);
2375 
2376 	if (lreq->register_gen) {
2377 		lreq->map_dne_bound = map->epoch;
2378 		dout("%s lreq %p linger_id %llu pool disappeared\n", __func__,
2379 		     lreq, lreq->linger_id);
2380 	} else {
2381 		dout("%s lreq %p linger_id %llu map_dne_bound %u have %u\n",
2382 		     __func__, lreq, lreq->linger_id, lreq->map_dne_bound,
2383 		     map->epoch);
2384 	}
2385 
2386 	if (lreq->map_dne_bound) {
2387 		if (map->epoch >= lreq->map_dne_bound) {
2388 			/* we had a new enough map */
2389 			pr_info("linger_id %llu pool does not exist\n",
2390 				lreq->linger_id);
2391 			linger_reg_commit_complete(lreq, -ENOENT);
2392 			__linger_cancel(lreq);
2393 		}
2394 	} else {
2395 		send_linger_map_check(lreq);
2396 	}
2397 }
2398 
2399 static void linger_map_check_cb(struct ceph_mon_generic_request *greq)
2400 {
2401 	struct ceph_osd_client *osdc = &greq->monc->client->osdc;
2402 	struct ceph_osd_linger_request *lreq;
2403 	u64 linger_id = greq->private_data;
2404 
2405 	WARN_ON(greq->result || !greq->u.newest);
2406 
2407 	down_write(&osdc->lock);
2408 	lreq = lookup_linger_mc(&osdc->linger_map_checks, linger_id);
2409 	if (!lreq) {
2410 		dout("%s linger_id %llu dne\n", __func__, linger_id);
2411 		goto out_unlock;
2412 	}
2413 
2414 	dout("%s lreq %p linger_id %llu map_dne_bound %u newest %llu\n",
2415 	     __func__, lreq, lreq->linger_id, lreq->map_dne_bound,
2416 	     greq->u.newest);
2417 	if (!lreq->map_dne_bound)
2418 		lreq->map_dne_bound = greq->u.newest;
2419 	erase_linger_mc(&osdc->linger_map_checks, lreq);
2420 	check_linger_pool_dne(lreq);
2421 
2422 	linger_put(lreq);
2423 out_unlock:
2424 	up_write(&osdc->lock);
2425 }
2426 
2427 static void send_linger_map_check(struct ceph_osd_linger_request *lreq)
2428 {
2429 	struct ceph_osd_client *osdc = lreq->osdc;
2430 	struct ceph_osd_linger_request *lookup_lreq;
2431 	int ret;
2432 
2433 	verify_osdc_wrlocked(osdc);
2434 
2435 	lookup_lreq = lookup_linger_mc(&osdc->linger_map_checks,
2436 				       lreq->linger_id);
2437 	if (lookup_lreq) {
2438 		WARN_ON(lookup_lreq != lreq);
2439 		return;
2440 	}
2441 
2442 	linger_get(lreq);
2443 	insert_linger_mc(&osdc->linger_map_checks, lreq);
2444 	ret = ceph_monc_get_version_async(&osdc->client->monc, "osdmap",
2445 					  linger_map_check_cb, lreq->linger_id);
2446 	WARN_ON(ret);
2447 }
2448 
2449 static int linger_reg_commit_wait(struct ceph_osd_linger_request *lreq)
2450 {
2451 	int ret;
2452 
2453 	dout("%s lreq %p linger_id %llu\n", __func__, lreq, lreq->linger_id);
2454 	ret = wait_for_completion_interruptible(&lreq->reg_commit_wait);
2455 	return ret ?: lreq->reg_commit_error;
2456 }
2457 
2458 static int linger_notify_finish_wait(struct ceph_osd_linger_request *lreq)
2459 {
2460 	int ret;
2461 
2462 	dout("%s lreq %p linger_id %llu\n", __func__, lreq, lreq->linger_id);
2463 	ret = wait_for_completion_interruptible(&lreq->notify_finish_wait);
2464 	return ret ?: lreq->notify_finish_error;
2465 }
2466 
2467 /*
2468  * Timeout callback, called every N seconds.  When 1 or more OSD
2469  * requests has been active for more than N seconds, we send a keepalive
2470  * (tag + timestamp) to its OSD to ensure any communications channel
2471  * reset is detected.
2472  */
2473 static void handle_timeout(struct work_struct *work)
2474 {
2475 	struct ceph_osd_client *osdc =
2476 		container_of(work, struct ceph_osd_client, timeout_work.work);
2477 	struct ceph_options *opts = osdc->client->options;
2478 	unsigned long cutoff = jiffies - opts->osd_keepalive_timeout;
2479 	LIST_HEAD(slow_osds);
2480 	struct rb_node *n, *p;
2481 
2482 	dout("%s osdc %p\n", __func__, osdc);
2483 	down_write(&osdc->lock);
2484 
2485 	/*
2486 	 * ping osds that are a bit slow.  this ensures that if there
2487 	 * is a break in the TCP connection we will notice, and reopen
2488 	 * a connection with that osd (from the fault callback).
2489 	 */
2490 	for (n = rb_first(&osdc->osds); n; n = rb_next(n)) {
2491 		struct ceph_osd *osd = rb_entry(n, struct ceph_osd, o_node);
2492 		bool found = false;
2493 
2494 		for (p = rb_first(&osd->o_requests); p; p = rb_next(p)) {
2495 			struct ceph_osd_request *req =
2496 			    rb_entry(p, struct ceph_osd_request, r_node);
2497 
2498 			if (time_before(req->r_stamp, cutoff)) {
2499 				dout(" req %p tid %llu on osd%d is laggy\n",
2500 				     req, req->r_tid, osd->o_osd);
2501 				found = true;
2502 			}
2503 		}
2504 		for (p = rb_first(&osd->o_linger_requests); p; p = rb_next(p)) {
2505 			struct ceph_osd_linger_request *lreq =
2506 			    rb_entry(p, struct ceph_osd_linger_request, node);
2507 
2508 			dout(" lreq %p linger_id %llu is served by osd%d\n",
2509 			     lreq, lreq->linger_id, osd->o_osd);
2510 			found = true;
2511 
2512 			mutex_lock(&lreq->lock);
2513 			if (lreq->is_watch && lreq->committed && !lreq->last_error)
2514 				send_linger_ping(lreq);
2515 			mutex_unlock(&lreq->lock);
2516 		}
2517 
2518 		if (found)
2519 			list_move_tail(&osd->o_keepalive_item, &slow_osds);
2520 	}
2521 
2522 	if (atomic_read(&osdc->num_homeless) || !list_empty(&slow_osds))
2523 		maybe_request_map(osdc);
2524 
2525 	while (!list_empty(&slow_osds)) {
2526 		struct ceph_osd *osd = list_first_entry(&slow_osds,
2527 							struct ceph_osd,
2528 							o_keepalive_item);
2529 		list_del_init(&osd->o_keepalive_item);
2530 		ceph_con_keepalive(&osd->o_con);
2531 	}
2532 
2533 	up_write(&osdc->lock);
2534 	schedule_delayed_work(&osdc->timeout_work,
2535 			      osdc->client->options->osd_keepalive_timeout);
2536 }
2537 
2538 static void handle_osds_timeout(struct work_struct *work)
2539 {
2540 	struct ceph_osd_client *osdc =
2541 		container_of(work, struct ceph_osd_client,
2542 			     osds_timeout_work.work);
2543 	unsigned long delay = osdc->client->options->osd_idle_ttl / 4;
2544 	struct ceph_osd *osd, *nosd;
2545 
2546 	dout("%s osdc %p\n", __func__, osdc);
2547 	down_write(&osdc->lock);
2548 	list_for_each_entry_safe(osd, nosd, &osdc->osd_lru, o_osd_lru) {
2549 		if (time_before(jiffies, osd->lru_ttl))
2550 			break;
2551 
2552 		WARN_ON(!RB_EMPTY_ROOT(&osd->o_requests));
2553 		WARN_ON(!RB_EMPTY_ROOT(&osd->o_linger_requests));
2554 		close_osd(osd);
2555 	}
2556 
2557 	up_write(&osdc->lock);
2558 	schedule_delayed_work(&osdc->osds_timeout_work,
2559 			      round_jiffies_relative(delay));
2560 }
2561 
2562 static int ceph_oloc_decode(void **p, void *end,
2563 			    struct ceph_object_locator *oloc)
2564 {
2565 	u8 struct_v, struct_cv;
2566 	u32 len;
2567 	void *struct_end;
2568 	int ret = 0;
2569 
2570 	ceph_decode_need(p, end, 1 + 1 + 4, e_inval);
2571 	struct_v = ceph_decode_8(p);
2572 	struct_cv = ceph_decode_8(p);
2573 	if (struct_v < 3) {
2574 		pr_warn("got v %d < 3 cv %d of ceph_object_locator\n",
2575 			struct_v, struct_cv);
2576 		goto e_inval;
2577 	}
2578 	if (struct_cv > 6) {
2579 		pr_warn("got v %d cv %d > 6 of ceph_object_locator\n",
2580 			struct_v, struct_cv);
2581 		goto e_inval;
2582 	}
2583 	len = ceph_decode_32(p);
2584 	ceph_decode_need(p, end, len, e_inval);
2585 	struct_end = *p + len;
2586 
2587 	oloc->pool = ceph_decode_64(p);
2588 	*p += 4; /* skip preferred */
2589 
2590 	len = ceph_decode_32(p);
2591 	if (len > 0) {
2592 		pr_warn("ceph_object_locator::key is set\n");
2593 		goto e_inval;
2594 	}
2595 
2596 	if (struct_v >= 5) {
2597 		len = ceph_decode_32(p);
2598 		if (len > 0) {
2599 			pr_warn("ceph_object_locator::nspace is set\n");
2600 			goto e_inval;
2601 		}
2602 	}
2603 
2604 	if (struct_v >= 6) {
2605 		s64 hash = ceph_decode_64(p);
2606 		if (hash != -1) {
2607 			pr_warn("ceph_object_locator::hash is set\n");
2608 			goto e_inval;
2609 		}
2610 	}
2611 
2612 	/* skip the rest */
2613 	*p = struct_end;
2614 out:
2615 	return ret;
2616 
2617 e_inval:
2618 	ret = -EINVAL;
2619 	goto out;
2620 }
2621 
2622 static int ceph_redirect_decode(void **p, void *end,
2623 				struct ceph_request_redirect *redir)
2624 {
2625 	u8 struct_v, struct_cv;
2626 	u32 len;
2627 	void *struct_end;
2628 	int ret;
2629 
2630 	ceph_decode_need(p, end, 1 + 1 + 4, e_inval);
2631 	struct_v = ceph_decode_8(p);
2632 	struct_cv = ceph_decode_8(p);
2633 	if (struct_cv > 1) {
2634 		pr_warn("got v %d cv %d > 1 of ceph_request_redirect\n",
2635 			struct_v, struct_cv);
2636 		goto e_inval;
2637 	}
2638 	len = ceph_decode_32(p);
2639 	ceph_decode_need(p, end, len, e_inval);
2640 	struct_end = *p + len;
2641 
2642 	ret = ceph_oloc_decode(p, end, &redir->oloc);
2643 	if (ret)
2644 		goto out;
2645 
2646 	len = ceph_decode_32(p);
2647 	if (len > 0) {
2648 		pr_warn("ceph_request_redirect::object_name is set\n");
2649 		goto e_inval;
2650 	}
2651 
2652 	len = ceph_decode_32(p);
2653 	*p += len; /* skip osd_instructions */
2654 
2655 	/* skip the rest */
2656 	*p = struct_end;
2657 out:
2658 	return ret;
2659 
2660 e_inval:
2661 	ret = -EINVAL;
2662 	goto out;
2663 }
2664 
2665 struct MOSDOpReply {
2666 	struct ceph_pg pgid;
2667 	u64 flags;
2668 	int result;
2669 	u32 epoch;
2670 	int num_ops;
2671 	u32 outdata_len[CEPH_OSD_MAX_OPS];
2672 	s32 rval[CEPH_OSD_MAX_OPS];
2673 	int retry_attempt;
2674 	struct ceph_eversion replay_version;
2675 	u64 user_version;
2676 	struct ceph_request_redirect redirect;
2677 };
2678 
2679 static int decode_MOSDOpReply(const struct ceph_msg *msg, struct MOSDOpReply *m)
2680 {
2681 	void *p = msg->front.iov_base;
2682 	void *const end = p + msg->front.iov_len;
2683 	u16 version = le16_to_cpu(msg->hdr.version);
2684 	struct ceph_eversion bad_replay_version;
2685 	u8 decode_redir;
2686 	u32 len;
2687 	int ret;
2688 	int i;
2689 
2690 	ceph_decode_32_safe(&p, end, len, e_inval);
2691 	ceph_decode_need(&p, end, len, e_inval);
2692 	p += len; /* skip oid */
2693 
2694 	ret = ceph_decode_pgid(&p, end, &m->pgid);
2695 	if (ret)
2696 		return ret;
2697 
2698 	ceph_decode_64_safe(&p, end, m->flags, e_inval);
2699 	ceph_decode_32_safe(&p, end, m->result, e_inval);
2700 	ceph_decode_need(&p, end, sizeof(bad_replay_version), e_inval);
2701 	memcpy(&bad_replay_version, p, sizeof(bad_replay_version));
2702 	p += sizeof(bad_replay_version);
2703 	ceph_decode_32_safe(&p, end, m->epoch, e_inval);
2704 
2705 	ceph_decode_32_safe(&p, end, m->num_ops, e_inval);
2706 	if (m->num_ops > ARRAY_SIZE(m->outdata_len))
2707 		goto e_inval;
2708 
2709 	ceph_decode_need(&p, end, m->num_ops * sizeof(struct ceph_osd_op),
2710 			 e_inval);
2711 	for (i = 0; i < m->num_ops; i++) {
2712 		struct ceph_osd_op *op = p;
2713 
2714 		m->outdata_len[i] = le32_to_cpu(op->payload_len);
2715 		p += sizeof(*op);
2716 	}
2717 
2718 	ceph_decode_32_safe(&p, end, m->retry_attempt, e_inval);
2719 	for (i = 0; i < m->num_ops; i++)
2720 		ceph_decode_32_safe(&p, end, m->rval[i], e_inval);
2721 
2722 	if (version >= 5) {
2723 		ceph_decode_need(&p, end, sizeof(m->replay_version), e_inval);
2724 		memcpy(&m->replay_version, p, sizeof(m->replay_version));
2725 		p += sizeof(m->replay_version);
2726 		ceph_decode_64_safe(&p, end, m->user_version, e_inval);
2727 	} else {
2728 		m->replay_version = bad_replay_version; /* struct */
2729 		m->user_version = le64_to_cpu(m->replay_version.version);
2730 	}
2731 
2732 	if (version >= 6) {
2733 		if (version >= 7)
2734 			ceph_decode_8_safe(&p, end, decode_redir, e_inval);
2735 		else
2736 			decode_redir = 1;
2737 	} else {
2738 		decode_redir = 0;
2739 	}
2740 
2741 	if (decode_redir) {
2742 		ret = ceph_redirect_decode(&p, end, &m->redirect);
2743 		if (ret)
2744 			return ret;
2745 	} else {
2746 		ceph_oloc_init(&m->redirect.oloc);
2747 	}
2748 
2749 	return 0;
2750 
2751 e_inval:
2752 	return -EINVAL;
2753 }
2754 
2755 /*
2756  * We are done with @req if
2757  *   - @m is a safe reply, or
2758  *   - @m is an unsafe reply and we didn't want a safe one
2759  */
2760 static bool done_request(const struct ceph_osd_request *req,
2761 			 const struct MOSDOpReply *m)
2762 {
2763 	return (m->result < 0 ||
2764 		(m->flags & CEPH_OSD_FLAG_ONDISK) ||
2765 		!(req->r_flags & CEPH_OSD_FLAG_ONDISK));
2766 }
2767 
2768 /*
2769  * handle osd op reply.  either call the callback if it is specified,
2770  * or do the completion to wake up the waiting thread.
2771  *
2772  * ->r_unsafe_callback is set?	yes			no
2773  *
2774  * first reply is OK (needed	r_cb/r_completion,	r_cb/r_completion,
2775  * any or needed/got safe)	r_safe_completion	r_safe_completion
2776  *
2777  * first reply is unsafe	r_unsafe_cb(true)	(nothing)
2778  *
2779  * when we get the safe reply	r_unsafe_cb(false),	r_cb/r_completion,
2780  *				r_safe_completion	r_safe_completion
2781  */
2782 static void handle_reply(struct ceph_osd *osd, struct ceph_msg *msg)
2783 {
2784 	struct ceph_osd_client *osdc = osd->o_osdc;
2785 	struct ceph_osd_request *req;
2786 	struct MOSDOpReply m;
2787 	u64 tid = le64_to_cpu(msg->hdr.tid);
2788 	u32 data_len = 0;
2789 	bool already_acked;
2790 	int ret;
2791 	int i;
2792 
2793 	dout("%s msg %p tid %llu\n", __func__, msg, tid);
2794 
2795 	down_read(&osdc->lock);
2796 	if (!osd_registered(osd)) {
2797 		dout("%s osd%d unknown\n", __func__, osd->o_osd);
2798 		goto out_unlock_osdc;
2799 	}
2800 	WARN_ON(osd->o_osd != le64_to_cpu(msg->hdr.src.num));
2801 
2802 	mutex_lock(&osd->lock);
2803 	req = lookup_request(&osd->o_requests, tid);
2804 	if (!req) {
2805 		dout("%s osd%d tid %llu unknown\n", __func__, osd->o_osd, tid);
2806 		goto out_unlock_session;
2807 	}
2808 
2809 	ret = decode_MOSDOpReply(msg, &m);
2810 	if (ret) {
2811 		pr_err("failed to decode MOSDOpReply for tid %llu: %d\n",
2812 		       req->r_tid, ret);
2813 		ceph_msg_dump(msg);
2814 		goto fail_request;
2815 	}
2816 	dout("%s req %p tid %llu flags 0x%llx pgid %llu.%x epoch %u attempt %d v %u'%llu uv %llu\n",
2817 	     __func__, req, req->r_tid, m.flags, m.pgid.pool, m.pgid.seed,
2818 	     m.epoch, m.retry_attempt, le32_to_cpu(m.replay_version.epoch),
2819 	     le64_to_cpu(m.replay_version.version), m.user_version);
2820 
2821 	if (m.retry_attempt >= 0) {
2822 		if (m.retry_attempt != req->r_attempts - 1) {
2823 			dout("req %p tid %llu retry_attempt %d != %d, ignoring\n",
2824 			     req, req->r_tid, m.retry_attempt,
2825 			     req->r_attempts - 1);
2826 			goto out_unlock_session;
2827 		}
2828 	} else {
2829 		WARN_ON(1); /* MOSDOpReply v4 is assumed */
2830 	}
2831 
2832 	if (!ceph_oloc_empty(&m.redirect.oloc)) {
2833 		dout("req %p tid %llu redirect pool %lld\n", req, req->r_tid,
2834 		     m.redirect.oloc.pool);
2835 		unlink_request(osd, req);
2836 		mutex_unlock(&osd->lock);
2837 
2838 		ceph_oloc_copy(&req->r_t.target_oloc, &m.redirect.oloc);
2839 		req->r_flags |= CEPH_OSD_FLAG_REDIRECTED;
2840 		req->r_tid = 0;
2841 		__submit_request(req, false);
2842 		goto out_unlock_osdc;
2843 	}
2844 
2845 	if (m.num_ops != req->r_num_ops) {
2846 		pr_err("num_ops %d != %d for tid %llu\n", m.num_ops,
2847 		       req->r_num_ops, req->r_tid);
2848 		goto fail_request;
2849 	}
2850 	for (i = 0; i < req->r_num_ops; i++) {
2851 		dout(" req %p tid %llu op %d rval %d len %u\n", req,
2852 		     req->r_tid, i, m.rval[i], m.outdata_len[i]);
2853 		req->r_ops[i].rval = m.rval[i];
2854 		req->r_ops[i].outdata_len = m.outdata_len[i];
2855 		data_len += m.outdata_len[i];
2856 	}
2857 	if (data_len != le32_to_cpu(msg->hdr.data_len)) {
2858 		pr_err("sum of lens %u != %u for tid %llu\n", data_len,
2859 		       le32_to_cpu(msg->hdr.data_len), req->r_tid);
2860 		goto fail_request;
2861 	}
2862 	dout("%s req %p tid %llu acked %d result %d data_len %u\n", __func__,
2863 	     req, req->r_tid, req->r_got_reply, m.result, data_len);
2864 
2865 	already_acked = req->r_got_reply;
2866 	if (!already_acked) {
2867 		req->r_result = m.result ?: data_len;
2868 		req->r_replay_version = m.replay_version; /* struct */
2869 		req->r_got_reply = true;
2870 	} else if (!(m.flags & CEPH_OSD_FLAG_ONDISK)) {
2871 		dout("req %p tid %llu dup ack\n", req, req->r_tid);
2872 		goto out_unlock_session;
2873 	}
2874 
2875 	if (done_request(req, &m)) {
2876 		__finish_request(req);
2877 		if (req->r_linger) {
2878 			WARN_ON(req->r_unsafe_callback);
2879 			dout("req %p tid %llu cb (locked)\n", req, req->r_tid);
2880 			__complete_request(req);
2881 		}
2882 	}
2883 
2884 	mutex_unlock(&osd->lock);
2885 	up_read(&osdc->lock);
2886 
2887 	if (done_request(req, &m)) {
2888 		if (already_acked && req->r_unsafe_callback) {
2889 			dout("req %p tid %llu safe-cb\n", req, req->r_tid);
2890 			req->r_unsafe_callback(req, false);
2891 		} else if (!req->r_linger) {
2892 			dout("req %p tid %llu cb\n", req, req->r_tid);
2893 			__complete_request(req);
2894 		}
2895 		if (m.flags & CEPH_OSD_FLAG_ONDISK)
2896 			complete_all(&req->r_safe_completion);
2897 		ceph_osdc_put_request(req);
2898 	} else {
2899 		if (req->r_unsafe_callback) {
2900 			dout("req %p tid %llu unsafe-cb\n", req, req->r_tid);
2901 			req->r_unsafe_callback(req, true);
2902 		} else {
2903 			WARN_ON(1);
2904 		}
2905 	}
2906 
2907 	return;
2908 
2909 fail_request:
2910 	complete_request(req, -EIO);
2911 out_unlock_session:
2912 	mutex_unlock(&osd->lock);
2913 out_unlock_osdc:
2914 	up_read(&osdc->lock);
2915 }
2916 
2917 static void set_pool_was_full(struct ceph_osd_client *osdc)
2918 {
2919 	struct rb_node *n;
2920 
2921 	for (n = rb_first(&osdc->osdmap->pg_pools); n; n = rb_next(n)) {
2922 		struct ceph_pg_pool_info *pi =
2923 		    rb_entry(n, struct ceph_pg_pool_info, node);
2924 
2925 		pi->was_full = __pool_full(pi);
2926 	}
2927 }
2928 
2929 static bool pool_cleared_full(struct ceph_osd_client *osdc, s64 pool_id)
2930 {
2931 	struct ceph_pg_pool_info *pi;
2932 
2933 	pi = ceph_pg_pool_by_id(osdc->osdmap, pool_id);
2934 	if (!pi)
2935 		return false;
2936 
2937 	return pi->was_full && !__pool_full(pi);
2938 }
2939 
2940 static enum calc_target_result
2941 recalc_linger_target(struct ceph_osd_linger_request *lreq)
2942 {
2943 	struct ceph_osd_client *osdc = lreq->osdc;
2944 	enum calc_target_result ct_res;
2945 
2946 	ct_res = calc_target(osdc, &lreq->t, &lreq->last_force_resend, true);
2947 	if (ct_res == CALC_TARGET_NEED_RESEND) {
2948 		struct ceph_osd *osd;
2949 
2950 		osd = lookup_create_osd(osdc, lreq->t.osd, true);
2951 		if (osd != lreq->osd) {
2952 			unlink_linger(lreq->osd, lreq);
2953 			link_linger(osd, lreq);
2954 		}
2955 	}
2956 
2957 	return ct_res;
2958 }
2959 
2960 /*
2961  * Requeue requests whose mapping to an OSD has changed.
2962  */
2963 static void scan_requests(struct ceph_osd *osd,
2964 			  bool force_resend,
2965 			  bool cleared_full,
2966 			  bool check_pool_cleared_full,
2967 			  struct rb_root *need_resend,
2968 			  struct list_head *need_resend_linger)
2969 {
2970 	struct ceph_osd_client *osdc = osd->o_osdc;
2971 	struct rb_node *n;
2972 	bool force_resend_writes;
2973 
2974 	for (n = rb_first(&osd->o_linger_requests); n; ) {
2975 		struct ceph_osd_linger_request *lreq =
2976 		    rb_entry(n, struct ceph_osd_linger_request, node);
2977 		enum calc_target_result ct_res;
2978 
2979 		n = rb_next(n); /* recalc_linger_target() */
2980 
2981 		dout("%s lreq %p linger_id %llu\n", __func__, lreq,
2982 		     lreq->linger_id);
2983 		ct_res = recalc_linger_target(lreq);
2984 		switch (ct_res) {
2985 		case CALC_TARGET_NO_ACTION:
2986 			force_resend_writes = cleared_full ||
2987 			    (check_pool_cleared_full &&
2988 			     pool_cleared_full(osdc, lreq->t.base_oloc.pool));
2989 			if (!force_resend && !force_resend_writes)
2990 				break;
2991 
2992 			/* fall through */
2993 		case CALC_TARGET_NEED_RESEND:
2994 			cancel_linger_map_check(lreq);
2995 			/*
2996 			 * scan_requests() for the previous epoch(s)
2997 			 * may have already added it to the list, since
2998 			 * it's not unlinked here.
2999 			 */
3000 			if (list_empty(&lreq->scan_item))
3001 				list_add_tail(&lreq->scan_item, need_resend_linger);
3002 			break;
3003 		case CALC_TARGET_POOL_DNE:
3004 			check_linger_pool_dne(lreq);
3005 			break;
3006 		}
3007 	}
3008 
3009 	for (n = rb_first(&osd->o_requests); n; ) {
3010 		struct ceph_osd_request *req =
3011 		    rb_entry(n, struct ceph_osd_request, r_node);
3012 		enum calc_target_result ct_res;
3013 
3014 		n = rb_next(n); /* unlink_request(), check_pool_dne() */
3015 
3016 		dout("%s req %p tid %llu\n", __func__, req, req->r_tid);
3017 		ct_res = calc_target(osdc, &req->r_t,
3018 				     &req->r_last_force_resend, false);
3019 		switch (ct_res) {
3020 		case CALC_TARGET_NO_ACTION:
3021 			force_resend_writes = cleared_full ||
3022 			    (check_pool_cleared_full &&
3023 			     pool_cleared_full(osdc, req->r_t.base_oloc.pool));
3024 			if (!force_resend &&
3025 			    (!(req->r_flags & CEPH_OSD_FLAG_WRITE) ||
3026 			     !force_resend_writes))
3027 				break;
3028 
3029 			/* fall through */
3030 		case CALC_TARGET_NEED_RESEND:
3031 			cancel_map_check(req);
3032 			unlink_request(osd, req);
3033 			insert_request(need_resend, req);
3034 			break;
3035 		case CALC_TARGET_POOL_DNE:
3036 			check_pool_dne(req);
3037 			break;
3038 		}
3039 	}
3040 }
3041 
3042 static int handle_one_map(struct ceph_osd_client *osdc,
3043 			  void *p, void *end, bool incremental,
3044 			  struct rb_root *need_resend,
3045 			  struct list_head *need_resend_linger)
3046 {
3047 	struct ceph_osdmap *newmap;
3048 	struct rb_node *n;
3049 	bool skipped_map = false;
3050 	bool was_full;
3051 
3052 	was_full = ceph_osdmap_flag(osdc, CEPH_OSDMAP_FULL);
3053 	set_pool_was_full(osdc);
3054 
3055 	if (incremental)
3056 		newmap = osdmap_apply_incremental(&p, end, osdc->osdmap);
3057 	else
3058 		newmap = ceph_osdmap_decode(&p, end);
3059 	if (IS_ERR(newmap))
3060 		return PTR_ERR(newmap);
3061 
3062 	if (newmap != osdc->osdmap) {
3063 		/*
3064 		 * Preserve ->was_full before destroying the old map.
3065 		 * For pools that weren't in the old map, ->was_full
3066 		 * should be false.
3067 		 */
3068 		for (n = rb_first(&newmap->pg_pools); n; n = rb_next(n)) {
3069 			struct ceph_pg_pool_info *pi =
3070 			    rb_entry(n, struct ceph_pg_pool_info, node);
3071 			struct ceph_pg_pool_info *old_pi;
3072 
3073 			old_pi = ceph_pg_pool_by_id(osdc->osdmap, pi->id);
3074 			if (old_pi)
3075 				pi->was_full = old_pi->was_full;
3076 			else
3077 				WARN_ON(pi->was_full);
3078 		}
3079 
3080 		if (osdc->osdmap->epoch &&
3081 		    osdc->osdmap->epoch + 1 < newmap->epoch) {
3082 			WARN_ON(incremental);
3083 			skipped_map = true;
3084 		}
3085 
3086 		ceph_osdmap_destroy(osdc->osdmap);
3087 		osdc->osdmap = newmap;
3088 	}
3089 
3090 	was_full &= !ceph_osdmap_flag(osdc, CEPH_OSDMAP_FULL);
3091 	scan_requests(&osdc->homeless_osd, skipped_map, was_full, true,
3092 		      need_resend, need_resend_linger);
3093 
3094 	for (n = rb_first(&osdc->osds); n; ) {
3095 		struct ceph_osd *osd = rb_entry(n, struct ceph_osd, o_node);
3096 
3097 		n = rb_next(n); /* close_osd() */
3098 
3099 		scan_requests(osd, skipped_map, was_full, true, need_resend,
3100 			      need_resend_linger);
3101 		if (!ceph_osd_is_up(osdc->osdmap, osd->o_osd) ||
3102 		    memcmp(&osd->o_con.peer_addr,
3103 			   ceph_osd_addr(osdc->osdmap, osd->o_osd),
3104 			   sizeof(struct ceph_entity_addr)))
3105 			close_osd(osd);
3106 	}
3107 
3108 	return 0;
3109 }
3110 
3111 static void kick_requests(struct ceph_osd_client *osdc,
3112 			  struct rb_root *need_resend,
3113 			  struct list_head *need_resend_linger)
3114 {
3115 	struct ceph_osd_linger_request *lreq, *nlreq;
3116 	struct rb_node *n;
3117 
3118 	for (n = rb_first(need_resend); n; ) {
3119 		struct ceph_osd_request *req =
3120 		    rb_entry(n, struct ceph_osd_request, r_node);
3121 		struct ceph_osd *osd;
3122 
3123 		n = rb_next(n);
3124 		erase_request(need_resend, req); /* before link_request() */
3125 
3126 		WARN_ON(req->r_osd);
3127 		calc_target(osdc, &req->r_t, NULL, false);
3128 		osd = lookup_create_osd(osdc, req->r_t.osd, true);
3129 		link_request(osd, req);
3130 		if (!req->r_linger) {
3131 			if (!osd_homeless(osd) && !req->r_t.paused)
3132 				send_request(req);
3133 		} else {
3134 			cancel_linger_request(req);
3135 		}
3136 	}
3137 
3138 	list_for_each_entry_safe(lreq, nlreq, need_resend_linger, scan_item) {
3139 		if (!osd_homeless(lreq->osd))
3140 			send_linger(lreq);
3141 
3142 		list_del_init(&lreq->scan_item);
3143 	}
3144 }
3145 
3146 /*
3147  * Process updated osd map.
3148  *
3149  * The message contains any number of incremental and full maps, normally
3150  * indicating some sort of topology change in the cluster.  Kick requests
3151  * off to different OSDs as needed.
3152  */
3153 void ceph_osdc_handle_map(struct ceph_osd_client *osdc, struct ceph_msg *msg)
3154 {
3155 	void *p = msg->front.iov_base;
3156 	void *const end = p + msg->front.iov_len;
3157 	u32 nr_maps, maplen;
3158 	u32 epoch;
3159 	struct ceph_fsid fsid;
3160 	struct rb_root need_resend = RB_ROOT;
3161 	LIST_HEAD(need_resend_linger);
3162 	bool handled_incremental = false;
3163 	bool was_pauserd, was_pausewr;
3164 	bool pauserd, pausewr;
3165 	int err;
3166 
3167 	dout("%s have %u\n", __func__, osdc->osdmap->epoch);
3168 	down_write(&osdc->lock);
3169 
3170 	/* verify fsid */
3171 	ceph_decode_need(&p, end, sizeof(fsid), bad);
3172 	ceph_decode_copy(&p, &fsid, sizeof(fsid));
3173 	if (ceph_check_fsid(osdc->client, &fsid) < 0)
3174 		goto bad;
3175 
3176 	was_pauserd = ceph_osdmap_flag(osdc, CEPH_OSDMAP_PAUSERD);
3177 	was_pausewr = ceph_osdmap_flag(osdc, CEPH_OSDMAP_PAUSEWR) ||
3178 		      ceph_osdmap_flag(osdc, CEPH_OSDMAP_FULL) ||
3179 		      have_pool_full(osdc);
3180 
3181 	/* incremental maps */
3182 	ceph_decode_32_safe(&p, end, nr_maps, bad);
3183 	dout(" %d inc maps\n", nr_maps);
3184 	while (nr_maps > 0) {
3185 		ceph_decode_need(&p, end, 2*sizeof(u32), bad);
3186 		epoch = ceph_decode_32(&p);
3187 		maplen = ceph_decode_32(&p);
3188 		ceph_decode_need(&p, end, maplen, bad);
3189 		if (osdc->osdmap->epoch &&
3190 		    osdc->osdmap->epoch + 1 == epoch) {
3191 			dout("applying incremental map %u len %d\n",
3192 			     epoch, maplen);
3193 			err = handle_one_map(osdc, p, p + maplen, true,
3194 					     &need_resend, &need_resend_linger);
3195 			if (err)
3196 				goto bad;
3197 			handled_incremental = true;
3198 		} else {
3199 			dout("ignoring incremental map %u len %d\n",
3200 			     epoch, maplen);
3201 		}
3202 		p += maplen;
3203 		nr_maps--;
3204 	}
3205 	if (handled_incremental)
3206 		goto done;
3207 
3208 	/* full maps */
3209 	ceph_decode_32_safe(&p, end, nr_maps, bad);
3210 	dout(" %d full maps\n", nr_maps);
3211 	while (nr_maps) {
3212 		ceph_decode_need(&p, end, 2*sizeof(u32), bad);
3213 		epoch = ceph_decode_32(&p);
3214 		maplen = ceph_decode_32(&p);
3215 		ceph_decode_need(&p, end, maplen, bad);
3216 		if (nr_maps > 1) {
3217 			dout("skipping non-latest full map %u len %d\n",
3218 			     epoch, maplen);
3219 		} else if (osdc->osdmap->epoch >= epoch) {
3220 			dout("skipping full map %u len %d, "
3221 			     "older than our %u\n", epoch, maplen,
3222 			     osdc->osdmap->epoch);
3223 		} else {
3224 			dout("taking full map %u len %d\n", epoch, maplen);
3225 			err = handle_one_map(osdc, p, p + maplen, false,
3226 					     &need_resend, &need_resend_linger);
3227 			if (err)
3228 				goto bad;
3229 		}
3230 		p += maplen;
3231 		nr_maps--;
3232 	}
3233 
3234 done:
3235 	/*
3236 	 * subscribe to subsequent osdmap updates if full to ensure
3237 	 * we find out when we are no longer full and stop returning
3238 	 * ENOSPC.
3239 	 */
3240 	pauserd = ceph_osdmap_flag(osdc, CEPH_OSDMAP_PAUSERD);
3241 	pausewr = ceph_osdmap_flag(osdc, CEPH_OSDMAP_PAUSEWR) ||
3242 		  ceph_osdmap_flag(osdc, CEPH_OSDMAP_FULL) ||
3243 		  have_pool_full(osdc);
3244 	if (was_pauserd || was_pausewr || pauserd || pausewr)
3245 		maybe_request_map(osdc);
3246 
3247 	kick_requests(osdc, &need_resend, &need_resend_linger);
3248 
3249 	ceph_monc_got_map(&osdc->client->monc, CEPH_SUB_OSDMAP,
3250 			  osdc->osdmap->epoch);
3251 	up_write(&osdc->lock);
3252 	wake_up_all(&osdc->client->auth_wq);
3253 	return;
3254 
3255 bad:
3256 	pr_err("osdc handle_map corrupt msg\n");
3257 	ceph_msg_dump(msg);
3258 	up_write(&osdc->lock);
3259 }
3260 
3261 /*
3262  * Resubmit requests pending on the given osd.
3263  */
3264 static void kick_osd_requests(struct ceph_osd *osd)
3265 {
3266 	struct rb_node *n;
3267 
3268 	for (n = rb_first(&osd->o_requests); n; ) {
3269 		struct ceph_osd_request *req =
3270 		    rb_entry(n, struct ceph_osd_request, r_node);
3271 
3272 		n = rb_next(n); /* cancel_linger_request() */
3273 
3274 		if (!req->r_linger) {
3275 			if (!req->r_t.paused)
3276 				send_request(req);
3277 		} else {
3278 			cancel_linger_request(req);
3279 		}
3280 	}
3281 	for (n = rb_first(&osd->o_linger_requests); n; n = rb_next(n)) {
3282 		struct ceph_osd_linger_request *lreq =
3283 		    rb_entry(n, struct ceph_osd_linger_request, node);
3284 
3285 		send_linger(lreq);
3286 	}
3287 }
3288 
3289 /*
3290  * If the osd connection drops, we need to resubmit all requests.
3291  */
3292 static void osd_fault(struct ceph_connection *con)
3293 {
3294 	struct ceph_osd *osd = con->private;
3295 	struct ceph_osd_client *osdc = osd->o_osdc;
3296 
3297 	dout("%s osd %p osd%d\n", __func__, osd, osd->o_osd);
3298 
3299 	down_write(&osdc->lock);
3300 	if (!osd_registered(osd)) {
3301 		dout("%s osd%d unknown\n", __func__, osd->o_osd);
3302 		goto out_unlock;
3303 	}
3304 
3305 	if (!reopen_osd(osd))
3306 		kick_osd_requests(osd);
3307 	maybe_request_map(osdc);
3308 
3309 out_unlock:
3310 	up_write(&osdc->lock);
3311 }
3312 
3313 /*
3314  * Process osd watch notifications
3315  */
3316 static void handle_watch_notify(struct ceph_osd_client *osdc,
3317 				struct ceph_msg *msg)
3318 {
3319 	void *p = msg->front.iov_base;
3320 	void *const end = p + msg->front.iov_len;
3321 	struct ceph_osd_linger_request *lreq;
3322 	struct linger_work *lwork;
3323 	u8 proto_ver, opcode;
3324 	u64 cookie, notify_id;
3325 	u64 notifier_id = 0;
3326 	s32 return_code = 0;
3327 	void *payload = NULL;
3328 	u32 payload_len = 0;
3329 
3330 	ceph_decode_8_safe(&p, end, proto_ver, bad);
3331 	ceph_decode_8_safe(&p, end, opcode, bad);
3332 	ceph_decode_64_safe(&p, end, cookie, bad);
3333 	p += 8; /* skip ver */
3334 	ceph_decode_64_safe(&p, end, notify_id, bad);
3335 
3336 	if (proto_ver >= 1) {
3337 		ceph_decode_32_safe(&p, end, payload_len, bad);
3338 		ceph_decode_need(&p, end, payload_len, bad);
3339 		payload = p;
3340 		p += payload_len;
3341 	}
3342 
3343 	if (le16_to_cpu(msg->hdr.version) >= 2)
3344 		ceph_decode_32_safe(&p, end, return_code, bad);
3345 
3346 	if (le16_to_cpu(msg->hdr.version) >= 3)
3347 		ceph_decode_64_safe(&p, end, notifier_id, bad);
3348 
3349 	down_read(&osdc->lock);
3350 	lreq = lookup_linger_osdc(&osdc->linger_requests, cookie);
3351 	if (!lreq) {
3352 		dout("%s opcode %d cookie %llu dne\n", __func__, opcode,
3353 		     cookie);
3354 		goto out_unlock_osdc;
3355 	}
3356 
3357 	mutex_lock(&lreq->lock);
3358 	dout("%s opcode %d cookie %llu lreq %p is_watch %d\n", __func__,
3359 	     opcode, cookie, lreq, lreq->is_watch);
3360 	if (opcode == CEPH_WATCH_EVENT_DISCONNECT) {
3361 		if (!lreq->last_error) {
3362 			lreq->last_error = -ENOTCONN;
3363 			queue_watch_error(lreq);
3364 		}
3365 	} else if (!lreq->is_watch) {
3366 		/* CEPH_WATCH_EVENT_NOTIFY_COMPLETE */
3367 		if (lreq->notify_id && lreq->notify_id != notify_id) {
3368 			dout("lreq %p notify_id %llu != %llu, ignoring\n", lreq,
3369 			     lreq->notify_id, notify_id);
3370 		} else if (!completion_done(&lreq->notify_finish_wait)) {
3371 			struct ceph_msg_data *data =
3372 			    list_first_entry_or_null(&msg->data,
3373 						     struct ceph_msg_data,
3374 						     links);
3375 
3376 			if (data) {
3377 				if (lreq->preply_pages) {
3378 					WARN_ON(data->type !=
3379 							CEPH_MSG_DATA_PAGES);
3380 					*lreq->preply_pages = data->pages;
3381 					*lreq->preply_len = data->length;
3382 				} else {
3383 					ceph_release_page_vector(data->pages,
3384 					       calc_pages_for(0, data->length));
3385 				}
3386 			}
3387 			lreq->notify_finish_error = return_code;
3388 			complete_all(&lreq->notify_finish_wait);
3389 		}
3390 	} else {
3391 		/* CEPH_WATCH_EVENT_NOTIFY */
3392 		lwork = lwork_alloc(lreq, do_watch_notify);
3393 		if (!lwork) {
3394 			pr_err("failed to allocate notify-lwork\n");
3395 			goto out_unlock_lreq;
3396 		}
3397 
3398 		lwork->notify.notify_id = notify_id;
3399 		lwork->notify.notifier_id = notifier_id;
3400 		lwork->notify.payload = payload;
3401 		lwork->notify.payload_len = payload_len;
3402 		lwork->notify.msg = ceph_msg_get(msg);
3403 		lwork_queue(lwork);
3404 	}
3405 
3406 out_unlock_lreq:
3407 	mutex_unlock(&lreq->lock);
3408 out_unlock_osdc:
3409 	up_read(&osdc->lock);
3410 	return;
3411 
3412 bad:
3413 	pr_err("osdc handle_watch_notify corrupt msg\n");
3414 }
3415 
3416 /*
3417  * Register request, send initial attempt.
3418  */
3419 int ceph_osdc_start_request(struct ceph_osd_client *osdc,
3420 			    struct ceph_osd_request *req,
3421 			    bool nofail)
3422 {
3423 	down_read(&osdc->lock);
3424 	submit_request(req, false);
3425 	up_read(&osdc->lock);
3426 
3427 	return 0;
3428 }
3429 EXPORT_SYMBOL(ceph_osdc_start_request);
3430 
3431 /*
3432  * Unregister a registered request.  The request is not completed (i.e.
3433  * no callbacks or wakeups) - higher layers are supposed to know what
3434  * they are canceling.
3435  */
3436 void ceph_osdc_cancel_request(struct ceph_osd_request *req)
3437 {
3438 	struct ceph_osd_client *osdc = req->r_osdc;
3439 
3440 	down_write(&osdc->lock);
3441 	if (req->r_osd)
3442 		cancel_request(req);
3443 	up_write(&osdc->lock);
3444 }
3445 EXPORT_SYMBOL(ceph_osdc_cancel_request);
3446 
3447 /*
3448  * @timeout: in jiffies, 0 means "wait forever"
3449  */
3450 static int wait_request_timeout(struct ceph_osd_request *req,
3451 				unsigned long timeout)
3452 {
3453 	long left;
3454 
3455 	dout("%s req %p tid %llu\n", __func__, req, req->r_tid);
3456 	left = wait_for_completion_killable_timeout(&req->r_completion,
3457 						ceph_timeout_jiffies(timeout));
3458 	if (left <= 0) {
3459 		left = left ?: -ETIMEDOUT;
3460 		ceph_osdc_cancel_request(req);
3461 
3462 		/* kludge - need to to wake ceph_osdc_sync() */
3463 		complete_all(&req->r_safe_completion);
3464 	} else {
3465 		left = req->r_result; /* completed */
3466 	}
3467 
3468 	return left;
3469 }
3470 
3471 /*
3472  * wait for a request to complete
3473  */
3474 int ceph_osdc_wait_request(struct ceph_osd_client *osdc,
3475 			   struct ceph_osd_request *req)
3476 {
3477 	return wait_request_timeout(req, 0);
3478 }
3479 EXPORT_SYMBOL(ceph_osdc_wait_request);
3480 
3481 /*
3482  * sync - wait for all in-flight requests to flush.  avoid starvation.
3483  */
3484 void ceph_osdc_sync(struct ceph_osd_client *osdc)
3485 {
3486 	struct rb_node *n, *p;
3487 	u64 last_tid = atomic64_read(&osdc->last_tid);
3488 
3489 again:
3490 	down_read(&osdc->lock);
3491 	for (n = rb_first(&osdc->osds); n; n = rb_next(n)) {
3492 		struct ceph_osd *osd = rb_entry(n, struct ceph_osd, o_node);
3493 
3494 		mutex_lock(&osd->lock);
3495 		for (p = rb_first(&osd->o_requests); p; p = rb_next(p)) {
3496 			struct ceph_osd_request *req =
3497 			    rb_entry(p, struct ceph_osd_request, r_node);
3498 
3499 			if (req->r_tid > last_tid)
3500 				break;
3501 
3502 			if (!(req->r_flags & CEPH_OSD_FLAG_WRITE))
3503 				continue;
3504 
3505 			ceph_osdc_get_request(req);
3506 			mutex_unlock(&osd->lock);
3507 			up_read(&osdc->lock);
3508 			dout("%s waiting on req %p tid %llu last_tid %llu\n",
3509 			     __func__, req, req->r_tid, last_tid);
3510 			wait_for_completion(&req->r_safe_completion);
3511 			ceph_osdc_put_request(req);
3512 			goto again;
3513 		}
3514 
3515 		mutex_unlock(&osd->lock);
3516 	}
3517 
3518 	up_read(&osdc->lock);
3519 	dout("%s done last_tid %llu\n", __func__, last_tid);
3520 }
3521 EXPORT_SYMBOL(ceph_osdc_sync);
3522 
3523 static struct ceph_osd_request *
3524 alloc_linger_request(struct ceph_osd_linger_request *lreq)
3525 {
3526 	struct ceph_osd_request *req;
3527 
3528 	req = ceph_osdc_alloc_request(lreq->osdc, NULL, 1, false, GFP_NOIO);
3529 	if (!req)
3530 		return NULL;
3531 
3532 	ceph_oid_copy(&req->r_base_oid, &lreq->t.base_oid);
3533 	ceph_oloc_copy(&req->r_base_oloc, &lreq->t.base_oloc);
3534 
3535 	if (ceph_osdc_alloc_messages(req, GFP_NOIO)) {
3536 		ceph_osdc_put_request(req);
3537 		return NULL;
3538 	}
3539 
3540 	return req;
3541 }
3542 
3543 /*
3544  * Returns a handle, caller owns a ref.
3545  */
3546 struct ceph_osd_linger_request *
3547 ceph_osdc_watch(struct ceph_osd_client *osdc,
3548 		struct ceph_object_id *oid,
3549 		struct ceph_object_locator *oloc,
3550 		rados_watchcb2_t wcb,
3551 		rados_watcherrcb_t errcb,
3552 		void *data)
3553 {
3554 	struct ceph_osd_linger_request *lreq;
3555 	int ret;
3556 
3557 	lreq = linger_alloc(osdc);
3558 	if (!lreq)
3559 		return ERR_PTR(-ENOMEM);
3560 
3561 	lreq->is_watch = true;
3562 	lreq->wcb = wcb;
3563 	lreq->errcb = errcb;
3564 	lreq->data = data;
3565 	lreq->watch_valid_thru = jiffies;
3566 
3567 	ceph_oid_copy(&lreq->t.base_oid, oid);
3568 	ceph_oloc_copy(&lreq->t.base_oloc, oloc);
3569 	lreq->t.flags = CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK;
3570 	lreq->mtime = CURRENT_TIME;
3571 
3572 	lreq->reg_req = alloc_linger_request(lreq);
3573 	if (!lreq->reg_req) {
3574 		ret = -ENOMEM;
3575 		goto err_put_lreq;
3576 	}
3577 
3578 	lreq->ping_req = alloc_linger_request(lreq);
3579 	if (!lreq->ping_req) {
3580 		ret = -ENOMEM;
3581 		goto err_put_lreq;
3582 	}
3583 
3584 	down_write(&osdc->lock);
3585 	linger_register(lreq); /* before osd_req_op_* */
3586 	osd_req_op_watch_init(lreq->reg_req, 0, lreq->linger_id,
3587 			      CEPH_OSD_WATCH_OP_WATCH);
3588 	osd_req_op_watch_init(lreq->ping_req, 0, lreq->linger_id,
3589 			      CEPH_OSD_WATCH_OP_PING);
3590 	linger_submit(lreq);
3591 	up_write(&osdc->lock);
3592 
3593 	ret = linger_reg_commit_wait(lreq);
3594 	if (ret) {
3595 		linger_cancel(lreq);
3596 		goto err_put_lreq;
3597 	}
3598 
3599 	return lreq;
3600 
3601 err_put_lreq:
3602 	linger_put(lreq);
3603 	return ERR_PTR(ret);
3604 }
3605 EXPORT_SYMBOL(ceph_osdc_watch);
3606 
3607 /*
3608  * Releases a ref.
3609  *
3610  * Times out after mount_timeout to preserve rbd unmap behaviour
3611  * introduced in 2894e1d76974 ("rbd: timeout watch teardown on unmap
3612  * with mount_timeout").
3613  */
3614 int ceph_osdc_unwatch(struct ceph_osd_client *osdc,
3615 		      struct ceph_osd_linger_request *lreq)
3616 {
3617 	struct ceph_options *opts = osdc->client->options;
3618 	struct ceph_osd_request *req;
3619 	int ret;
3620 
3621 	req = ceph_osdc_alloc_request(osdc, NULL, 1, false, GFP_NOIO);
3622 	if (!req)
3623 		return -ENOMEM;
3624 
3625 	ceph_oid_copy(&req->r_base_oid, &lreq->t.base_oid);
3626 	ceph_oloc_copy(&req->r_base_oloc, &lreq->t.base_oloc);
3627 	req->r_flags = CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK;
3628 	req->r_mtime = CURRENT_TIME;
3629 	osd_req_op_watch_init(req, 0, lreq->linger_id,
3630 			      CEPH_OSD_WATCH_OP_UNWATCH);
3631 
3632 	ret = ceph_osdc_alloc_messages(req, GFP_NOIO);
3633 	if (ret)
3634 		goto out_put_req;
3635 
3636 	ceph_osdc_start_request(osdc, req, false);
3637 	linger_cancel(lreq);
3638 	linger_put(lreq);
3639 	ret = wait_request_timeout(req, opts->mount_timeout);
3640 
3641 out_put_req:
3642 	ceph_osdc_put_request(req);
3643 	return ret;
3644 }
3645 EXPORT_SYMBOL(ceph_osdc_unwatch);
3646 
3647 static int osd_req_op_notify_ack_init(struct ceph_osd_request *req, int which,
3648 				      u64 notify_id, u64 cookie, void *payload,
3649 				      size_t payload_len)
3650 {
3651 	struct ceph_osd_req_op *op;
3652 	struct ceph_pagelist *pl;
3653 	int ret;
3654 
3655 	op = _osd_req_op_init(req, which, CEPH_OSD_OP_NOTIFY_ACK, 0);
3656 
3657 	pl = kmalloc(sizeof(*pl), GFP_NOIO);
3658 	if (!pl)
3659 		return -ENOMEM;
3660 
3661 	ceph_pagelist_init(pl);
3662 	ret = ceph_pagelist_encode_64(pl, notify_id);
3663 	ret |= ceph_pagelist_encode_64(pl, cookie);
3664 	if (payload) {
3665 		ret |= ceph_pagelist_encode_32(pl, payload_len);
3666 		ret |= ceph_pagelist_append(pl, payload, payload_len);
3667 	} else {
3668 		ret |= ceph_pagelist_encode_32(pl, 0);
3669 	}
3670 	if (ret) {
3671 		ceph_pagelist_release(pl);
3672 		return -ENOMEM;
3673 	}
3674 
3675 	ceph_osd_data_pagelist_init(&op->notify_ack.request_data, pl);
3676 	op->indata_len = pl->length;
3677 	return 0;
3678 }
3679 
3680 int ceph_osdc_notify_ack(struct ceph_osd_client *osdc,
3681 			 struct ceph_object_id *oid,
3682 			 struct ceph_object_locator *oloc,
3683 			 u64 notify_id,
3684 			 u64 cookie,
3685 			 void *payload,
3686 			 size_t payload_len)
3687 {
3688 	struct ceph_osd_request *req;
3689 	int ret;
3690 
3691 	req = ceph_osdc_alloc_request(osdc, NULL, 1, false, GFP_NOIO);
3692 	if (!req)
3693 		return -ENOMEM;
3694 
3695 	ceph_oid_copy(&req->r_base_oid, oid);
3696 	ceph_oloc_copy(&req->r_base_oloc, oloc);
3697 	req->r_flags = CEPH_OSD_FLAG_READ;
3698 
3699 	ret = ceph_osdc_alloc_messages(req, GFP_NOIO);
3700 	if (ret)
3701 		goto out_put_req;
3702 
3703 	ret = osd_req_op_notify_ack_init(req, 0, notify_id, cookie, payload,
3704 					 payload_len);
3705 	if (ret)
3706 		goto out_put_req;
3707 
3708 	ceph_osdc_start_request(osdc, req, false);
3709 	ret = ceph_osdc_wait_request(osdc, req);
3710 
3711 out_put_req:
3712 	ceph_osdc_put_request(req);
3713 	return ret;
3714 }
3715 EXPORT_SYMBOL(ceph_osdc_notify_ack);
3716 
3717 static int osd_req_op_notify_init(struct ceph_osd_request *req, int which,
3718 				  u64 cookie, u32 prot_ver, u32 timeout,
3719 				  void *payload, size_t payload_len)
3720 {
3721 	struct ceph_osd_req_op *op;
3722 	struct ceph_pagelist *pl;
3723 	int ret;
3724 
3725 	op = _osd_req_op_init(req, which, CEPH_OSD_OP_NOTIFY, 0);
3726 	op->notify.cookie = cookie;
3727 
3728 	pl = kmalloc(sizeof(*pl), GFP_NOIO);
3729 	if (!pl)
3730 		return -ENOMEM;
3731 
3732 	ceph_pagelist_init(pl);
3733 	ret = ceph_pagelist_encode_32(pl, 1); /* prot_ver */
3734 	ret |= ceph_pagelist_encode_32(pl, timeout);
3735 	ret |= ceph_pagelist_encode_32(pl, payload_len);
3736 	ret |= ceph_pagelist_append(pl, payload, payload_len);
3737 	if (ret) {
3738 		ceph_pagelist_release(pl);
3739 		return -ENOMEM;
3740 	}
3741 
3742 	ceph_osd_data_pagelist_init(&op->notify.request_data, pl);
3743 	op->indata_len = pl->length;
3744 	return 0;
3745 }
3746 
3747 /*
3748  * @timeout: in seconds
3749  *
3750  * @preply_{pages,len} are initialized both on success and error.
3751  * The caller is responsible for:
3752  *
3753  *     ceph_release_page_vector(reply_pages, calc_pages_for(0, reply_len))
3754  */
3755 int ceph_osdc_notify(struct ceph_osd_client *osdc,
3756 		     struct ceph_object_id *oid,
3757 		     struct ceph_object_locator *oloc,
3758 		     void *payload,
3759 		     size_t payload_len,
3760 		     u32 timeout,
3761 		     struct page ***preply_pages,
3762 		     size_t *preply_len)
3763 {
3764 	struct ceph_osd_linger_request *lreq;
3765 	struct page **pages;
3766 	int ret;
3767 
3768 	WARN_ON(!timeout);
3769 	if (preply_pages) {
3770 		*preply_pages = NULL;
3771 		*preply_len = 0;
3772 	}
3773 
3774 	lreq = linger_alloc(osdc);
3775 	if (!lreq)
3776 		return -ENOMEM;
3777 
3778 	lreq->preply_pages = preply_pages;
3779 	lreq->preply_len = preply_len;
3780 
3781 	ceph_oid_copy(&lreq->t.base_oid, oid);
3782 	ceph_oloc_copy(&lreq->t.base_oloc, oloc);
3783 	lreq->t.flags = CEPH_OSD_FLAG_READ;
3784 
3785 	lreq->reg_req = alloc_linger_request(lreq);
3786 	if (!lreq->reg_req) {
3787 		ret = -ENOMEM;
3788 		goto out_put_lreq;
3789 	}
3790 
3791 	/* for notify_id */
3792 	pages = ceph_alloc_page_vector(1, GFP_NOIO);
3793 	if (IS_ERR(pages)) {
3794 		ret = PTR_ERR(pages);
3795 		goto out_put_lreq;
3796 	}
3797 
3798 	down_write(&osdc->lock);
3799 	linger_register(lreq); /* before osd_req_op_* */
3800 	ret = osd_req_op_notify_init(lreq->reg_req, 0, lreq->linger_id, 1,
3801 				     timeout, payload, payload_len);
3802 	if (ret) {
3803 		linger_unregister(lreq);
3804 		up_write(&osdc->lock);
3805 		ceph_release_page_vector(pages, 1);
3806 		goto out_put_lreq;
3807 	}
3808 	ceph_osd_data_pages_init(osd_req_op_data(lreq->reg_req, 0, notify,
3809 						 response_data),
3810 				 pages, PAGE_SIZE, 0, false, true);
3811 	linger_submit(lreq);
3812 	up_write(&osdc->lock);
3813 
3814 	ret = linger_reg_commit_wait(lreq);
3815 	if (!ret)
3816 		ret = linger_notify_finish_wait(lreq);
3817 	else
3818 		dout("lreq %p failed to initiate notify %d\n", lreq, ret);
3819 
3820 	linger_cancel(lreq);
3821 out_put_lreq:
3822 	linger_put(lreq);
3823 	return ret;
3824 }
3825 EXPORT_SYMBOL(ceph_osdc_notify);
3826 
3827 /*
3828  * Return the number of milliseconds since the watch was last
3829  * confirmed, or an error.  If there is an error, the watch is no
3830  * longer valid, and should be destroyed with ceph_osdc_unwatch().
3831  */
3832 int ceph_osdc_watch_check(struct ceph_osd_client *osdc,
3833 			  struct ceph_osd_linger_request *lreq)
3834 {
3835 	unsigned long stamp, age;
3836 	int ret;
3837 
3838 	down_read(&osdc->lock);
3839 	mutex_lock(&lreq->lock);
3840 	stamp = lreq->watch_valid_thru;
3841 	if (!list_empty(&lreq->pending_lworks)) {
3842 		struct linger_work *lwork =
3843 		    list_first_entry(&lreq->pending_lworks,
3844 				     struct linger_work,
3845 				     pending_item);
3846 
3847 		if (time_before(lwork->queued_stamp, stamp))
3848 			stamp = lwork->queued_stamp;
3849 	}
3850 	age = jiffies - stamp;
3851 	dout("%s lreq %p linger_id %llu age %lu last_error %d\n", __func__,
3852 	     lreq, lreq->linger_id, age, lreq->last_error);
3853 	/* we are truncating to msecs, so return a safe upper bound */
3854 	ret = lreq->last_error ?: 1 + jiffies_to_msecs(age);
3855 
3856 	mutex_unlock(&lreq->lock);
3857 	up_read(&osdc->lock);
3858 	return ret;
3859 }
3860 
3861 /*
3862  * Call all pending notify callbacks - for use after a watch is
3863  * unregistered, to make sure no more callbacks for it will be invoked
3864  */
3865 void ceph_osdc_flush_notifies(struct ceph_osd_client *osdc)
3866 {
3867 	flush_workqueue(osdc->notify_wq);
3868 }
3869 EXPORT_SYMBOL(ceph_osdc_flush_notifies);
3870 
3871 void ceph_osdc_maybe_request_map(struct ceph_osd_client *osdc)
3872 {
3873 	down_read(&osdc->lock);
3874 	maybe_request_map(osdc);
3875 	up_read(&osdc->lock);
3876 }
3877 EXPORT_SYMBOL(ceph_osdc_maybe_request_map);
3878 
3879 /*
3880  * init, shutdown
3881  */
3882 int ceph_osdc_init(struct ceph_osd_client *osdc, struct ceph_client *client)
3883 {
3884 	int err;
3885 
3886 	dout("init\n");
3887 	osdc->client = client;
3888 	init_rwsem(&osdc->lock);
3889 	osdc->osds = RB_ROOT;
3890 	INIT_LIST_HEAD(&osdc->osd_lru);
3891 	spin_lock_init(&osdc->osd_lru_lock);
3892 	osd_init(&osdc->homeless_osd);
3893 	osdc->homeless_osd.o_osdc = osdc;
3894 	osdc->homeless_osd.o_osd = CEPH_HOMELESS_OSD;
3895 	osdc->linger_requests = RB_ROOT;
3896 	osdc->map_checks = RB_ROOT;
3897 	osdc->linger_map_checks = RB_ROOT;
3898 	INIT_DELAYED_WORK(&osdc->timeout_work, handle_timeout);
3899 	INIT_DELAYED_WORK(&osdc->osds_timeout_work, handle_osds_timeout);
3900 
3901 	err = -ENOMEM;
3902 	osdc->osdmap = ceph_osdmap_alloc();
3903 	if (!osdc->osdmap)
3904 		goto out;
3905 
3906 	osdc->req_mempool = mempool_create_slab_pool(10,
3907 						     ceph_osd_request_cache);
3908 	if (!osdc->req_mempool)
3909 		goto out_map;
3910 
3911 	err = ceph_msgpool_init(&osdc->msgpool_op, CEPH_MSG_OSD_OP,
3912 				PAGE_SIZE, 10, true, "osd_op");
3913 	if (err < 0)
3914 		goto out_mempool;
3915 	err = ceph_msgpool_init(&osdc->msgpool_op_reply, CEPH_MSG_OSD_OPREPLY,
3916 				PAGE_SIZE, 10, true, "osd_op_reply");
3917 	if (err < 0)
3918 		goto out_msgpool;
3919 
3920 	err = -ENOMEM;
3921 	osdc->notify_wq = create_singlethread_workqueue("ceph-watch-notify");
3922 	if (!osdc->notify_wq)
3923 		goto out_msgpool_reply;
3924 
3925 	schedule_delayed_work(&osdc->timeout_work,
3926 			      osdc->client->options->osd_keepalive_timeout);
3927 	schedule_delayed_work(&osdc->osds_timeout_work,
3928 	    round_jiffies_relative(osdc->client->options->osd_idle_ttl));
3929 
3930 	return 0;
3931 
3932 out_msgpool_reply:
3933 	ceph_msgpool_destroy(&osdc->msgpool_op_reply);
3934 out_msgpool:
3935 	ceph_msgpool_destroy(&osdc->msgpool_op);
3936 out_mempool:
3937 	mempool_destroy(osdc->req_mempool);
3938 out_map:
3939 	ceph_osdmap_destroy(osdc->osdmap);
3940 out:
3941 	return err;
3942 }
3943 
3944 void ceph_osdc_stop(struct ceph_osd_client *osdc)
3945 {
3946 	flush_workqueue(osdc->notify_wq);
3947 	destroy_workqueue(osdc->notify_wq);
3948 	cancel_delayed_work_sync(&osdc->timeout_work);
3949 	cancel_delayed_work_sync(&osdc->osds_timeout_work);
3950 
3951 	down_write(&osdc->lock);
3952 	while (!RB_EMPTY_ROOT(&osdc->osds)) {
3953 		struct ceph_osd *osd = rb_entry(rb_first(&osdc->osds),
3954 						struct ceph_osd, o_node);
3955 		close_osd(osd);
3956 	}
3957 	up_write(&osdc->lock);
3958 	WARN_ON(atomic_read(&osdc->homeless_osd.o_ref) != 1);
3959 	osd_cleanup(&osdc->homeless_osd);
3960 
3961 	WARN_ON(!list_empty(&osdc->osd_lru));
3962 	WARN_ON(!RB_EMPTY_ROOT(&osdc->linger_requests));
3963 	WARN_ON(!RB_EMPTY_ROOT(&osdc->map_checks));
3964 	WARN_ON(!RB_EMPTY_ROOT(&osdc->linger_map_checks));
3965 	WARN_ON(atomic_read(&osdc->num_requests));
3966 	WARN_ON(atomic_read(&osdc->num_homeless));
3967 
3968 	ceph_osdmap_destroy(osdc->osdmap);
3969 	mempool_destroy(osdc->req_mempool);
3970 	ceph_msgpool_destroy(&osdc->msgpool_op);
3971 	ceph_msgpool_destroy(&osdc->msgpool_op_reply);
3972 }
3973 
3974 /*
3975  * Read some contiguous pages.  If we cross a stripe boundary, shorten
3976  * *plen.  Return number of bytes read, or error.
3977  */
3978 int ceph_osdc_readpages(struct ceph_osd_client *osdc,
3979 			struct ceph_vino vino, struct ceph_file_layout *layout,
3980 			u64 off, u64 *plen,
3981 			u32 truncate_seq, u64 truncate_size,
3982 			struct page **pages, int num_pages, int page_align)
3983 {
3984 	struct ceph_osd_request *req;
3985 	int rc = 0;
3986 
3987 	dout("readpages on ino %llx.%llx on %llu~%llu\n", vino.ino,
3988 	     vino.snap, off, *plen);
3989 	req = ceph_osdc_new_request(osdc, layout, vino, off, plen, 0, 1,
3990 				    CEPH_OSD_OP_READ, CEPH_OSD_FLAG_READ,
3991 				    NULL, truncate_seq, truncate_size,
3992 				    false);
3993 	if (IS_ERR(req))
3994 		return PTR_ERR(req);
3995 
3996 	/* it may be a short read due to an object boundary */
3997 	osd_req_op_extent_osd_data_pages(req, 0,
3998 				pages, *plen, page_align, false, false);
3999 
4000 	dout("readpages  final extent is %llu~%llu (%llu bytes align %d)\n",
4001 	     off, *plen, *plen, page_align);
4002 
4003 	rc = ceph_osdc_start_request(osdc, req, false);
4004 	if (!rc)
4005 		rc = ceph_osdc_wait_request(osdc, req);
4006 
4007 	ceph_osdc_put_request(req);
4008 	dout("readpages result %d\n", rc);
4009 	return rc;
4010 }
4011 EXPORT_SYMBOL(ceph_osdc_readpages);
4012 
4013 /*
4014  * do a synchronous write on N pages
4015  */
4016 int ceph_osdc_writepages(struct ceph_osd_client *osdc, struct ceph_vino vino,
4017 			 struct ceph_file_layout *layout,
4018 			 struct ceph_snap_context *snapc,
4019 			 u64 off, u64 len,
4020 			 u32 truncate_seq, u64 truncate_size,
4021 			 struct timespec *mtime,
4022 			 struct page **pages, int num_pages)
4023 {
4024 	struct ceph_osd_request *req;
4025 	int rc = 0;
4026 	int page_align = off & ~PAGE_MASK;
4027 
4028 	req = ceph_osdc_new_request(osdc, layout, vino, off, &len, 0, 1,
4029 				    CEPH_OSD_OP_WRITE,
4030 				    CEPH_OSD_FLAG_ONDISK | CEPH_OSD_FLAG_WRITE,
4031 				    snapc, truncate_seq, truncate_size,
4032 				    true);
4033 	if (IS_ERR(req))
4034 		return PTR_ERR(req);
4035 
4036 	/* it may be a short write due to an object boundary */
4037 	osd_req_op_extent_osd_data_pages(req, 0, pages, len, page_align,
4038 				false, false);
4039 	dout("writepages %llu~%llu (%llu bytes)\n", off, len, len);
4040 
4041 	req->r_mtime = *mtime;
4042 	rc = ceph_osdc_start_request(osdc, req, true);
4043 	if (!rc)
4044 		rc = ceph_osdc_wait_request(osdc, req);
4045 
4046 	ceph_osdc_put_request(req);
4047 	if (rc == 0)
4048 		rc = len;
4049 	dout("writepages result %d\n", rc);
4050 	return rc;
4051 }
4052 EXPORT_SYMBOL(ceph_osdc_writepages);
4053 
4054 int ceph_osdc_setup(void)
4055 {
4056 	size_t size = sizeof(struct ceph_osd_request) +
4057 	    CEPH_OSD_SLAB_OPS * sizeof(struct ceph_osd_req_op);
4058 
4059 	BUG_ON(ceph_osd_request_cache);
4060 	ceph_osd_request_cache = kmem_cache_create("ceph_osd_request", size,
4061 						   0, 0, NULL);
4062 
4063 	return ceph_osd_request_cache ? 0 : -ENOMEM;
4064 }
4065 EXPORT_SYMBOL(ceph_osdc_setup);
4066 
4067 void ceph_osdc_cleanup(void)
4068 {
4069 	BUG_ON(!ceph_osd_request_cache);
4070 	kmem_cache_destroy(ceph_osd_request_cache);
4071 	ceph_osd_request_cache = NULL;
4072 }
4073 EXPORT_SYMBOL(ceph_osdc_cleanup);
4074 
4075 /*
4076  * handle incoming message
4077  */
4078 static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
4079 {
4080 	struct ceph_osd *osd = con->private;
4081 	struct ceph_osd_client *osdc = osd->o_osdc;
4082 	int type = le16_to_cpu(msg->hdr.type);
4083 
4084 	switch (type) {
4085 	case CEPH_MSG_OSD_MAP:
4086 		ceph_osdc_handle_map(osdc, msg);
4087 		break;
4088 	case CEPH_MSG_OSD_OPREPLY:
4089 		handle_reply(osd, msg);
4090 		break;
4091 	case CEPH_MSG_WATCH_NOTIFY:
4092 		handle_watch_notify(osdc, msg);
4093 		break;
4094 
4095 	default:
4096 		pr_err("received unknown message type %d %s\n", type,
4097 		       ceph_msg_type_name(type));
4098 	}
4099 
4100 	ceph_msg_put(msg);
4101 }
4102 
4103 /*
4104  * Lookup and return message for incoming reply.  Don't try to do
4105  * anything about a larger than preallocated data portion of the
4106  * message at the moment - for now, just skip the message.
4107  */
4108 static struct ceph_msg *get_reply(struct ceph_connection *con,
4109 				  struct ceph_msg_header *hdr,
4110 				  int *skip)
4111 {
4112 	struct ceph_osd *osd = con->private;
4113 	struct ceph_osd_client *osdc = osd->o_osdc;
4114 	struct ceph_msg *m = NULL;
4115 	struct ceph_osd_request *req;
4116 	int front_len = le32_to_cpu(hdr->front_len);
4117 	int data_len = le32_to_cpu(hdr->data_len);
4118 	u64 tid = le64_to_cpu(hdr->tid);
4119 
4120 	down_read(&osdc->lock);
4121 	if (!osd_registered(osd)) {
4122 		dout("%s osd%d unknown, skipping\n", __func__, osd->o_osd);
4123 		*skip = 1;
4124 		goto out_unlock_osdc;
4125 	}
4126 	WARN_ON(osd->o_osd != le64_to_cpu(hdr->src.num));
4127 
4128 	mutex_lock(&osd->lock);
4129 	req = lookup_request(&osd->o_requests, tid);
4130 	if (!req) {
4131 		dout("%s osd%d tid %llu unknown, skipping\n", __func__,
4132 		     osd->o_osd, tid);
4133 		*skip = 1;
4134 		goto out_unlock_session;
4135 	}
4136 
4137 	ceph_msg_revoke_incoming(req->r_reply);
4138 
4139 	if (front_len > req->r_reply->front_alloc_len) {
4140 		pr_warn("%s osd%d tid %llu front %d > preallocated %d\n",
4141 			__func__, osd->o_osd, req->r_tid, front_len,
4142 			req->r_reply->front_alloc_len);
4143 		m = ceph_msg_new(CEPH_MSG_OSD_OPREPLY, front_len, GFP_NOFS,
4144 				 false);
4145 		if (!m)
4146 			goto out_unlock_session;
4147 		ceph_msg_put(req->r_reply);
4148 		req->r_reply = m;
4149 	}
4150 
4151 	if (data_len > req->r_reply->data_length) {
4152 		pr_warn("%s osd%d tid %llu data %d > preallocated %zu, skipping\n",
4153 			__func__, osd->o_osd, req->r_tid, data_len,
4154 			req->r_reply->data_length);
4155 		m = NULL;
4156 		*skip = 1;
4157 		goto out_unlock_session;
4158 	}
4159 
4160 	m = ceph_msg_get(req->r_reply);
4161 	dout("get_reply tid %lld %p\n", tid, m);
4162 
4163 out_unlock_session:
4164 	mutex_unlock(&osd->lock);
4165 out_unlock_osdc:
4166 	up_read(&osdc->lock);
4167 	return m;
4168 }
4169 
4170 /*
4171  * TODO: switch to a msg-owned pagelist
4172  */
4173 static struct ceph_msg *alloc_msg_with_page_vector(struct ceph_msg_header *hdr)
4174 {
4175 	struct ceph_msg *m;
4176 	int type = le16_to_cpu(hdr->type);
4177 	u32 front_len = le32_to_cpu(hdr->front_len);
4178 	u32 data_len = le32_to_cpu(hdr->data_len);
4179 
4180 	m = ceph_msg_new(type, front_len, GFP_NOIO, false);
4181 	if (!m)
4182 		return NULL;
4183 
4184 	if (data_len) {
4185 		struct page **pages;
4186 		struct ceph_osd_data osd_data;
4187 
4188 		pages = ceph_alloc_page_vector(calc_pages_for(0, data_len),
4189 					       GFP_NOIO);
4190 		if (!pages) {
4191 			ceph_msg_put(m);
4192 			return NULL;
4193 		}
4194 
4195 		ceph_osd_data_pages_init(&osd_data, pages, data_len, 0, false,
4196 					 false);
4197 		ceph_osdc_msg_data_add(m, &osd_data);
4198 	}
4199 
4200 	return m;
4201 }
4202 
4203 static struct ceph_msg *alloc_msg(struct ceph_connection *con,
4204 				  struct ceph_msg_header *hdr,
4205 				  int *skip)
4206 {
4207 	struct ceph_osd *osd = con->private;
4208 	int type = le16_to_cpu(hdr->type);
4209 
4210 	*skip = 0;
4211 	switch (type) {
4212 	case CEPH_MSG_OSD_MAP:
4213 	case CEPH_MSG_WATCH_NOTIFY:
4214 		return alloc_msg_with_page_vector(hdr);
4215 	case CEPH_MSG_OSD_OPREPLY:
4216 		return get_reply(con, hdr, skip);
4217 	default:
4218 		pr_warn("%s osd%d unknown msg type %d, skipping\n", __func__,
4219 			osd->o_osd, type);
4220 		*skip = 1;
4221 		return NULL;
4222 	}
4223 }
4224 
4225 /*
4226  * Wrappers to refcount containing ceph_osd struct
4227  */
4228 static struct ceph_connection *get_osd_con(struct ceph_connection *con)
4229 {
4230 	struct ceph_osd *osd = con->private;
4231 	if (get_osd(osd))
4232 		return con;
4233 	return NULL;
4234 }
4235 
4236 static void put_osd_con(struct ceph_connection *con)
4237 {
4238 	struct ceph_osd *osd = con->private;
4239 	put_osd(osd);
4240 }
4241 
4242 /*
4243  * authentication
4244  */
4245 /*
4246  * Note: returned pointer is the address of a structure that's
4247  * managed separately.  Caller must *not* attempt to free it.
4248  */
4249 static struct ceph_auth_handshake *get_authorizer(struct ceph_connection *con,
4250 					int *proto, int force_new)
4251 {
4252 	struct ceph_osd *o = con->private;
4253 	struct ceph_osd_client *osdc = o->o_osdc;
4254 	struct ceph_auth_client *ac = osdc->client->monc.auth;
4255 	struct ceph_auth_handshake *auth = &o->o_auth;
4256 
4257 	if (force_new && auth->authorizer) {
4258 		ceph_auth_destroy_authorizer(auth->authorizer);
4259 		auth->authorizer = NULL;
4260 	}
4261 	if (!auth->authorizer) {
4262 		int ret = ceph_auth_create_authorizer(ac, CEPH_ENTITY_TYPE_OSD,
4263 						      auth);
4264 		if (ret)
4265 			return ERR_PTR(ret);
4266 	} else {
4267 		int ret = ceph_auth_update_authorizer(ac, CEPH_ENTITY_TYPE_OSD,
4268 						     auth);
4269 		if (ret)
4270 			return ERR_PTR(ret);
4271 	}
4272 	*proto = ac->protocol;
4273 
4274 	return auth;
4275 }
4276 
4277 
4278 static int verify_authorizer_reply(struct ceph_connection *con, int len)
4279 {
4280 	struct ceph_osd *o = con->private;
4281 	struct ceph_osd_client *osdc = o->o_osdc;
4282 	struct ceph_auth_client *ac = osdc->client->monc.auth;
4283 
4284 	return ceph_auth_verify_authorizer_reply(ac, o->o_auth.authorizer, len);
4285 }
4286 
4287 static int invalidate_authorizer(struct ceph_connection *con)
4288 {
4289 	struct ceph_osd *o = con->private;
4290 	struct ceph_osd_client *osdc = o->o_osdc;
4291 	struct ceph_auth_client *ac = osdc->client->monc.auth;
4292 
4293 	ceph_auth_invalidate_authorizer(ac, CEPH_ENTITY_TYPE_OSD);
4294 	return ceph_monc_validate_auth(&osdc->client->monc);
4295 }
4296 
4297 static int osd_sign_message(struct ceph_msg *msg)
4298 {
4299 	struct ceph_osd *o = msg->con->private;
4300 	struct ceph_auth_handshake *auth = &o->o_auth;
4301 
4302 	return ceph_auth_sign_message(auth, msg);
4303 }
4304 
4305 static int osd_check_message_signature(struct ceph_msg *msg)
4306 {
4307 	struct ceph_osd *o = msg->con->private;
4308 	struct ceph_auth_handshake *auth = &o->o_auth;
4309 
4310 	return ceph_auth_check_message_signature(auth, msg);
4311 }
4312 
4313 static const struct ceph_connection_operations osd_con_ops = {
4314 	.get = get_osd_con,
4315 	.put = put_osd_con,
4316 	.dispatch = dispatch,
4317 	.get_authorizer = get_authorizer,
4318 	.verify_authorizer_reply = verify_authorizer_reply,
4319 	.invalidate_authorizer = invalidate_authorizer,
4320 	.alloc_msg = alloc_msg,
4321 	.sign_message = osd_sign_message,
4322 	.check_message_signature = osd_check_message_signature,
4323 	.fault = osd_fault,
4324 };
4325