drivers/block/rbd.c

   1 /*
   2    rbd.c -- Export ceph rados objects as a Linux block device
   3
   4
   5    based on drivers/block/osdblk.c:
   6
   7    Copyright 2009 Red Hat, Inc.
   8
   9    This program is free software; you can redistribute it and/or modify
  10    it under the terms of the GNU General Public License as published by
  11    the Free Software Foundation.
  12
  13    This program is distributed in the hope that it will be useful,
  14    but WITHOUT ANY WARRANTY; without even the implied warranty of
  15    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16    GNU General Public License for more details.
  17
  18    You should have received a copy of the GNU General Public License
  19    along with this program; see the file COPYING.  If not, write to
  20    the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
  21
  22
  23
  24    For usage instructions, please refer to:
  25
  26                  Documentation/ABI/testing/sysfs-bus-rbd
  27
  28  */
  29
  30 #include <linux/ceph/libceph.h>
  31 #include <linux/ceph/osd_client.h>
  32 #include <linux/ceph/mon_client.h>
  33 #include <linux/ceph/decode.h>
  34 #include <linux/parser.h>
  35
  36 #include <linux/kernel.h>
  37 #include <linux/device.h>
  38 #include <linux/module.h>
  39 #include <linux/fs.h>
  40 #include <linux/blkdev.h>
  41
  42 #include "rbd_types.h"
  43
  44 /*
  45  * The basic unit of block I/O is a sector.  It is interpreted in a
  46  * number of contexts in Linux (blk, bio, genhd), but the default is
  47  * universally 512 bytes.  These symbols are just slightly more
  48  * meaningful than the bare numbers they represent.
  49  */
  50 #define SECTOR_SHIFT    9
  51 #define SECTOR_SIZE     (1ULL << SECTOR_SHIFT)
  52
  53 #define RBD_DRV_NAME "rbd"
  54 #define RBD_DRV_NAME_LONG "rbd (rados block device)"
  55
  56 #define RBD_MINORS_PER_MAJOR    256             /* max minors per blkdev */
  57
  58 #define RBD_MAX_MD_NAME_LEN     (RBD_MAX_OBJ_NAME_LEN + sizeof(RBD_SUFFIX))
  59 #define RBD_MAX_POOL_NAME_LEN   64
  60 #define RBD_MAX_SNAP_NAME_LEN   32
  61 #define RBD_MAX_OPT_LEN         1024
  62
  63 #define RBD_SNAP_HEAD_NAME      "-"
  64
  65 /*
  66  * An RBD device name will be "rbd#", where the "rbd" comes from
  67  * RBD_DRV_NAME above, and # is a unique integer identifier.
  68  * MAX_INT_FORMAT_WIDTH is used in ensuring DEV_NAME_LEN is big
  69  * enough to hold all possible device names.
  70  */
  71 #define DEV_NAME_LEN            32
  72 #define MAX_INT_FORMAT_WIDTH    ((5 * sizeof (int)) / 2 + 1)
  73
  74 #define RBD_NOTIFY_TIMEOUT_DEFAULT 10
  75
  76 /*
  77  * block device image metadata (in-memory version)
  78  */
  79 struct rbd_image_header {
  80         u64 image_size;
  81         char object_prefix[32];
  82         __u8 obj_order;
  83         __u8 crypt_type;
  84         __u8 comp_type;
  85         struct ceph_snap_context *snapc;
  86         size_t snap_names_len;
  87         u64 snap_seq;
  88         u32 total_snaps;
  89
  90         char *snap_names;
  91         u64 *snap_sizes;
  92
  93         u64 obj_version;
  94 };
  95
  96 struct rbd_options {
  97         int     notify_timeout;
  98 };
  99
 100 /*
 101  * an instance of the client.  multiple devices may share an rbd client.
 102  */
 103 struct rbd_client {
 104         struct ceph_client      *client;
 105         struct rbd_options      *rbd_opts;
 106         struct kref             kref;
 107         struct list_head        node;
 108 };
 109
 110 /*
 111  * a request completion status
 112  */
 113 struct rbd_req_status {
 114         int done;
 115         int rc;
 116         u64 bytes;
 117 };
 118
 119 /*
 120  * a collection of requests
 121  */
 122 struct rbd_req_coll {
 123         int                     total;
 124         int                     num_done;
 125         struct kref             kref;
 126         struct rbd_req_status   status[0];
 127 };
 128
 129 /*
 130  * a single io request
 131  */
 132 struct rbd_request {
 133         struct request          *rq;            /* blk layer request */
 134         struct bio              *bio;           /* cloned bio */
 135         struct page             **pages;        /* list of used pages */
 136         u64                     len;
 137         int                     coll_index;
 138         struct rbd_req_coll     *coll;
 139 };
 140
 141 struct rbd_snap {
 142         struct  device          dev;
 143         const char              *name;
 144         u64                     size;
 145         struct list_head        node;
 146         u64                     id;
 147 };
 148
 149 /*
 150  * a single device
 151  */
 152 struct rbd_device {
 153         int                     id;             /* blkdev unique id */
 154
 155         int                     major;          /* blkdev assigned major */
 156         struct gendisk          *disk;          /* blkdev's gendisk and rq */
 157         struct request_queue    *q;
 158
 159         struct rbd_client       *rbd_client;
 160
 161         char                    name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */
 162
 163         spinlock_t              lock;           /* queue lock */
 164
 165         struct rbd_image_header header;
 166         char                    obj[RBD_MAX_OBJ_NAME_LEN]; /* rbd image name */
 167         int                     obj_len;
 168         char                    obj_md_name[RBD_MAX_MD_NAME_LEN]; /* hdr nm. */
 169         char                    pool_name[RBD_MAX_POOL_NAME_LEN];
 170         int                     poolid;
 171
 172         struct ceph_osd_event   *watch_event;
 173         struct ceph_osd_request *watch_request;
 174
 175         /* protects updating the header */
 176         struct rw_semaphore     header_rwsem;
 177         char                    snap_name[RBD_MAX_SNAP_NAME_LEN];
 178         u64                     snap_id;        /* current snapshot id */
 179         int read_only;
 180
 181         struct list_head        node;
 182
 183         /* list of snapshots */
 184         struct list_head        snaps;
 185
 186         /* sysfs related */
 187         struct device           dev;
 188 };
 189
 190 static DEFINE_MUTEX(ctl_mutex);   /* Serialize open/close/setup/teardown */
 191
 192 static LIST_HEAD(rbd_dev_list);    /* devices */
 193 static DEFINE_SPINLOCK(rbd_dev_list_lock);
 194
 195 static LIST_HEAD(rbd_client_list);              /* clients */
 196 static DEFINE_SPINLOCK(rbd_client_list_lock);
 197
 198 static int __rbd_init_snaps_header(struct rbd_device *rbd_dev);
 199 static void rbd_dev_release(struct device *dev);
 200 static ssize_t rbd_snap_add(struct device *dev,
 201                             struct device_attribute *attr,
 202                             const char *buf,
 203                             size_t count);
 204 static void __rbd_remove_snap_dev(struct rbd_device *rbd_dev,
 205                                   struct rbd_snap *snap);
 206
 207 static ssize_t rbd_add(struct bus_type *bus, const char *buf,
 208                        size_t count);
 209 static ssize_t rbd_remove(struct bus_type *bus, const char *buf,
 210                           size_t count);
 211
 212 static struct bus_attribute rbd_bus_attrs[] = {
 213         __ATTR(add, S_IWUSR, NULL, rbd_add),
 214         __ATTR(remove, S_IWUSR, NULL, rbd_remove),
 215         __ATTR_NULL
 216 };
 217
 218 static struct bus_type rbd_bus_type = {
 219         .name           = "rbd",
 220         .bus_attrs      = rbd_bus_attrs,
 221 };
 222
 223 static void rbd_root_dev_release(struct device *dev)
 224 {
 225 }
 226
 227 static struct device rbd_root_dev = {
 228         .init_name =    "rbd",
 229         .release =      rbd_root_dev_release,
 230 };
 231
 232
 233 static struct device *rbd_get_dev(struct rbd_device *rbd_dev)
 234 {
 235         return get_device(&rbd_dev->dev);
 236 }
 237
 238 static void rbd_put_dev(struct rbd_device *rbd_dev)
 239 {
 240         put_device(&rbd_dev->dev);
 241 }
 242
 243 static int __rbd_refresh_header(struct rbd_device *rbd_dev);
 244
 245 static int rbd_open(struct block_device *bdev, fmode_t mode)
 246 {
 247         struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
 248
 249         rbd_get_dev(rbd_dev);
 250
 251         set_device_ro(bdev, rbd_dev->read_only);
 252
 253         if ((mode & FMODE_WRITE) && rbd_dev->read_only)
 254                 return -EROFS;
 255
 256         return 0;
 257 }
 258
 259 static int rbd_release(struct gendisk *disk, fmode_t mode)
 260 {
 261         struct rbd_device *rbd_dev = disk->private_data;
 262
 263         rbd_put_dev(rbd_dev);
 264
 265         return 0;
 266 }
 267
 268 static const struct block_device_operations rbd_bd_ops = {
 269         .owner                  = THIS_MODULE,
 270         .open                   = rbd_open,
 271         .release                = rbd_release,
 272 };
 273
 274 /*
 275  * Initialize an rbd client instance.
 276  * We own *opt.
 277  */
 278 static struct rbd_client *rbd_client_create(struct ceph_options *opt,
 279                                             struct rbd_options *rbd_opts)
 280 {
 281         struct rbd_client *rbdc;
 282         int ret = -ENOMEM;
 283
 284         dout("rbd_client_create\n");
 285         rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
 286         if (!rbdc)
 287                 goto out_opt;
 288
 289         kref_init(&rbdc->kref);
 290         INIT_LIST_HEAD(&rbdc->node);
 291
 292         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
 293
 294         rbdc->client = ceph_create_client(opt, rbdc, 0, 0);
 295         if (IS_ERR(rbdc->client))
 296                 goto out_mutex;
 297         opt = NULL; /* Now rbdc->client is responsible for opt */
 298
 299         ret = ceph_open_session(rbdc->client);
 300         if (ret < 0)
 301                 goto out_err;
 302
 303         rbdc->rbd_opts = rbd_opts;
 304
 305         spin_lock(&rbd_client_list_lock);
 306         list_add_tail(&rbdc->node, &rbd_client_list);
 307         spin_unlock(&rbd_client_list_lock);
 308
 309         mutex_unlock(&ctl_mutex);
 310
 311         dout("rbd_client_create created %p\n", rbdc);
 312         return rbdc;
 313
 314 out_err:
 315         ceph_destroy_client(rbdc->client);
 316 out_mutex:
 317         mutex_unlock(&ctl_mutex);
 318         kfree(rbdc);
 319 out_opt:
 320         if (opt)
 321                 ceph_destroy_options(opt);
 322         return ERR_PTR(ret);
 323 }
 324
 325 /*
 326  * Find a ceph client with specific addr and configuration.
 327  */
 328 static struct rbd_client *__rbd_client_find(struct ceph_options *opt)
 329 {
 330         struct rbd_client *client_node;
 331
 332         if (opt->flags & CEPH_OPT_NOSHARE)
 333                 return NULL;
 334
 335         list_for_each_entry(client_node, &rbd_client_list, node)
 336                 if (ceph_compare_options(opt, client_node->client) == 0)
 337                         return client_node;
 338         return NULL;
 339 }
 340
 341 /*
 342  * mount options
 343  */
 344 enum {
 345         Opt_notify_timeout,
 346         Opt_last_int,
 347         /* int args above */
 348         Opt_last_string,
 349         /* string args above */
 350 };
 351
 352 static match_table_t rbdopt_tokens = {
 353         {Opt_notify_timeout, "notify_timeout=%d"},
 354         /* int args above */
 355         /* string args above */
 356         {-1, NULL}
 357 };
 358
 359 static int parse_rbd_opts_token(char *c, void *private)
 360 {
 361         struct rbd_options *rbdopt = private;
 362         substring_t argstr[MAX_OPT_ARGS];
 363         int token, intval, ret;
 364
 365         token = match_token(c, rbdopt_tokens, argstr);
 366         if (token < 0)
 367                 return -EINVAL;
 368
 369         if (token < Opt_last_int) {
 370                 ret = match_int(&argstr[0], &intval);
 371                 if (ret < 0) {
 372                         pr_err("bad mount option arg (not int) "
 373                                "at '%s'\n", c);
 374                         return ret;
 375                 }
 376                 dout("got int token %d val %d\n", token, intval);
 377         } else if (token > Opt_last_int && token < Opt_last_string) {
 378                 dout("got string token %d val %s\n", token,
 379                      argstr[0].from);
 380         } else {
 381                 dout("got token %d\n", token);
 382         }
 383
 384         switch (token) {
 385         case Opt_notify_timeout:
 386                 rbdopt->notify_timeout = intval;
 387                 break;
 388         default:
 389                 BUG_ON(token);
 390         }
 391         return 0;
 392 }
 393
 394 /*
 395  * Get a ceph client with specific addr and configuration, if one does
 396  * not exist create it.
 397  */
 398 static struct rbd_client *rbd_get_client(const char *mon_addr,
 399                                          size_t mon_addr_len,
 400                                          char *options)
 401 {
 402         struct rbd_client *rbdc;
 403         struct ceph_options *opt;
 404         struct rbd_options *rbd_opts;
 405
 406         rbd_opts = kzalloc(sizeof(*rbd_opts), GFP_KERNEL);
 407         if (!rbd_opts)
 408                 return ERR_PTR(-ENOMEM);
 409
 410         rbd_opts->notify_timeout = RBD_NOTIFY_TIMEOUT_DEFAULT;
 411
 412         opt = ceph_parse_options(options, mon_addr,
 413                                 mon_addr + mon_addr_len,
 414                                 parse_rbd_opts_token, rbd_opts);
 415         if (IS_ERR(opt)) {
 416                 kfree(rbd_opts);
 417                 return ERR_CAST(opt);
 418         }
 419
 420         spin_lock(&rbd_client_list_lock);
 421         rbdc = __rbd_client_find(opt);
 422         if (rbdc) {
 423                 /* using an existing client */
 424                 kref_get(&rbdc->kref);
 425                 spin_unlock(&rbd_client_list_lock);
 426
 427                 ceph_destroy_options(opt);
 428                 kfree(rbd_opts);
 429
 430                 return rbdc;
 431         }
 432         spin_unlock(&rbd_client_list_lock);
 433
 434         rbdc = rbd_client_create(opt, rbd_opts);
 435
 436         if (IS_ERR(rbdc))
 437                 kfree(rbd_opts);
 438
 439         return rbdc;
 440 }
 441
 442 /*
 443  * Destroy ceph client
 444  *
 445  * Caller must hold rbd_client_list_lock.
 446  */
 447 static void rbd_client_release(struct kref *kref)
 448 {
 449         struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
 450
 451         dout("rbd_release_client %p\n", rbdc);
 452         spin_lock(&rbd_client_list_lock);
 453         list_del(&rbdc->node);
 454         spin_unlock(&rbd_client_list_lock);
 455
 456         ceph_destroy_client(rbdc->client);
 457         kfree(rbdc->rbd_opts);
 458         kfree(rbdc);
 459 }
 460
 461 /*
 462  * Drop reference to ceph client node. If it's not referenced anymore, release
 463  * it.
 464  */
 465 static void rbd_put_client(struct rbd_device *rbd_dev)
 466 {
 467         kref_put(&rbd_dev->rbd_client->kref, rbd_client_release);
 468         rbd_dev->rbd_client = NULL;
 469 }
 470
 471 /*
 472  * Destroy requests collection
 473  */
 474 static void rbd_coll_release(struct kref *kref)
 475 {
 476         struct rbd_req_coll *coll =
 477                 container_of(kref, struct rbd_req_coll, kref);
 478
 479         dout("rbd_coll_release %p\n", coll);
 480         kfree(coll);
 481 }
 482
 483 /*
 484  * Create a new header structure, translate header format from the on-disk
 485  * header.
 486  */
 487 static int rbd_header_from_disk(struct rbd_image_header *header,
 488                                  struct rbd_image_header_ondisk *ondisk,
 489                                  u32 allocated_snaps,
 490                                  gfp_t gfp_flags)
 491 {
 492         u32 i, snap_count;
 493
 494         if (memcmp(ondisk, RBD_HEADER_TEXT, sizeof(RBD_HEADER_TEXT)))
 495                 return -ENXIO;
 496
 497         snap_count = le32_to_cpu(ondisk->snap_count);
 498         if (snap_count > (UINT_MAX - sizeof(struct ceph_snap_context))
 499                          / sizeof (*ondisk))
 500                 return -EINVAL;
 501         header->snapc = kmalloc(sizeof(struct ceph_snap_context) +
 502                                 snap_count * sizeof(u64),
 503                                 gfp_flags);
 504         if (!header->snapc)
 505                 return -ENOMEM;
 506
 507         header->snap_names_len = le64_to_cpu(ondisk->snap_names_len);
 508         if (snap_count) {
 509                 header->snap_names = kmalloc(header->snap_names_len,
 510                                              gfp_flags);
 511                 if (!header->snap_names)
 512                         goto err_snapc;
 513                 header->snap_sizes = kmalloc(snap_count * sizeof(u64),
 514                                              gfp_flags);
 515                 if (!header->snap_sizes)
 516                         goto err_names;
 517         } else {
 518                 header->snap_names = NULL;
 519                 header->snap_sizes = NULL;
 520         }
 521         memcpy(header->object_prefix, ondisk->block_name,
 522                sizeof(ondisk->block_name));
 523
 524         header->image_size = le64_to_cpu(ondisk->image_size);
 525         header->obj_order = ondisk->options.order;
 526         header->crypt_type = ondisk->options.crypt_type;
 527         header->comp_type = ondisk->options.comp_type;
 528
 529         atomic_set(&header->snapc->nref, 1);
 530         header->snap_seq = le64_to_cpu(ondisk->snap_seq);
 531         header->snapc->num_snaps = snap_count;
 532         header->total_snaps = snap_count;
 533
 534         if (snap_count && allocated_snaps == snap_count) {
 535                 for (i = 0; i < snap_count; i++) {
 536                         header->snapc->snaps[i] =
 537                                 le64_to_cpu(ondisk->snaps[i].id);
 538                         header->snap_sizes[i] =
 539                                 le64_to_cpu(ondisk->snaps[i].image_size);
 540                 }
 541
 542                 /* copy snapshot names */
 543                 memcpy(header->snap_names, &ondisk->snaps[i],
 544                         header->snap_names_len);
 545         }
 546
 547         return 0;
 548
 549 err_names:
 550         kfree(header->snap_names);
 551 err_snapc:
 552         kfree(header->snapc);
 553         return -ENOMEM;
 554 }
 555
 556 static int snap_by_name(struct rbd_image_header *header, const char *snap_name,
 557                         u64 *seq, u64 *size)
 558 {
 559         int i;
 560         char *p = header->snap_names;
 561
 562         for (i = 0; i < header->total_snaps; i++) {
 563                 if (!strcmp(snap_name, p)) {
 564
 565                         /* Found it.  Pass back its id and/or size */
 566
 567                         if (seq)
 568                                 *seq = header->snapc->snaps[i];
 569                         if (size)
 570                                 *size = header->snap_sizes[i];
 571                         return i;
 572                 }
 573                 p += strlen(p) + 1;     /* Skip ahead to the next name */
 574         }
 575         return -ENOENT;
 576 }
 577
 578 static int rbd_header_set_snap(struct rbd_device *dev, u64 *size)
 579 {
 580         struct rbd_image_header *header = &dev->header;
 581         struct ceph_snap_context *snapc = header->snapc;
 582         int ret = -ENOENT;
 583
 584         BUILD_BUG_ON(sizeof (dev->snap_name) < sizeof (RBD_SNAP_HEAD_NAME));
 585
 586         down_write(&dev->header_rwsem);
 587
 588         if (!memcmp(dev->snap_name, RBD_SNAP_HEAD_NAME,
 589                     sizeof (RBD_SNAP_HEAD_NAME))) {
 590                 if (header->total_snaps)
 591                         snapc->seq = header->snap_seq;
 592                 else
 593                         snapc->seq = 0;
 594                 dev->snap_id = CEPH_NOSNAP;
 595                 dev->read_only = 0;
 596                 if (size)
 597                         *size = header->image_size;
 598         } else {
 599                 ret = snap_by_name(header, dev->snap_name, &snapc->seq, size);
 600                 if (ret < 0)
 601                         goto done;
 602                 dev->snap_id = snapc->seq;
 603                 dev->read_only = 1;
 604         }
 605
 606         ret = 0;
 607 done:
 608         up_write(&dev->header_rwsem);
 609         return ret;
 610 }
 611
 612 static void rbd_header_free(struct rbd_image_header *header)
 613 {
 614         kfree(header->snapc);
 615         kfree(header->snap_names);
 616         kfree(header->snap_sizes);
 617 }
 618
 619 /*
 620  * get the actual striped segment name, offset and length
 621  */
 622 static u64 rbd_get_segment(struct rbd_image_header *header,
 623                            const char *object_prefix,
 624                            u64 ofs, u64 len,
 625                            char *seg_name, u64 *segofs)
 626 {
 627         u64 seg = ofs >> header->obj_order;
 628
 629         if (seg_name)
 630                 snprintf(seg_name, RBD_MAX_SEG_NAME_LEN,
 631                          "%s.%012llx", object_prefix, seg);
 632
 633         ofs = ofs & ((1 << header->obj_order) - 1);
 634         len = min_t(u64, len, (1 << header->obj_order) - ofs);
 635
 636         if (segofs)
 637                 *segofs = ofs;
 638
 639         return len;
 640 }
 641
 642 static int rbd_get_num_segments(struct rbd_image_header *header,
 643                                 u64 ofs, u64 len)
 644 {
 645         u64 start_seg = ofs >> header->obj_order;
 646         u64 end_seg = (ofs + len - 1) >> header->obj_order;
 647         return end_seg - start_seg + 1;
 648 }
 649
 650 /*
 651  * returns the size of an object in the image
 652  */
 653 static u64 rbd_obj_bytes(struct rbd_image_header *header)
 654 {
 655         return 1 << header->obj_order;
 656 }
 657
 658 /*
 659  * bio helpers
 660  */
 661
 662 static void bio_chain_put(struct bio *chain)
 663 {
 664         struct bio *tmp;
 665
 666         while (chain) {
 667                 tmp = chain;
 668                 chain = chain->bi_next;
 669                 bio_put(tmp);
 670         }
 671 }
 672
 673 /*
 674  * zeros a bio chain, starting at specific offset
 675  */
 676 static void zero_bio_chain(struct bio *chain, int start_ofs)
 677 {
 678         struct bio_vec *bv;
 679         unsigned long flags;
 680         void *buf;
 681         int i;
 682         int pos = 0;
 683
 684         while (chain) {
 685                 bio_for_each_segment(bv, chain, i) {
 686                         if (pos + bv->bv_len > start_ofs) {
 687                                 int remainder = max(start_ofs - pos, 0);
 688                                 buf = bvec_kmap_irq(bv, &flags);
 689                                 memset(buf + remainder, 0,
 690                                        bv->bv_len - remainder);
 691                                 bvec_kunmap_irq(buf, &flags);
 692                         }
 693                         pos += bv->bv_len;
 694                 }
 695
 696                 chain = chain->bi_next;
 697         }
 698 }
 699
 700 /*
 701  * bio_chain_clone - clone a chain of bios up to a certain length.
 702  * might return a bio_pair that will need to be released.
 703  */
 704 static struct bio *bio_chain_clone(struct bio **old, struct bio **next,
 705                                    struct bio_pair **bp,
 706                                    int len, gfp_t gfpmask)
 707 {
 708         struct bio *tmp, *old_chain = *old, *new_chain = NULL, *tail = NULL;
 709         int total = 0;
 710
 711         if (*bp) {
 712                 bio_pair_release(*bp);
 713                 *bp = NULL;
 714         }
 715
 716         while (old_chain && (total < len)) {
 717                 tmp = bio_kmalloc(gfpmask, old_chain->bi_max_vecs);
 718                 if (!tmp)
 719                         goto err_out;
 720
 721                 if (total + old_chain->bi_size > len) {
 722                         struct bio_pair *bp;
 723
 724                         /*
 725                          * this split can only happen with a single paged bio,
 726                          * split_bio will BUG_ON if this is not the case
 727                          */
 728                         dout("bio_chain_clone split! total=%d remaining=%d"
 729                              "bi_size=%d\n",
 730                              (int)total, (int)len-total,
 731                              (int)old_chain->bi_size);
 732
 733                         /* split the bio. We'll release it either in the next
 734                            call, or it will have to be released outside */
 735                         bp = bio_split(old_chain, (len - total) / SECTOR_SIZE);
 736                         if (!bp)
 737                                 goto err_out;
 738
 739                         __bio_clone(tmp, &bp->bio1);
 740
 741                         *next = &bp->bio2;
 742                 } else {
 743                         __bio_clone(tmp, old_chain);
 744                         *next = old_chain->bi_next;
 745                 }
 746
 747                 tmp->bi_bdev = NULL;
 748                 gfpmask &= ~__GFP_WAIT;
 749                 tmp->bi_next = NULL;
 750
 751                 if (!new_chain) {
 752                         new_chain = tail = tmp;
 753                 } else {
 754                         tail->bi_next = tmp;
 755                         tail = tmp;
 756                 }
 757                 old_chain = old_chain->bi_next;
 758
 759                 total += tmp->bi_size;
 760         }
 761
 762         BUG_ON(total < len);
 763
 764         if (tail)
 765                 tail->bi_next = NULL;
 766
 767         *old = old_chain;
 768
 769         return new_chain;
 770
 771 err_out:
 772         dout("bio_chain_clone with err\n");
 773         bio_chain_put(new_chain);
 774         return NULL;
 775 }
 776
 777 /*
 778  * helpers for osd request op vectors.
 779  */
 780 static int rbd_create_rw_ops(struct ceph_osd_req_op **ops,
 781                             int num_ops,
 782                             int opcode,
 783                             u32 payload_len)
 784 {
 785         *ops = kzalloc(sizeof(struct ceph_osd_req_op) * (num_ops + 1),
 786                        GFP_NOIO);
 787         if (!*ops)
 788                 return -ENOMEM;
 789         (*ops)[0].op = opcode;
 790         /*
 791          * op extent offset and length will be set later on
 792          * in calc_raw_layout()
 793          */
 794         (*ops)[0].payload_len = payload_len;
 795         return 0;
 796 }
 797
 798 static void rbd_destroy_ops(struct ceph_osd_req_op *ops)
 799 {
 800         kfree(ops);
 801 }
 802
 803 static void rbd_coll_end_req_index(struct request *rq,
 804                                    struct rbd_req_coll *coll,
 805                                    int index,
 806                                    int ret, u64 len)
 807 {
 808         struct request_queue *q;
 809         int min, max, i;
 810
 811         dout("rbd_coll_end_req_index %p index %d ret %d len %lld\n",
 812              coll, index, ret, len);
 813
 814         if (!rq)
 815                 return;
 816
 817         if (!coll) {
 818                 blk_end_request(rq, ret, len);
 819                 return;
 820         }
 821
 822         q = rq->q;
 823
 824         spin_lock_irq(q->queue_lock);
 825         coll->status[index].done = 1;
 826         coll->status[index].rc = ret;
 827         coll->status[index].bytes = len;
 828         max = min = coll->num_done;
 829         while (max < coll->total && coll->status[max].done)
 830                 max++;
 831
 832         for (i = min; i<max; i++) {
 833                 __blk_end_request(rq, coll->status[i].rc,
 834                                   coll->status[i].bytes);
 835                 coll->num_done++;
 836                 kref_put(&coll->kref, rbd_coll_release);
 837         }
 838         spin_unlock_irq(q->queue_lock);
 839 }
 840
 841 static void rbd_coll_end_req(struct rbd_request *req,
 842                              int ret, u64 len)
 843 {
 844         rbd_coll_end_req_index(req->rq, req->coll, req->coll_index, ret, len);
 845 }
 846
 847 /*
 848  * Send ceph osd request
 849  */
 850 static int rbd_do_request(struct request *rq,
 851                           struct rbd_device *dev,
 852                           struct ceph_snap_context *snapc,
 853                           u64 snapid,
 854                           const char *obj, u64 ofs, u64 len,
 855                           struct bio *bio,
 856                           struct page **pages,
 857                           int num_pages,
 858                           int flags,
 859                           struct ceph_osd_req_op *ops,
 860                           int num_reply,
 861                           struct rbd_req_coll *coll,
 862                           int coll_index,
 863                           void (*rbd_cb)(struct ceph_osd_request *req,
 864                                          struct ceph_msg *msg),
 865                           struct ceph_osd_request **linger_req,
 866                           u64 *ver)
 867 {
 868         struct ceph_osd_request *req;
 869         struct ceph_file_layout *layout;
 870         int ret;
 871         u64 bno;
 872         struct timespec mtime = CURRENT_TIME;
 873         struct rbd_request *req_data;
 874         struct ceph_osd_request_head *reqhead;
 875         struct ceph_osd_client *osdc;
 876
 877         req_data = kzalloc(sizeof(*req_data), GFP_NOIO);
 878         if (!req_data) {
 879                 if (coll)
 880                         rbd_coll_end_req_index(rq, coll, coll_index,
 881                                                -ENOMEM, len);
 882                 return -ENOMEM;
 883         }
 884
 885         if (coll) {
 886                 req_data->coll = coll;
 887                 req_data->coll_index = coll_index;
 888         }
 889
 890         dout("rbd_do_request obj=%s ofs=%lld len=%lld\n", obj, len, ofs);
 891
 892         down_read(&dev->header_rwsem);
 893
 894         osdc = &dev->rbd_client->client->osdc;
 895         req = ceph_osdc_alloc_request(osdc, flags, snapc, ops,
 896                                         false, GFP_NOIO, pages, bio);
 897         if (!req) {
 898                 up_read(&dev->header_rwsem);
 899                 ret = -ENOMEM;
 900                 goto done_pages;
 901         }
 902
 903         req->r_callback = rbd_cb;
 904
 905         req_data->rq = rq;
 906         req_data->bio = bio;
 907         req_data->pages = pages;
 908         req_data->len = len;
 909
 910         req->r_priv = req_data;
 911
 912         reqhead = req->r_request->front.iov_base;
 913         reqhead->snapid = cpu_to_le64(CEPH_NOSNAP);
 914
 915         strncpy(req->r_oid, obj, sizeof(req->r_oid));
 916         req->r_oid_len = strlen(req->r_oid);
 917
 918         layout = &req->r_file_layout;
 919         memset(layout, 0, sizeof(*layout));
 920         layout->fl_stripe_unit = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
 921         layout->fl_stripe_count = cpu_to_le32(1);
 922         layout->fl_object_size = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
 923         layout->fl_pg_pool = cpu_to_le32(dev->poolid);
 924         ceph_calc_raw_layout(osdc, layout, snapid, ofs, &len, &bno,
 925                                 req, ops);
 926
 927         ceph_osdc_build_request(req, ofs, &len,
 928                                 ops,
 929                                 snapc,
 930                                 &mtime,
 931                                 req->r_oid, req->r_oid_len);
 932         up_read(&dev->header_rwsem);
 933
 934         if (linger_req) {
 935                 ceph_osdc_set_request_linger(osdc, req);
 936                 *linger_req = req;
 937         }
 938
 939         ret = ceph_osdc_start_request(osdc, req, false);
 940         if (ret < 0)
 941                 goto done_err;
 942
 943         if (!rbd_cb) {
 944                 ret = ceph_osdc_wait_request(osdc, req);
 945                 if (ver)
 946                         *ver = le64_to_cpu(req->r_reassert_version.version);
 947                 dout("reassert_ver=%lld\n",
 948                      le64_to_cpu(req->r_reassert_version.version));
 949                 ceph_osdc_put_request(req);
 950         }
 951         return ret;
 952
 953 done_err:
 954         bio_chain_put(req_data->bio);
 955         ceph_osdc_put_request(req);
 956 done_pages:
 957         rbd_coll_end_req(req_data, ret, len);
 958         kfree(req_data);
 959         return ret;
 960 }
 961
 962 /*
 963  * Ceph osd op callback
 964  */
 965 static void rbd_req_cb(struct ceph_osd_request *req, struct ceph_msg *msg)
 966 {
 967         struct rbd_request *req_data = req->r_priv;
 968         struct ceph_osd_reply_head *replyhead;
 969         struct ceph_osd_op *op;
 970         __s32 rc;
 971         u64 bytes;
 972         int read_op;
 973
 974         /* parse reply */
 975         replyhead = msg->front.iov_base;
 976         WARN_ON(le32_to_cpu(replyhead->num_ops) == 0);
 977         op = (void *)(replyhead + 1);
 978         rc = le32_to_cpu(replyhead->result);
 979         bytes = le64_to_cpu(op->extent.length);
 980         read_op = (le16_to_cpu(op->op) == CEPH_OSD_OP_READ);
 981
 982         dout("rbd_req_cb bytes=%lld readop=%d rc=%d\n", bytes, read_op, rc);
 983
 984         if (rc == -ENOENT && read_op) {
 985                 zero_bio_chain(req_data->bio, 0);
 986                 rc = 0;
 987         } else if (rc == 0 && read_op && bytes < req_data->len) {
 988                 zero_bio_chain(req_data->bio, bytes);
 989                 bytes = req_data->len;
 990         }
 991
 992         rbd_coll_end_req(req_data, rc, bytes);
 993
 994         if (req_data->bio)
 995                 bio_chain_put(req_data->bio);
 996
 997         ceph_osdc_put_request(req);
 998         kfree(req_data);
 999 }
