net/ceph/osd_client.c

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