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