1000
1001 static void rbd_simple_req_cb(struct ceph_osd_request *req, struct ceph_msg *msg)
1002 {
1003         ceph_osdc_put_request(req);
1004 }
1005
1006 /*
1007  * Do a synchronous ceph osd operation
1008  */
1009 static int rbd_req_sync_op(struct rbd_device *dev,
1010                            struct ceph_snap_context *snapc,
1011                            u64 snapid,
1012                            int opcode,
1013                            int flags,
1014                            struct ceph_osd_req_op *orig_ops,
1015                            int num_reply,
1016                            const char *obj,
1017                            u64 ofs, u64 len,
1018                            char *buf,
1019                            struct ceph_osd_request **linger_req,
1020                            u64 *ver)
1021 {
1022         int ret;
1023         struct page **pages;
1024         int num_pages;
1025         struct ceph_osd_req_op *ops = orig_ops;
1026         u32 payload_len;
1027
1028         num_pages = calc_pages_for(ofs , len);
1029         pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
1030         if (IS_ERR(pages))
1031                 return PTR_ERR(pages);
1032
1033         if (!orig_ops) {
1034                 payload_len = (flags & CEPH_OSD_FLAG_WRITE ? len : 0);
1035                 ret = rbd_create_rw_ops(&ops, 1, opcode, payload_len);
1036                 if (ret < 0)
1037                         goto done;
1038
1039                 if ((flags & CEPH_OSD_FLAG_WRITE) && buf) {
1040                         ret = ceph_copy_to_page_vector(pages, buf, ofs, len);
1041                         if (ret < 0)
1042                                 goto done_ops;
1043                 }
1044         }
1045
1046         ret = rbd_do_request(NULL, dev, snapc, snapid,
1047                           obj, ofs, len, NULL,
1048                           pages, num_pages,
1049                           flags,
1050                           ops,
1051                           2,
1052                           NULL, 0,
1053                           NULL,
1054                           linger_req, ver);
1055         if (ret < 0)
1056                 goto done_ops;
1057
1058         if ((flags & CEPH_OSD_FLAG_READ) && buf)
1059                 ret = ceph_copy_from_page_vector(pages, buf, ofs, ret);
1060
1061 done_ops:
1062         if (!orig_ops)
1063                 rbd_destroy_ops(ops);
1064 done:
1065         ceph_release_page_vector(pages, num_pages);
1066         return ret;
1067 }
1068
1069 /*
1070  * Do an asynchronous ceph osd operation
1071  */
1072 static int rbd_do_op(struct request *rq,
1073                      struct rbd_device *rbd_dev ,
1074                      struct ceph_snap_context *snapc,
1075                      u64 snapid,
1076                      int opcode, int flags, int num_reply,
1077                      u64 ofs, u64 len,
1078                      struct bio *bio,
1079                      struct rbd_req_coll *coll,
1080                      int coll_index)
1081 {
1082         char *seg_name;
1083         u64 seg_ofs;
1084         u64 seg_len;
1085         int ret;
1086         struct ceph_osd_req_op *ops;
1087         u32 payload_len;
1088
1089         seg_name = kmalloc(RBD_MAX_SEG_NAME_LEN + 1, GFP_NOIO);
1090         if (!seg_name)
1091                 return -ENOMEM;
1092
1093         seg_len = rbd_get_segment(&rbd_dev->header,
1094                                   rbd_dev->header.object_prefix,
1095                                   ofs, len,
1096                                   seg_name, &seg_ofs);
1097
1098         payload_len = (flags & CEPH_OSD_FLAG_WRITE ? seg_len : 0);
1099
1100         ret = rbd_create_rw_ops(&ops, 1, opcode, payload_len);
1101         if (ret < 0)
1102                 goto done;
1103
1104         /* we've taken care of segment sizes earlier when we
1105            cloned the bios. We should never have a segment
1106            truncated at this point */
1107         BUG_ON(seg_len < len);
1108
1109         ret = rbd_do_request(rq, rbd_dev, snapc, snapid,
1110                              seg_name, seg_ofs, seg_len,
1111                              bio,
1112                              NULL, 0,
1113                              flags,
1114                              ops,
1115                              num_reply,
1116                              coll, coll_index,
1117                              rbd_req_cb, 0, NULL);
1118
1119         rbd_destroy_ops(ops);
1120 done:
1121         kfree(seg_name);
1122         return ret;
1123 }
1124
1125 /*
1126  * Request async osd write
1127  */
1128 static int rbd_req_write(struct request *rq,
1129                          struct rbd_device *rbd_dev,
1130                          struct ceph_snap_context *snapc,
1131                          u64 ofs, u64 len,
1132                          struct bio *bio,
1133                          struct rbd_req_coll *coll,
1134                          int coll_index)
1135 {
1136         return rbd_do_op(rq, rbd_dev, snapc, CEPH_NOSNAP,
1137                          CEPH_OSD_OP_WRITE,
1138                          CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1139                          2,
1140                          ofs, len, bio, coll, coll_index);
1141 }
1142
1143 /*
1144  * Request async osd read
1145  */
1146 static int rbd_req_read(struct request *rq,
1147                          struct rbd_device *rbd_dev,
1148                          u64 snapid,
1149                          u64 ofs, u64 len,
1150                          struct bio *bio,
1151                          struct rbd_req_coll *coll,
1152                          int coll_index)
1153 {
1154         return rbd_do_op(rq, rbd_dev, NULL,
1155                          snapid,
1156                          CEPH_OSD_OP_READ,
1157                          CEPH_OSD_FLAG_READ,
1158                          2,
1159                          ofs, len, bio, coll, coll_index);
1160 }
1161
1162 /*
1163  * Request sync osd read
1164  */
1165 static int rbd_req_sync_read(struct rbd_device *dev,
1166                           struct ceph_snap_context *snapc,
1167                           u64 snapid,
1168                           const char *obj,
1169                           u64 ofs, u64 len,
1170                           char *buf,
1171                           u64 *ver)
1172 {
1173         return rbd_req_sync_op(dev, NULL,
1174                                snapid,
1175                                CEPH_OSD_OP_READ,
1176                                CEPH_OSD_FLAG_READ,
1177                                NULL,
1178                                1, obj, ofs, len, buf, NULL, ver);
1179 }
1180
1181 /*
1182  * Request sync osd watch
1183  */
1184 static int rbd_req_sync_notify_ack(struct rbd_device *dev,
1185                                    u64 ver,
1186                                    u64 notify_id,
1187                                    const char *obj)
1188 {
1189         struct ceph_osd_req_op *ops;
1190         int ret;
1191
1192         ret = rbd_create_rw_ops(&ops, 1, CEPH_OSD_OP_NOTIFY_ACK, 0);
1193         if (ret < 0)
1194                 return ret;
1195
1196         ops[0].watch.ver = cpu_to_le64(dev->header.obj_version);
1197         ops[0].watch.cookie = notify_id;
1198         ops[0].watch.flag = 0;
1199
1200         ret = rbd_do_request(NULL, dev, NULL, CEPH_NOSNAP,
1201                           obj, 0, 0, NULL,
1202                           NULL, 0,
1203                           CEPH_OSD_FLAG_READ,
1204                           ops,
1205                           1,
1206                           NULL, 0,
1207                           rbd_simple_req_cb, 0, NULL);
1208
1209         rbd_destroy_ops(ops);
1210         return ret;
1211 }
1212
1213 static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
1214 {
1215         struct rbd_device *dev = (struct rbd_device *)data;
1216         int rc;
1217
1218         if (!dev)
1219                 return;
1220
1221         dout("rbd_watch_cb %s notify_id=%lld opcode=%d\n", dev->obj_md_name,
1222                 notify_id, (int)opcode);
1223         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
1224         rc = __rbd_refresh_header(dev);
1225         mutex_unlock(&ctl_mutex);
1226         if (rc)
1227                 pr_warning(RBD_DRV_NAME "%d got notification but failed to "
1228                            " update snaps: %d\n", dev->major, rc);
1229
1230         rbd_req_sync_notify_ack(dev, ver, notify_id, dev->obj_md_name);
1231 }
1232
1233 /*
1234  * Request sync osd watch
1235  */
1236 static int rbd_req_sync_watch(struct rbd_device *dev,
1237                               const char *obj,
1238                               u64 ver)
1239 {
1240         struct ceph_osd_req_op *ops;
1241         struct ceph_osd_client *osdc = &dev->rbd_client->client->osdc;
1242
1243         int ret = rbd_create_rw_ops(&ops, 1, CEPH_OSD_OP_WATCH, 0);
1244         if (ret < 0)
1245                 return ret;
1246
1247         ret = ceph_osdc_create_event(osdc, rbd_watch_cb, 0,
1248                                      (void *)dev, &dev->watch_event);
1249         if (ret < 0)
1250                 goto fail;
1251
1252         ops[0].watch.ver = cpu_to_le64(ver);
1253         ops[0].watch.cookie = cpu_to_le64(dev->watch_event->cookie);
1254         ops[0].watch.flag = 1;
1255
1256         ret = rbd_req_sync_op(dev, NULL,
1257                               CEPH_NOSNAP,
1258                               0,
1259                               CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1260                               ops,
1261                               1, obj, 0, 0, NULL,
1262                               &dev->watch_request, NULL);
1263
1264         if (ret < 0)
1265                 goto fail_event;
1266
1267         rbd_destroy_ops(ops);
1268         return 0;
1269
1270 fail_event:
1271         ceph_osdc_cancel_event(dev->watch_event);
1272         dev->watch_event = NULL;
1273 fail:
1274         rbd_destroy_ops(ops);
1275         return ret;
1276 }
1277
1278 /*
1279  * Request sync osd unwatch
1280  */
1281 static int rbd_req_sync_unwatch(struct rbd_device *dev,
1282                                 const char *obj)
1283 {
1284         struct ceph_osd_req_op *ops;
1285
1286         int ret = rbd_create_rw_ops(&ops, 1, CEPH_OSD_OP_WATCH, 0);
1287         if (ret < 0)
1288                 return ret;
1289
1290         ops[0].watch.ver = 0;
1291         ops[0].watch.cookie = cpu_to_le64(dev->watch_event->cookie);
1292         ops[0].watch.flag = 0;
1293
1294         ret = rbd_req_sync_op(dev, NULL,
1295                               CEPH_NOSNAP,
1296                               0,
1297                               CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1298                               ops,
1299                               1, obj, 0, 0, NULL, NULL, NULL);
1300
1301         rbd_destroy_ops(ops);
1302         ceph_osdc_cancel_event(dev->watch_event);
1303         dev->watch_event = NULL;
1304         return ret;
1305 }
1306
1307 struct rbd_notify_info {
1308         struct rbd_device *dev;
1309 };
1310
1311 static void rbd_notify_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
1312 {
1313         struct rbd_device *dev = (struct rbd_device *)data;
1314         if (!dev)
1315                 return;
1316
1317         dout("rbd_notify_cb %s notify_id=%lld opcode=%d\n", dev->obj_md_name,
1318                 notify_id, (int)opcode);
1319 }
1320
1321 /*
1322  * Request sync osd notify
1323  */
1324 static int rbd_req_sync_notify(struct rbd_device *dev,
1325                           const char *obj)
1326 {
1327         struct ceph_osd_req_op *ops;
1328         struct ceph_osd_client *osdc = &dev->rbd_client->client->osdc;
1329         struct ceph_osd_event *event;
1330         struct rbd_notify_info info;
1331         int payload_len = sizeof(u32) + sizeof(u32);
1332         int ret;
1333
1334         ret = rbd_create_rw_ops(&ops, 1, CEPH_OSD_OP_NOTIFY, payload_len);
1335         if (ret < 0)
1336                 return ret;
1337
1338         info.dev = dev;
1339
1340         ret = ceph_osdc_create_event(osdc, rbd_notify_cb, 1,
1341                                      (void *)&info, &event);
1342         if (ret < 0)
1343                 goto fail;
1344
1345         ops[0].watch.ver = 1;
1346         ops[0].watch.flag = 1;
1347         ops[0].watch.cookie = event->cookie;
1348         ops[0].watch.prot_ver = RADOS_NOTIFY_VER;
1349         ops[0].watch.timeout = 12;
1350
1351         ret = rbd_req_sync_op(dev, NULL,
1352                                CEPH_NOSNAP,
1353                                0,
1354                                CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1355                                ops,
1356                                1, obj, 0, 0, NULL, NULL, NULL);
1357         if (ret < 0)
1358                 goto fail_event;
1359
1360         ret = ceph_osdc_wait_event(event, CEPH_OSD_TIMEOUT_DEFAULT);
1361         dout("ceph_osdc_wait_event returned %d\n", ret);
1362         rbd_destroy_ops(ops);
1363         return 0;
1364
1365 fail_event:
1366         ceph_osdc_cancel_event(event);
1367 fail:
1368         rbd_destroy_ops(ops);
1369         return ret;
1370 }
1371
1372 /*
1373  * Request sync osd read
1374  */
1375 static int rbd_req_sync_exec(struct rbd_device *dev,
1376                              const char *obj,
1377                              const char *cls,
1378                              const char *method,
1379                              const char *data,
1380                              int len,
1381                              u64 *ver)
1382 {
1383         struct ceph_osd_req_op *ops;
1384         int cls_len = strlen(cls);
1385         int method_len = strlen(method);
1386         int ret = rbd_create_rw_ops(&ops, 1, CEPH_OSD_OP_CALL,
1387                                     cls_len + method_len + len);
1388         if (ret < 0)
1389                 return ret;
1390
1391         ops[0].cls.class_name = cls;
1392         ops[0].cls.class_len = (__u8)cls_len;
1393         ops[0].cls.method_name = method;
1394         ops[0].cls.method_len = (__u8)method_len;
1395         ops[0].cls.argc = 0;
1396         ops[0].cls.indata = data;
1397         ops[0].cls.indata_len = len;
1398
1399         ret = rbd_req_sync_op(dev, NULL,
1400                                CEPH_NOSNAP,
1401                                0,
1402                                CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1403                                ops,
1404                                1, obj, 0, 0, NULL, NULL, ver);
1405
1406         rbd_destroy_ops(ops);
1407
1408         dout("cls_exec returned %d\n", ret);
1409         return ret;
1410 }
1411
1412 static struct rbd_req_coll *rbd_alloc_coll(int num_reqs)
1413 {
1414         struct rbd_req_coll *coll =
1415                         kzalloc(sizeof(struct rbd_req_coll) +
1416                                 sizeof(struct rbd_req_status) * num_reqs,
1417                                 GFP_ATOMIC);
1418
1419         if (!coll)
1420                 return NULL;
1421         coll->total = num_reqs;
1422         kref_init(&coll->kref);
1423         return coll;
1424 }
1425
1426 /*
1427  * block device queue callback
1428  */
1429 static void rbd_rq_fn(struct request_queue *q)
1430 {
1431         struct rbd_device *rbd_dev = q->queuedata;
1432         struct request *rq;
1433         struct bio_pair *bp = NULL;
1434
1435         while ((rq = blk_fetch_request(q))) {
1436                 struct bio *bio;
1437                 struct bio *rq_bio, *next_bio = NULL;
1438                 bool do_write;
1439                 int size, op_size = 0;
1440                 u64 ofs;
1441                 int num_segs, cur_seg = 0;
1442                 struct rbd_req_coll *coll;
1443
1444                 /* peek at request from block layer */
1445                 if (!rq)
1446                         break;
1447
1448                 dout("fetched request\n");
1449
1450                 /* filter out block requests we don't understand */
1451                 if ((rq->cmd_type != REQ_TYPE_FS)) {
1452                         __blk_end_request_all(rq, 0);
1453                         continue;
1454                 }
1455
1456                 /* deduce our operation (read, write) */
1457                 do_write = (rq_data_dir(rq) == WRITE);
1458
1459                 size = blk_rq_bytes(rq);
1460                 ofs = blk_rq_pos(rq) * SECTOR_SIZE;
1461                 rq_bio = rq->bio;
1462                 if (do_write && rbd_dev->read_only) {
1463                         __blk_end_request_all(rq, -EROFS);
1464                         continue;
1465                 }
1466
1467                 spin_unlock_irq(q->queue_lock);
1468
1469                 dout("%s 0x%x bytes at 0x%llx\n",
1470                      do_write ? "write" : "read",
1471                      size, blk_rq_pos(rq) * SECTOR_SIZE);
1472
1473                 num_segs = rbd_get_num_segments(&rbd_dev->header, ofs, size);
1474                 coll = rbd_alloc_coll(num_segs);
1475                 if (!coll) {
1476                         spin_lock_irq(q->queue_lock);
1477                         __blk_end_request_all(rq, -ENOMEM);
1478                         continue;
1479                 }
1480
1481                 do {
1482                         /* a bio clone to be passed down to OSD req */
1483                         dout("rq->bio->bi_vcnt=%d\n", rq->bio->bi_vcnt);
1484                         op_size = rbd_get_segment(&rbd_dev->header,
1485                                                   rbd_dev->header.object_prefix,
1486                                                   ofs, size,
1487                                                   NULL, NULL);
1488                         kref_get(&coll->kref);
1489                         bio = bio_chain_clone(&rq_bio, &next_bio, &bp,
1490                                               op_size, GFP_ATOMIC);
1491                         if (!bio) {
1492                                 rbd_coll_end_req_index(rq, coll, cur_seg,
1493                                                        -ENOMEM, op_size);
1494                                 goto next_seg;
1495                         }
1496
1497
1498                         /* init OSD command: write or read */
1499                         if (do_write)
1500                                 rbd_req_write(rq, rbd_dev,
1501                                               rbd_dev->header.snapc,
1502                                               ofs,
1503                                               op_size, bio,
1504                                               coll, cur_seg);
1505                         else
1506                                 rbd_req_read(rq, rbd_dev,
1507                                              rbd_dev->snap_id,
1508                                              ofs,
1509                                              op_size, bio,
1510                                              coll, cur_seg);
1511
1512 next_seg:
1513                         size -= op_size;
1514                         ofs += op_size;
1515
1516                         cur_seg++;
1517                         rq_bio = next_bio;
1518                 } while (size > 0);
1519                 kref_put(&coll->kref, rbd_coll_release);
1520
1521                 if (bp)
1522                         bio_pair_release(bp);
1523                 spin_lock_irq(q->queue_lock);
1524         }
1525 }
1526
1527 /*
1528  * a queue callback. Makes sure that we don't create a bio that spans across
1529  * multiple osd objects. One exception would be with a single page bios,
1530  * which we handle later at bio_chain_clone
1531  */
1532 static int rbd_merge_bvec(struct request_queue *q, struct bvec_merge_data *bmd,
1533                           struct bio_vec *bvec)
1534 {
1535         struct rbd_device *rbd_dev = q->queuedata;
1536         unsigned int chunk_sectors;
1537         sector_t sector;
1538         unsigned int bio_sectors;
1539         int max;
1540
1541         chunk_sectors = 1 << (rbd_dev->header.obj_order - SECTOR_SHIFT);
1542         sector = bmd->bi_sector + get_start_sect(bmd->bi_bdev);
1543         bio_sectors = bmd->bi_size >> SECTOR_SHIFT;
1544
1545         max =  (chunk_sectors - ((sector & (chunk_sectors - 1))
1546                                  + bio_sectors)) << SECTOR_SHIFT;
1547         if (max < 0)
1548                 max = 0; /* bio_add cannot handle a negative return */
1549         if (max <= bvec->bv_len && bio_sectors == 0)
1550                 return bvec->bv_len;
1551         return max;
1552 }
1553
1554 static void rbd_free_disk(struct rbd_device *rbd_dev)
1555 {
1556         struct gendisk *disk = rbd_dev->disk;
1557
1558         if (!disk)
1559                 return;
1560
1561         rbd_header_free(&rbd_dev->header);
1562
1563         if (disk->flags & GENHD_FL_UP)
1564                 del_gendisk(disk);
1565         if (disk->queue)
1566                 blk_cleanup_queue(disk->queue);
1567         put_disk(disk);
1568 }
1569
1570 /*
1571  * reload the ondisk the header
1572  */
1573 static int rbd_read_header(struct rbd_device *rbd_dev,
1574                            struct rbd_image_header *header)
1575 {
1576         ssize_t rc;
1577         struct rbd_image_header_ondisk *dh;
1578         u32 snap_count = 0;
1579         u64 ver;
1580         size_t len;
1581
1582         /*
1583          * First reads the fixed-size header to determine the number
1584          * of snapshots, then re-reads it, along with all snapshot
1585          * records as well as their stored names.
1586          */
1587         len = sizeof (*dh);
1588         while (1) {
1589                 dh = kmalloc(len, GFP_KERNEL);
1590                 if (!dh)
1591                         return -ENOMEM;
1592
1593                 rc = rbd_req_sync_read(rbd_dev,
1594                                        NULL, CEPH_NOSNAP,
1595                                        rbd_dev->obj_md_name,
1596                                        0, len,
1597                                        (char *)dh, &ver);
1598                 if (rc < 0)
1599                         goto out_dh;
1600
1601                 rc = rbd_header_from_disk(header, dh, snap_count, GFP_KERNEL);
1602                 if (rc < 0) {
1603                         if (rc == -ENXIO)
1604                                 pr_warning("unrecognized header format"
1605                                            " for image %s", rbd_dev->obj);
1606                         goto out_dh;
1607                 }
1608
1609                 if (snap_count == header->total_snaps)
1610                         break;
1611
1612                 snap_count = header->total_snaps;
1613                 len = sizeof (*dh) +
1614                         snap_count * sizeof(struct rbd_image_snap_ondisk) +
1615                         header->snap_names_len;
1616
1617                 rbd_header_free(header);
1618                 kfree(dh);
1619         }
1620         header->obj_version = ver;
1621
1622 out_dh:
1623         kfree(dh);
1624         return rc;
1625 }
1626
1627 /*
1628  * create a snapshot
1629  */
1630 static int rbd_header_add_snap(struct rbd_device *dev,
1631                                const char *snap_name,
1632                                gfp_t gfp_flags)
1633 {
1634         int name_len = strlen(snap_name);
1635         u64 new_snapid;
1636         int ret;
1637         void *data, *p, *e;
1638         u64 ver;
1639         struct ceph_mon_client *monc;
1640
1641         /* we should create a snapshot only if we're pointing at the head */
1642         if (dev->snap_id != CEPH_NOSNAP)
1643                 return -EINVAL;
1644
1645         monc = &dev->rbd_client->client->monc;
1646         ret = ceph_monc_create_snapid(monc, dev->poolid, &new_snapid);
1647         dout("created snapid=%lld\n", new_snapid);
1648         if (ret < 0)
1649                 return ret;
1650
1651         data = kmalloc(name_len + 16, gfp_flags);
1652         if (!data)
1653                 return -ENOMEM;
1654
1655         p = data;
1656         e = data + name_len + 16;
1657
1658         ceph_encode_string_safe(&p, e, snap_name, name_len, bad);
1659         ceph_encode_64_safe(&p, e, new_snapid, bad);
1660
1661         ret = rbd_req_sync_exec(dev, dev->obj_md_name, "rbd", "snap_add",
1662                                 data, p - data, &ver);
1663
1664         kfree(data);
1665
1666         if (ret < 0)
1667                 return ret;
1668
1669         down_write(&dev->header_rwsem);
1670         dev->header.snapc->seq = new_snapid;
1671         up_write(&dev->header_rwsem);
1672
1673         return 0;
1674 bad:
1675         return -ERANGE;
1676 }
1677
1678 static void __rbd_remove_all_snaps(struct rbd_device *rbd_dev)
1679 {
1680         struct rbd_snap *snap;
1681
1682         while (!list_empty(&rbd_dev->snaps)) {
1683                 snap = list_first_entry(&rbd_dev->snaps, struct rbd_snap, node);
1684                 __rbd_remove_snap_dev(rbd_dev, snap);
1685         }
1686 }
1687
1688 /*
1689  * only read the first part of the ondisk header, without the snaps info
1690  */
1691 static int __rbd_refresh_header(struct rbd_device *rbd_dev)
1692 {
1693         int ret;
1694         struct rbd_image_header h;
1695         u64 snap_seq;
1696         int follow_seq = 0;
1697
1698         ret = rbd_read_header(rbd_dev, &h);
1699         if (ret < 0)
1700                 return ret;
1701
1702         /* resized? */
1703         set_capacity(rbd_dev->disk, h.image_size / SECTOR_SIZE);
1704
1705         down_write(&rbd_dev->header_rwsem);
1706
1707         snap_seq = rbd_dev->header.snapc->seq;
1708         if (rbd_dev->header.total_snaps &&
1709             rbd_dev->header.snapc->snaps[0] == snap_seq)
1710                 /* pointing at the head, will need to follow that
1711                    if head moves */
1712                 follow_seq = 1;
1713
1714         kfree(rbd_dev->header.snapc);
1715         kfree(rbd_dev->header.snap_names);
1716         kfree(rbd_dev->header.snap_sizes);
1717
1718         rbd_dev->header.total_snaps = h.total_snaps;
1719         rbd_dev->header.snapc = h.snapc;
1720         rbd_dev->header.snap_names = h.snap_names;
1721         rbd_dev->header.snap_names_len = h.snap_names_len;
1722         rbd_dev->header.snap_sizes = h.snap_sizes;
1723         if (follow_seq)
1724                 rbd_dev->header.snapc->seq = rbd_dev->header.snapc->snaps[0];
1725         else
1726                 rbd_dev->header.snapc->seq = snap_seq;
1727
1728         ret = __rbd_init_snaps_header(rbd_dev);
1729
1730         up_write(&rbd_dev->header_rwsem);
1731
1732         return ret;
1733 }
1734
1735 static int rbd_init_disk(struct rbd_device *rbd_dev)
1736 {
1737         struct gendisk *disk;
1738         struct request_queue *q;
1739         int rc;
1740         u64 segment_size;
1741         u64 total_size = 0;
1742
1743         /* contact OSD, request size info about the object being mapped */
1744         rc = rbd_read_header(rbd_dev, &rbd_dev->header);
1745         if (rc)
1746                 return rc;
1747
1748         /* no need to lock here, as rbd_dev is not registered yet */
1749         rc = __rbd_init_snaps_header(rbd_dev);
1750         if (rc)
1751                 return rc;
1752
1753         rc = rbd_header_set_snap(rbd_dev, &total_size);
1754         if (rc)
1755                 return rc;
1756
1757         /* create gendisk info */
1758         rc = -ENOMEM;
1759         disk = alloc_disk(RBD_MINORS_PER_MAJOR);
1760         if (!disk)
1761                 goto out;
1762
1763         snprintf(disk->disk_name, sizeof(disk->disk_name), RBD_DRV_NAME "%d",
1764                  rbd_dev->id);
1765         disk->major = rbd_dev->major;
1766         disk->first_minor = 0;
1767         disk->fops = &rbd_bd_ops;
1768         disk->private_data = rbd_dev;
1769
1770         /* init rq */
1771         rc = -ENOMEM;
1772         q = blk_init_queue(rbd_rq_fn, &rbd_dev->lock);
1773         if (!q)
1774                 goto out_disk;
1775
1776         /* We use the default size, but let's be explicit about it. */
1777         blk_queue_physical_block_size(q, SECTOR_SIZE);
1778
1779         /* set io sizes to object size */
1780         segment_size = rbd_obj_bytes(&rbd_dev->header);
1781         blk_queue_max_hw_sectors(q, segment_size / SECTOR_SIZE);
1782         blk_queue_max_segment_size(q, segment_size);
1783         blk_queue_io_min(q, segment_size);
1784         blk_queue_io_opt(q, segment_size);
1785
1786         blk_queue_merge_bvec(q, rbd_merge_bvec);
1787         disk->queue = q;
1788
1789         q->queuedata = rbd_dev;
1790
1791         rbd_dev->disk = disk;
1792         rbd_dev->q = q;
1793
1794         /* finally, announce the disk to the world */
1795         set_capacity(disk, total_size / SECTOR_SIZE);
1796         add_disk(disk);
1797
1798         pr_info("%s: added with size 0x%llx\n",
1799                 disk->disk_name, (unsigned long long)total_size);
1800         return 0;
1801
1802 out_disk:
1803         put_disk(disk);
1804 out:
1805         return rc;
1806 }
1807
1808 /*
1809   sysfs
1810 */
1811
1812 static struct rbd_device *dev_to_rbd_dev(struct device *dev)
1813 {
1814         return container_of(dev, struct rbd_device, dev);
1815 }
1816
1817 static ssize_t rbd_size_show(struct device *dev,
1818                              struct device_attribute *attr, char *buf)
1819 {
1820         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1821
1822         return sprintf(buf, "%llu\n", (unsigned long long)rbd_dev->header.image_size);
1823 }
1824
1825 static ssize_t rbd_major_show(struct device *dev,
1826                               struct device_attribute *attr, char *buf)
1827 {
1828         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1829
1830         return sprintf(buf, "%d\n", rbd_dev->major);
1831 }
1832
1833 static ssize_t rbd_client_id_show(struct device *dev,
1834                                   struct device_attribute *attr, char *buf)
1835 {
1836         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1837
1838         return sprintf(buf, "client%lld\n",
1839                         ceph_client_id(rbd_dev->rbd_client->client));
1840 }
1841
1842 static ssize_t rbd_pool_show(struct device *dev,
1843                              struct device_attribute *attr, char *buf)
1844 {
1845         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1846
1847         return sprintf(buf, "%s\n", rbd_dev->pool_name);
1848 }
1849
1850 static ssize_t rbd_name_show(struct device *dev,
1851                              struct device_attribute *attr, char *buf)
1852 {
1853         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1854
1855         return sprintf(buf, "%s\n", rbd_dev->obj);
1856 }
1857
1858 static ssize_t rbd_snap_show(struct device *dev,
1859                              struct device_attribute *attr,
1860                              char *buf)
1861 {
1862         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1863
1864         return sprintf(buf, "%s\n", rbd_dev->snap_name);
1865 }
1866
1867 static ssize_t rbd_image_refresh(struct device *dev,
1868                                  struct device_attribute *attr,
1869                                  const char *buf,
1870                                  size_t size)
1871 {
1872         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1873         int rc;
1874         int ret = size;
1875
1876         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
1877
1878         rc = __rbd_refresh_header(rbd_dev);
1879         if (rc < 0)
1880                 ret = rc;
1881
1882         mutex_unlock(&ctl_mutex);
1883         return ret;
1884 }
1885
1886 static DEVICE_ATTR(size, S_IRUGO, rbd_size_show, NULL);
1887 static DEVICE_ATTR(major, S_IRUGO, rbd_major_show, NULL);
1888 static DEVICE_ATTR(client_id, S_IRUGO, rbd_client_id_show, NULL);
1889 static DEVICE_ATTR(pool, S_IRUGO, rbd_pool_show, NULL);
1890 static DEVICE_ATTR(name, S_IRUGO, rbd_name_show, NULL);
1891 static DEVICE_ATTR(refresh, S_IWUSR, NULL, rbd_image_refresh);
1892 static DEVICE_ATTR(current_snap, S_IRUGO, rbd_snap_show, NULL);
1893 static DEVICE_ATTR(create_snap, S_IWUSR, NULL, rbd_snap_add);
1894
1895 static struct attribute *rbd_attrs[] = {
1896         &dev_attr_size.attr,
1897         &dev_attr_major.attr,
1898         &dev_attr_client_id.attr,
1899         &dev_attr_pool.attr,
1900         &dev_attr_name.attr,
1901         &dev_attr_current_snap.attr,
1902         &dev_attr_refresh.attr,
1903         &dev_attr_create_snap.attr,
1904         NULL
1905 };
1906
1907 static struct attribute_group rbd_attr_group = {
1908         .attrs = rbd_attrs,
1909 };
1910
1911 static const struct attribute_group *rbd_attr_groups[] = {
1912         &rbd_attr_group,
1913         NULL
1914 };
1915
1916 static void rbd_sysfs_dev_release(struct device *dev)
1917 {
1918 }
1919
1920 static struct device_type rbd_device_type = {
1921         .name           = "rbd",
1922         .groups         = rbd_attr_groups,
1923         .release        = rbd_sysfs_dev_release,
1924 };
1925
1926
1927 /*
1928   sysfs - snapshots
1929 */
1930
1931 static ssize_t rbd_snap_size_show(struct device *dev,
1932                                   struct device_attribute *attr,
1933                                   char *buf)
1934 {
1935         struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
1936
1937         return sprintf(buf, "%llu\n", (unsigned long long)snap->size);
1938 }
1939
1940 static ssize_t rbd_snap_id_show(struct device *dev,
1941                                 struct device_attribute *attr,
1942                                 char *buf)
1943 {
1944         struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
1945
1946         return sprintf(buf, "%llu\n", (unsigned long long)snap->id);
1947 }
1948
1949 static DEVICE_ATTR(snap_size, S_IRUGO, rbd_snap_size_show, NULL);
1950 static DEVICE_ATTR(snap_id, S_IRUGO, rbd_snap_id_show, NULL);
1951
1952 static struct attribute *rbd_snap_attrs[] = {
1953         &dev_attr_snap_size.attr,
1954         &dev_attr_snap_id.attr,
1955         NULL,
1956 };
1957
1958 static struct attribute_group rbd_snap_attr_group = {
1959         .attrs = rbd_snap_attrs,
1960 };
1961
1962 static void rbd_snap_dev_release(struct device *dev)
1963 {
1964         struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
1965         kfree(snap->name);
1966         kfree(snap);
1967 }
1968
1969 static const struct attribute_group *rbd_snap_attr_groups[] = {
1970         &rbd_snap_attr_group,
1971         NULL
1972 };
1973
1974 static struct device_type rbd_snap_device_type = {
1975         .groups         = rbd_snap_attr_groups,
1976         .release        = rbd_snap_dev_release,
1977 };
1978
1979 static void __rbd_remove_snap_dev(struct rbd_device *rbd_dev,
1980                                   struct rbd_snap *snap)
1981 {
1982         list_del(&snap->node);
1983         device_unregister(&snap->dev);
1984 }
1985
1986 static int rbd_register_snap_dev(struct rbd_device *rbd_dev,
1987                                   struct rbd_snap *snap,
1988                                   struct device *parent)
1989 {
1990         struct device *dev = &snap->dev;
1991         int ret;
1992
1993         dev->type = &rbd_snap_device_type;
1994         dev->parent = parent;
1995         dev->release = rbd_snap_dev_release;
1996         dev_set_name(dev, "snap_%s", snap->name);
1997         ret = device_register(dev);
1998
1999         return ret;
2000 }
2001
2002 static int __rbd_add_snap_dev(struct rbd_device *rbd_dev,
2003                               int i, const char *name,
2004                               struct rbd_snap **snapp)
2005 {
2006         int ret;
2007         struct rbd_snap *snap = kzalloc(sizeof(*snap), GFP_KERNEL);
2008         if (!snap)
2009                 return -ENOMEM;
2010         snap->name = kstrdup(name, GFP_KERNEL);
2011         snap->size = rbd_dev->header.snap_sizes[i];
2012         snap->id = rbd_dev->header.snapc->snaps[i];
2013         if (device_is_registered(&rbd_dev->dev)) {
2014                 ret = rbd_register_snap_dev(rbd_dev, snap,
2015                                              &rbd_dev->dev);
2016                 if (ret < 0)
2017                         goto err;
2018         }
2019         *snapp = snap;
2020         return 0;
2021 err:
2022         kfree(snap->name);
2023         kfree(snap);
2024         return ret;
2025 }
2026
2027 /*
2028  * search for the previous snap in a null delimited string list
2029  */
2030 const char *rbd_prev_snap_name(const char *name, const char *start)
2031 {
2032         if (name < start + 2)
2033                 return NULL;
2034
2035         name -= 2;
2036         while (*name) {
2037                 if (name == start)
2038                         return start;
2039                 name--;
2040         }
2041         return name + 1;
2042 }
2043
2044 /*
2045  * compare the old list of snapshots that we have to what's in the header
2046  * and update it accordingly. Note that the header holds the snapshots
2047  * in a reverse order (from newest to oldest) and we need to go from
2048  * older to new so that we don't get a duplicate snap name when
2049  * doing the process (e.g., removed snapshot and recreated a new
2050  * one with the same name.
2051  */
2052 static int __rbd_init_snaps_header(struct rbd_device *rbd_dev)
2053 {
2054         const char *name, *first_name;
2055         int i = rbd_dev->header.total_snaps;
2056         struct rbd_snap *snap, *old_snap = NULL;
2057         int ret;
2058         struct list_head *p, *n;
2059
2060         first_name = rbd_dev->header.snap_names;
2061         name = first_name + rbd_dev->header.snap_names_len;
2062
2063         list_for_each_prev_safe(p, n, &rbd_dev->snaps) {
2064                 u64 cur_id;
2065
2066                 old_snap = list_entry(p, struct rbd_snap, node);
2067
2068                 if (i)
2069                         cur_id = rbd_dev->header.snapc->snaps[i - 1];
2070
2071                 if (!i || old_snap->id < cur_id) {
2072                         /* old_snap->id was skipped, thus was removed */
2073                         __rbd_remove_snap_dev(rbd_dev, old_snap);
2074                         continue;
2075                 }
2076                 if (old_snap->id == cur_id) {
2077                         /* we have this snapshot already */
2078                         i--;
2079                         name = rbd_prev_snap_name(name, first_name);
2080                         continue;
2081                 }
2082                 for (; i > 0;
2083                      i--, name = rbd_prev_snap_name(name, first_name)) {
2084                         if (!name) {
2085                                 WARN_ON(1);
2086                                 return -EINVAL;
2087                         }
2088                         cur_id = rbd_dev->header.snapc->snaps[i];
2089                         /* snapshot removal? handle it above */
2090                         if (cur_id >= old_snap->id)
2091                                 break;
2092                         /* a new snapshot */
2093                         ret = __rbd_add_snap_dev(rbd_dev, i - 1, name, &snap);
2094                         if (ret < 0)
2095                                 return ret;
2096
2097                         /* note that we add it backward so using n and not p */
2098                         list_add(&snap->node, n);
2099                         p = &snap->node;
2100                 }
2101         }
2102         /* we're done going over the old snap list, just add what's left */
2103         for (; i > 0; i--) {
2104                 name = rbd_prev_snap_name(name, first_name);
2105                 if (!name) {
2106                         WARN_ON(1);
2107                         return -EINVAL;
2108                 }
2109                 ret = __rbd_add_snap_dev(rbd_dev, i - 1, name, &snap);
2110                 if (ret < 0)
2111                         return ret;
2112                 list_add(&snap->node, &rbd_dev->snaps);
2113         }
2114
2115         return 0;
2116 }
2117
2118 static int rbd_bus_add_dev(struct rbd_device *rbd_dev)
2119 {
2120         int ret;
2121         struct device *dev;
2122         struct rbd_snap *snap;
2123
2124         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2125         dev = &rbd_dev->dev;
2126
2127         dev->bus = &rbd_bus_type;
2128         dev->type = &rbd_device_type;
2129         dev->parent = &rbd_root_dev;
2130         dev->release = rbd_dev_release;
2131         dev_set_name(dev, "%d", rbd_dev->id);
2132         ret = device_register(dev);
2133         if (ret < 0)
2134                 goto out;
2135
2136         list_for_each_entry(snap, &rbd_dev->snaps, node) {
2137                 ret = rbd_register_snap_dev(rbd_dev, snap,
2138                                              &rbd_dev->dev);
2139                 if (ret < 0)
2140                         break;
2141         }
2142 out:
2143         mutex_unlock(&ctl_mutex);
2144         return ret;
2145 }
2146
2147 static void rbd_bus_del_dev(struct rbd_device *rbd_dev)
2148 {
2149         device_unregister(&rbd_dev->dev);
2150 }
2151
2152 static int rbd_init_watch_dev(struct rbd_device *rbd_dev)
2153 {
2154         int ret, rc;
2155
2156         do {
2157                 ret = rbd_req_sync_watch(rbd_dev, rbd_dev->obj_md_name,
2158                                          rbd_dev->header.obj_version);
2159                 if (ret == -ERANGE) {
2160                         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2161                         rc = __rbd_refresh_header(rbd_dev);
2162                         mutex_unlock(&ctl_mutex);
2163                         if (rc < 0)
2164                                 return rc;
2165                 }
2166         } while (ret == -ERANGE);
2167
2168         return ret;
2169 }
2170
2171 static atomic64_t rbd_id_max = ATOMIC64_INIT(0);
2172
2173 /*
2174  * Get a unique rbd identifier for the given new rbd_dev, and add
2175  * the rbd_dev to the global list.  The minimum rbd id is 1.
2176  */
2177 static void rbd_id_get(struct rbd_device *rbd_dev)
2178 {
2179         rbd_dev->id = atomic64_inc_return(&rbd_id_max);
2180
2181         spin_lock(&rbd_dev_list_lock);
2182         list_add_tail(&rbd_dev->node, &rbd_dev_list);
2183         spin_unlock(&rbd_dev_list_lock);
2184 }
2185
2186 /*
2187  * Remove an rbd_dev from the global list, and record that its
2188  * identifier is no longer in use.
2189  */
2190 static void rbd_id_put(struct rbd_device *rbd_dev)
2191 {
2192         struct list_head *tmp;
2193         int rbd_id = rbd_dev->id;
2194         int max_id;
2195
2196         BUG_ON(rbd_id < 1);
2197
2198         spin_lock(&rbd_dev_list_lock);
2199         list_del_init(&rbd_dev->node);
2200
2201         /*
2202          * If the id being "put" is not the current maximum, there
2203          * is nothing special we need to do.
2204          */
2205         if (rbd_id != atomic64_read(&rbd_id_max)) {
2206                 spin_unlock(&rbd_dev_list_lock);
2207                 return;
2208         }
2209
2210         /*
2211          * We need to update the current maximum id.  Search the
2212          * list to find out what it is.  We're more likely to find
2213          * the maximum at the end, so search the list backward.
2214          */
2215         max_id = 0;
2216         list_for_each_prev(tmp, &rbd_dev_list) {
2217                 struct rbd_device *rbd_dev;
2218
2219                 rbd_dev = list_entry(tmp, struct rbd_device, node);
2220                 if (rbd_id > max_id)
2221                         max_id = rbd_id;
2222         }
2223         spin_unlock(&rbd_dev_list_lock);
2224
2225         /*
2226          * The max id could have been updated by rbd_id_get(), in
2227          * which case it now accurately reflects the new maximum.
2228          * Be careful not to overwrite the maximum value in that
2229          * case.
2230          */
2231         atomic64_cmpxchg(&rbd_id_max, rbd_id, max_id);
2232 }
2233
2234 /*
2235  * Skips over white space at *buf, and updates *buf to point to the
2236  * first found non-space character (if any). Returns the length of
2237  * the token (string of non-white space characters) found.  Note
2238  * that *buf must be terminated with '\0'.
2239  */
2240 static inline size_t next_token(const char **buf)
2241 {
2242         /*
2243         * These are the characters that produce nonzero for
2244         * isspace() in the "C" and "POSIX" locales.
2245         */
2246         const char *spaces = " \f\n\r\t\v";
2247
2248         *buf += strspn(*buf, spaces);   /* Find start of token */
2249
2250         return strcspn(*buf, spaces);   /* Return token length */
2251 }
2252
2253 /*
2254  * Finds the next token in *buf, and if the provided token buffer is
2255  * big enough, copies the found token into it.  The result, if
2256  * copied, is guaranteed to be terminated with '\0'.  Note that *buf
2257  * must be terminated with '\0' on entry.
2258  *
2259  * Returns the length of the token found (not including the '\0').
2260  * Return value will be 0 if no token is found, and it will be >=
2261  * token_size if the token would not fit.
2262  *
2263  * The *buf pointer will be updated to point beyond the end of the
2264  * found token.  Note that this occurs even if the token buffer is
2265  * too small to hold it.
2266  */
2267 static inline size_t copy_token(const char **buf,
2268                                 char *token,
2269                                 size_t token_size)
2270 {
2271         size_t len;
2272
2273         len = next_token(buf);
2274         if (len < token_size) {
2275                 memcpy(token, *buf, len);
2276                 *(token + len) = '\0';
2277         }
2278         *buf += len;
2279
2280         return len;
2281 }
2282
2283 /*
2284  * Finds the next token in *buf, dynamically allocates a buffer big
2285  * enough to hold a copy of it, and copies the token into the new
2286  * buffer.  The copy is guaranteed to be terminated with '\0'.  Note
2287  * that a duplicate buffer is created even for a zero-length token.
2288  *
2289  * Returns a pointer to the newly-allocated duplicate, or a null
2290  * pointer if memory for the duplicate was not available.  If
2291  * the lenp argument is a non-null pointer, the length of the token
2292  * (not including the '\0') is returned in *lenp.
2293  *
2294  * If successful, the *buf pointer will be updated to point beyond
2295  * the end of the found token.
2296  *
2297  * Note: uses GFP_KERNEL for allocation.
2298  */
2299 static inline char *dup_token(const char **buf, size_t *lenp)
2300 {
2301         char *dup;
2302         size_t len;
2303
2304         len = next_token(buf);
2305         dup = kmalloc(len + 1, GFP_KERNEL);
2306         if (!dup)
2307                 return NULL;
2308
2309         memcpy(dup, *buf, len);
2310         *(dup + len) = '\0';
2311         *buf += len;
2312
2313         if (lenp)
2314                 *lenp = len;
2315
2316         return dup;
2317 }
2318
2319 /*
2320  * This fills in the pool_name, obj, obj_len, snap_name, obj_len,
2321  * rbd_dev, rbd_md_name, and name fields of the given rbd_dev, based
2322  * on the list of monitor addresses and other options provided via
2323  * /sys/bus/rbd/add.
2324  */
2325 static int rbd_add_parse_args(struct rbd_device *rbd_dev,
2326                               const char *buf,
2327                               const char **mon_addrs,
2328                               size_t *mon_addrs_size,
2329                               char *options,
2330                               size_t options_size)
2331 {
2332         size_t  len;
2333
2334         /* The first four tokens are required */
2335
2336         len = next_token(&buf);
2337         if (!len)
2338                 return -EINVAL;
2339         *mon_addrs_size = len + 1;
2340         *mon_addrs = buf;
2341
2342         buf += len;
2343
2344         len = copy_token(&buf, options, options_size);
2345         if (!len || len >= options_size)
2346                 return -EINVAL;
2347
2348         len = copy_token(&buf, rbd_dev->pool_name, sizeof (rbd_dev->pool_name));
2349         if (!len || len >= sizeof (rbd_dev->pool_name))
2350                 return -EINVAL;
2351
2352         len = copy_token(&buf, rbd_dev->obj, sizeof (rbd_dev->obj));
2353         if (!len || len >= sizeof (rbd_dev->obj))
2354                 return -EINVAL;
2355
2356         /* We have the object length in hand, save it. */
2357
2358         rbd_dev->obj_len = len;
2359
2360         BUILD_BUG_ON(RBD_MAX_MD_NAME_LEN
2361                                 < RBD_MAX_OBJ_NAME_LEN + sizeof (RBD_SUFFIX));
2362         sprintf(rbd_dev->obj_md_name, "%s%s", rbd_dev->obj, RBD_SUFFIX);
2363
2364         /*
2365          * The snapshot name is optional, but it's an error if it's
2366          * too long.  If no snapshot is supplied, fill in the default.
2367          */
2368         len = copy_token(&buf, rbd_dev->snap_name, sizeof (rbd_dev->snap_name));
2369         if (!len)
2370                 memcpy(rbd_dev->snap_name, RBD_SNAP_HEAD_NAME,
2371                         sizeof (RBD_SNAP_HEAD_NAME));
2372         else if (len >= sizeof (rbd_dev->snap_name))
2373                 return -EINVAL;
2374
2375         return 0;
2376 }
2377
2378 static ssize_t rbd_add(struct bus_type *bus,
2379                        const char *buf,
2380                        size_t count)
2381 {
2382         struct rbd_device *rbd_dev;
2383         const char *mon_addrs = NULL;
2384         size_t mon_addrs_size = 0;
2385         char *options = NULL;
2386         struct ceph_osd_client *osdc;
2387         int rc = -ENOMEM;
2388
2389         if (!try_module_get(THIS_MODULE))
2390                 return -ENODEV;
2391
2392         rbd_dev = kzalloc(sizeof(*rbd_dev), GFP_KERNEL);
2393         if (!rbd_dev)
2394                 goto err_nomem;
2395         options = kmalloc(count, GFP_KERNEL);
2396         if (!options)
2397                 goto err_nomem;
2398
2399         /* static rbd_device initialization */
2400         spin_lock_init(&rbd_dev->lock);
2401         INIT_LIST_HEAD(&rbd_dev->node);
2402         INIT_LIST_HEAD(&rbd_dev->snaps);
2403         init_rwsem(&rbd_dev->header_rwsem);
2404
2405         init_rwsem(&rbd_dev->header_rwsem);
2406
2407         /* generate unique id: find highest unique id, add one */
2408         rbd_id_get(rbd_dev);
2409
2410         /* Fill in the device name, now that we have its id. */
2411         BUILD_BUG_ON(DEV_NAME_LEN
2412                         < sizeof (RBD_DRV_NAME) + MAX_INT_FORMAT_WIDTH);
2413         sprintf(rbd_dev->name, "%s%d", RBD_DRV_NAME, rbd_dev->id);
2414
2415         /* parse add command */
2416         rc = rbd_add_parse_args(rbd_dev, buf, &mon_addrs, &mon_addrs_size,
2417                                 options, count);
2418         if (rc)
2419                 goto err_put_id;
2420
2421         rbd_dev->rbd_client = rbd_get_client(mon_addrs, mon_addrs_size - 1,
2422                                                 options);
2423         if (IS_ERR(rbd_dev->rbd_client)) {
2424                 rc = PTR_ERR(rbd_dev->rbd_client);
2425                 goto err_put_id;
2426         }
2427
2428         /* pick the pool */
2429         osdc = &rbd_dev->rbd_client->client->osdc;
2430         rc = ceph_pg_poolid_by_name(osdc->osdmap, rbd_dev->pool_name);
2431         if (rc < 0)
2432                 goto err_out_client;
2433         rbd_dev->poolid = rc;
2434
2435         /* register our block device */
2436         rc = register_blkdev(0, rbd_dev->name);
2437         if (rc < 0)
2438                 goto err_out_client;
2439         rbd_dev->major = rc;
2440
2441         rc = rbd_bus_add_dev(rbd_dev);
2442         if (rc)
2443                 goto err_out_blkdev;
2444
2445         /*
2446          * At this point cleanup in the event of an error is the job
2447          * of the sysfs code (initiated by rbd_bus_del_dev()).
2448          *
2449          * Set up and announce blkdev mapping.
2450          */
2451         rc = rbd_init_disk(rbd_dev);
2452         if (rc)
2453                 goto err_out_bus;
2454
2455         rc = rbd_init_watch_dev(rbd_dev);
2456         if (rc)
2457                 goto err_out_bus;
2458
2459         return count;
2460
2461 err_out_bus:
2462         /* this will also clean up rest of rbd_dev stuff */
2463
2464         rbd_bus_del_dev(rbd_dev);
2465         kfree(options);
2466         return rc;
2467
2468 err_out_blkdev:
2469         unregister_blkdev(rbd_dev->major, rbd_dev->name);
2470 err_out_client:
2471         rbd_put_client(rbd_dev);
2472 err_put_id:
2473         rbd_id_put(rbd_dev);
2474 err_nomem:
2475         kfree(options);
2476         kfree(rbd_dev);
2477
2478         dout("Error adding device %s\n", buf);
2479         module_put(THIS_MODULE);
2480
2481         return (ssize_t) rc;
2482 }
2483
2484 static struct rbd_device *__rbd_get_dev(unsigned long id)
2485 {
2486         struct list_head *tmp;
2487         struct rbd_device *rbd_dev;
2488
2489         spin_lock(&rbd_dev_list_lock);
2490         list_for_each(tmp, &rbd_dev_list) {
2491                 rbd_dev = list_entry(tmp, struct rbd_device, node);
2492                 if (rbd_dev->id == id) {
2493                         spin_unlock(&rbd_dev_list_lock);
2494                         return rbd_dev;
2495                 }
2496         }
2497         spin_unlock(&rbd_dev_list_lock);
2498         return NULL;
2499 }
2500
2501 static void rbd_dev_release(struct device *dev)
2502 {
2503         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2504
2505         if (rbd_dev->watch_request) {
2506                 struct ceph_client *client = rbd_dev->rbd_client->client;
2507
2508                 ceph_osdc_unregister_linger_request(&client->osdc,
2509                                                     rbd_dev->watch_request);
2510         }
2511         if (rbd_dev->watch_event)
2512                 rbd_req_sync_unwatch(rbd_dev, rbd_dev->obj_md_name);
2513
2514         rbd_put_client(rbd_dev);
2515
2516         /* clean up and free blkdev */
2517         rbd_free_disk(rbd_dev);
2518         unregister_blkdev(rbd_dev->major, rbd_dev->name);
2519
2520         /* done with the id, and with the rbd_dev */
2521         rbd_id_put(rbd_dev);
2522         kfree(rbd_dev);
2523
2524         /* release module ref */
2525         module_put(THIS_MODULE);
2526 }
2527
2528 static ssize_t rbd_remove(struct bus_type *bus,
2529                           const char *buf,
2530                           size_t count)
2531 {
2532         struct rbd_device *rbd_dev = NULL;
2533         int target_id, rc;
2534         unsigned long ul;
2535         int ret = count;
2536
2537         rc = strict_strtoul(buf, 10, &ul);
2538         if (rc)
2539                 return rc;
2540
2541         /* convert to int; abort if we lost anything in the conversion */
2542         target_id = (int) ul;
2543         if (target_id != ul)
2544                 return -EINVAL;
2545
2546         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2547
2548         rbd_dev = __rbd_get_dev(target_id);
2549         if (!rbd_dev) {
2550                 ret = -ENOENT;
2551                 goto done;
2552         }
2553
2554         __rbd_remove_all_snaps(rbd_dev);
2555         rbd_bus_del_dev(rbd_dev);
2556
2557 done:
2558         mutex_unlock(&ctl_mutex);
2559         return ret;
2560 }
2561
2562 static ssize_t rbd_snap_add(struct device *dev,
2563                             struct device_attribute *attr,
2564                             const char *buf,
2565                             size_t count)
2566 {
2567         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2568         int ret;
2569         char *name = kmalloc(count + 1, GFP_KERNEL);
2570         if (!name)
2571                 return -ENOMEM;
2572
2573         snprintf(name, count, "%s", buf);
2574
2575         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2576
2577         ret = rbd_header_add_snap(rbd_dev,
2578                                   name, GFP_KERNEL);
2579         if (ret < 0)
2580                 goto err_unlock;
2581
2582         ret = __rbd_refresh_header(rbd_dev);
2583         if (ret < 0)
2584                 goto err_unlock;
2585
2586         /* shouldn't hold ctl_mutex when notifying.. notify might
2587            trigger a watch callback that would need to get that mutex */
2588         mutex_unlock(&ctl_mutex);
2589
2590         /* make a best effort, don't error if failed */
2591         rbd_req_sync_notify(rbd_dev, rbd_dev->obj_md_name);
2592
2593         ret = count;
2594         kfree(name);
2595         return ret;
2596
2597 err_unlock:
2598         mutex_unlock(&ctl_mutex);
2599         kfree(name);
2600         return ret;
2601 }
2602
2603 /*
2604  * create control files in sysfs
2605  * /sys/bus/rbd/...
2606  */
2607 static int rbd_sysfs_init(void)
2608 {
2609         int ret;
2610
2611         ret = device_register(&rbd_root_dev);
2612         if (ret < 0)
2613                 return ret;
2614
2615         ret = bus_register(&rbd_bus_type);
2616         if (ret < 0)
2617                 device_unregister(&rbd_root_dev);
2618
2619         return ret;
2620 }
2621
2622 static void rbd_sysfs_cleanup(void)
2623 {
2624         bus_unregister(&rbd_bus_type);
2625         device_unregister(&rbd_root_dev);
2626 }
2627
2628 int __init rbd_init(void)
2629 {
2630         int rc;
2631
2632         rc = rbd_sysfs_init();
2633         if (rc)
2634                 return rc;
2635         pr_info("loaded " RBD_DRV_NAME_LONG "\n");
2636         return 0;
2637 }
2638
2639 void __exit rbd_exit(void)
2640 {
2641         rbd_sysfs_cleanup();
2642 }
2643
2644 module_init(rbd_init);
2645 module_exit(rbd_exit);
2646
2647 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
2648 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
2649 MODULE_DESCRIPTION("rados block device");
2650
2651 /* following authorship retained from original osdblk.c */
2652 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
2653
2654 MODULE_LICENSE("GPL");