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 #define RBD_DEBUG       /* Activate rbd_assert() calls */
  45
  46 /*
  47  * The basic unit of block I/O is a sector.  It is interpreted in a
  48  * number of contexts in Linux (blk, bio, genhd), but the default is
  49  * universally 512 bytes.  These symbols are just slightly more
  50  * meaningful than the bare numbers they represent.
  51  */
  52 #define SECTOR_SHIFT    9
  53 #define SECTOR_SIZE     (1ULL << SECTOR_SHIFT)
  54
  55 /* It might be useful to have this defined elsewhere too */
  56
  57 #define U64_MAX ((u64) (~0ULL))
  58
  59 #define RBD_DRV_NAME "rbd"
  60 #define RBD_DRV_NAME_LONG "rbd (rados block device)"
  61
  62 #define RBD_MINORS_PER_MAJOR    256             /* max minors per blkdev */
  63
  64 #define RBD_MAX_SNAP_NAME_LEN   32
  65 #define RBD_MAX_SNAP_COUNT      510     /* allows max snapc to fit in 4KB */
  66 #define RBD_MAX_OPT_LEN         1024
  67
  68 #define RBD_SNAP_HEAD_NAME      "-"
  69
  70 #define RBD_IMAGE_ID_LEN_MAX    64
  71 #define RBD_OBJ_PREFIX_LEN_MAX  64
  72
  73 /* Feature bits */
  74
  75 #define RBD_FEATURE_LAYERING      1
  76
  77 /* Features supported by this (client software) implementation. */
  78
  79 #define RBD_FEATURES_ALL          (0)
  80
  81 /*
  82  * An RBD device name will be "rbd#", where the "rbd" comes from
  83  * RBD_DRV_NAME above, and # is a unique integer identifier.
  84  * MAX_INT_FORMAT_WIDTH is used in ensuring DEV_NAME_LEN is big
  85  * enough to hold all possible device names.
  86  */
  87 #define DEV_NAME_LEN            32
  88 #define MAX_INT_FORMAT_WIDTH    ((5 * sizeof (int)) / 2 + 1)
  89
  90 #define RBD_READ_ONLY_DEFAULT           false
  91
  92 /*
  93  * block device image metadata (in-memory version)
  94  */
  95 struct rbd_image_header {
  96         /* These four fields never change for a given rbd image */
  97         char *object_prefix;
  98         u64 features;
  99         __u8 obj_order;
 100         __u8 crypt_type;
 101         __u8 comp_type;
 102
 103         /* The remaining fields need to be updated occasionally */
 104         u64 image_size;
 105         struct ceph_snap_context *snapc;
 106         char *snap_names;
 107         u64 *snap_sizes;
 108
 109         u64 obj_version;
 110 };
 111
 112 struct rbd_options {
 113         bool    read_only;
 114 };
 115
 116 /*
 117  * an instance of the client.  multiple devices may share an rbd client.
 118  */
 119 struct rbd_client {
 120         struct ceph_client      *client;
 121         struct kref             kref;
 122         struct list_head        node;
 123 };
 124
 125 /*
 126  * a request completion status
 127  */
 128 struct rbd_req_status {
 129         int done;
 130         int rc;
 131         u64 bytes;
 132 };
 133
 134 /*
 135  * a collection of requests
 136  */
 137 struct rbd_req_coll {
 138         int                     total;
 139         int                     num_done;
 140         struct kref             kref;
 141         struct rbd_req_status   status[0];
 142 };
 143
 144 /*
 145  * a single io request
 146  */
 147 struct rbd_request {
 148         struct request          *rq;            /* blk layer request */
 149         struct bio              *bio;           /* cloned bio */
 150         struct page             **pages;        /* list of used pages */
 151         u64                     len;
 152         int                     coll_index;
 153         struct rbd_req_coll     *coll;
 154 };
 155
 156 struct rbd_snap {
 157         struct  device          dev;
 158         const char              *name;
 159         u64                     size;
 160         struct list_head        node;
 161         u64                     id;
 162         u64                     features;
 163 };
 164
 165 struct rbd_mapping {
 166         char                    *snap_name;
 167         u64                     snap_id;
 168         u64                     size;
 169         u64                     features;
 170         bool                    snap_exists;
 171         bool                    read_only;
 172 };
 173
 174 /*
 175  * a single device
 176  */
 177 struct rbd_device {
 178         int                     dev_id;         /* blkdev unique id */
 179
 180         int                     major;          /* blkdev assigned major */
 181         struct gendisk          *disk;          /* blkdev's gendisk and rq */
 182
 183         u32                     image_format;   /* Either 1 or 2 */
 184         struct rbd_options      rbd_opts;
 185         struct rbd_client       *rbd_client;
 186
 187         char                    name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */
 188
 189         spinlock_t              lock;           /* queue lock */
 190
 191         struct rbd_image_header header;
 192         char                    *image_id;
 193         size_t                  image_id_len;
 194         char                    *image_name;
 195         size_t                  image_name_len;
 196         char                    *header_name;
 197         char                    *pool_name;
 198         int                     pool_id;
 199
 200         struct ceph_osd_event   *watch_event;
 201         struct ceph_osd_request *watch_request;
 202
 203         /* protects updating the header */
 204         struct rw_semaphore     header_rwsem;
 205
 206         struct rbd_mapping      mapping;
 207
 208         struct list_head        node;
 209
 210         /* list of snapshots */
 211         struct list_head        snaps;
 212
 213         /* sysfs related */
 214         struct device           dev;
 215 };
 216
 217 static DEFINE_MUTEX(ctl_mutex);   /* Serialize open/close/setup/teardown */
 218
 219 static LIST_HEAD(rbd_dev_list);    /* devices */
 220 static DEFINE_SPINLOCK(rbd_dev_list_lock);
 221
 222 static LIST_HEAD(rbd_client_list);              /* clients */
 223 static DEFINE_SPINLOCK(rbd_client_list_lock);
 224
 225 static int rbd_dev_snaps_update(struct rbd_device *rbd_dev);
 226 static int rbd_dev_snaps_register(struct rbd_device *rbd_dev);
 227
 228 static void rbd_dev_release(struct device *dev);
 229 static void __rbd_remove_snap_dev(struct rbd_snap *snap);
 230
 231 static ssize_t rbd_add(struct bus_type *bus, const char *buf,
 232                        size_t count);
 233 static ssize_t rbd_remove(struct bus_type *bus, const char *buf,
 234                           size_t count);
 235
 236 static struct bus_attribute rbd_bus_attrs[] = {
 237         __ATTR(add, S_IWUSR, NULL, rbd_add),
 238         __ATTR(remove, S_IWUSR, NULL, rbd_remove),
 239         __ATTR_NULL
 240 };
 241
 242 static struct bus_type rbd_bus_type = {
 243         .name           = "rbd",
 244         .bus_attrs      = rbd_bus_attrs,
 245 };
 246
 247 static void rbd_root_dev_release(struct device *dev)
 248 {
 249 }
 250
 251 static struct device rbd_root_dev = {
 252         .init_name =    "rbd",
 253         .release =      rbd_root_dev_release,
 254 };
 255
 256 #ifdef RBD_DEBUG
 257 #define rbd_assert(expr)                                                \
 258                 if (unlikely(!(expr))) {                                \
 259                         printk(KERN_ERR "\nAssertion failure in %s() "  \
 260                                                 "at line %d:\n\n"       \
 261                                         "\trbd_assert(%s);\n\n",        \
 262                                         __func__, __LINE__, #expr);     \
 263                         BUG();                                          \
 264                 }
 265 #else /* !RBD_DEBUG */
 266 #  define rbd_assert(expr)      ((void) 0)
 267 #endif /* !RBD_DEBUG */
 268
 269 static struct device *rbd_get_dev(struct rbd_device *rbd_dev)
 270 {
 271         return get_device(&rbd_dev->dev);
 272 }
 273
 274 static void rbd_put_dev(struct rbd_device *rbd_dev)
 275 {
 276         put_device(&rbd_dev->dev);
 277 }
 278
 279 static int rbd_dev_refresh(struct rbd_device *rbd_dev, u64 *hver);
 280 static int rbd_dev_v2_refresh(struct rbd_device *rbd_dev, u64 *hver);
 281
 282 static int rbd_open(struct block_device *bdev, fmode_t mode)
 283 {
 284         struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
 285
 286         if ((mode & FMODE_WRITE) && rbd_dev->mapping.read_only)
 287                 return -EROFS;
 288
 289         rbd_get_dev(rbd_dev);
 290         set_device_ro(bdev, rbd_dev->mapping.read_only);
 291
 292         return 0;
 293 }
 294
 295 static int rbd_release(struct gendisk *disk, fmode_t mode)
 296 {
 297         struct rbd_device *rbd_dev = disk->private_data;
 298
 299         rbd_put_dev(rbd_dev);
 300
 301         return 0;
 302 }
 303
 304 static const struct block_device_operations rbd_bd_ops = {
 305         .owner                  = THIS_MODULE,
 306         .open                   = rbd_open,
 307         .release                = rbd_release,
 308 };
 309
 310 /*
 311  * Initialize an rbd client instance.
 312  * We own *ceph_opts.
 313  */
 314 static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
 315 {
 316         struct rbd_client *rbdc;
 317         int ret = -ENOMEM;
 318
 319         dout("rbd_client_create\n");
 320         rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
 321         if (!rbdc)
 322                 goto out_opt;
 323
 324         kref_init(&rbdc->kref);
 325         INIT_LIST_HEAD(&rbdc->node);
 326
 327         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
 328
 329         rbdc->client = ceph_create_client(ceph_opts, rbdc, 0, 0);
 330         if (IS_ERR(rbdc->client))
 331                 goto out_mutex;
 332         ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
 333
 334         ret = ceph_open_session(rbdc->client);
 335         if (ret < 0)
 336                 goto out_err;
 337
 338         spin_lock(&rbd_client_list_lock);
 339         list_add_tail(&rbdc->node, &rbd_client_list);
 340         spin_unlock(&rbd_client_list_lock);
 341
 342         mutex_unlock(&ctl_mutex);
 343
 344         dout("rbd_client_create created %p\n", rbdc);
 345         return rbdc;
 346
 347 out_err:
 348         ceph_destroy_client(rbdc->client);
 349 out_mutex:
 350         mutex_unlock(&ctl_mutex);
 351         kfree(rbdc);
 352 out_opt:
 353         if (ceph_opts)
 354                 ceph_destroy_options(ceph_opts);
 355         return ERR_PTR(ret);
 356 }
 357
 358 /*
 359  * Find a ceph client with specific addr and configuration.  If
 360  * found, bump its reference count.
 361  */
 362 static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
 363 {
 364         struct rbd_client *client_node;
 365         bool found = false;
 366
 367         if (ceph_opts->flags & CEPH_OPT_NOSHARE)
 368                 return NULL;
 369
 370         spin_lock(&rbd_client_list_lock);
 371         list_for_each_entry(client_node, &rbd_client_list, node) {
 372                 if (!ceph_compare_options(ceph_opts, client_node->client)) {
 373                         kref_get(&client_node->kref);
 374                         found = true;
 375                         break;
 376                 }
 377         }
 378         spin_unlock(&rbd_client_list_lock);
 379
 380         return found ? client_node : NULL;
 381 }
 382
 383 /*
 384  * mount options
 385  */
 386 enum {
 387         Opt_last_int,
 388         /* int args above */
 389         Opt_last_string,
 390         /* string args above */
 391         Opt_read_only,
 392         Opt_read_write,
 393         /* Boolean args above */
 394         Opt_last_bool,
 395 };
 396
 397 static match_table_t rbd_opts_tokens = {
 398         /* int args above */
 399         /* string args above */
 400         {Opt_read_only, "mapping.read_only"},
 401         {Opt_read_only, "ro"},          /* Alternate spelling */
 402         {Opt_read_write, "read_write"},
 403         {Opt_read_write, "rw"},         /* Alternate spelling */
 404         /* Boolean args above */
 405         {-1, NULL}
 406 };
 407
 408 static int parse_rbd_opts_token(char *c, void *private)
 409 {
 410         struct rbd_options *rbd_opts = private;
 411         substring_t argstr[MAX_OPT_ARGS];
 412         int token, intval, ret;
 413
 414         token = match_token(c, rbd_opts_tokens, argstr);
 415         if (token < 0)
 416                 return -EINVAL;
 417
 418         if (token < Opt_last_int) {
 419                 ret = match_int(&argstr[0], &intval);
 420                 if (ret < 0) {
 421                         pr_err("bad mount option arg (not int) "
 422                                "at '%s'\n", c);
 423                         return ret;
 424                 }
 425                 dout("got int token %d val %d\n", token, intval);
 426         } else if (token > Opt_last_int && token < Opt_last_string) {
 427                 dout("got string token %d val %s\n", token,
 428                      argstr[0].from);
 429         } else if (token > Opt_last_string && token < Opt_last_bool) {
 430                 dout("got Boolean token %d\n", token);
 431         } else {
 432                 dout("got token %d\n", token);
 433         }
 434
 435         switch (token) {
 436         case Opt_read_only:
 437                 rbd_opts->read_only = true;
 438                 break;
 439         case Opt_read_write:
 440                 rbd_opts->read_only = false;
 441                 break;
 442         default:
 443                 rbd_assert(false);
 444                 break;
 445         }
 446         return 0;
 447 }
 448
 449 /*
 450  * Get a ceph client with specific addr and configuration, if one does
 451  * not exist create it.
 452  */
 453 static int rbd_get_client(struct rbd_device *rbd_dev, const char *mon_addr,
 454                                 size_t mon_addr_len, char *options)
 455 {
 456         struct rbd_options *rbd_opts = &rbd_dev->rbd_opts;
 457         struct ceph_options *ceph_opts;
 458         struct rbd_client *rbdc;
 459
 460         rbd_opts->read_only = RBD_READ_ONLY_DEFAULT;
 461
 462         ceph_opts = ceph_parse_options(options, mon_addr,
 463                                         mon_addr + mon_addr_len,
 464                                         parse_rbd_opts_token, rbd_opts);
 465         if (IS_ERR(ceph_opts))
 466                 return PTR_ERR(ceph_opts);
 467
 468         rbdc = rbd_client_find(ceph_opts);
 469         if (rbdc) {
 470                 /* using an existing client */
 471                 ceph_destroy_options(ceph_opts);
 472         } else {
 473                 rbdc = rbd_client_create(ceph_opts);
 474                 if (IS_ERR(rbdc))
 475                         return PTR_ERR(rbdc);
 476         }
 477         rbd_dev->rbd_client = rbdc;
 478
 479         return 0;
 480 }
 481
 482 /*
 483  * Destroy ceph client
 484  *
 485  * Caller must hold rbd_client_list_lock.
 486  */
 487 static void rbd_client_release(struct kref *kref)
 488 {
 489         struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
 490
 491         dout("rbd_release_client %p\n", rbdc);
 492         spin_lock(&rbd_client_list_lock);
 493         list_del(&rbdc->node);
 494         spin_unlock(&rbd_client_list_lock);
 495
 496         ceph_destroy_client(rbdc->client);
 497         kfree(rbdc);
 498 }
 499
 500 /*
 501  * Drop reference to ceph client node. If it's not referenced anymore, release
 502  * it.
 503  */
 504 static void rbd_put_client(struct rbd_device *rbd_dev)
 505 {
 506         kref_put(&rbd_dev->rbd_client->kref, rbd_client_release);
 507         rbd_dev->rbd_client = NULL;
 508 }
 509
 510 /*
 511  * Destroy requests collection
 512  */
 513 static void rbd_coll_release(struct kref *kref)
 514 {
 515         struct rbd_req_coll *coll =
 516                 container_of(kref, struct rbd_req_coll, kref);
 517
 518         dout("rbd_coll_release %p\n", coll);
 519         kfree(coll);
 520 }
 521
 522 static bool rbd_image_format_valid(u32 image_format)
 523 {
 524         return image_format == 1 || image_format == 2;
 525 }
 526
 527 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
 528 {
 529         size_t size;
 530         u32 snap_count;
 531
 532         /* The header has to start with the magic rbd header text */
 533         if (memcmp(&ondisk->text, RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT)))
 534                 return false;
 535
 536         /*
 537          * The size of a snapshot header has to fit in a size_t, and
 538          * that limits the number of snapshots.
 539          */
 540         snap_count = le32_to_cpu(ondisk->snap_count);
 541         size = SIZE_MAX - sizeof (struct ceph_snap_context);
 542         if (snap_count > size / sizeof (__le64))
 543                 return false;
 544
 545         /*
 546          * Not only that, but the size of the entire the snapshot
 547          * header must also be representable in a size_t.
 548          */
 549         size -= snap_count * sizeof (__le64);
 550         if ((u64) size < le64_to_cpu(ondisk->snap_names_len))
 551                 return false;
 552
 553         return true;
 554 }
 555
 556 /*
 557  * Create a new header structure, translate header format from the on-disk
 558  * header.
 559  */
 560 static int rbd_header_from_disk(struct rbd_image_header *header,
 561                                  struct rbd_image_header_ondisk *ondisk)
 562 {
 563         u32 snap_count;
 564         size_t len;
 565         size_t size;
 566         u32 i;
 567
 568         memset(header, 0, sizeof (*header));
 569
 570         snap_count = le32_to_cpu(ondisk->snap_count);
 571
 572         len = strnlen(ondisk->object_prefix, sizeof (ondisk->object_prefix));
 573         header->object_prefix = kmalloc(len + 1, GFP_KERNEL);
 574         if (!header->object_prefix)
 575                 return -ENOMEM;
 576         memcpy(header->object_prefix, ondisk->object_prefix, len);
 577         header->object_prefix[len] = '\0';
 578
 579         if (snap_count) {
 580                 u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len);
 581
 582                 /* Save a copy of the snapshot names */
 583
 584                 if (snap_names_len > (u64) SIZE_MAX)
 585                         return -EIO;
 586                 header->snap_names = kmalloc(snap_names_len, GFP_KERNEL);
 587                 if (!header->snap_names)
 588                         goto out_err;
 589                 /*
 590                  * Note that rbd_dev_v1_header_read() guarantees
 591                  * the ondisk buffer we're working with has
 592                  * snap_names_len bytes beyond the end of the
 593                  * snapshot id array, this memcpy() is safe.
 594                  */
 595                 memcpy(header->snap_names, &ondisk->snaps[snap_count],
 596                         snap_names_len);
 597
 598                 /* Record each snapshot's size */
 599
 600                 size = snap_count * sizeof (*header->snap_sizes);
 601                 header->snap_sizes = kmalloc(size, GFP_KERNEL);
 602                 if (!header->snap_sizes)
 603                         goto out_err;
 604                 for (i = 0; i < snap_count; i++)
 605                         header->snap_sizes[i] =
 606                                 le64_to_cpu(ondisk->snaps[i].image_size);
 607         } else {
 608                 WARN_ON(ondisk->snap_names_len);
 609                 header->snap_names = NULL;
 610                 header->snap_sizes = NULL;
 611         }
 612
 613         header->features = 0;   /* No features support in v1 images */
 614         header->obj_order = ondisk->options.order;
 615         header->crypt_type = ondisk->options.crypt_type;
 616         header->comp_type = ondisk->options.comp_type;
 617
 618         /* Allocate and fill in the snapshot context */
 619
 620         header->image_size = le64_to_cpu(ondisk->image_size);
 621         size = sizeof (struct ceph_snap_context);
 622         size += snap_count * sizeof (header->snapc->snaps[0]);
 623         header->snapc = kzalloc(size, GFP_KERNEL);
 624         if (!header->snapc)
 625                 goto out_err;
 626
 627         atomic_set(&header->snapc->nref, 1);
 628         header->snapc->seq = le64_to_cpu(ondisk->snap_seq);
 629         header->snapc->num_snaps = snap_count;
 630         for (i = 0; i < snap_count; i++)
 631                 header->snapc->snaps[i] =
 632                         le64_to_cpu(ondisk->snaps[i].id);
 633
 634         return 0;
 635
 636 out_err:
 637         kfree(header->snap_sizes);
 638         header->snap_sizes = NULL;
 639         kfree(header->snap_names);
 640         header->snap_names = NULL;
 641         kfree(header->object_prefix);
 642         header->object_prefix = NULL;
 643
 644         return -ENOMEM;
 645 }
 646
 647 static int snap_by_name(struct rbd_device *rbd_dev, const char *snap_name)
 648 {
 649
 650         struct rbd_snap *snap;
 651
 652         list_for_each_entry(snap, &rbd_dev->snaps, node) {
 653                 if (!strcmp(snap_name, snap->name)) {
 654                         rbd_dev->mapping.snap_id = snap->id;
 655                         rbd_dev->mapping.size = snap->size;
 656                         rbd_dev->mapping.features = snap->features;
 657
 658                         return 0;
 659                 }
 660         }
 661
 662         return -ENOENT;
 663 }
 664
 665 static int rbd_dev_set_mapping(struct rbd_device *rbd_dev, char *snap_name)
 666 {
 667         int ret;
 668
 669         if (!memcmp(snap_name, RBD_SNAP_HEAD_NAME,
 670                     sizeof (RBD_SNAP_HEAD_NAME))) {
 671                 rbd_dev->mapping.snap_id = CEPH_NOSNAP;
 672                 rbd_dev->mapping.size = rbd_dev->header.image_size;
 673                 rbd_dev->mapping.features = rbd_dev->header.features;
 674                 rbd_dev->mapping.snap_exists = false;
 675                 rbd_dev->mapping.read_only = rbd_dev->rbd_opts.read_only;
 676                 ret = 0;
 677         } else {
 678                 ret = snap_by_name(rbd_dev, snap_name);
 679                 if (ret < 0)
 680                         goto done;
 681                 rbd_dev->mapping.snap_exists = true;
 682                 rbd_dev->mapping.read_only = true;
 683         }
 684         rbd_dev->mapping.snap_name = snap_name;
 685 done:
 686         return ret;
 687 }
 688
 689 static void rbd_header_free(struct rbd_image_header *header)
 690 {
 691         kfree(header->object_prefix);
 692         header->object_prefix = NULL;
 693         kfree(header->snap_sizes);
 694         header->snap_sizes = NULL;
 695         kfree(header->snap_names);
 696         header->snap_names = NULL;
 697         ceph_put_snap_context(header->snapc);
 698         header->snapc = NULL;
 699 }
 700
 701 static char *rbd_segment_name(struct rbd_device *rbd_dev, u64 offset)
 702 {
 703         char *name;
 704         u64 segment;
 705         int ret;
 706
 707         name = kmalloc(RBD_MAX_SEG_NAME_LEN + 1, GFP_NOIO);
 708         if (!name)
 709                 return NULL;
 710         segment = offset >> rbd_dev->header.obj_order;
 711         ret = snprintf(name, RBD_MAX_SEG_NAME_LEN, "%s.%012llx",
 712                         rbd_dev->header.object_prefix, segment);
 713         if (ret < 0 || ret >= RBD_MAX_SEG_NAME_LEN) {
 714                 pr_err("error formatting segment name for #%llu (%d)\n",
 715                         segment, ret);
 716                 kfree(name);
 717                 name = NULL;
 718         }
 719
 720         return name;
 721 }
 722
 723 static u64 rbd_segment_offset(struct rbd_device *rbd_dev, u64 offset)
 724 {
 725         u64 segment_size = (u64) 1 << rbd_dev->header.obj_order;
 726
 727         return offset & (segment_size - 1);
 728 }
 729
 730 static u64 rbd_segment_length(struct rbd_device *rbd_dev,
 731                                 u64 offset, u64 length)
 732 {
 733         u64 segment_size = (u64) 1 << rbd_dev->header.obj_order;
 734
 735         offset &= segment_size - 1;
 736
 737         rbd_assert(length <= U64_MAX - offset);
 738         if (offset + length > segment_size)
 739                 length = segment_size - offset;
 740
 741         return length;
 742 }
 743
 744 static int rbd_get_num_segments(struct rbd_image_header *header,
 745                                 u64 ofs, u64 len)
 746 {
 747         u64 start_seg;
 748         u64 end_seg;
 749
 750         if (!len)
 751                 return 0;
 752         if (len - 1 > U64_MAX - ofs)
 753                 return -ERANGE;
 754
 755         start_seg = ofs >> header->obj_order;
 756         end_seg = (ofs + len - 1) >> header->obj_order;
 757
 758         return end_seg - start_seg + 1;
 759 }
 760
 761 /*
 762  * returns the size of an object in the image
 763  */
 764 static u64 rbd_obj_bytes(struct rbd_image_header *header)
 765 {
 766         return 1 << header->obj_order;
 767 }
 768
 769 /*
 770  * bio helpers
 771  */
 772
 773 static void bio_chain_put(struct bio *chain)
 774 {
 775         struct bio *tmp;
 776
 777         while (chain) {
 778                 tmp = chain;
 779                 chain = chain->bi_next;
 780                 bio_put(tmp);
 781         }
 782 }
 783
 784 /*
 785  * zeros a bio chain, starting at specific offset
 786  */
 787 static void zero_bio_chain(struct bio *chain, int start_ofs)
 788 {
 789         struct bio_vec *bv;
 790         unsigned long flags;
 791         void *buf;
 792         int i;
 793         int pos = 0;
 794
 795         while (chain) {
 796                 bio_for_each_segment(bv, chain, i) {
 797                         if (pos + bv->bv_len > start_ofs) {
 798                                 int remainder = max(start_ofs - pos, 0);
 799                                 buf = bvec_kmap_irq(bv, &flags);
 800                                 memset(buf + remainder, 0,
 801                                        bv->bv_len - remainder);
 802                                 bvec_kunmap_irq(buf, &flags);
 803                         }
 804                         pos += bv->bv_len;
 805                 }
 806
 807                 chain = chain->bi_next;
 808         }
 809 }
 810
 811 /*
 812  * bio_chain_clone - clone a chain of bios up to a certain length.
 813  * might return a bio_pair that will need to be released.
 814  */
 815 static struct bio *bio_chain_clone(struct bio **old, struct bio **next,
 816                                    struct bio_pair **bp,
 817                                    int len, gfp_t gfpmask)
 818 {
 819         struct bio *old_chain = *old;
 820         struct bio *new_chain = NULL;
 821         struct bio *tail;
 822         int total = 0;
 823
 824         if (*bp) {
 825                 bio_pair_release(*bp);
 826                 *bp = NULL;
 827         }
 828
 829         while (old_chain && (total < len)) {
 830                 struct bio *tmp;
 831
 832                 tmp = bio_kmalloc(gfpmask, old_chain->bi_max_vecs);
 833                 if (!tmp)
 834                         goto err_out;
 835                 gfpmask &= ~__GFP_WAIT; /* can't wait after the first */
 836
 837                 if (total + old_chain->bi_size > len) {
 838                         struct bio_pair *bp;
 839
 840                         /*
 841                          * this split can only happen with a single paged bio,
 842                          * split_bio will BUG_ON if this is not the case
 843                          */
 844                         dout("bio_chain_clone split! total=%d remaining=%d"
 845                              "bi_size=%u\n",
 846                              total, len - total, old_chain->bi_size);
 847
 848                         /* split the bio. We'll release it either in the next
 849                            call, or it will have to be released outside */
 850                         bp = bio_split(old_chain, (len - total) / SECTOR_SIZE);
 851                         if (!bp)
 852                                 goto err_out;
 853
 854                         __bio_clone(tmp, &bp->bio1);
 855
 856                         *next = &bp->bio2;
 857                 } else {
 858                         __bio_clone(tmp, old_chain);
 859                         *next = old_chain->bi_next;
 860                 }
 861
 862                 tmp->bi_bdev = NULL;
 863                 tmp->bi_next = NULL;
 864                 if (new_chain)
 865                         tail->bi_next = tmp;
 866                 else
 867                         new_chain = tmp;
 868                 tail = tmp;
 869                 old_chain = old_chain->bi_next;
 870
 871                 total += tmp->bi_size;
 872         }
 873
 874         rbd_assert(total == len);
 875
 876         *old = old_chain;
 877
 878         return new_chain;
 879
 880 err_out:
 881         dout("bio_chain_clone with err\n");
 882         bio_chain_put(new_chain);
 883         return NULL;
 884 }
 885
 886 /*
 887  * helpers for osd request op vectors.
 888  */
 889 static struct ceph_osd_req_op *rbd_create_rw_ops(int num_ops,
 890                                         int opcode, u32 payload_len)
 891 {
 892         struct ceph_osd_req_op *ops;
 893
 894         ops = kzalloc(sizeof (*ops) * (num_ops + 1), GFP_NOIO);
 895         if (!ops)
 896                 return NULL;
 897
 898         ops[0].op = opcode;
 899
 900         /*
 901          * op extent offset and length will be set later on
 902          * in calc_raw_layout()
 903          */
 904         ops[0].payload_len = payload_len;
 905
 906         return ops;
 907 }
 908
 909 static void rbd_destroy_ops(struct ceph_osd_req_op *ops)
 910 {
 911         kfree(ops);
 912 }
 913
 914 static void rbd_coll_end_req_index(struct request *rq,
 915                                    struct rbd_req_coll *coll,
 916                                    int index,
 917                                    int ret, u64 len)
 918 {
 919         struct request_queue *q;
 920         int min, max, i;
 921
 922         dout("rbd_coll_end_req_index %p index %d ret %d len %llu\n",
 923              coll, index, ret, (unsigned long long) len);
 924
 925         if (!rq)
 926                 return;
 927
 928         if (!coll) {
 929                 blk_end_request(rq, ret, len);
 930                 return;
 931         }
 932
 933         q = rq->q;
 934
 935         spin_lock_irq(q->queue_lock);
 936         coll->status[index].done = 1;
 937         coll->status[index].rc = ret;
 938         coll->status[index].bytes = len;
 939         max = min = coll->num_done;
 940         while (max < coll->total && coll->status[max].done)
 941                 max++;
 942
 943         for (i = min; i<max; i++) {
 944                 __blk_end_request(rq, coll->status[i].rc,
 945                                   coll->status[i].bytes);
 946                 coll->num_done++;
 947                 kref_put(&coll->kref, rbd_coll_release);
 948         }
 949         spin_unlock_irq(q->queue_lock);
 950 }
 951
 952 static void rbd_coll_end_req(struct rbd_request *req,
 953                              int ret, u64 len)
 954 {
 955         rbd_coll_end_req_index(req->rq, req->coll, req->coll_index, ret, len);
 956 }
 957
 958 /*
 959  * Send ceph osd request
 960  */
 961 static int rbd_do_request(struct request *rq,
 962                           struct rbd_device *rbd_dev,
 963                           struct ceph_snap_context *snapc,
 964                           u64 snapid,
 965                           const char *object_name, u64 ofs, u64 len,
 966                           struct bio *bio,
 967                           struct page **pages,
 968                           int num_pages,
 969                           int flags,
 970                           struct ceph_osd_req_op *ops,
 971                           struct rbd_req_coll *coll,
 972                           int coll_index,
 973                           void (*rbd_cb)(struct ceph_osd_request *req,
 974                                          struct ceph_msg *msg),
 975                           struct ceph_osd_request **linger_req,
 976                           u64 *ver)
 977 {
 978         struct ceph_osd_request *req;
 979         struct ceph_file_layout *layout;
 980         int ret;
 981         u64 bno;
 982         struct timespec mtime = CURRENT_TIME;
 983         struct rbd_request *req_data;
 984         struct ceph_osd_request_head *reqhead;
 985         struct ceph_osd_client *osdc;
 986
 987         req_data = kzalloc(sizeof(*req_data), GFP_NOIO);
 988         if (!req_data) {
 989                 if (coll)
 990                         rbd_coll_end_req_index(rq, coll, coll_index,
 991                                                -ENOMEM, len);
 992                 return -ENOMEM;
 993         }
 994
 995         if (coll) {
 996                 req_data->coll = coll;
 997                 req_data->coll_index = coll_index;
 998         }
 999
1000         dout("rbd_do_request object_name=%s ofs=%llu len=%llu\n", object_name,
1001                 (unsigned long long) ofs, (unsigned long long) len);
1002
1003         osdc = &rbd_dev->rbd_client->client->osdc;
1004         req = ceph_osdc_alloc_request(osdc, flags, snapc, ops,
1005                                         false, GFP_NOIO, pages, bio);
1006         if (!req) {
1007                 ret = -ENOMEM;
1008                 goto done_pages;
1009         }
1010
1011         req->r_callback = rbd_cb;
1012
1013         req_data->rq = rq;
1014         req_data->bio = bio;
1015         req_data->pages = pages;
1016         req_data->len = len;
1017
1018         req->r_priv = req_data;
1019
1020         reqhead = req->r_request->front.iov_base;
1021         reqhead->snapid = cpu_to_le64(CEPH_NOSNAP);
1022
1023         strncpy(req->r_oid, object_name, sizeof(req->r_oid));
1024         req->r_oid_len = strlen(req->r_oid);
1025
1026         layout = &req->r_file_layout;
1027         memset(layout, 0, sizeof(*layout));
1028         layout->fl_stripe_unit = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
1029         layout->fl_stripe_count = cpu_to_le32(1);
1030         layout->fl_object_size = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
1031         layout->fl_pg_pool = cpu_to_le32(rbd_dev->pool_id);
1032         ret = ceph_calc_raw_layout(osdc, layout, snapid, ofs, &len, &bno,
1033                                    req, ops);
1034         rbd_assert(ret == 0);
1035
1036         ceph_osdc_build_request(req, ofs, &len,
1037                                 ops,
1038                                 snapc,
1039                                 &mtime,
1040                                 req->r_oid, req->r_oid_len);
1041
1042         if (linger_req) {
1043                 ceph_osdc_set_request_linger(osdc, req);
1044                 *linger_req = req;
1045         }
1046
1047         ret = ceph_osdc_start_request(osdc, req, false);
1048         if (ret < 0)
1049                 goto done_err;
1050
1051         if (!rbd_cb) {
1052                 ret = ceph_osdc_wait_request(osdc, req);
1053                 if (ver)
1054                         *ver = le64_to_cpu(req->r_reassert_version.version);
1055                 dout("reassert_ver=%llu\n",
1056                         (unsigned long long)
1057                                 le64_to_cpu(req->r_reassert_version.version));
1058                 ceph_osdc_put_request(req);
1059         }
1060         return ret;
1061
1062 done_err:
1063         bio_chain_put(req_data->bio);
1064         ceph_osdc_put_request(req);
1065 done_pages:
1066         rbd_coll_end_req(req_data, ret, len);
1067         kfree(req_data);
1068         return ret;
1069 }
1070
1071 /*
1072  * Ceph osd op callback
1073  */
1074 static void rbd_req_cb(struct ceph_osd_request *req, struct ceph_msg *msg)
1075 {
1076         struct rbd_request *req_data = req->r_priv;
1077         struct ceph_osd_reply_head *replyhead;
1078         struct ceph_osd_op *op;
1079         __s32 rc;
1080         u64 bytes;
1081         int read_op;
1082
1083         /* parse reply */
1084         replyhead = msg->front.iov_base;
1085         WARN_ON(le32_to_cpu(replyhead->num_ops) == 0);
1086         op = (void *)(replyhead + 1);
1087         rc = le32_to_cpu(replyhead->result);
1088         bytes = le64_to_cpu(op->extent.length);
1089         read_op = (le16_to_cpu(op->op) == CEPH_OSD_OP_READ);
1090
1091         dout("rbd_req_cb bytes=%llu readop=%d rc=%d\n",
1092                 (unsigned long long) bytes, read_op, (int) rc);
1093
1094         if (rc == -ENOENT && read_op) {
1095                 zero_bio_chain(req_data->bio, 0);
1096                 rc = 0;
1097         } else if (rc == 0 && read_op && bytes < req_data->len) {
1098                 zero_bio_chain(req_data->bio, bytes);
1099                 bytes = req_data->len;
1100         }
1101
1102         rbd_coll_end_req(req_data, rc, bytes);
1103
1104         if (req_data->bio)
1105                 bio_chain_put(req_data->bio);
1106
1107         ceph_osdc_put_request(req);
1108         kfree(req_data);
1109 }
1110
1111 static void rbd_simple_req_cb(struct ceph_osd_request *req, struct ceph_msg *msg)
1112 {
1113         ceph_osdc_put_request(req);
1114 }
1115
1116 /*
1117  * Do a synchronous ceph osd operation
1118  */
1119 static int rbd_req_sync_op(struct rbd_device *rbd_dev,
1120                            struct ceph_snap_context *snapc,
1121                            u64 snapid,
1122                            int flags,
1123                            struct ceph_osd_req_op *ops,
1124                            const char *object_name,
1125                            u64 ofs, u64 inbound_size,
1126                            char *inbound,
1127                            struct ceph_osd_request **linger_req,
1128                            u64 *ver)
1129 {
1130         int ret;
1131         struct page **pages;
1132         int num_pages;
1133
1134         rbd_assert(ops != NULL);
1135
1136         num_pages = calc_pages_for(ofs, inbound_size);
1137         pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
1138         if (IS_ERR(pages))
1139                 return PTR_ERR(pages);
1140
1141         ret = rbd_do_request(NULL, rbd_dev, snapc, snapid,
1142                           object_name, ofs, inbound_size, NULL,
1143                           pages, num_pages,
1144                           flags,
1145                           ops,
1146                           NULL, 0,
1147                           NULL,
1148                           linger_req, ver);
1149         if (ret < 0)
1150                 goto done;
1151
1152         if ((flags & CEPH_OSD_FLAG_READ) && inbound)
1153                 ret = ceph_copy_from_page_vector(pages, inbound, ofs, ret);
1154
1155 done:
1156         ceph_release_page_vector(pages, num_pages);
1157         return ret;
1158 }
1159
1160 /*
1161  * Do an asynchronous ceph osd operation
1162  */
1163 static int rbd_do_op(struct request *rq,
1164                      struct rbd_device *rbd_dev,
1165                      struct ceph_snap_context *snapc,
1166                      u64 snapid,
1167                      int opcode, int flags,
1168                      u64 ofs, u64 len,
1169                      struct bio *bio,
1170                      struct rbd_req_coll *coll,
1171                      int coll_index)
1172 {
1173         char *seg_name;
1174         u64 seg_ofs;
1175         u64 seg_len;
1176         int ret;
1177         struct ceph_osd_req_op *ops;
1178         u32 payload_len;
1179
1180         seg_name = rbd_segment_name(rbd_dev, ofs);
1181         if (!seg_name)
1182                 return -ENOMEM;
1183         seg_len = rbd_segment_length(rbd_dev, ofs, len);
1184         seg_ofs = rbd_segment_offset(rbd_dev, ofs);
1185
1186         payload_len = (flags & CEPH_OSD_FLAG_WRITE ? seg_len : 0);
1187
1188         ret = -ENOMEM;
1189         ops = rbd_create_rw_ops(1, opcode, payload_len);
1190         if (!ops)
1191                 goto done;
1192
1193         /* we've taken care of segment sizes earlier when we
1194            cloned the bios. We should never have a segment
1195            truncated at this point */
1196         rbd_assert(seg_len == len);
1197
1198         ret = rbd_do_request(rq, rbd_dev, snapc, snapid,
1199                              seg_name, seg_ofs, seg_len,
1200                              bio,
1201                              NULL, 0,
1202                              flags,
1203                              ops,
1204                              coll, coll_index,
1205                              rbd_req_cb, 0, NULL);
1206
1207         rbd_destroy_ops(ops);
1208 done:
1209         kfree(seg_name);
1210         return ret;
1211 }
1212
1213 /*
1214  * Request async osd write
1215  */
1216 static int rbd_req_write(struct request *rq,
1217                          struct rbd_device *rbd_dev,
1218                          struct ceph_snap_context *snapc,
1219                          u64 ofs, u64 len,
1220                          struct bio *bio,
1221                          struct rbd_req_coll *coll,
1222                          int coll_index)
1223 {
1224         return rbd_do_op(rq, rbd_dev, snapc, CEPH_NOSNAP,
1225                          CEPH_OSD_OP_WRITE,
1226                          CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1227                          ofs, len, bio, coll, coll_index);
1228 }
1229
1230 /*
1231  * Request async osd read
1232  */
1233 static int rbd_req_read(struct request *rq,
1234                          struct rbd_device *rbd_dev,
1235                          u64 snapid,
1236                          u64 ofs, u64 len,
1237                          struct bio *bio,
1238                          struct rbd_req_coll *coll,
1239                          int coll_index)
1240 {
1241         return rbd_do_op(rq, rbd_dev, NULL,
1242                          snapid,
1243                          CEPH_OSD_OP_READ,
1244                          CEPH_OSD_FLAG_READ,
1245                          ofs, len, bio, coll, coll_index);
1246 }
1247
1248 /*
1249  * Request sync osd read
1250  */
1251 static int rbd_req_sync_read(struct rbd_device *rbd_dev,
1252                           u64 snapid,
1253                           const char *object_name,
1254                           u64 ofs, u64 len,
1255                           char *buf,
1256                           u64 *ver)
1257 {
1258         struct ceph_osd_req_op *ops;
1259         int ret;
1260
1261         ops = rbd_create_rw_ops(1, CEPH_OSD_OP_READ, 0);
1262         if (!ops)
1263                 return -ENOMEM;
1264
1265         ret = rbd_req_sync_op(rbd_dev, NULL,
1266                                snapid,
1267                                CEPH_OSD_FLAG_READ,
1268                                ops, object_name, ofs, len, buf, NULL, ver);
1269         rbd_destroy_ops(ops);
1270
1271         return ret;
1272 }
1273
1274 /*
1275  * Request sync osd watch
1276  */
1277 static int rbd_req_sync_notify_ack(struct rbd_device *rbd_dev,
1278                                    u64 ver,
1279                                    u64 notify_id)
1280 {
1281         struct ceph_osd_req_op *ops;
1282         int ret;
1283
1284         ops = rbd_create_rw_ops(1, CEPH_OSD_OP_NOTIFY_ACK, 0);
1285         if (!ops)
1286                 return -ENOMEM;
1287
1288         ops[0].watch.ver = cpu_to_le64(ver);
1289         ops[0].watch.cookie = notify_id;
1290         ops[0].watch.flag = 0;
1291
1292         ret = rbd_do_request(NULL, rbd_dev, NULL, CEPH_NOSNAP,
1293                           rbd_dev->header_name, 0, 0, NULL,
1294                           NULL, 0,
1295                           CEPH_OSD_FLAG_READ,
1296                           ops,
1297                           NULL, 0,
1298                           rbd_simple_req_cb, 0, NULL);
1299
1300         rbd_destroy_ops(ops);
1301         return ret;
1302 }
1303
1304 static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
1305 {
1306         struct rbd_device *rbd_dev = (struct rbd_device *)data;
1307         u64 hver;
1308         int rc;
1309
1310         if (!rbd_dev)
1311                 return;
1312
1313         dout("rbd_watch_cb %s notify_id=%llu opcode=%u\n",
1314                 rbd_dev->header_name, (unsigned long long) notify_id,
1315                 (unsigned int) opcode);
1316         rc = rbd_dev_refresh(rbd_dev, &hver);
1317         if (rc)
1318                 pr_warning(RBD_DRV_NAME "%d got notification but failed to "
1319                            " update snaps: %d\n", rbd_dev->major, rc);
1320
1321         rbd_req_sync_notify_ack(rbd_dev, hver, notify_id);
1322 }
1323
1324 /*
1325  * Request sync osd watch
1326  */
1327 static int rbd_req_sync_watch(struct rbd_device *rbd_dev)
1328 {
1329         struct ceph_osd_req_op *ops;
1330         struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1331         int ret;
1332
1333         ops = rbd_create_rw_ops(1, CEPH_OSD_OP_WATCH, 0);
1334         if (!ops)
1335                 return -ENOMEM;
1336
1337         ret = ceph_osdc_create_event(osdc, rbd_watch_cb, 0,
1338                                      (void *)rbd_dev, &rbd_dev->watch_event);
1339         if (ret < 0)
1340                 goto fail;
1341
1342         ops[0].watch.ver = cpu_to_le64(rbd_dev->header.obj_version);
1343         ops[0].watch.cookie = cpu_to_le64(rbd_dev->watch_event->cookie);
1344         ops[0].watch.flag = 1;
1345
1346         ret = rbd_req_sync_op(rbd_dev, NULL,
1347                               CEPH_NOSNAP,
1348                               CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1349                               ops,
1350                               rbd_dev->header_name,
1351                               0, 0, NULL,
1352                               &rbd_dev->watch_request, NULL);
1353
1354         if (ret < 0)
1355                 goto fail_event;
1356
1357         rbd_destroy_ops(ops);
1358         return 0;
1359
1360 fail_event:
1361         ceph_osdc_cancel_event(rbd_dev->watch_event);
1362         rbd_dev->watch_event = NULL;
1363 fail:
1364         rbd_destroy_ops(ops);
1365         return ret;
1366 }
1367
1368 /*
1369  * Request sync osd unwatch
1370  */
1371 static int rbd_req_sync_unwatch(struct rbd_device *rbd_dev)
1372 {
1373         struct ceph_osd_req_op *ops;
1374         int ret;
1375
1376         ops = rbd_create_rw_ops(1, CEPH_OSD_OP_WATCH, 0);
1377         if (!ops)
1378                 return -ENOMEM;
1379
1380         ops[0].watch.ver = 0;
1381         ops[0].watch.cookie = cpu_to_le64(rbd_dev->watch_event->cookie);
1382         ops[0].watch.flag = 0;
1383
1384         ret = rbd_req_sync_op(rbd_dev, NULL,
1385                               CEPH_NOSNAP,
1386                               CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1387                               ops,
1388                               rbd_dev->header_name,
1389                               0, 0, NULL, NULL, NULL);
1390
1391
1392         rbd_destroy_ops(ops);
1393         ceph_osdc_cancel_event(rbd_dev->watch_event);
1394         rbd_dev->watch_event = NULL;
1395         return ret;
1396 }
1397
1398 /*
1399  * Synchronous osd object method call
1400  */
1401 static int rbd_req_sync_exec(struct rbd_device *rbd_dev,
1402                              const char *object_name,
1403                              const char *class_name,
1404                              const char *method_name,
1405                              const char *outbound,
1406                              size_t outbound_size,
1407                              char *inbound,
1408                              size_t inbound_size,
1409                              int flags,
1410                              u64 *ver)
1411 {
1412         struct ceph_osd_req_op *ops;
1413         int class_name_len = strlen(class_name);
1414         int method_name_len = strlen(method_name);
1415         int payload_size;
1416         int ret;
1417
1418         /*
1419          * Any input parameters required by the method we're calling
1420          * will be sent along with the class and method names as
1421          * part of the message payload.  That data and its size are
1422          * supplied via the indata and indata_len fields (named from
1423          * the perspective of the server side) in the OSD request
1424          * operation.
1425          */
1426         payload_size = class_name_len + method_name_len + outbound_size;
1427         ops = rbd_create_rw_ops(1, CEPH_OSD_OP_CALL, payload_size);
1428         if (!ops)
1429                 return -ENOMEM;
1430
1431         ops[0].cls.class_name = class_name;
1432         ops[0].cls.class_len = (__u8) class_name_len;
1433         ops[0].cls.method_name = method_name;
1434         ops[0].cls.method_len = (__u8) method_name_len;
1435         ops[0].cls.argc = 0;
1436         ops[0].cls.indata = outbound;
1437         ops[0].cls.indata_len = outbound_size;
1438
1439         ret = rbd_req_sync_op(rbd_dev, NULL,
1440                                CEPH_NOSNAP,
1441                                flags, ops,
1442                                object_name, 0, inbound_size, inbound,
1443                                NULL, ver);
1444
1445         rbd_destroy_ops(ops);
1446
1447         dout("cls_exec returned %d\n", ret);
1448         return ret;
1449 }
1450
1451 static struct rbd_req_coll *rbd_alloc_coll(int num_reqs)
1452 {
1453         struct rbd_req_coll *coll =
1454                         kzalloc(sizeof(struct rbd_req_coll) +
1455                                 sizeof(struct rbd_req_status) * num_reqs,
1456                                 GFP_ATOMIC);
1457
1458         if (!coll)
1459                 return NULL;
1460         coll->total = num_reqs;
1461         kref_init(&coll->kref);
1462         return coll;
1463 }
1464
1465 /*
1466  * block device queue callback
1467  */
1468 static void rbd_rq_fn(struct request_queue *q)
1469 {
1470         struct rbd_device *rbd_dev = q->queuedata;
1471         struct request *rq;
1472         struct bio_pair *bp = NULL;
1473
1474         while ((rq = blk_fetch_request(q))) {
1475                 struct bio *bio;
1476                 struct bio *rq_bio, *next_bio = NULL;
1477                 bool do_write;
1478                 unsigned int size;
1479                 u64 op_size = 0;
1480                 u64 ofs;
1481                 int num_segs, cur_seg = 0;
1482                 struct rbd_req_coll *coll;
1483                 struct ceph_snap_context *snapc;
1484
1485                 dout("fetched request\n");
1486
1487                 /* filter out block requests we don't understand */
1488                 if ((rq->cmd_type != REQ_TYPE_FS)) {
1489                         __blk_end_request_all(rq, 0);
1490                         continue;
1491                 }
1492
1493                 /* deduce our operation (read, write) */
1494                 do_write = (rq_data_dir(rq) == WRITE);
1495
1496                 size = blk_rq_bytes(rq);
1497                 ofs = blk_rq_pos(rq) * SECTOR_SIZE;
1498                 rq_bio = rq->bio;
1499                 if (do_write && rbd_dev->mapping.read_only) {
1500                         __blk_end_request_all(rq, -EROFS);
1501                         continue;
1502                 }
1503
1504                 spin_unlock_irq(q->queue_lock);
1505
1506                 down_read(&rbd_dev->header_rwsem);
1507
1508                 if (rbd_dev->mapping.snap_id != CEPH_NOSNAP &&
1509                                 !rbd_dev->mapping.snap_exists) {
1510                         up_read(&rbd_dev->header_rwsem);
1511                         dout("request for non-existent snapshot");
1512                         spin_lock_irq(q->queue_lock);
1513                         __blk_end_request_all(rq, -ENXIO);
1514                         continue;
1515                 }
1516
1517                 snapc = ceph_get_snap_context(rbd_dev->header.snapc);
1518
1519                 up_read(&rbd_dev->header_rwsem);
1520
1521                 dout("%s 0x%x bytes at 0x%llx\n",
1522                      do_write ? "write" : "read",
1523                      size, (unsigned long long) blk_rq_pos(rq) * SECTOR_SIZE);
1524
1525                 num_segs = rbd_get_num_segments(&rbd_dev->header, ofs, size);
1526                 if (num_segs <= 0) {
1527                         spin_lock_irq(q->queue_lock);
1528                         __blk_end_request_all(rq, num_segs);
1529                         ceph_put_snap_context(snapc);
1530                         continue;
1531                 }
1532                 coll = rbd_alloc_coll(num_segs);
1533                 if (!coll) {
1534                         spin_lock_irq(q->queue_lock);
1535                         __blk_end_request_all(rq, -ENOMEM);
1536                         ceph_put_snap_context(snapc);
1537                         continue;
1538                 }
1539
1540                 do {
1541                         /* a bio clone to be passed down to OSD req */
1542                         dout("rq->bio->bi_vcnt=%hu\n", rq->bio->bi_vcnt);
1543                         op_size = rbd_segment_length(rbd_dev, ofs, size);
1544                         kref_get(&coll->kref);
1545                         bio = bio_chain_clone(&rq_bio, &next_bio, &bp,
1546                                               op_size, GFP_ATOMIC);
1547                         if (!bio) {
1548                                 rbd_coll_end_req_index(rq, coll, cur_seg,
1549                                                        -ENOMEM, op_size);
1550                                 goto next_seg;
1551                         }
1552
1553
1554                         /* init OSD command: write or read */
1555                         if (do_write)
1556                                 rbd_req_write(rq, rbd_dev,
1557                                               snapc,
1558                                               ofs,
1559                                               op_size, bio,
1560                                               coll, cur_seg);
1561                         else
1562                                 rbd_req_read(rq, rbd_dev,
1563                                              rbd_dev->mapping.snap_id,
1564                                              ofs,
1565                                              op_size, bio,
1566                                              coll, cur_seg);
1567
1568 next_seg:
1569                         size -= op_size;
1570                         ofs += op_size;
1571
1572                         cur_seg++;
1573                         rq_bio = next_bio;
1574                 } while (size > 0);
1575                 kref_put(&coll->kref, rbd_coll_release);
1576
1577                 if (bp)
1578                         bio_pair_release(bp);
1579                 spin_lock_irq(q->queue_lock);
1580
1581                 ceph_put_snap_context(snapc);
1582         }
1583 }
1584
1585 /*
1586  * a queue callback. Makes sure that we don't create a bio that spans across
1587  * multiple osd objects. One exception would be with a single page bios,
1588  * which we handle later at bio_chain_clone
1589  */
1590 static int rbd_merge_bvec(struct request_queue *q, struct bvec_merge_data *bmd,
1591                           struct bio_vec *bvec)
1592 {
1593         struct rbd_device *rbd_dev = q->queuedata;
1594         unsigned int chunk_sectors;
1595         sector_t sector;
1596         unsigned int bio_sectors;
1597         int max;
1598
1599         chunk_sectors = 1 << (rbd_dev->header.obj_order - SECTOR_SHIFT);
1600         sector = bmd->bi_sector + get_start_sect(bmd->bi_bdev);
1601         bio_sectors = bmd->bi_size >> SECTOR_SHIFT;
1602
1603         max =  (chunk_sectors - ((sector & (chunk_sectors - 1))
1604                                  + bio_sectors)) << SECTOR_SHIFT;
1605         if (max < 0)
1606                 max = 0; /* bio_add cannot handle a negative return */
1607         if (max <= bvec->bv_len && bio_sectors == 0)
1608                 return bvec->bv_len;
1609         return max;
1610 }
1611
1612 static void rbd_free_disk(struct rbd_device *rbd_dev)
1613 {
1614         struct gendisk *disk = rbd_dev->disk;
1615
1616         if (!disk)
1617                 return;
1618
1619         if (disk->flags & GENHD_FL_UP)
1620                 del_gendisk(disk);
1621         if (disk->queue)
1622                 blk_cleanup_queue(disk->queue);
1623         put_disk(disk);
1624 }
1625
1626 /*
1627  * Read the complete header for the given rbd device.
1628  *
1629  * Returns a pointer to a dynamically-allocated buffer containing
1630  * the complete and validated header.  Caller can pass the address
1631  * of a variable that will be filled in with the version of the
1632  * header object at the time it was read.
1633  *
1634  * Returns a pointer-coded errno if a failure occurs.
1635  */
1636 static struct rbd_image_header_ondisk *
1637 rbd_dev_v1_header_read(struct rbd_device *rbd_dev, u64 *version)
1638 {
1639         struct rbd_image_header_ondisk *ondisk = NULL;
1640         u32 snap_count = 0;
1641         u64 names_size = 0;
1642         u32 want_count;
1643         int ret;
1644
1645         /*
1646          * The complete header will include an array of its 64-bit
1647          * snapshot ids, followed by the names of those snapshots as
1648          * a contiguous block of NUL-terminated strings.  Note that
1649          * the number of snapshots could change by the time we read
1650          * it in, in which case we re-read it.
1651          */
1652         do {
1653                 size_t size;
1654
1655                 kfree(ondisk);
1656
1657                 size = sizeof (*ondisk);
1658                 size += snap_count * sizeof (struct rbd_image_snap_ondisk);
1659                 size += names_size;
1660                 ondisk = kmalloc(size, GFP_KERNEL);
1661                 if (!ondisk)
1662                         return ERR_PTR(-ENOMEM);
1663
1664                 ret = rbd_req_sync_read(rbd_dev, CEPH_NOSNAP,
1665                                        rbd_dev->header_name,
1666                                        0, size,
1667                                        (char *) ondisk, version);
1668
1669                 if (ret < 0)
1670                         goto out_err;
1671                 if (WARN_ON((size_t) ret < size)) {
1672                         ret = -ENXIO;
1673                         pr_warning("short header read for image %s"
1674                                         " (want %zd got %d)\n",
1675                                 rbd_dev->image_name, size, ret);
1676                         goto out_err;
1677                 }
1678                 if (!rbd_dev_ondisk_valid(ondisk)) {
1679                         ret = -ENXIO;
1680                         pr_warning("invalid header for image %s\n",
1681                                 rbd_dev->image_name);
1682                         goto out_err;
1683                 }
1684
1685                 names_size = le64_to_cpu(ondisk->snap_names_len);
1686                 want_count = snap_count;
1687                 snap_count = le32_to_cpu(ondisk->snap_count);
1688         } while (snap_count != want_count);
1689
1690         return ondisk;
1691
1692 out_err:
1693         kfree(ondisk);
1694
1695         return ERR_PTR(ret);
1696 }
1697
1698 /*
1699  * reload the ondisk the header
1700  */
1701 static int rbd_read_header(struct rbd_device *rbd_dev,
1702                            struct rbd_image_header *header)
1703 {
1704         struct rbd_image_header_ondisk *ondisk;
1705         u64 ver = 0;
1706         int ret;
1707
1708         ondisk = rbd_dev_v1_header_read(rbd_dev, &ver);
1709         if (IS_ERR(ondisk))
1710                 return PTR_ERR(ondisk);
1711         ret = rbd_header_from_disk(header, ondisk);
1712         if (ret >= 0)
1713                 header->obj_version = ver;
1714         kfree(ondisk);
1715
1716         return ret;
1717 }
1718
1719 static void __rbd_remove_all_snaps(struct rbd_device *rbd_dev)
1720 {
1721         struct rbd_snap *snap;
1722         struct rbd_snap *next;
1723
1724         list_for_each_entry_safe(snap, next, &rbd_dev->snaps, node)
1725                 __rbd_remove_snap_dev(snap);
1726 }
1727
1728 static void rbd_update_mapping_size(struct rbd_device *rbd_dev)
1729 {
1730         sector_t size;
1731
1732         if (rbd_dev->mapping.snap_id != CEPH_NOSNAP)
1733                 return;
1734
1735         size = (sector_t) rbd_dev->header.image_size / SECTOR_SIZE;
1736         dout("setting size to %llu sectors", (unsigned long long) size);
1737         rbd_dev->mapping.size = (u64) size;
1738         set_capacity(rbd_dev->disk, size);
1739 }
1740
1741 /*
1742  * only read the first part of the ondisk header, without the snaps info
1743  */
1744 static int rbd_dev_v1_refresh(struct rbd_device *rbd_dev, u64 *hver)
1745 {
1746         int ret;
1747         struct rbd_image_header h;
1748
1749         ret = rbd_read_header(rbd_dev, &h);
1750         if (ret < 0)
1751                 return ret;
1752
1753         down_write(&rbd_dev->header_rwsem);
1754
1755         /* Update image size, and check for resize of mapped image */
1756         rbd_dev->header.image_size = h.image_size;
1757         rbd_update_mapping_size(rbd_dev);
1758
1759         /* rbd_dev->header.object_prefix shouldn't change */
1760         kfree(rbd_dev->header.snap_sizes);
1761         kfree(rbd_dev->header.snap_names);
1762         /* osd requests may still refer to snapc */
1763         ceph_put_snap_context(rbd_dev->header.snapc);
1764
1765         if (hver)
1766                 *hver = h.obj_version;
1767         rbd_dev->header.obj_version = h.obj_version;
1768         rbd_dev->header.image_size = h.image_size;
1769         rbd_dev->header.snapc = h.snapc;
1770         rbd_dev->header.snap_names = h.snap_names;
1771         rbd_dev->header.snap_sizes = h.snap_sizes;
1772         /* Free the extra copy of the object prefix */
1773         WARN_ON(strcmp(rbd_dev->header.object_prefix, h.object_prefix));
1774         kfree(h.object_prefix);
1775
1776         ret = rbd_dev_snaps_update(rbd_dev);
1777         if (!ret)
1778                 ret = rbd_dev_snaps_register(rbd_dev);
1779
1780         up_write(&rbd_dev->header_rwsem);
1781
1782         return ret;
1783 }
1784
1785 static int rbd_dev_refresh(struct rbd_device *rbd_dev, u64 *hver)
1786 {
1787         int ret;
1788
1789         rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1790         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
1791         if (rbd_dev->image_format == 1)
1792                 ret = rbd_dev_v1_refresh(rbd_dev, hver);
1793         else
1794                 ret = rbd_dev_v2_refresh(rbd_dev, hver);
1795         mutex_unlock(&ctl_mutex);
1796
1797         return ret;
1798 }
1799
1800 static int rbd_init_disk(struct rbd_device *rbd_dev)
1801 {
1802         struct gendisk *disk;
1803         struct request_queue *q;
1804         u64 segment_size;
1805
1806         /* create gendisk info */
1807         disk = alloc_disk(RBD_MINORS_PER_MAJOR);
1808         if (!disk)
1809                 return -ENOMEM;
1810
1811         snprintf(disk->disk_name, sizeof(disk->disk_name), RBD_DRV_NAME "%d",
1812                  rbd_dev->dev_id);
1813         disk->major = rbd_dev->major;
1814         disk->first_minor = 0;
1815         disk->fops = &rbd_bd_ops;
1816         disk->private_data = rbd_dev;
1817
1818         /* init rq */
1819         q = blk_init_queue(rbd_rq_fn, &rbd_dev->lock);
1820         if (!q)
1821                 goto out_disk;
1822
1823         /* We use the default size, but let's be explicit about it. */
1824         blk_queue_physical_block_size(q, SECTOR_SIZE);
1825
1826         /* set io sizes to object size */
1827         segment_size = rbd_obj_bytes(&rbd_dev->header);
1828         blk_queue_max_hw_sectors(q, segment_size / SECTOR_SIZE);
1829         blk_queue_max_segment_size(q, segment_size);
1830         blk_queue_io_min(q, segment_size);
1831         blk_queue_io_opt(q, segment_size);
1832
1833         blk_queue_merge_bvec(q, rbd_merge_bvec);
1834         disk->queue = q;
1835
1836         q->queuedata = rbd_dev;
1837
1838         rbd_dev->disk = disk;
1839
1840         set_capacity(rbd_dev->disk, rbd_dev->mapping.size / SECTOR_SIZE);
1841
1842         return 0;
1843 out_disk:
1844         put_disk(disk);
1845
1846         return -ENOMEM;
1847 }
1848
1849 /*
1850   sysfs
1851 */
1852
1853 static struct rbd_device *dev_to_rbd_dev(struct device *dev)
1854 {
1855         return container_of(dev, struct rbd_device, dev);
1856 }
1857
1858 static ssize_t rbd_size_show(struct device *dev,
1859                              struct device_attribute *attr, char *buf)
1860 {
1861         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1862         sector_t size;
1863
1864         down_read(&rbd_dev->header_rwsem);
1865         size = get_capacity(rbd_dev->disk);
1866         up_read(&rbd_dev->header_rwsem);
1867
1868         return sprintf(buf, "%llu\n", (unsigned long long) size * SECTOR_SIZE);
1869 }
1870
1871 /*
1872  * Note this shows the features for whatever's mapped, which is not
1873  * necessarily the base image.
1874  */
1875 static ssize_t rbd_features_show(struct device *dev,
1876                              struct device_attribute *attr, char *buf)
1877 {
1878         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1879
1880         return sprintf(buf, "0x%016llx\n",
1881                         (unsigned long long) rbd_dev->mapping.features);
1882 }
1883
1884 static ssize_t rbd_major_show(struct device *dev,
1885                               struct device_attribute *attr, char *buf)
1886 {
1887         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1888
1889         return sprintf(buf, "%d\n", rbd_dev->major);
1890 }
1891
1892 static ssize_t rbd_client_id_show(struct device *dev,
1893                                   struct device_attribute *attr, char *buf)
1894 {
1895         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1896
1897         return sprintf(buf, "client%lld\n",
1898                         ceph_client_id(rbd_dev->rbd_client->client));
1899 }
1900
1901 static ssize_t rbd_pool_show(struct device *dev,
1902                              struct device_attribute *attr, char *buf)
1903 {
1904         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1905
1906         return sprintf(buf, "%s\n", rbd_dev->pool_name);
1907 }
1908
1909 static ssize_t rbd_pool_id_show(struct device *dev,
1910                              struct device_attribute *attr, char *buf)
1911 {
1912         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1913
1914         return sprintf(buf, "%d\n", rbd_dev->pool_id);
1915 }
1916
1917 static ssize_t rbd_name_show(struct device *dev,
1918                              struct device_attribute *attr, char *buf)
1919 {
1920         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1921
1922         return sprintf(buf, "%s\n", rbd_dev->image_name);
1923 }
1924
1925 static ssize_t rbd_image_id_show(struct device *dev,
1926                              struct device_attribute *attr, char *buf)
1927 {
1928         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1929
1930         return sprintf(buf, "%s\n", rbd_dev->image_id);
1931 }
1932
1933 /*
1934  * Shows the name of the currently-mapped snapshot (or
1935  * RBD_SNAP_HEAD_NAME for the base image).
1936  */
1937 static ssize_t rbd_snap_show(struct device *dev,
1938                              struct device_attribute *attr,
1939                              char *buf)
1940 {
1941         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1942
1943         return sprintf(buf, "%s\n", rbd_dev->mapping.snap_name);
1944 }
1945
1946 static ssize_t rbd_image_refresh(struct device *dev,
1947                                  struct device_attribute *attr,
1948                                  const char *buf,
1949                                  size_t size)
1950 {
1951         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
1952         int ret;
1953
1954         ret = rbd_dev_refresh(rbd_dev, NULL);
1955
1956         return ret < 0 ? ret : size;
1957 }
1958
1959 static DEVICE_ATTR(size, S_IRUGO, rbd_size_show, NULL);
1960 static DEVICE_ATTR(features, S_IRUGO, rbd_features_show, NULL);
1961 static DEVICE_ATTR(major, S_IRUGO, rbd_major_show, NULL);
1962 static DEVICE_ATTR(client_id, S_IRUGO, rbd_client_id_show, NULL);
1963 static DEVICE_ATTR(pool, S_IRUGO, rbd_pool_show, NULL);
1964 static DEVICE_ATTR(pool_id, S_IRUGO, rbd_pool_id_show, NULL);
1965 static DEVICE_ATTR(name, S_IRUGO, rbd_name_show, NULL);
1966 static DEVICE_ATTR(image_id, S_IRUGO, rbd_image_id_show, NULL);
1967 static DEVICE_ATTR(refresh, S_IWUSR, NULL, rbd_image_refresh);
1968 static DEVICE_ATTR(current_snap, S_IRUGO, rbd_snap_show, NULL);
1969
1970 static struct attribute *rbd_attrs[] = {
1971         &dev_attr_size.attr,
1972         &dev_attr_features.attr,
1973         &dev_attr_major.attr,
1974         &dev_attr_client_id.attr,
1975         &dev_attr_pool.attr,
1976         &dev_attr_pool_id.attr,
1977         &dev_attr_name.attr,
1978         &dev_attr_image_id.attr,
1979         &dev_attr_current_snap.attr,
1980         &dev_attr_refresh.attr,
1981         NULL
1982 };
1983
1984 static struct attribute_group rbd_attr_group = {
1985         .attrs = rbd_attrs,
1986 };
1987
1988 static const struct attribute_group *rbd_attr_groups[] = {
1989         &rbd_attr_group,
1990         NULL
1991 };
1992
1993 static void rbd_sysfs_dev_release(struct device *dev)
1994 {
1995 }
1996
1997 static struct device_type rbd_device_type = {
1998         .name           = "rbd",
1999         .groups         = rbd_attr_groups,
2000         .release        = rbd_sysfs_dev_release,
2001 };
2002
2003
2004 /*
2005   sysfs - snapshots
2006 */
2007
2008 static ssize_t rbd_snap_size_show(struct device *dev,
2009                                   struct device_attribute *attr,
2010                                   char *buf)
2011 {
2012         struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
2013
2014         return sprintf(buf, "%llu\n", (unsigned long long)snap->size);
2015 }
2016
2017 static ssize_t rbd_snap_id_show(struct device *dev,
2018                                 struct device_attribute *attr,
2019                                 char *buf)
2020 {
2021         struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
2022
2023         return sprintf(buf, "%llu\n", (unsigned long long)snap->id);
2024 }
2025
2026 static ssize_t rbd_snap_features_show(struct device *dev,
2027                                 struct device_attribute *attr,
2028                                 char *buf)
2029 {
2030         struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
2031
2032         return sprintf(buf, "0x%016llx\n",
2033                         (unsigned long long) snap->features);
2034 }
2035
2036 static DEVICE_ATTR(snap_size, S_IRUGO, rbd_snap_size_show, NULL);
2037 static DEVICE_ATTR(snap_id, S_IRUGO, rbd_snap_id_show, NULL);
2038 static DEVICE_ATTR(snap_features, S_IRUGO, rbd_snap_features_show, NULL);
2039
2040 static struct attribute *rbd_snap_attrs[] = {
2041         &dev_attr_snap_size.attr,
2042         &dev_attr_snap_id.attr,
2043         &dev_attr_snap_features.attr,
2044         NULL,
2045 };
2046
2047 static struct attribute_group rbd_snap_attr_group = {
2048         .attrs = rbd_snap_attrs,
2049 };
2050
2051 static void rbd_snap_dev_release(struct device *dev)
2052 {
2053         struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
2054         kfree(snap->name);
2055         kfree(snap);
2056 }
2057
2058 static const struct attribute_group *rbd_snap_attr_groups[] = {
2059         &rbd_snap_attr_group,
2060         NULL
2061 };
2062
2063 static struct device_type rbd_snap_device_type = {
2064         .groups         = rbd_snap_attr_groups,
2065         .release        = rbd_snap_dev_release,
2066 };
2067
2068 static bool rbd_snap_registered(struct rbd_snap *snap)
2069 {
2070         bool ret = snap->dev.type == &rbd_snap_device_type;
2071         bool reg = device_is_registered(&snap->dev);
2072
2073         rbd_assert(!ret ^ reg);
2074
2075         return ret;
2076 }
2077
2078 static void __rbd_remove_snap_dev(struct rbd_snap *snap)
2079 {
2080         list_del(&snap->node);
2081         if (device_is_registered(&snap->dev))
2082                 device_unregister(&snap->dev);
2083 }
2084
2085 static int rbd_register_snap_dev(struct rbd_snap *snap,
2086                                   struct device *parent)
2087 {
2088         struct device *dev = &snap->dev;
2089         int ret;
2090
2091         dev->type = &rbd_snap_device_type;
2092         dev->parent = parent;
2093         dev->release = rbd_snap_dev_release;
2094         dev_set_name(dev, "snap_%s", snap->name);
2095         dout("%s: registering device for snapshot %s\n", __func__, snap->name);
2096
2097         ret = device_register(dev);
2098
2099         return ret;
2100 }
2101
2102 static struct rbd_snap *__rbd_add_snap_dev(struct rbd_device *rbd_dev,
2103                                                 const char *snap_name,
2104                                                 u64 snap_id, u64 snap_size,
2105                                                 u64 snap_features)
2106 {
2107         struct rbd_snap *snap;
2108         int ret;
2109
2110         snap = kzalloc(sizeof (*snap), GFP_KERNEL);
2111         if (!snap)
2112                 return ERR_PTR(-ENOMEM);
2113
2114         ret = -ENOMEM;
2115         snap->name = kstrdup(snap_name, GFP_KERNEL);
2116         if (!snap->name)
2117                 goto err;
2118
2119         snap->id = snap_id;
2120         snap->size = snap_size;
2121         snap->features = snap_features;
2122
2123         return snap;
2124
2125 err:
2126         kfree(snap->name);
2127         kfree(snap);
2128
2129         return ERR_PTR(ret);
2130 }
2131
2132 static char *rbd_dev_v1_snap_info(struct rbd_device *rbd_dev, u32 which,
2133                 u64 *snap_size, u64 *snap_features)
2134 {
2135         char *snap_name;
2136
2137         rbd_assert(which < rbd_dev->header.snapc->num_snaps);
2138
2139         *snap_size = rbd_dev->header.snap_sizes[which];
2140         *snap_features = 0;     /* No features for v1 */
2141
2142         /* Skip over names until we find the one we are looking for */
2143
2144         snap_name = rbd_dev->header.snap_names;
2145         while (which--)
2146                 snap_name += strlen(snap_name) + 1;
2147
2148         return snap_name;
2149 }
2150
2151 /*
2152  * Get the size and object order for an image snapshot, or if
2153  * snap_id is CEPH_NOSNAP, gets this information for the base
2154  * image.
2155  */
2156 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
2157                                 u8 *order, u64 *snap_size)
2158 {
2159         __le64 snapid = cpu_to_le64(snap_id);
2160         int ret;
2161         struct {
2162                 u8 order;
2163                 __le64 size;
2164         } __attribute__ ((packed)) size_buf = { 0 };
2165
2166         ret = rbd_req_sync_exec(rbd_dev, rbd_dev->header_name,
2167                                 "rbd", "get_size",
2168                                 (char *) &snapid, sizeof (snapid),
2169                                 (char *) &size_buf, sizeof (size_buf),
2170                                 CEPH_OSD_FLAG_READ, NULL);
2171         dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
2172         if (ret < 0)
2173                 return ret;
2174
2175         *order = size_buf.order;
2176         *snap_size = le64_to_cpu(size_buf.size);
2177
2178         dout("  snap_id 0x%016llx order = %u, snap_size = %llu\n",
2179                 (unsigned long long) snap_id, (unsigned int) *order,
2180                 (unsigned long long) *snap_size);
2181
2182         return 0;
2183 }
2184
2185 static int rbd_dev_v2_image_size(struct rbd_device *rbd_dev)
2186 {
2187         return _rbd_dev_v2_snap_size(rbd_dev, CEPH_NOSNAP,
2188                                         &rbd_dev->header.obj_order,
2189                                         &rbd_dev->header.image_size);
2190 }
2191
2192 static int rbd_dev_v2_object_prefix(struct rbd_device *rbd_dev)
2193 {
2194         void *reply_buf;
2195         int ret;
2196         void *p;
2197
2198         reply_buf = kzalloc(RBD_OBJ_PREFIX_LEN_MAX, GFP_KERNEL);
2199         if (!reply_buf)
2200                 return -ENOMEM;
2201
2202         ret = rbd_req_sync_exec(rbd_dev, rbd_dev->header_name,
2203                                 "rbd", "get_object_prefix",
2204                                 NULL, 0,
2205                                 reply_buf, RBD_OBJ_PREFIX_LEN_MAX,
2206                                 CEPH_OSD_FLAG_READ, NULL);
2207         dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
2208         if (ret < 0)
2209                 goto out;
2210         ret = 0;    /* rbd_req_sync_exec() can return positive */
2211
2212         p = reply_buf;
2213         rbd_dev->header.object_prefix = ceph_extract_encoded_string(&p,
2214                                                 p + RBD_OBJ_PREFIX_LEN_MAX,
2215                                                 NULL, GFP_NOIO);
2216
2217         if (IS_ERR(rbd_dev->header.object_prefix)) {
2218                 ret = PTR_ERR(rbd_dev->header.object_prefix);
2219                 rbd_dev->header.object_prefix = NULL;
2220         } else {
2221                 dout("  object_prefix = %s\n", rbd_dev->header.object_prefix);
2222         }
2223
2224 out:
2225         kfree(reply_buf);
2226
2227         return ret;
2228 }
2229
2230 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
2231                 u64 *snap_features)
2232 {
2233         __le64 snapid = cpu_to_le64(snap_id);
2234         struct {
2235                 __le64 features;
2236                 __le64 incompat;
2237         } features_buf = { 0 };
2238         u64 incompat;
2239         int ret;
2240
2241         ret = rbd_req_sync_exec(rbd_dev, rbd_dev->header_name,
2242                                 "rbd", "get_features",
2243                                 (char *) &snapid, sizeof (snapid),
2244                                 (char *) &features_buf, sizeof (features_buf),
2245                                 CEPH_OSD_FLAG_READ, NULL);
2246         dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
2247         if (ret < 0)
2248                 return ret;
2249
2250         incompat = le64_to_cpu(features_buf.incompat);
2251         if (incompat & ~RBD_FEATURES_ALL)
2252                 return -ENOTSUPP;
2253
2254         *snap_features = le64_to_cpu(features_buf.features);
2255
2256         dout("  snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
2257                 (unsigned long long) snap_id,
2258                 (unsigned long long) *snap_features,
2259                 (unsigned long long) le64_to_cpu(features_buf.incompat));
2260
2261         return 0;
2262 }
2263
2264 static int rbd_dev_v2_features(struct rbd_device *rbd_dev)
2265 {
2266         return _rbd_dev_v2_snap_features(rbd_dev, CEPH_NOSNAP,
2267                                                 &rbd_dev->header.features);
2268 }
2269
2270 static int rbd_dev_v2_snap_context(struct rbd_device *rbd_dev, u64 *ver)
2271 {
2272         size_t size;
2273         int ret;
2274         void *reply_buf;
2275         void *p;
2276         void *end;
2277         u64 seq;
2278         u32 snap_count;
2279         struct ceph_snap_context *snapc;
2280         u32 i;
2281
2282         /*
2283          * We'll need room for the seq value (maximum snapshot id),
2284          * snapshot count, and array of that many snapshot ids.
2285          * For now we have a fixed upper limit on the number we're
2286          * prepared to receive.
2287          */
2288         size = sizeof (__le64) + sizeof (__le32) +
2289                         RBD_MAX_SNAP_COUNT * sizeof (__le64);
2290         reply_buf = kzalloc(size, GFP_KERNEL);
2291         if (!reply_buf)
2292                 return -ENOMEM;
2293
2294         ret = rbd_req_sync_exec(rbd_dev, rbd_dev->header_name,
2295                                 "rbd", "get_snapcontext",
2296                                 NULL, 0,
2297                                 reply_buf, size,
2298                                 CEPH_OSD_FLAG_READ, ver);
2299         dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
2300         if (ret < 0)
2301                 goto out;
2302
2303         ret = -ERANGE;
2304         p = reply_buf;
2305         end = (char *) reply_buf + size;
2306         ceph_decode_64_safe(&p, end, seq, out);
2307         ceph_decode_32_safe(&p, end, snap_count, out);
2308
2309         /*
2310          * Make sure the reported number of snapshot ids wouldn't go
2311          * beyond the end of our buffer.  But before checking that,
2312          * make sure the computed size of the snapshot context we
2313          * allocate is representable in a size_t.
2314          */
2315         if (snap_count > (SIZE_MAX - sizeof (struct ceph_snap_context))
2316                                  / sizeof (u64)) {
2317                 ret = -EINVAL;
2318                 goto out;
2319         }
2320         if (!ceph_has_room(&p, end, snap_count * sizeof (__le64)))
2321                 goto out;
2322
2323         size = sizeof (struct ceph_snap_context) +
2324                                 snap_count * sizeof (snapc->snaps[0]);
2325         snapc = kmalloc(size, GFP_KERNEL);
2326         if (!snapc) {
2327                 ret = -ENOMEM;
2328                 goto out;
2329         }
2330
2331         atomic_set(&snapc->nref, 1);
2332         snapc->seq = seq;
2333         snapc->num_snaps = snap_count;
2334         for (i = 0; i < snap_count; i++)
2335                 snapc->snaps[i] = ceph_decode_64(&p);
2336
2337         rbd_dev->header.snapc = snapc;
2338
2339         dout("  snap context seq = %llu, snap_count = %u\n",
2340                 (unsigned long long) seq, (unsigned int) snap_count);
2341
2342 out:
2343         kfree(reply_buf);
2344
2345         return 0;
2346 }
2347
2348 static char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev, u32 which)
2349 {
2350         size_t size;
2351         void *reply_buf;
2352         __le64 snap_id;
2353         int ret;
2354         void *p;
2355         void *end;
2356         size_t snap_name_len;
2357         char *snap_name;
2358
2359         size = sizeof (__le32) + RBD_MAX_SNAP_NAME_LEN;
2360         reply_buf = kmalloc(size, GFP_KERNEL);
2361         if (!reply_buf)
2362                 return ERR_PTR(-ENOMEM);
2363
2364         snap_id = cpu_to_le64(rbd_dev->header.snapc->snaps[which]);
2365         ret = rbd_req_sync_exec(rbd_dev, rbd_dev->header_name,
2366                                 "rbd", "get_snapshot_name",
2367                                 (char *) &snap_id, sizeof (snap_id),
2368                                 reply_buf, size,
2369                                 CEPH_OSD_FLAG_READ, NULL);
2370         dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
2371         if (ret < 0)
2372                 goto out;
2373
2374         p = reply_buf;
2375         end = (char *) reply_buf + size;
2376         snap_name_len = 0;
2377         snap_name = ceph_extract_encoded_string(&p, end, &snap_name_len,
2378                                 GFP_KERNEL);
2379         if (IS_ERR(snap_name)) {
2380                 ret = PTR_ERR(snap_name);
2381                 goto out;
2382         } else {
2383                 dout("  snap_id 0x%016llx snap_name = %s\n",
2384                         (unsigned long long) le64_to_cpu(snap_id), snap_name);
2385         }
2386         kfree(reply_buf);
2387
2388         return snap_name;
2389 out:
2390         kfree(reply_buf);
2391
2392         return ERR_PTR(ret);
2393 }
2394
2395 static char *rbd_dev_v2_snap_info(struct rbd_device *rbd_dev, u32 which,
2396                 u64 *snap_size, u64 *snap_features)
2397 {
2398         __le64 snap_id;
2399         u8 order;
2400         int ret;
2401
2402         snap_id = rbd_dev->header.snapc->snaps[which];
2403         ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, &order, snap_size);
2404         if (ret)
2405                 return ERR_PTR(ret);
2406         ret = _rbd_dev_v2_snap_features(rbd_dev, snap_id, snap_features);
2407         if (ret)
2408                 return ERR_PTR(ret);
2409
2410         return rbd_dev_v2_snap_name(rbd_dev, which);
2411 }
2412
2413 static char *rbd_dev_snap_info(struct rbd_device *rbd_dev, u32 which,
2414                 u64 *snap_size, u64 *snap_features)
2415 {
2416         if (rbd_dev->image_format == 1)
2417                 return rbd_dev_v1_snap_info(rbd_dev, which,
2418                                         snap_size, snap_features);
2419         if (rbd_dev->image_format == 2)
2420                 return rbd_dev_v2_snap_info(rbd_dev, which,
2421                                         snap_size, snap_features);
2422         return ERR_PTR(-EINVAL);
2423 }
2424
2425 static int rbd_dev_v2_refresh(struct rbd_device *rbd_dev, u64 *hver)
2426 {
2427         int ret;
2428         __u8 obj_order;
2429
2430         down_write(&rbd_dev->header_rwsem);
2431
2432         /* Grab old order first, to see if it changes */
2433
2434         obj_order = rbd_dev->header.obj_order,
2435         ret = rbd_dev_v2_image_size(rbd_dev);
2436         if (ret)
2437                 goto out;
2438         if (rbd_dev->header.obj_order != obj_order) {
2439                 ret = -EIO;
2440                 goto out;
2441         }
2442         rbd_update_mapping_size(rbd_dev);
2443
2444         ret = rbd_dev_v2_snap_context(rbd_dev, hver);
2445         dout("rbd_dev_v2_snap_context returned %d\n", ret);
2446         if (ret)
2447                 goto out;
2448         ret = rbd_dev_snaps_update(rbd_dev);
2449         dout("rbd_dev_snaps_update returned %d\n", ret);
2450         if (ret)
2451                 goto out;
2452         ret = rbd_dev_snaps_register(rbd_dev);
2453         dout("rbd_dev_snaps_register returned %d\n", ret);
2454 out:
2455         up_write(&rbd_dev->header_rwsem);
2456
2457         return ret;
2458 }
2459
2460 /*
2461  * Scan the rbd device's current snapshot list and compare it to the
2462  * newly-received snapshot context.  Remove any existing snapshots
2463  * not present in the new snapshot context.  Add a new snapshot for
2464  * any snaphots in the snapshot context not in the current list.
2465  * And verify there are no changes to snapshots we already know
2466  * about.
2467  *
2468  * Assumes the snapshots in the snapshot context are sorted by
2469  * snapshot id, highest id first.  (Snapshots in the rbd_dev's list
2470  * are also maintained in that order.)
2471  */
2472 static int rbd_dev_snaps_update(struct rbd_device *rbd_dev)
2473 {
2474         struct ceph_snap_context *snapc = rbd_dev->header.snapc;
2475         const u32 snap_count = snapc->num_snaps;
2476         struct list_head *head = &rbd_dev->snaps;
2477         struct list_head *links = head->next;
2478         u32 index = 0;
2479
2480         dout("%s: snap count is %u\n", __func__, (unsigned int) snap_count);
2481         while (index < snap_count || links != head) {
2482                 u64 snap_id;
2483                 struct rbd_snap *snap;
2484                 char *snap_name;
2485                 u64 snap_size = 0;
2486                 u64 snap_features = 0;
2487
2488                 snap_id = index < snap_count ? snapc->snaps[index]
2489                                              : CEPH_NOSNAP;
2490                 snap = links != head ? list_entry(links, struct rbd_snap, node)
2491                                      : NULL;
2492                 rbd_assert(!snap || snap->id != CEPH_NOSNAP);
2493
2494                 if (snap_id == CEPH_NOSNAP || (snap && snap->id > snap_id)) {
2495                         struct list_head *next = links->next;
2496
2497                         /* Existing snapshot not in the new snap context */
2498
2499                         if (rbd_dev->mapping.snap_id == snap->id)
2500                                 rbd_dev->mapping.snap_exists = false;
2501                         __rbd_remove_snap_dev(snap);
2502                         dout("%ssnap id %llu has been removed\n",
2503                                 rbd_dev->mapping.snap_id == snap->id ?
2504                                                                 "mapped " : "",
2505                                 (unsigned long long) snap->id);
2506
2507                         /* Done with this list entry; advance */
2508
2509                         links = next;
2510                         continue;
2511                 }
2512
2513                 snap_name = rbd_dev_snap_info(rbd_dev, index,
2514                                         &snap_size, &snap_features);
2515                 if (IS_ERR(snap_name))
2516                         return PTR_ERR(snap_name);
2517
2518                 dout("entry %u: snap_id = %llu\n", (unsigned int) snap_count,
2519                         (unsigned long long) snap_id);
2520                 if (!snap || (snap_id != CEPH_NOSNAP && snap->id < snap_id)) {
2521                         struct rbd_snap *new_snap;
2522
2523                         /* We haven't seen this snapshot before */
2524
2525                         new_snap = __rbd_add_snap_dev(rbd_dev, snap_name,
2526                                         snap_id, snap_size, snap_features);
2527                         if (IS_ERR(new_snap)) {
2528                                 int err = PTR_ERR(new_snap);
2529
2530                                 dout("  failed to add dev, error %d\n", err);
2531
2532                                 return err;
2533                         }
2534
2535                         /* New goes before existing, or at end of list */
2536
2537                         dout("  added dev%s\n", snap ? "" : " at end\n");
2538                         if (snap)
2539                                 list_add_tail(&new_snap->node, &snap->node);
2540                         else
2541                                 list_add_tail(&new_snap->node, head);
2542                 } else {
2543                         /* Already have this one */
2544
2545                         dout("  already present\n");
2546
2547                         rbd_assert(snap->size == snap_size);
2548                         rbd_assert(!strcmp(snap->name, snap_name));
2549                         rbd_assert(snap->features == snap_features);
2550
2551                         /* Done with this list entry; advance */
2552
2553                         links = links->next;
2554                 }
2555
2556                 /* Advance to the next entry in the snapshot context */
2557
2558                 index++;
2559         }
2560         dout("%s: done\n", __func__);
2561
2562         return 0;
2563 }
2564
2565 /*
2566  * Scan the list of snapshots and register the devices for any that
2567  * have not already been registered.
2568  */
2569 static int rbd_dev_snaps_register(struct rbd_device *rbd_dev)
2570 {
2571         struct rbd_snap *snap;
2572         int ret = 0;
2573
2574         dout("%s called\n", __func__);
2575         if (WARN_ON(!device_is_registered(&rbd_dev->dev)))
2576                 return -EIO;
2577
2578         list_for_each_entry(snap, &rbd_dev->snaps, node) {
2579                 if (!rbd_snap_registered(snap)) {
2580                         ret = rbd_register_snap_dev(snap, &rbd_dev->dev);
2581                         if (ret < 0)
2582                                 break;
2583                 }
2584         }
2585         dout("%s: returning %d\n", __func__, ret);
2586
2587         return ret;
2588 }
2589
2590 static int rbd_bus_add_dev(struct rbd_device *rbd_dev)
2591 {
2592         struct device *dev;
2593         int ret;
2594
2595         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2596
2597         dev = &rbd_dev->dev;
2598         dev->bus = &rbd_bus_type;
2599         dev->type = &rbd_device_type;
2600         dev->parent = &rbd_root_dev;
2601         dev->release = rbd_dev_release;
2602         dev_set_name(dev, "%d", rbd_dev->dev_id);
2603         ret = device_register(dev);
2604
2605         mutex_unlock(&ctl_mutex);
2606
2607         return ret;
2608 }
2609
2610 static void rbd_bus_del_dev(struct rbd_device *rbd_dev)
2611 {
2612         device_unregister(&rbd_dev->dev);
2613 }
2614
2615 static int rbd_init_watch_dev(struct rbd_device *rbd_dev)
2616 {
2617         int ret, rc;
2618
2619         do {
2620                 ret = rbd_req_sync_watch(rbd_dev);
2621                 if (ret == -ERANGE) {
2622                         rc = rbd_dev_refresh(rbd_dev, NULL);
2623                         if (rc < 0)
2624                                 return rc;
2625                 }
2626         } while (ret == -ERANGE);
2627
2628         return ret;
2629 }
2630
2631 static atomic64_t rbd_dev_id_max = ATOMIC64_INIT(0);
2632
2633 /*
2634  * Get a unique rbd identifier for the given new rbd_dev, and add
2635  * the rbd_dev to the global list.  The minimum rbd id is 1.
2636  */
2637 static void rbd_dev_id_get(struct rbd_device *rbd_dev)
2638 {
2639         rbd_dev->dev_id = atomic64_inc_return(&rbd_dev_id_max);
2640
2641         spin_lock(&rbd_dev_list_lock);
2642         list_add_tail(&rbd_dev->node, &rbd_dev_list);
2643         spin_unlock(&rbd_dev_list_lock);
2644         dout("rbd_dev %p given dev id %llu\n", rbd_dev,
2645                 (unsigned long long) rbd_dev->dev_id);
2646 }
2647
2648 /*
2649  * Remove an rbd_dev from the global list, and record that its
2650  * identifier is no longer in use.
2651  */
2652 static void rbd_dev_id_put(struct rbd_device *rbd_dev)
2653 {
2654         struct list_head *tmp;
2655         int rbd_id = rbd_dev->dev_id;
2656         int max_id;
2657
2658         rbd_assert(rbd_id > 0);
2659
2660         dout("rbd_dev %p released dev id %llu\n", rbd_dev,
2661                 (unsigned long long) rbd_dev->dev_id);
2662         spin_lock(&rbd_dev_list_lock);
2663         list_del_init(&rbd_dev->node);
2664
2665         /*
2666          * If the id being "put" is not the current maximum, there
2667          * is nothing special we need to do.
2668          */
2669         if (rbd_id != atomic64_read(&rbd_dev_id_max)) {
2670                 spin_unlock(&rbd_dev_list_lock);
2671                 return;
2672         }
2673
2674         /*
2675          * We need to update the current maximum id.  Search the
2676          * list to find out what it is.  We're more likely to find
2677          * the maximum at the end, so search the list backward.
2678          */
2679         max_id = 0;
2680         list_for_each_prev(tmp, &rbd_dev_list) {
2681                 struct rbd_device *rbd_dev;
2682
2683                 rbd_dev = list_entry(tmp, struct rbd_device, node);
2684                 if (rbd_dev->dev_id > max_id)
2685                         max_id = rbd_dev->dev_id;
2686         }
2687         spin_unlock(&rbd_dev_list_lock);
2688
2689         /*
2690          * The max id could have been updated by rbd_dev_id_get(), in
2691          * which case it now accurately reflects the new maximum.
2692          * Be careful not to overwrite the maximum value in that
2693          * case.
2694          */
2695         atomic64_cmpxchg(&rbd_dev_id_max, rbd_id, max_id);
2696         dout("  max dev id has been reset\n");
2697 }
2698
2699 /*
2700  * Skips over white space at *buf, and updates *buf to point to the
2701  * first found non-space character (if any). Returns the length of
2702  * the token (string of non-white space characters) found.  Note
2703  * that *buf must be terminated with '\0'.
2704  */
2705 static inline size_t next_token(const char **buf)
2706 {
2707         /*
2708         * These are the characters that produce nonzero for
2709         * isspace() in the "C" and "POSIX" locales.
2710         */
2711         const char *spaces = " \f\n\r\t\v";
2712
2713         *buf += strspn(*buf, spaces);   /* Find start of token */
2714
2715         return strcspn(*buf, spaces);   /* Return token length */
2716 }
2717
2718 /*
2719  * Finds the next token in *buf, and if the provided token buffer is
2720  * big enough, copies the found token into it.  The result, if
2721  * copied, is guaranteed to be terminated with '\0'.  Note that *buf
2722  * must be terminated with '\0' on entry.
2723  *
2724  * Returns the length of the token found (not including the '\0').
2725  * Return value will be 0 if no token is found, and it will be >=
2726  * token_size if the token would not fit.
2727  *
2728  * The *buf pointer will be updated to point beyond the end of the
2729  * found token.  Note that this occurs even if the token buffer is
2730  * too small to hold it.
2731  */
2732 static inline size_t copy_token(const char **buf,
2733                                 char *token,
2734                                 size_t token_size)
2735 {
2736         size_t len;
2737
2738         len = next_token(buf);
2739         if (len < token_size) {
2740                 memcpy(token, *buf, len);
2741                 *(token + len) = '\0';
2742         }
2743         *buf += len;
2744
2745         return len;
2746 }
2747
2748 /*
2749  * Finds the next token in *buf, dynamically allocates a buffer big
2750  * enough to hold a copy of it, and copies the token into the new
2751  * buffer.  The copy is guaranteed to be terminated with '\0'.  Note
2752  * that a duplicate buffer is created even for a zero-length token.
2753  *
2754  * Returns a pointer to the newly-allocated duplicate, or a null
2755  * pointer if memory for the duplicate was not available.  If
2756  * the lenp argument is a non-null pointer, the length of the token
2757  * (not including the '\0') is returned in *lenp.
2758  *
2759  * If successful, the *buf pointer will be updated to point beyond
2760  * the end of the found token.
2761  *
2762  * Note: uses GFP_KERNEL for allocation.
2763  */
2764 static inline char *dup_token(const char **buf, size_t *lenp)
2765 {
2766         char *dup;
2767         size_t len;
2768
2769         len = next_token(buf);
2770         dup = kmalloc(len + 1, GFP_KERNEL);
2771         if (!dup)
2772                 return NULL;
2773
2774         memcpy(dup, *buf, len);
2775         *(dup + len) = '\0';
2776         *buf += len;
2777
2778         if (lenp)
2779                 *lenp = len;
2780
2781         return dup;
2782 }
2783
2784 /*
2785  * This fills in the pool_name, image_name, image_name_len, rbd_dev,
2786  * rbd_md_name, and name fields of the given rbd_dev, based on the
2787  * list of monitor addresses and other options provided via
2788  * /sys/bus/rbd/add.  Returns a pointer to a dynamically-allocated
2789  * copy of the snapshot name to map if successful, or a
2790  * pointer-coded error otherwise.
2791  *
2792  * Note: rbd_dev is assumed to have been initially zero-filled.
2793  */
2794 static char *rbd_add_parse_args(struct rbd_device *rbd_dev,
2795                                 const char *buf,
2796                                 const char **mon_addrs,
2797                                 size_t *mon_addrs_size,
2798                                 char *options,
2799                                 size_t options_size)
2800 {
2801         size_t len;
2802         char *err_ptr = ERR_PTR(-EINVAL);
2803         char *snap_name;
2804
2805         /* The first four tokens are required */
2806
2807         len = next_token(&buf);
2808         if (!len)
2809                 return err_ptr;
2810         *mon_addrs_size = len + 1;
2811         *mon_addrs = buf;
2812
2813         buf += len;
2814
2815         len = copy_token(&buf, options, options_size);
2816         if (!len || len >= options_size)
2817                 return err_ptr;
2818
2819         err_ptr = ERR_PTR(-ENOMEM);
2820         rbd_dev->pool_name = dup_token(&buf, NULL);
2821         if (!rbd_dev->pool_name)
2822                 goto out_err;
2823
2824         rbd_dev->image_name = dup_token(&buf, &rbd_dev->image_name_len);
2825         if (!rbd_dev->image_name)
2826                 goto out_err;
2827
2828         /* Snapshot name is optional */
2829         len = next_token(&buf);
2830         if (!len) {
2831                 buf = RBD_SNAP_HEAD_NAME; /* No snapshot supplied */
2832                 len = sizeof (RBD_SNAP_HEAD_NAME) - 1;
2833         }
2834         snap_name = kmalloc(len + 1, GFP_KERNEL);
2835         if (!snap_name)
2836                 goto out_err;
2837         memcpy(snap_name, buf, len);
2838         *(snap_name + len) = '\0';
2839
2840 dout("    SNAP_NAME is <%s>, len is %zd\n", snap_name, len);
2841
2842         return snap_name;
2843
2844 out_err:
2845         kfree(rbd_dev->image_name);
2846         rbd_dev->image_name = NULL;
2847         rbd_dev->image_name_len = 0;
2848         kfree(rbd_dev->pool_name);
2849         rbd_dev->pool_name = NULL;
2850
2851         return err_ptr;
2852 }
2853
2854 /*
2855  * An rbd format 2 image has a unique identifier, distinct from the
2856  * name given to it by the user.  Internally, that identifier is
2857  * what's used to specify the names of objects related to the image.
2858  *
2859  * A special "rbd id" object is used to map an rbd image name to its
2860  * id.  If that object doesn't exist, then there is no v2 rbd image
2861  * with the supplied name.
2862  *
2863  * This function will record the given rbd_dev's image_id field if
2864  * it can be determined, and in that case will return 0.  If any
2865  * errors occur a negative errno will be returned and the rbd_dev's
2866  * image_id field will be unchanged (and should be NULL).
2867  */
2868 static int rbd_dev_image_id(struct rbd_device *rbd_dev)
2869 {
2870         int ret;
2871         size_t size;
2872         char *object_name;
2873         void *response;
2874         void *p;
2875
2876         /*
2877          * First, see if the format 2 image id file exists, and if
2878          * so, get the image's persistent id from it.
2879          */
2880         size = sizeof (RBD_ID_PREFIX) + rbd_dev->image_name_len;
2881         object_name = kmalloc(size, GFP_NOIO);
2882         if (!object_name)
2883                 return -ENOMEM;
2884         sprintf(object_name, "%s%s", RBD_ID_PREFIX, rbd_dev->image_name);
2885         dout("rbd id object name is %s\n", object_name);
2886
2887         /* Response will be an encoded string, which includes a length */
2888
2889         size = sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX;
2890         response = kzalloc(size, GFP_NOIO);
2891         if (!response) {
2892                 ret = -ENOMEM;
2893                 goto out;
2894         }
2895
2896         ret = rbd_req_sync_exec(rbd_dev, object_name,
2897                                 "rbd", "get_id",
2898                                 NULL, 0,
2899                                 response, RBD_IMAGE_ID_LEN_MAX,
2900                                 CEPH_OSD_FLAG_READ, NULL);
2901         dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
2902         if (ret < 0)
2903                 goto out;
2904         ret = 0;    /* rbd_req_sync_exec() can return positive */
2905
2906         p = response;
2907         rbd_dev->image_id = ceph_extract_encoded_string(&p,
2908                                                 p + RBD_IMAGE_ID_LEN_MAX,
2909                                                 &rbd_dev->image_id_len,
2910                                                 GFP_NOIO);
2911         if (IS_ERR(rbd_dev->image_id)) {
2912                 ret = PTR_ERR(rbd_dev->image_id);
2913                 rbd_dev->image_id = NULL;
2914         } else {
2915                 dout("image_id is %s\n", rbd_dev->image_id);
2916         }
2917 out:
2918         kfree(response);
2919         kfree(object_name);
2920
2921         return ret;
2922 }
2923
2924 static int rbd_dev_v1_probe(struct rbd_device *rbd_dev)
2925 {
2926         int ret;
2927         size_t size;
2928
2929         /* Version 1 images have no id; empty string is used */
2930
2931         rbd_dev->image_id = kstrdup("", GFP_KERNEL);
2932         if (!rbd_dev->image_id)
2933                 return -ENOMEM;
2934         rbd_dev->image_id_len = 0;
2935
2936         /* Record the header object name for this rbd image. */
2937
2938         size = rbd_dev->image_name_len + sizeof (RBD_SUFFIX);
2939         rbd_dev->header_name = kmalloc(size, GFP_KERNEL);
2940         if (!rbd_dev->header_name) {
2941                 ret = -ENOMEM;
2942                 goto out_err;
2943         }
2944         sprintf(rbd_dev->header_name, "%s%s", rbd_dev->image_name, RBD_SUFFIX);
2945
2946         /* Populate rbd image metadata */
2947
2948         ret = rbd_read_header(rbd_dev, &rbd_dev->header);
2949         if (ret < 0)
2950                 goto out_err;
2951         rbd_dev->image_format = 1;
2952
2953         dout("discovered version 1 image, header name is %s\n",
2954                 rbd_dev->header_name);
2955
2956         return 0;
2957
2958 out_err:
2959         kfree(rbd_dev->header_name);
2960         rbd_dev->header_name = NULL;
2961         kfree(rbd_dev->image_id);
2962         rbd_dev->image_id = NULL;
2963
2964         return ret;
2965 }
2966
2967 static int rbd_dev_v2_probe(struct rbd_device *rbd_dev)
2968 {
2969         size_t size;
2970         int ret;
2971         u64 ver = 0;
2972
2973         /*
2974          * Image id was filled in by the caller.  Record the header
2975          * object name for this rbd image.
2976          */
2977         size = sizeof (RBD_HEADER_PREFIX) + rbd_dev->image_id_len;
2978         rbd_dev->header_name = kmalloc(size, GFP_KERNEL);
2979         if (!rbd_dev->header_name)
2980                 return -ENOMEM;
2981         sprintf(rbd_dev->header_name, "%s%s",
2982                         RBD_HEADER_PREFIX, rbd_dev->image_id);
2983
2984         /* Get the size and object order for the image */
2985
2986         ret = rbd_dev_v2_image_size(rbd_dev);
2987         if (ret < 0)
2988                 goto out_err;
2989
2990         /* Get the object prefix (a.k.a. block_name) for the image */
2991
2992         ret = rbd_dev_v2_object_prefix(rbd_dev);
2993         if (ret < 0)
2994                 goto out_err;
2995
2996         /* Get the and check features for the image */
2997
2998         ret = rbd_dev_v2_features(rbd_dev);
2999         if (ret < 0)
3000                 goto out_err;
3001
3002         /* crypto and compression type aren't (yet) supported for v2 images */
3003
3004         rbd_dev->header.crypt_type = 0;
3005         rbd_dev->header.comp_type = 0;
3006
3007         /* Get the snapshot context, plus the header version */
3008
3009         ret = rbd_dev_v2_snap_context(rbd_dev, &ver);
3010         if (ret)
3011                 goto out_err;
3012         rbd_dev->header.obj_version = ver;
3013
3014         rbd_dev->image_format = 2;
3015
3016         dout("discovered version 2 image, header name is %s\n",
3017                 rbd_dev->header_name);
3018
3019         return 0;
3020 out_err:
3021         kfree(rbd_dev->header_name);
3022         rbd_dev->header_name = NULL;
3023         kfree(rbd_dev->header.object_prefix);
3024         rbd_dev->header.object_prefix = NULL;
3025
3026         return ret;
3027 }
3028
3029 /*
3030  * Probe for the existence of the header object for the given rbd
3031  * device.  For format 2 images this includes determining the image
3032  * id.
3033  */
3034 static int rbd_dev_probe(struct rbd_device *rbd_dev)
3035 {
3036         int ret;
3037
3038         /*
3039          * Get the id from the image id object.  If it's not a
3040          * format 2 image, we'll get ENOENT back, and we'll assume
3041          * it's a format 1 image.
3042          */
3043         ret = rbd_dev_image_id(rbd_dev);
3044         if (ret)
3045                 ret = rbd_dev_v1_probe(rbd_dev);
3046         else
3047                 ret = rbd_dev_v2_probe(rbd_dev);
3048         if (ret)
3049                 dout("probe failed, returning %d\n", ret);
3050
3051         return ret;
3052 }
3053
3054 static ssize_t rbd_add(struct bus_type *bus,
3055                        const char *buf,
3056                        size_t count)
3057 {
3058         char *options;
3059         struct rbd_device *rbd_dev = NULL;
3060         const char *mon_addrs = NULL;
3061         size_t mon_addrs_size = 0;
3062         struct ceph_osd_client *osdc;
3063         int rc = -ENOMEM;
3064         char *snap_name;
3065
3066         if (!try_module_get(THIS_MODULE))
3067                 return -ENODEV;
3068
3069         options = kmalloc(count, GFP_KERNEL);
3070         if (!options)
3071                 goto err_out_mem;
3072         rbd_dev = kzalloc(sizeof(*rbd_dev), GFP_KERNEL);
3073         if (!rbd_dev)
3074                 goto err_out_mem;
3075
3076         /* static rbd_device initialization */
3077         spin_lock_init(&rbd_dev->lock);
3078         INIT_LIST_HEAD(&rbd_dev->node);
3079         INIT_LIST_HEAD(&rbd_dev->snaps);
3080         init_rwsem(&rbd_dev->header_rwsem);
3081
3082         /* parse add command */
3083         snap_name = rbd_add_parse_args(rbd_dev, buf,
3084                                 &mon_addrs, &mon_addrs_size, options, count);
3085         if (IS_ERR(snap_name)) {
3086                 rc = PTR_ERR(snap_name);
3087                 goto err_out_mem;
3088         }
3089
3090         rc = rbd_get_client(rbd_dev, mon_addrs, mon_addrs_size - 1, options);
3091         if (rc < 0)
3092                 goto err_out_args;
3093
3094         /* pick the pool */
3095         osdc = &rbd_dev->rbd_client->client->osdc;
3096         rc = ceph_pg_poolid_by_name(osdc->osdmap, rbd_dev->pool_name);
3097         if (rc < 0)
3098                 goto err_out_client;
3099         rbd_dev->pool_id = rc;
3100
3101         rc = rbd_dev_probe(rbd_dev);
3102         if (rc < 0)
3103                 goto err_out_client;
3104
3105         /* no need to lock here, as rbd_dev is not registered yet */
3106         rc = rbd_dev_snaps_update(rbd_dev);
3107         if (rc)
3108                 goto err_out_header;
3109
3110         rc = rbd_dev_set_mapping(rbd_dev, snap_name);
3111         if (rc)
3112                 goto err_out_header;
3113
3114         /* generate unique id: find highest unique id, add one */
3115         rbd_dev_id_get(rbd_dev);
3116
3117         /* Fill in the device name, now that we have its id. */
3118         BUILD_BUG_ON(DEV_NAME_LEN
3119                         < sizeof (RBD_DRV_NAME) + MAX_INT_FORMAT_WIDTH);
3120         sprintf(rbd_dev->name, "%s%d", RBD_DRV_NAME, rbd_dev->dev_id);
3121
3122         /* Get our block major device number. */
3123
3124         rc = register_blkdev(0, rbd_dev->name);
3125         if (rc < 0)
3126                 goto err_out_id;
3127         rbd_dev->major = rc;
3128
3129         /* Set up the blkdev mapping. */
3130
3131         rc = rbd_init_disk(rbd_dev);
3132         if (rc)
3133                 goto err_out_blkdev;
3134
3135         rc = rbd_bus_add_dev(rbd_dev);
3136         if (rc)
3137                 goto err_out_disk;
3138
3139         /*
3140          * At this point cleanup in the event of an error is the job
3141          * of the sysfs code (initiated by rbd_bus_del_dev()).
3142          */
3143
3144         down_write(&rbd_dev->header_rwsem);
3145         rc = rbd_dev_snaps_register(rbd_dev);
3146         up_write(&rbd_dev->header_rwsem);
3147         if (rc)
3148                 goto err_out_bus;
3149
3150         rc = rbd_init_watch_dev(rbd_dev);
3151         if (rc)
3152                 goto err_out_bus;
3153
3154         /* Everything's ready.  Announce the disk to the world. */
3155
3156         add_disk(rbd_dev->disk);
3157
3158         pr_info("%s: added with size 0x%llx\n", rbd_dev->disk->disk_name,
3159                 (unsigned long long) rbd_dev->mapping.size);
3160
3161         return count;
3162
3163 err_out_bus:
3164         /* this will also clean up rest of rbd_dev stuff */
3165
3166         rbd_bus_del_dev(rbd_dev);
3167         kfree(options);
3168         return rc;
3169
3170 err_out_disk:
3171         rbd_free_disk(rbd_dev);
3172 err_out_blkdev:
3173         unregister_blkdev(rbd_dev->major, rbd_dev->name);
3174 err_out_id:
3175         rbd_dev_id_put(rbd_dev);
3176 err_out_header:
3177         rbd_header_free(&rbd_dev->header);
3178 err_out_client:
3179         kfree(rbd_dev->header_name);
3180         rbd_put_client(rbd_dev);
3181         kfree(rbd_dev->image_id);
3182 err_out_args:
3183         kfree(rbd_dev->mapping.snap_name);
3184         kfree(rbd_dev->image_name);
3185         kfree(rbd_dev->pool_name);
3186 err_out_mem:
3187         kfree(rbd_dev);
3188         kfree(options);
3189
3190         dout("Error adding device %s\n", buf);
3191         module_put(THIS_MODULE);
3192
3193         return (ssize_t) rc;
3194 }
3195
3196 static struct rbd_device *__rbd_get_dev(unsigned long dev_id)
3197 {
3198         struct list_head *tmp;
3199         struct rbd_device *rbd_dev;
3200
3201         spin_lock(&rbd_dev_list_lock);
3202         list_for_each(tmp, &rbd_dev_list) {
3203                 rbd_dev = list_entry(tmp, struct rbd_device, node);
3204                 if (rbd_dev->dev_id == dev_id) {
3205                         spin_unlock(&rbd_dev_list_lock);
3206                         return rbd_dev;
3207                 }
3208         }
3209         spin_unlock(&rbd_dev_list_lock);
3210         return NULL;
3211 }
3212
3213 static void rbd_dev_release(struct device *dev)
3214 {
3215         struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
3216
3217         if (rbd_dev->watch_request) {
3218                 struct ceph_client *client = rbd_dev->rbd_client->client;
3219
3220                 ceph_osdc_unregister_linger_request(&client->osdc,
3221                                                     rbd_dev->watch_request);
3222         }
3223         if (rbd_dev->watch_event)
3224                 rbd_req_sync_unwatch(rbd_dev);
3225
3226         rbd_put_client(rbd_dev);
3227
3228         /* clean up and free blkdev */
3229         rbd_free_disk(rbd_dev);
3230         unregister_blkdev(rbd_dev->major, rbd_dev->name);
3231
3232         /* release allocated disk header fields */
3233         rbd_header_free(&rbd_dev->header);
3234
3235         /* done with the id, and with the rbd_dev */
3236         kfree(rbd_dev->mapping.snap_name);
3237         kfree(rbd_dev->image_id);
3238         kfree(rbd_dev->header_name);
3239         kfree(rbd_dev->pool_name);
3240         kfree(rbd_dev->image_name);
3241         rbd_dev_id_put(rbd_dev);
3242         kfree(rbd_dev);
3243
3244         /* release module ref */
3245         module_put(THIS_MODULE);
3246 }
3247
3248 static ssize_t rbd_remove(struct bus_type *bus,
3249                           const char *buf,
3250                           size_t count)
3251 {
3252         struct rbd_device *rbd_dev = NULL;
3253         int target_id, rc;
3254         unsigned long ul;
3255         int ret = count;
3256
3257         rc = strict_strtoul(buf, 10, &ul);
3258         if (rc)
3259                 return rc;
3260
3261         /* convert to int; abort if we lost anything in the conversion */
3262         target_id = (int) ul;
3263         if (target_id != ul)
3264                 return -EINVAL;
3265
3266         mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
3267
3268         rbd_dev = __rbd_get_dev(target_id);
3269         if (!rbd_dev) {
3270                 ret = -ENOENT;
3271                 goto done;
3272         }
3273
3274         __rbd_remove_all_snaps(rbd_dev);
3275         rbd_bus_del_dev(rbd_dev);
3276
3277 done:
3278         mutex_unlock(&ctl_mutex);
3279
3280         return ret;
3281 }
3282
3283 /*
3284  * create control files in sysfs
3285  * /sys/bus/rbd/...
3286  */
3287 static int rbd_sysfs_init(void)
3288 {
3289         int ret;
3290
3291         ret = device_register(&rbd_root_dev);
3292         if (ret < 0)
3293                 return ret;
3294
3295         ret = bus_register(&rbd_bus_type);
3296         if (ret < 0)
3297                 device_unregister(&rbd_root_dev);
3298
3299         return ret;
3300 }
3301
3302 static void rbd_sysfs_cleanup(void)
3303 {
3304         bus_unregister(&rbd_bus_type);
3305         device_unregister(&rbd_root_dev);
3306 }
3307
3308 int __init rbd_init(void)
3309 {
3310         int rc;
3311
3312         rc = rbd_sysfs_init();
3313         if (rc)
3314                 return rc;
3315         pr_info("loaded " RBD_DRV_NAME_LONG "\n");
3316         return 0;
3317 }
3318
3319 void __exit rbd_exit(void)
3320 {
3321         rbd_sysfs_cleanup();
3322 }
3323
3324 module_init(rbd_init);
3325 module_exit(rbd_exit);
3326
3327 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
3328 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
3329 MODULE_DESCRIPTION("rados block device");
3330
3331 /* following authorship retained from original osdblk.c */
3332 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
3333
3334 MODULE_LICENSE("GPL");