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