ceph: allow renewal of auth credentials
[linux-3.10.git] / fs / ceph / osd_client.c
1 #include "ceph_debug.h"
2
3 #include <linux/err.h>
4 #include <linux/highmem.h>
5 #include <linux/mm.h>
6 #include <linux/pagemap.h>
7 #include <linux/slab.h>
8 #include <linux/uaccess.h>
9
10 #include "super.h"
11 #include "osd_client.h"
12 #include "messenger.h"
13 #include "decode.h"
14 #include "auth.h"
15
16 #define OSD_REPLY_RESERVE_FRONT_LEN     512
17
18 const static struct ceph_connection_operations osd_con_ops;
19
20 static void kick_requests(struct ceph_osd_client *osdc, struct ceph_osd *osd);
21
22 /*
23  * Implement client access to distributed object storage cluster.
24  *
25  * All data objects are stored within a cluster/cloud of OSDs, or
26  * "object storage devices."  (Note that Ceph OSDs have _nothing_ to
27  * do with the T10 OSD extensions to SCSI.)  Ceph OSDs are simply
28  * remote daemons serving up and coordinating consistent and safe
29  * access to storage.
30  *
31  * Cluster membership and the mapping of data objects onto storage devices
32  * are described by the osd map.
33  *
34  * We keep track of pending OSD requests (read, write), resubmit
35  * requests to different OSDs when the cluster topology/data layout
36  * change, or retry the affected requests when the communications
37  * channel with an OSD is reset.
38  */
39
40 /*
41  * calculate the mapping of a file extent onto an object, and fill out the
42  * request accordingly.  shorten extent as necessary if it crosses an
43  * object boundary.
44  *
45  * fill osd op in request message.
46  */
47 static void calc_layout(struct ceph_osd_client *osdc,
48                         struct ceph_vino vino, struct ceph_file_layout *layout,
49                         u64 off, u64 *plen,
50                         struct ceph_osd_request *req)
51 {
52         struct ceph_osd_request_head *reqhead = req->r_request->front.iov_base;
53         struct ceph_osd_op *op = (void *)(reqhead + 1);
54         u64 orig_len = *plen;
55         u64 objoff, objlen;    /* extent in object */
56         u64 bno;
57
58         reqhead->snapid = cpu_to_le64(vino.snap);
59
60         /* object extent? */
61         ceph_calc_file_object_mapping(layout, off, plen, &bno,
62                                       &objoff, &objlen);
63         if (*plen < orig_len)
64                 dout(" skipping last %llu, final file extent %llu~%llu\n",
65                      orig_len - *plen, off, *plen);
66
67         sprintf(req->r_oid, "%llx.%08llx", vino.ino, bno);
68         req->r_oid_len = strlen(req->r_oid);
69
70         op->extent.offset = cpu_to_le64(objoff);
71         op->extent.length = cpu_to_le64(objlen);
72         req->r_num_pages = calc_pages_for(off, *plen);
73
74         dout("calc_layout %s (%d) %llu~%llu (%d pages)\n",
75              req->r_oid, req->r_oid_len, objoff, objlen, req->r_num_pages);
76 }
77
78 static void remove_replies(struct ceph_osd_request *req)
79 {
80         int i;
81         int max = ARRAY_SIZE(req->replies);
82
83         for (i=0; i<max; i++) {
84                 if (req->replies[i])
85                         ceph_msg_put(req->replies[i]);
86         }
87 }
88
89 /*
90  * requests
91  */
92 void ceph_osdc_release_request(struct kref *kref)
93 {
94         struct ceph_osd_request *req = container_of(kref,
95                                                     struct ceph_osd_request,
96                                                     r_kref);
97
98         if (req->r_request)
99                 ceph_msg_put(req->r_request);
100         if (req->r_reply)
101                 ceph_msg_put(req->r_reply);
102         remove_replies(req);
103         if (req->r_con_filling_msg) {
104                 dout("release_request revoking pages %p from con %p\n",
105                      req->r_pages, req->r_con_filling_msg);
106                 ceph_con_revoke_message(req->r_con_filling_msg,
107                                       req->r_reply);
108                 ceph_con_put(req->r_con_filling_msg);
109         }
110         if (req->r_own_pages)
111                 ceph_release_page_vector(req->r_pages,
112                                          req->r_num_pages);
113         ceph_put_snap_context(req->r_snapc);
114         if (req->r_mempool)
115                 mempool_free(req, req->r_osdc->req_mempool);
116         else
117                 kfree(req);
118 }
119
120 static int alloc_replies(struct ceph_osd_request *req, int num_reply)
121 {
122         int i;
123         int max = ARRAY_SIZE(req->replies);
124
125         BUG_ON(num_reply > max);
126
127         for (i=0; i<num_reply; i++) {
128                 req->replies[i] = ceph_msg_new(0, OSD_REPLY_RESERVE_FRONT_LEN, 0, 0, NULL);
129                 if (IS_ERR(req->replies[i])) {
130                         int j;
131                         int err = PTR_ERR(req->replies[i]);
132                         for (j = 0; j<=i; j++) {
133                                 ceph_msg_put(req->replies[j]);
134                         }
135                         return err;
136                 }
137         }
138
139         for (; i<max; i++) {
140                 req->replies[i] = NULL;
141         }
142
143         req->cur_reply = 0;
144
145         return 0;
146 }
147
148 static struct ceph_msg *__get_next_reply(struct ceph_connection *con,
149                                        struct ceph_osd_request *req,
150                                        int front_len)
151 {
152         struct ceph_msg *reply;
153         if (req->r_con_filling_msg) {
154                 dout("revoking reply msg %p from old con %p\n", req->r_reply,
155                      req->r_con_filling_msg);
156                 ceph_con_revoke_message(req->r_con_filling_msg, req->r_reply);
157                 ceph_con_put(req->r_con_filling_msg);
158                 req->cur_reply = 0;
159         }
160         reply = req->replies[req->cur_reply];
161         if (!reply || front_len > OSD_REPLY_RESERVE_FRONT_LEN) {
162                 /* maybe we can allocate it now? */
163                 reply = ceph_msg_new(0, front_len, 0, 0, NULL);
164                 if (!reply || IS_ERR(reply)) {
165                         pr_err(" reply alloc failed, front_len=%d\n", front_len);
166                         return ERR_PTR(-ENOMEM);
167                 }
168         }
169         req->r_con_filling_msg = ceph_con_get(con);
170         req->r_reply = ceph_msg_get(reply); /* for duration of read over socket */
171         return ceph_msg_get(reply);
172 }
173
174 /*
175  * build new request AND message, calculate layout, and adjust file
176  * extent as needed.
177  *
178  * if the file was recently truncated, we include information about its
179  * old and new size so that the object can be updated appropriately.  (we
180  * avoid synchronously deleting truncated objects because it's slow.)
181  *
182  * if @do_sync, include a 'startsync' command so that the osd will flush
183  * data quickly.
184  */
185 struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *osdc,
186                                                struct ceph_file_layout *layout,
187                                                struct ceph_vino vino,
188                                                u64 off, u64 *plen,
189                                                int opcode, int flags,
190                                                struct ceph_snap_context *snapc,
191                                                int do_sync,
192                                                u32 truncate_seq,
193                                                u64 truncate_size,
194                                                struct timespec *mtime,
195                                                bool use_mempool, int num_reply)
196 {
197         struct ceph_osd_request *req;
198         struct ceph_msg *msg;
199         struct ceph_osd_request_head *head;
200         struct ceph_osd_op *op;
201         void *p;
202         int num_op = 1 + do_sync;
203         size_t msg_size = sizeof(*head) + num_op*sizeof(*op);
204         int err, i;
205
206         if (use_mempool) {
207                 req = mempool_alloc(osdc->req_mempool, GFP_NOFS);
208                 memset(req, 0, sizeof(*req));
209         } else {
210                 req = kzalloc(sizeof(*req), GFP_NOFS);
211         }
212         if (req == NULL)
213                 return ERR_PTR(-ENOMEM);
214
215         err = alloc_replies(req, num_reply);
216         if (err) {
217                 ceph_osdc_put_request(req);
218                 return ERR_PTR(-ENOMEM);
219         }
220         req->r_num_prealloc_reply = num_reply;
221
222         req->r_osdc = osdc;
223         req->r_mempool = use_mempool;
224         kref_init(&req->r_kref);
225         init_completion(&req->r_completion);
226         init_completion(&req->r_safe_completion);
227         INIT_LIST_HEAD(&req->r_unsafe_item);
228         req->r_flags = flags;
229
230         WARN_ON((flags & (CEPH_OSD_FLAG_READ|CEPH_OSD_FLAG_WRITE)) == 0);
231
232         /* create message; allow space for oid */
233         msg_size += 40;
234         if (snapc)
235                 msg_size += sizeof(u64) * snapc->num_snaps;
236         if (use_mempool)
237                 msg = ceph_msgpool_get(&osdc->msgpool_op, 0);
238         else
239                 msg = ceph_msg_new(CEPH_MSG_OSD_OP, msg_size, 0, 0, NULL);
240         if (IS_ERR(msg)) {
241                 ceph_osdc_put_request(req);
242                 return ERR_PTR(PTR_ERR(msg));
243         }
244         msg->hdr.type = cpu_to_le16(CEPH_MSG_OSD_OP);
245         memset(msg->front.iov_base, 0, msg->front.iov_len);
246         head = msg->front.iov_base;
247         op = (void *)(head + 1);
248         p = (void *)(op + num_op);
249
250         req->r_request = msg;
251         req->r_snapc = ceph_get_snap_context(snapc);
252
253         head->client_inc = cpu_to_le32(1); /* always, for now. */
254         head->flags = cpu_to_le32(flags);
255         if (flags & CEPH_OSD_FLAG_WRITE)
256                 ceph_encode_timespec(&head->mtime, mtime);
257         head->num_ops = cpu_to_le16(num_op);
258         op->op = cpu_to_le16(opcode);
259
260         /* calculate max write size */
261         calc_layout(osdc, vino, layout, off, plen, req);
262         req->r_file_layout = *layout;  /* keep a copy */
263
264         if (flags & CEPH_OSD_FLAG_WRITE) {
265                 req->r_request->hdr.data_off = cpu_to_le16(off);
266                 req->r_request->hdr.data_len = cpu_to_le32(*plen);
267                 op->payload_len = cpu_to_le32(*plen);
268         }
269         op->extent.truncate_size = cpu_to_le64(truncate_size);
270         op->extent.truncate_seq = cpu_to_le32(truncate_seq);
271
272         /* fill in oid */
273         head->object_len = cpu_to_le32(req->r_oid_len);
274         memcpy(p, req->r_oid, req->r_oid_len);
275         p += req->r_oid_len;
276
277         if (do_sync) {
278                 op++;
279                 op->op = cpu_to_le16(CEPH_OSD_OP_STARTSYNC);
280         }
281         if (snapc) {
282                 head->snap_seq = cpu_to_le64(snapc->seq);
283                 head->num_snaps = cpu_to_le32(snapc->num_snaps);
284                 for (i = 0; i < snapc->num_snaps; i++) {
285                         put_unaligned_le64(snapc->snaps[i], p);
286                         p += sizeof(u64);
287                 }
288         }
289
290         BUG_ON(p > msg->front.iov_base + msg->front.iov_len);
291         return req;
292 }
293
294 /*
295  * We keep osd requests in an rbtree, sorted by ->r_tid.
296  */
297 static void __insert_request(struct ceph_osd_client *osdc,
298                              struct ceph_osd_request *new)
299 {
300         struct rb_node **p = &osdc->requests.rb_node;
301         struct rb_node *parent = NULL;
302         struct ceph_osd_request *req = NULL;
303
304         while (*p) {
305                 parent = *p;
306                 req = rb_entry(parent, struct ceph_osd_request, r_node);
307                 if (new->r_tid < req->r_tid)
308                         p = &(*p)->rb_left;
309                 else if (new->r_tid > req->r_tid)
310                         p = &(*p)->rb_right;
311                 else
312                         BUG();
313         }
314
315         rb_link_node(&new->r_node, parent, p);
316         rb_insert_color(&new->r_node, &osdc->requests);
317 }
318
319 static struct ceph_osd_request *__lookup_request(struct ceph_osd_client *osdc,
320                                                  u64 tid)
321 {
322         struct ceph_osd_request *req;
323         struct rb_node *n = osdc->requests.rb_node;
324
325         while (n) {
326                 req = rb_entry(n, struct ceph_osd_request, r_node);
327                 if (tid < req->r_tid)
328                         n = n->rb_left;
329                 else if (tid > req->r_tid)
330                         n = n->rb_right;
331                 else
332                         return req;
333         }
334         return NULL;
335 }
336
337 static struct ceph_osd_request *
338 __lookup_request_ge(struct ceph_osd_client *osdc,
339                     u64 tid)
340 {
341         struct ceph_osd_request *req;
342         struct rb_node *n = osdc->requests.rb_node;
343
344         while (n) {
345                 req = rb_entry(n, struct ceph_osd_request, r_node);
346                 if (tid < req->r_tid) {
347                         if (!n->rb_left)
348                                 return req;
349                         n = n->rb_left;
350                 } else if (tid > req->r_tid) {
351                         n = n->rb_right;
352                 } else {
353                         return req;
354                 }
355         }
356         return NULL;
357 }
358
359
360 /*
361  * If the osd connection drops, we need to resubmit all requests.
362  */
363 static void osd_reset(struct ceph_connection *con)
364 {
365         struct ceph_osd *osd = con->private;
366         struct ceph_osd_client *osdc;
367
368         if (!osd)
369                 return;
370         dout("osd_reset osd%d\n", osd->o_osd);
371         osdc = osd->o_osdc;
372         osd->o_incarnation++;
373         down_read(&osdc->map_sem);
374         kick_requests(osdc, osd);
375         up_read(&osdc->map_sem);
376 }
377
378 /*
379  * Track open sessions with osds.
380  */
381 static struct ceph_osd *create_osd(struct ceph_osd_client *osdc)
382 {
383         struct ceph_osd *osd;
384
385         osd = kzalloc(sizeof(*osd), GFP_NOFS);
386         if (!osd)
387                 return NULL;
388
389         atomic_set(&osd->o_ref, 1);
390         osd->o_osdc = osdc;
391         INIT_LIST_HEAD(&osd->o_requests);
392         osd->o_incarnation = 1;
393
394         ceph_con_init(osdc->client->msgr, &osd->o_con);
395         osd->o_con.private = osd;
396         osd->o_con.ops = &osd_con_ops;
397         osd->o_con.peer_name.type = CEPH_ENTITY_TYPE_OSD;
398
399         return osd;
400 }
401
402 static struct ceph_osd *get_osd(struct ceph_osd *osd)
403 {
404         if (atomic_inc_not_zero(&osd->o_ref)) {
405                 dout("get_osd %p %d -> %d\n", osd, atomic_read(&osd->o_ref)-1,
406                      atomic_read(&osd->o_ref));
407                 return osd;
408         } else {
409                 dout("get_osd %p FAIL\n", osd);
410                 return NULL;
411         }
412 }
413
414 static void put_osd(struct ceph_osd *osd)
415 {
416         dout("put_osd %p %d -> %d\n", osd, atomic_read(&osd->o_ref),
417              atomic_read(&osd->o_ref) - 1);
418         if (atomic_dec_and_test(&osd->o_ref))
419                 kfree(osd);
420 }
421
422 /*
423  * remove an osd from our map
424  */
425 static void remove_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd)
426 {
427         dout("remove_osd %p\n", osd);
428         BUG_ON(!list_empty(&osd->o_requests));
429         rb_erase(&osd->o_node, &osdc->osds);
430         ceph_con_close(&osd->o_con);
431         put_osd(osd);
432 }
433
434 /*
435  * reset osd connect
436  */
437 static int reset_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd)
438 {
439         int ret = 0;
440
441         dout("reset_osd %p osd%d\n", osd, osd->o_osd);
442         if (list_empty(&osd->o_requests)) {
443                 remove_osd(osdc, osd);
444         } else {
445                 ceph_con_close(&osd->o_con);
446                 ceph_con_open(&osd->o_con, &osdc->osdmap->osd_addr[osd->o_osd]);
447                 osd->o_incarnation++;
448         }
449         return ret;
450 }
451
452 static void __insert_osd(struct ceph_osd_client *osdc, struct ceph_osd *new)
453 {
454         struct rb_node **p = &osdc->osds.rb_node;
455         struct rb_node *parent = NULL;
456         struct ceph_osd *osd = NULL;
457
458         while (*p) {
459                 parent = *p;
460                 osd = rb_entry(parent, struct ceph_osd, o_node);
461                 if (new->o_osd < osd->o_osd)
462                         p = &(*p)->rb_left;
463                 else if (new->o_osd > osd->o_osd)
464                         p = &(*p)->rb_right;
465                 else
466                         BUG();
467         }
468
469         rb_link_node(&new->o_node, parent, p);
470         rb_insert_color(&new->o_node, &osdc->osds);
471 }
472
473 static struct ceph_osd *__lookup_osd(struct ceph_osd_client *osdc, int o)
474 {
475         struct ceph_osd *osd;
476         struct rb_node *n = osdc->osds.rb_node;
477
478         while (n) {
479                 osd = rb_entry(n, struct ceph_osd, o_node);
480                 if (o < osd->o_osd)
481                         n = n->rb_left;
482                 else if (o > osd->o_osd)
483                         n = n->rb_right;
484                 else
485                         return osd;
486         }
487         return NULL;
488 }
489
490
491 /*
492  * Register request, assign tid.  If this is the first request, set up
493  * the timeout event.
494  */
495 static void register_request(struct ceph_osd_client *osdc,
496                              struct ceph_osd_request *req)
497 {
498         mutex_lock(&osdc->request_mutex);
499         req->r_tid = ++osdc->last_tid;
500         req->r_request->hdr.tid = cpu_to_le64(req->r_tid);
501
502         dout("register_request %p tid %lld\n", req, req->r_tid);
503         __insert_request(osdc, req);
504         ceph_osdc_get_request(req);
505         osdc->num_requests++;
506
507         req->r_timeout_stamp =
508                 jiffies + osdc->client->mount_args->osd_timeout*HZ;
509
510         if (osdc->num_requests == 1) {
511                 osdc->timeout_tid = req->r_tid;
512                 dout("  timeout on tid %llu at %lu\n", req->r_tid,
513                      req->r_timeout_stamp);
514                 schedule_delayed_work(&osdc->timeout_work,
515                       round_jiffies_relative(req->r_timeout_stamp - jiffies));
516         }
517         mutex_unlock(&osdc->request_mutex);
518 }
519
520 /*
521  * called under osdc->request_mutex
522  */
523 static void __unregister_request(struct ceph_osd_client *osdc,
524                                  struct ceph_osd_request *req)
525 {
526         dout("__unregister_request %p tid %lld\n", req, req->r_tid);
527         rb_erase(&req->r_node, &osdc->requests);
528         osdc->num_requests--;
529
530         if (req->r_osd) {
531                 /* make sure the original request isn't in flight. */
532                 ceph_con_revoke(&req->r_osd->o_con, req->r_request);
533
534                 list_del_init(&req->r_osd_item);
535                 if (list_empty(&req->r_osd->o_requests))
536                         remove_osd(osdc, req->r_osd);
537                 req->r_osd = NULL;
538         }
539
540         ceph_osdc_put_request(req);
541
542         if (req->r_tid == osdc->timeout_tid) {
543                 if (osdc->num_requests == 0) {
544                         dout("no requests, canceling timeout\n");
545                         osdc->timeout_tid = 0;
546                         cancel_delayed_work(&osdc->timeout_work);
547                 } else {
548                         req = rb_entry(rb_first(&osdc->requests),
549                                        struct ceph_osd_request, r_node);
550                         osdc->timeout_tid = req->r_tid;
551                         dout("rescheduled timeout on tid %llu at %lu\n",
552                              req->r_tid, req->r_timeout_stamp);
553                         schedule_delayed_work(&osdc->timeout_work,
554                               round_jiffies_relative(req->r_timeout_stamp -
555                                                      jiffies));
556                 }
557         }
558 }
559
560 /*
561  * Cancel a previously queued request message
562  */
563 static void __cancel_request(struct ceph_osd_request *req)
564 {
565         if (req->r_sent) {
566                 ceph_con_revoke(&req->r_osd->o_con, req->r_request);
567                 req->r_sent = 0;
568         }
569 }
570
571 /*
572  * Pick an osd (the first 'up' osd in the pg), allocate the osd struct
573  * (as needed), and set the request r_osd appropriately.  If there is
574  * no up osd, set r_osd to NULL.
575  *
576  * Return 0 if unchanged, 1 if changed, or negative on error.
577  *
578  * Caller should hold map_sem for read and request_mutex.
579  */
580 static int __map_osds(struct ceph_osd_client *osdc,
581                       struct ceph_osd_request *req)
582 {
583         struct ceph_osd_request_head *reqhead = req->r_request->front.iov_base;
584         struct ceph_pg pgid;
585         int o = -1;
586         int err;
587         struct ceph_osd *newosd = NULL;
588
589         dout("map_osds %p tid %lld\n", req, req->r_tid);
590         err = ceph_calc_object_layout(&reqhead->layout, req->r_oid,
591                                       &req->r_file_layout, osdc->osdmap);
592         if (err)
593                 return err;
594         pgid = reqhead->layout.ol_pgid;
595         req->r_pgid = pgid;
596
597         o = ceph_calc_pg_primary(osdc->osdmap, pgid);
598
599         if ((req->r_osd && req->r_osd->o_osd == o &&
600              req->r_sent >= req->r_osd->o_incarnation) ||
601             (req->r_osd == NULL && o == -1))
602                 return 0;  /* no change */
603
604         dout("map_osds tid %llu pgid %d.%x osd%d (was osd%d)\n",
605              req->r_tid, le32_to_cpu(pgid.pool), le16_to_cpu(pgid.ps), o,
606              req->r_osd ? req->r_osd->o_osd : -1);
607
608         if (req->r_osd) {
609                 __cancel_request(req);
610                 list_del_init(&req->r_osd_item);
611                 if (list_empty(&req->r_osd->o_requests)) {
612                         /* try to re-use r_osd if possible */
613                         newosd = get_osd(req->r_osd);
614                         remove_osd(osdc, newosd);
615                 }
616                 req->r_osd = NULL;
617         }
618
619         req->r_osd = __lookup_osd(osdc, o);
620         if (!req->r_osd && o >= 0) {
621                 if (newosd) {
622                         req->r_osd = newosd;
623                         newosd = NULL;
624                 } else {
625                         err = -ENOMEM;
626                         req->r_osd = create_osd(osdc);
627                         if (!req->r_osd)
628                                 goto out;
629                 }
630
631                 dout("map_osds osd %p is osd%d\n", req->r_osd, o);
632                 req->r_osd->o_osd = o;
633                 req->r_osd->o_con.peer_name.num = cpu_to_le64(o);
634                 __insert_osd(osdc, req->r_osd);
635
636                 ceph_con_open(&req->r_osd->o_con, &osdc->osdmap->osd_addr[o]);
637         }
638
639         if (req->r_osd)
640                 list_add(&req->r_osd_item, &req->r_osd->o_requests);
641         err = 1;   /* osd changed */
642
643 out:
644         if (newosd)
645                 put_osd(newosd);
646         return err;
647 }
648
649 /*
650  * caller should hold map_sem (for read) and request_mutex
651  */
652 static int __send_request(struct ceph_osd_client *osdc,
653                           struct ceph_osd_request *req)
654 {
655         struct ceph_osd_request_head *reqhead;
656         int err;
657
658         err = __map_osds(osdc, req);
659         if (err < 0)
660                 return err;
661         if (req->r_osd == NULL) {
662                 dout("send_request %p no up osds in pg\n", req);
663                 ceph_monc_request_next_osdmap(&osdc->client->monc);
664                 return 0;
665         }
666
667         dout("send_request %p tid %llu to osd%d flags %d\n",
668              req, req->r_tid, req->r_osd->o_osd, req->r_flags);
669
670         reqhead = req->r_request->front.iov_base;
671         reqhead->osdmap_epoch = cpu_to_le32(osdc->osdmap->epoch);
672         reqhead->flags |= cpu_to_le32(req->r_flags);  /* e.g., RETRY */
673         reqhead->reassert_version = req->r_reassert_version;
674
675         req->r_timeout_stamp = jiffies+osdc->client->mount_args->osd_timeout*HZ;
676
677         ceph_msg_get(req->r_request); /* send consumes a ref */
678         ceph_con_send(&req->r_osd->o_con, req->r_request);
679         req->r_sent = req->r_osd->o_incarnation;
680         return 0;
681 }
682
683 /*
684  * Timeout callback, called every N seconds when 1 or more osd
685  * requests has been active for more than N seconds.  When this
686  * happens, we ping all OSDs with requests who have timed out to
687  * ensure any communications channel reset is detected.  Reset the
688  * request timeouts another N seconds in the future as we go.
689  * Reschedule the timeout event another N seconds in future (unless
690  * there are no open requests).
691  */
692 static void handle_timeout(struct work_struct *work)
693 {
694         struct ceph_osd_client *osdc =
695                 container_of(work, struct ceph_osd_client, timeout_work.work);
696         struct ceph_osd_request *req;
697         struct ceph_osd *osd;
698         unsigned long timeout = osdc->client->mount_args->osd_timeout * HZ;
699         unsigned long next_timeout = timeout + jiffies;
700         struct rb_node *p;
701
702         dout("timeout\n");
703         down_read(&osdc->map_sem);
704
705         ceph_monc_request_next_osdmap(&osdc->client->monc);
706
707         mutex_lock(&osdc->request_mutex);
708         for (p = rb_first(&osdc->requests); p; p = rb_next(p)) {
709                 req = rb_entry(p, struct ceph_osd_request, r_node);
710
711                 if (req->r_resend) {
712                         int err;
713
714                         dout("osdc resending prev failed %lld\n", req->r_tid);
715                         err = __send_request(osdc, req);
716                         if (err)
717                                 dout("osdc failed again on %lld\n", req->r_tid);
718                         else
719                                 req->r_resend = false;
720                         continue;
721                 }
722         }
723         for (p = rb_first(&osdc->osds); p; p = rb_next(p)) {
724                 osd = rb_entry(p, struct ceph_osd, o_node);
725                 if (list_empty(&osd->o_requests))
726                         continue;
727                 req = list_first_entry(&osd->o_requests,
728                                        struct ceph_osd_request, r_osd_item);
729                 if (time_before(jiffies, req->r_timeout_stamp))
730                         continue;
731
732                 dout(" tid %llu (at least) timed out on osd%d\n",
733                      req->r_tid, osd->o_osd);
734                 req->r_timeout_stamp = next_timeout;
735                 ceph_con_keepalive(&osd->o_con);
736         }
737
738         if (osdc->timeout_tid)
739                 schedule_delayed_work(&osdc->timeout_work,
740                                       round_jiffies_relative(timeout));
741
742         mutex_unlock(&osdc->request_mutex);
743
744         up_read(&osdc->map_sem);
745 }
746
747 /*
748  * handle osd op reply.  either call the callback if it is specified,
749  * or do the completion to wake up the waiting thread.
750  */
751 static void handle_reply(struct ceph_osd_client *osdc, struct ceph_msg *msg,
752                          struct ceph_connection *con)
753 {
754         struct ceph_osd_reply_head *rhead = msg->front.iov_base;
755         struct ceph_osd_request *req;
756         u64 tid;
757         int numops, object_len, flags;
758
759         tid = le64_to_cpu(msg->hdr.tid);
760         if (msg->front.iov_len < sizeof(*rhead))
761                 goto bad;
762         numops = le32_to_cpu(rhead->num_ops);
763         object_len = le32_to_cpu(rhead->object_len);
764         if (msg->front.iov_len != sizeof(*rhead) + object_len +
765             numops * sizeof(struct ceph_osd_op))
766                 goto bad;
767         dout("handle_reply %p tid %llu\n", msg, tid);
768
769         /* lookup */
770         mutex_lock(&osdc->request_mutex);
771         req = __lookup_request(osdc, tid);
772         if (req == NULL) {
773                 dout("handle_reply tid %llu dne\n", tid);
774                 mutex_unlock(&osdc->request_mutex);
775                 return;
776         }
777         ceph_osdc_get_request(req);
778         flags = le32_to_cpu(rhead->flags);
779
780         /*
781          * if this connection filled our message, drop our reference now, to
782          * avoid a (safe but slower) revoke later.
783          */
784         if (req->r_con_filling_msg == con && req->r_reply == msg) {
785                 dout(" got pages, dropping con_filling_msg ref %p\n", con);
786                 req->r_con_filling_msg = NULL;
787                 ceph_con_put(con);
788         }
789
790         if (req->r_reply) {
791                 /*
792                  * once we see the message has been received, we don't
793                  * need a ref (which is only needed for revoking
794                  * pages)
795                  */
796                 ceph_msg_put(req->r_reply);
797                 req->r_reply = NULL;
798         }
799
800         if (!req->r_got_reply) {
801                 unsigned bytes;
802
803                 req->r_result = le32_to_cpu(rhead->result);
804                 bytes = le32_to_cpu(msg->hdr.data_len);
805                 dout("handle_reply result %d bytes %d\n", req->r_result,
806                      bytes);
807                 if (req->r_result == 0)
808                         req->r_result = bytes;
809
810                 /* in case this is a write and we need to replay, */
811                 req->r_reassert_version = rhead->reassert_version;
812
813                 req->r_got_reply = 1;
814         } else if ((flags & CEPH_OSD_FLAG_ONDISK) == 0) {
815                 dout("handle_reply tid %llu dup ack\n", tid);
816                 mutex_unlock(&osdc->request_mutex);
817                 goto done;
818         }
819
820         dout("handle_reply tid %llu flags %d\n", tid, flags);
821
822         /* either this is a read, or we got the safe response */
823         if ((flags & CEPH_OSD_FLAG_ONDISK) ||
824             ((flags & CEPH_OSD_FLAG_WRITE) == 0))
825                 __unregister_request(osdc, req);
826
827         mutex_unlock(&osdc->request_mutex);
828
829         if (req->r_callback)
830                 req->r_callback(req, msg);
831         else
832                 complete(&req->r_completion);
833
834         if (flags & CEPH_OSD_FLAG_ONDISK) {
835                 if (req->r_safe_callback)
836                         req->r_safe_callback(req, msg);
837                 complete(&req->r_safe_completion);  /* fsync waiter */
838         }
839
840 done:
841         ceph_osdc_put_request(req);
842         return;
843
844 bad:
845         pr_err("corrupt osd_op_reply got %d %d expected %d\n",
846                (int)msg->front.iov_len, le32_to_cpu(msg->hdr.front_len),
847                (int)sizeof(*rhead));
848         ceph_msg_dump(msg);
849 }
850
851
852 /*
853  * Resubmit osd requests whose osd or osd address has changed.  Request
854  * a new osd map if osds are down, or we are otherwise unable to determine
855  * how to direct a request.
856  *
857  * Close connections to down osds.
858  *
859  * If @who is specified, resubmit requests for that specific osd.
860  *
861  * Caller should hold map_sem for read and request_mutex.
862  */
863 static void kick_requests(struct ceph_osd_client *osdc,
864                           struct ceph_osd *kickosd)
865 {
866         struct ceph_osd_request *req;
867         struct rb_node *p, *n;
868         int needmap = 0;
869         int err;
870
871         dout("kick_requests osd%d\n", kickosd ? kickosd->o_osd : -1);
872         mutex_lock(&osdc->request_mutex);
873         if (!kickosd) {
874                 for (p = rb_first(&osdc->osds); p; p = n) {
875                         struct ceph_osd *osd =
876                                 rb_entry(p, struct ceph_osd, o_node);
877
878                         n = rb_next(p);
879                         if (!ceph_osd_is_up(osdc->osdmap, osd->o_osd) ||
880                             memcmp(&osd->o_con.peer_addr,
881                                    ceph_osd_addr(osdc->osdmap,
882                                                  osd->o_osd),
883                                    sizeof(struct ceph_entity_addr)) != 0)
884                                 reset_osd(osdc, osd);
885                 }
886         }
887
888         for (p = rb_first(&osdc->requests); p; p = rb_next(p)) {
889                 req = rb_entry(p, struct ceph_osd_request, r_node);
890
891                 if (req->r_resend) {
892                         dout(" r_resend set on tid %llu\n", req->r_tid);
893                         __cancel_request(req);
894                         goto kick;
895                 }
896                 if (req->r_osd && kickosd == req->r_osd) {
897                         __cancel_request(req);
898                         goto kick;
899                 }
900
901                 err = __map_osds(osdc, req);
902                 if (err == 0)
903                         continue;  /* no change */
904                 if (err < 0) {
905                         /*
906                          * FIXME: really, we should set the request
907                          * error and fail if this isn't a 'nofail'
908                          * request, but that's a fair bit more
909                          * complicated to do.  So retry!
910                          */
911                         dout(" setting r_resend on %llu\n", req->r_tid);
912                         req->r_resend = true;
913                         continue;
914                 }
915                 if (req->r_osd == NULL) {
916                         dout("tid %llu maps to no valid osd\n", req->r_tid);
917                         needmap++;  /* request a newer map */
918                         continue;
919                 }
920
921 kick:
922                 dout("kicking %p tid %llu osd%d\n", req, req->r_tid,
923                      req->r_osd->o_osd);
924                 req->r_flags |= CEPH_OSD_FLAG_RETRY;
925                 err = __send_request(osdc, req);
926                 if (err) {
927                         dout(" setting r_resend on %llu\n", req->r_tid);
928                         req->r_resend = true;
929                 }
930         }
931         mutex_unlock(&osdc->request_mutex);
932
933         if (needmap) {
934                 dout("%d requests for down osds, need new map\n", needmap);
935                 ceph_monc_request_next_osdmap(&osdc->client->monc);
936         }
937 }
938
939 /*
940  * Process updated osd map.
941  *
942  * The message contains any number of incremental and full maps, normally
943  * indicating some sort of topology change in the cluster.  Kick requests
944  * off to different OSDs as needed.
945  */
946 void ceph_osdc_handle_map(struct ceph_osd_client *osdc, struct ceph_msg *msg)
947 {
948         void *p, *end, *next;
949         u32 nr_maps, maplen;
950         u32 epoch;
951         struct ceph_osdmap *newmap = NULL, *oldmap;
952         int err;
953         struct ceph_fsid fsid;
954
955         dout("handle_map have %u\n", osdc->osdmap ? osdc->osdmap->epoch : 0);
956         p = msg->front.iov_base;
957         end = p + msg->front.iov_len;
958
959         /* verify fsid */
960         ceph_decode_need(&p, end, sizeof(fsid), bad);
961         ceph_decode_copy(&p, &fsid, sizeof(fsid));
962         if (ceph_check_fsid(osdc->client, &fsid) < 0)
963                 return;
964
965         down_write(&osdc->map_sem);
966
967         /* incremental maps */
968         ceph_decode_32_safe(&p, end, nr_maps, bad);
969         dout(" %d inc maps\n", nr_maps);
970         while (nr_maps > 0) {
971                 ceph_decode_need(&p, end, 2*sizeof(u32), bad);
972                 epoch = ceph_decode_32(&p);
973                 maplen = ceph_decode_32(&p);
974                 ceph_decode_need(&p, end, maplen, bad);
975                 next = p + maplen;
976                 if (osdc->osdmap && osdc->osdmap->epoch+1 == epoch) {
977                         dout("applying incremental map %u len %d\n",
978                              epoch, maplen);
979                         newmap = osdmap_apply_incremental(&p, next,
980                                                           osdc->osdmap,
981                                                           osdc->client->msgr);
982                         if (IS_ERR(newmap)) {
983                                 err = PTR_ERR(newmap);
984                                 goto bad;
985                         }
986                         BUG_ON(!newmap);
987                         if (newmap != osdc->osdmap) {
988                                 ceph_osdmap_destroy(osdc->osdmap);
989                                 osdc->osdmap = newmap;
990                         }
991                 } else {
992                         dout("ignoring incremental map %u len %d\n",
993                              epoch, maplen);
994                 }
995                 p = next;
996                 nr_maps--;
997         }
998         if (newmap)
999                 goto done;
1000
1001         /* full maps */
1002         ceph_decode_32_safe(&p, end, nr_maps, bad);
1003         dout(" %d full maps\n", nr_maps);
1004         while (nr_maps) {
1005                 ceph_decode_need(&p, end, 2*sizeof(u32), bad);
1006                 epoch = ceph_decode_32(&p);
1007                 maplen = ceph_decode_32(&p);
1008                 ceph_decode_need(&p, end, maplen, bad);
1009                 if (nr_maps > 1) {
1010                         dout("skipping non-latest full map %u len %d\n",
1011                              epoch, maplen);
1012                 } else if (osdc->osdmap && osdc->osdmap->epoch >= epoch) {
1013                         dout("skipping full map %u len %d, "
1014                              "older than our %u\n", epoch, maplen,
1015                              osdc->osdmap->epoch);
1016                 } else {
1017                         dout("taking full map %u len %d\n", epoch, maplen);
1018                         newmap = osdmap_decode(&p, p+maplen);
1019                         if (IS_ERR(newmap)) {
1020                                 err = PTR_ERR(newmap);
1021                                 goto bad;
1022                         }
1023                         BUG_ON(!newmap);
1024                         oldmap = osdc->osdmap;
1025                         osdc->osdmap = newmap;
1026                         if (oldmap)
1027                                 ceph_osdmap_destroy(oldmap);
1028                 }
1029                 p += maplen;
1030                 nr_maps--;
1031         }
1032
1033 done:
1034         downgrade_write(&osdc->map_sem);
1035         ceph_monc_got_osdmap(&osdc->client->monc, osdc->osdmap->epoch);
1036         if (newmap)
1037                 kick_requests(osdc, NULL);
1038         up_read(&osdc->map_sem);
1039         return;
1040
1041 bad:
1042         pr_err("osdc handle_map corrupt msg\n");
1043         ceph_msg_dump(msg);
1044         up_write(&osdc->map_sem);
1045         return;
1046 }
1047
1048
1049 /*
1050  * A read request prepares specific pages that data is to be read into.
1051  * When a message is being read off the wire, we call prepare_pages to
1052  * find those pages.
1053  *  0 = success, -1 failure.
1054  */
1055 static int __prepare_pages(struct ceph_connection *con,
1056                          struct ceph_msg_header *hdr,
1057                          struct ceph_osd_request *req,
1058                          u64 tid,
1059                          struct ceph_msg *m)
1060 {
1061         struct ceph_osd *osd = con->private;
1062         struct ceph_osd_client *osdc;
1063         int ret = -1;
1064         int data_len = le32_to_cpu(hdr->data_len);
1065         unsigned data_off = le16_to_cpu(hdr->data_off);
1066
1067         int want = calc_pages_for(data_off & ~PAGE_MASK, data_len);
1068
1069         if (!osd)
1070                 return -1;
1071
1072         osdc = osd->o_osdc;
1073
1074         dout("__prepare_pages on msg %p tid %llu, has %d pages, want %d\n", m,
1075              tid, req->r_num_pages, want);
1076         if (unlikely(req->r_num_pages < want))
1077                 goto out;
1078         m->pages = req->r_pages;
1079         m->nr_pages = req->r_num_pages;
1080         ret = 0; /* success */
1081 out:
1082         BUG_ON(ret < 0 || m->nr_pages < want);
1083
1084         return ret;
1085 }
1086
1087 /*
1088  * Register request, send initial attempt.
1089  */
1090 int ceph_osdc_start_request(struct ceph_osd_client *osdc,
1091                             struct ceph_osd_request *req,
1092                             bool nofail)
1093 {
1094         int rc = 0;
1095
1096         req->r_request->pages = req->r_pages;
1097         req->r_request->nr_pages = req->r_num_pages;
1098
1099         register_request(osdc, req);
1100
1101         down_read(&osdc->map_sem);
1102         mutex_lock(&osdc->request_mutex);
1103         /*
1104          * a racing kick_requests() may have sent the message for us
1105          * while we dropped request_mutex above, so only send now if
1106          * the request still han't been touched yet.
1107          */
1108         if (req->r_sent == 0) {
1109                 rc = __send_request(osdc, req);
1110                 if (rc) {
1111                         if (nofail) {
1112                                 dout("osdc_start_request failed send, "
1113                                      " marking %lld\n", req->r_tid);
1114                                 req->r_resend = true;
1115                                 rc = 0;
1116                         } else {
1117                                 __unregister_request(osdc, req);
1118                         }
1119                 }
1120         }
1121         mutex_unlock(&osdc->request_mutex);
1122         up_read(&osdc->map_sem);
1123         return rc;
1124 }
1125
1126 /*
1127  * wait for a request to complete
1128  */
1129 int ceph_osdc_wait_request(struct ceph_osd_client *osdc,
1130                            struct ceph_osd_request *req)
1131 {
1132         int rc;
1133
1134         rc = wait_for_completion_interruptible(&req->r_completion);
1135         if (rc < 0) {
1136                 mutex_lock(&osdc->request_mutex);
1137                 __cancel_request(req);
1138                 __unregister_request(osdc, req);
1139                 mutex_unlock(&osdc->request_mutex);
1140                 dout("wait_request tid %llu canceled/timed out\n", req->r_tid);
1141                 return rc;
1142         }
1143
1144         dout("wait_request tid %llu result %d\n", req->r_tid, req->r_result);
1145         return req->r_result;
1146 }
1147
1148 /*
1149  * sync - wait for all in-flight requests to flush.  avoid starvation.
1150  */
1151 void ceph_osdc_sync(struct ceph_osd_client *osdc)
1152 {
1153         struct ceph_osd_request *req;
1154         u64 last_tid, next_tid = 0;
1155
1156         mutex_lock(&osdc->request_mutex);
1157         last_tid = osdc->last_tid;
1158         while (1) {
1159                 req = __lookup_request_ge(osdc, next_tid);
1160                 if (!req)
1161                         break;
1162                 if (req->r_tid > last_tid)
1163                         break;
1164
1165                 next_tid = req->r_tid + 1;
1166                 if ((req->r_flags & CEPH_OSD_FLAG_WRITE) == 0)
1167                         continue;
1168
1169                 ceph_osdc_get_request(req);
1170                 mutex_unlock(&osdc->request_mutex);
1171                 dout("sync waiting on tid %llu (last is %llu)\n",
1172                      req->r_tid, last_tid);
1173                 wait_for_completion(&req->r_safe_completion);
1174                 mutex_lock(&osdc->request_mutex);
1175                 ceph_osdc_put_request(req);
1176         }
1177         mutex_unlock(&osdc->request_mutex);
1178         dout("sync done (thru tid %llu)\n", last_tid);
1179 }
1180
1181 /*
1182  * init, shutdown
1183  */
1184 int ceph_osdc_init(struct ceph_osd_client *osdc, struct ceph_client *client)
1185 {
1186         int err;
1187
1188         dout("init\n");
1189         osdc->client = client;
1190         osdc->osdmap = NULL;
1191         init_rwsem(&osdc->map_sem);
1192         init_completion(&osdc->map_waiters);
1193         osdc->last_requested_map = 0;
1194         mutex_init(&osdc->request_mutex);
1195         osdc->timeout_tid = 0;
1196         osdc->last_tid = 0;
1197         osdc->osds = RB_ROOT;
1198         osdc->requests = RB_ROOT;
1199         osdc->num_requests = 0;
1200         INIT_DELAYED_WORK(&osdc->timeout_work, handle_timeout);
1201
1202         err = -ENOMEM;
1203         osdc->req_mempool = mempool_create_kmalloc_pool(10,
1204                                         sizeof(struct ceph_osd_request));
1205         if (!osdc->req_mempool)
1206                 goto out;
1207
1208         err = ceph_msgpool_init(&osdc->msgpool_op, 4096, 10, true);
1209         if (err < 0)
1210                 goto out_mempool;
1211         return 0;
1212
1213 out_mempool:
1214         mempool_destroy(osdc->req_mempool);
1215 out:
1216         return err;
1217 }
1218
1219 void ceph_osdc_stop(struct ceph_osd_client *osdc)
1220 {
1221         cancel_delayed_work_sync(&osdc->timeout_work);
1222         if (osdc->osdmap) {
1223                 ceph_osdmap_destroy(osdc->osdmap);
1224                 osdc->osdmap = NULL;
1225         }
1226         mempool_destroy(osdc->req_mempool);
1227         ceph_msgpool_destroy(&osdc->msgpool_op);
1228 }
1229
1230 /*
1231  * Read some contiguous pages.  If we cross a stripe boundary, shorten
1232  * *plen.  Return number of bytes read, or error.
1233  */
1234 int ceph_osdc_readpages(struct ceph_osd_client *osdc,
1235                         struct ceph_vino vino, struct ceph_file_layout *layout,
1236                         u64 off, u64 *plen,
1237                         u32 truncate_seq, u64 truncate_size,
1238                         struct page **pages, int num_pages)
1239 {
1240         struct ceph_osd_request *req;
1241         int rc = 0;
1242
1243         dout("readpages on ino %llx.%llx on %llu~%llu\n", vino.ino,
1244              vino.snap, off, *plen);
1245         req = ceph_osdc_new_request(osdc, layout, vino, off, plen,
1246                                     CEPH_OSD_OP_READ, CEPH_OSD_FLAG_READ,
1247                                     NULL, 0, truncate_seq, truncate_size, NULL,
1248                                     false, 1);
1249         if (IS_ERR(req))
1250                 return PTR_ERR(req);
1251
1252         /* it may be a short read due to an object boundary */
1253         req->r_pages = pages;
1254         num_pages = calc_pages_for(off, *plen);
1255         req->r_num_pages = num_pages;
1256
1257         dout("readpages  final extent is %llu~%llu (%d pages)\n",
1258              off, *plen, req->r_num_pages);
1259
1260         rc = ceph_osdc_start_request(osdc, req, false);
1261         if (!rc)
1262                 rc = ceph_osdc_wait_request(osdc, req);
1263
1264         ceph_osdc_put_request(req);
1265         dout("readpages result %d\n", rc);
1266         return rc;
1267 }
1268
1269 /*
1270  * do a synchronous write on N pages
1271  */
1272 int ceph_osdc_writepages(struct ceph_osd_client *osdc, struct ceph_vino vino,
1273                          struct ceph_file_layout *layout,
1274                          struct ceph_snap_context *snapc,
1275                          u64 off, u64 len,
1276                          u32 truncate_seq, u64 truncate_size,
1277                          struct timespec *mtime,
1278                          struct page **pages, int num_pages,
1279                          int flags, int do_sync, bool nofail)
1280 {
1281         struct ceph_osd_request *req;
1282         int rc = 0;
1283
1284         BUG_ON(vino.snap != CEPH_NOSNAP);
1285         req = ceph_osdc_new_request(osdc, layout, vino, off, &len,
1286                                     CEPH_OSD_OP_WRITE,
1287                                     flags | CEPH_OSD_FLAG_ONDISK |
1288                                             CEPH_OSD_FLAG_WRITE,
1289                                     snapc, do_sync,
1290                                     truncate_seq, truncate_size, mtime,
1291                                     nofail, 1);
1292         if (IS_ERR(req))
1293                 return PTR_ERR(req);
1294
1295         /* it may be a short write due to an object boundary */
1296         req->r_pages = pages;
1297         req->r_num_pages = calc_pages_for(off, len);
1298         dout("writepages %llu~%llu (%d pages)\n", off, len,
1299              req->r_num_pages);
1300
1301         rc = ceph_osdc_start_request(osdc, req, nofail);
1302         if (!rc)
1303                 rc = ceph_osdc_wait_request(osdc, req);
1304
1305         ceph_osdc_put_request(req);
1306         if (rc == 0)
1307                 rc = len;
1308         dout("writepages result %d\n", rc);
1309         return rc;
1310 }
1311
1312 /*
1313  * handle incoming message
1314  */
1315 static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
1316 {
1317         struct ceph_osd *osd = con->private;
1318         struct ceph_osd_client *osdc;
1319         int type = le16_to_cpu(msg->hdr.type);
1320
1321         if (!osd)
1322                 return;
1323         osdc = osd->o_osdc;
1324
1325         switch (type) {
1326         case CEPH_MSG_OSD_MAP:
1327                 ceph_osdc_handle_map(osdc, msg);
1328                 break;
1329         case CEPH_MSG_OSD_OPREPLY:
1330                 handle_reply(osdc, msg, con);
1331                 break;
1332
1333         default:
1334                 pr_err("received unknown message type %d %s\n", type,
1335                        ceph_msg_type_name(type));
1336         }
1337         ceph_msg_put(msg);
1338 }
1339
1340 static struct ceph_msg *alloc_msg(struct ceph_connection *con,
1341                                   struct ceph_msg_header *hdr,
1342                                   int *skip)
1343 {
1344         struct ceph_osd *osd = con->private;
1345         struct ceph_osd_client *osdc = osd->o_osdc;
1346         int type = le16_to_cpu(hdr->type);
1347         int front = le32_to_cpu(hdr->front_len);
1348         int data_len = le32_to_cpu(hdr->data_len);
1349         struct ceph_msg *m;
1350         struct ceph_osd_request *req;
1351         u64 tid;
1352         int err;
1353
1354         *skip = 0;
1355         if (type != CEPH_MSG_OSD_OPREPLY)
1356                 return NULL;
1357
1358         tid = le64_to_cpu(hdr->tid);
1359         mutex_lock(&osdc->request_mutex);
1360         req = __lookup_request(osdc, tid);
1361         if (!req) {
1362                 *skip = 1;
1363                 m = NULL;
1364                 dout("alloc_msg unknown tid %llu\n", tid);
1365                 goto out;
1366         }
1367         m = __get_next_reply(con, req, front);
1368         if (!m || IS_ERR(m)) {
1369                 *skip = 1;
1370                 goto out;
1371         }
1372
1373         if (data_len > 0) {
1374                 err = __prepare_pages(con, hdr, req, tid, m);
1375                 if (err < 0) {
1376                         *skip = 1;
1377                         ceph_msg_put(m);
1378                         m = ERR_PTR(err);
1379                 }
1380         }
1381
1382 out:
1383         mutex_unlock(&osdc->request_mutex);
1384
1385         return m;
1386 }
1387
1388 /*
1389  * Wrappers to refcount containing ceph_osd struct
1390  */
1391 static struct ceph_connection *get_osd_con(struct ceph_connection *con)
1392 {
1393         struct ceph_osd *osd = con->private;
1394         if (get_osd(osd))
1395                 return con;
1396         return NULL;
1397 }
1398
1399 static void put_osd_con(struct ceph_connection *con)
1400 {
1401         struct ceph_osd *osd = con->private;
1402         put_osd(osd);
1403 }
1404
1405 /*
1406  * authentication
1407  */
1408 static int get_authorizer(struct ceph_connection *con,
1409                           void **buf, int *len, int *proto,
1410                           void **reply_buf, int *reply_len, int force_new)
1411 {
1412         struct ceph_osd *o = con->private;
1413         struct ceph_osd_client *osdc = o->o_osdc;
1414         struct ceph_auth_client *ac = osdc->client->monc.auth;
1415         int ret = 0;
1416
1417         if (force_new && o->o_authorizer) {
1418                 ac->ops->destroy_authorizer(ac, o->o_authorizer);
1419                 o->o_authorizer = NULL;
1420         }
1421         if (o->o_authorizer == NULL) {
1422                 ret = ac->ops->create_authorizer(
1423                         ac, CEPH_ENTITY_TYPE_OSD,
1424                         &o->o_authorizer,
1425                         &o->o_authorizer_buf,
1426                         &o->o_authorizer_buf_len,
1427                         &o->o_authorizer_reply_buf,
1428                         &o->o_authorizer_reply_buf_len);
1429                 if (ret)
1430                 return ret;
1431         }
1432
1433         *proto = ac->protocol;
1434         *buf = o->o_authorizer_buf;
1435         *len = o->o_authorizer_buf_len;
1436         *reply_buf = o->o_authorizer_reply_buf;
1437         *reply_len = o->o_authorizer_reply_buf_len;
1438         return 0;
1439 }
1440
1441
1442 static int verify_authorizer_reply(struct ceph_connection *con, int len)
1443 {
1444         struct ceph_osd *o = con->private;
1445         struct ceph_osd_client *osdc = o->o_osdc;
1446         struct ceph_auth_client *ac = osdc->client->monc.auth;
1447
1448         return ac->ops->verify_authorizer_reply(ac, o->o_authorizer, len);
1449 }
1450
1451 static int invalidate_authorizer(struct ceph_connection *con)
1452 {
1453         struct ceph_osd *o = con->private;
1454         struct ceph_osd_client *osdc = o->o_osdc;
1455         struct ceph_auth_client *ac = osdc->client->monc.auth;
1456
1457         if (ac->ops->invalidate_authorizer)
1458                 ac->ops->invalidate_authorizer(ac, CEPH_ENTITY_TYPE_OSD);
1459
1460         return ceph_monc_validate_auth(&osdc->client->monc);
1461 }
1462
1463 const static struct ceph_connection_operations osd_con_ops = {
1464         .get = get_osd_con,
1465         .put = put_osd_con,
1466         .dispatch = dispatch,
1467         .get_authorizer = get_authorizer,
1468         .verify_authorizer_reply = verify_authorizer_reply,
1469         .invalidate_authorizer = invalidate_authorizer,
1470         .alloc_msg = alloc_msg,
1471         .fault = osd_reset,
1472 };