1 2 /* 3 rbd.c -- Export ceph rados objects as a Linux block device 4 5 6 based on drivers/block/osdblk.c: 7 8 Copyright 2009 Red Hat, Inc. 9 10 This program is free software; you can redistribute it and/or modify 11 it under the terms of the GNU General Public License as published by 12 the Free Software Foundation. 13 14 This program is distributed in the hope that it will be useful, 15 but WITHOUT ANY WARRANTY; without even the implied warranty of 16 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 17 GNU General Public License for more details. 18 19 You should have received a copy of the GNU General Public License 20 along with this program; see the file COPYING. If not, write to 21 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA. 22 23 24 25 For usage instructions, please refer to: 26 27 Documentation/ABI/testing/sysfs-bus-rbd 28 29 */ 30 31 #include <linux/ceph/libceph.h> 32 #include <linux/ceph/osd_client.h> 33 #include <linux/ceph/mon_client.h> 34 #include <linux/ceph/decode.h> 35 #include <linux/parser.h> 36 #include <linux/bsearch.h> 37 38 #include <linux/kernel.h> 39 #include <linux/device.h> 40 #include <linux/module.h> 41 #include <linux/fs.h> 42 #include <linux/blkdev.h> 43 #include <linux/slab.h> 44 #include <linux/idr.h> 45 #include <linux/workqueue.h> 46 47 #include "rbd_types.h" 48 49 #define RBD_DEBUG /* Activate rbd_assert() calls */ 50 51 /* 52 * The basic unit of block I/O is a sector. It is interpreted in a 53 * number of contexts in Linux (blk, bio, genhd), but the default is 54 * universally 512 bytes. These symbols are just slightly more 55 * meaningful than the bare numbers they represent. 56 */ 57 #define SECTOR_SHIFT 9 58 #define SECTOR_SIZE (1ULL << SECTOR_SHIFT) 59 60 /* 61 * Increment the given counter and return its updated value. 62 * If the counter is already 0 it will not be incremented. 63 * If the counter is already at its maximum value returns 64 * -EINVAL without updating it. 65 */ 66 static int atomic_inc_return_safe(atomic_t *v) 67 { 68 unsigned int counter; 69 70 counter = (unsigned int)__atomic_add_unless(v, 1, 0); 71 if (counter <= (unsigned int)INT_MAX) 72 return (int)counter; 73 74 atomic_dec(v); 75 76 return -EINVAL; 77 } 78 79 /* Decrement the counter. Return the resulting value, or -EINVAL */ 80 static int atomic_dec_return_safe(atomic_t *v) 81 { 82 int counter; 83 84 counter = atomic_dec_return(v); 85 if (counter >= 0) 86 return counter; 87 88 atomic_inc(v); 89 90 return -EINVAL; 91 } 92 93 #define RBD_DRV_NAME "rbd" 94 95 #define RBD_MINORS_PER_MAJOR 256 96 #define RBD_SINGLE_MAJOR_PART_SHIFT 4 97 98 #define RBD_SNAP_DEV_NAME_PREFIX "snap_" 99 #define RBD_MAX_SNAP_NAME_LEN \ 100 (NAME_MAX - (sizeof (RBD_SNAP_DEV_NAME_PREFIX) - 1)) 101 102 #define RBD_MAX_SNAP_COUNT 510 /* allows max snapc to fit in 4KB */ 103 104 #define RBD_SNAP_HEAD_NAME "-" 105 106 #define BAD_SNAP_INDEX U32_MAX /* invalid index into snap array */ 107 108 /* This allows a single page to hold an image name sent by OSD */ 109 #define RBD_IMAGE_NAME_LEN_MAX (PAGE_SIZE - sizeof (__le32) - 1) 110 #define RBD_IMAGE_ID_LEN_MAX 64 111 112 #define RBD_OBJ_PREFIX_LEN_MAX 64 113 114 /* Feature bits */ 115 116 #define RBD_FEATURE_LAYERING (1<<0) 117 #define RBD_FEATURE_STRIPINGV2 (1<<1) 118 #define RBD_FEATURES_ALL \ 119 (RBD_FEATURE_LAYERING | RBD_FEATURE_STRIPINGV2) 120 121 /* Features supported by this (client software) implementation. */ 122 123 #define RBD_FEATURES_SUPPORTED (RBD_FEATURES_ALL) 124 125 /* 126 * An RBD device name will be "rbd#", where the "rbd" comes from 127 * RBD_DRV_NAME above, and # is a unique integer identifier. 128 * MAX_INT_FORMAT_WIDTH is used in ensuring DEV_NAME_LEN is big 129 * enough to hold all possible device names. 130 */ 131 #define DEV_NAME_LEN 32 132 #define MAX_INT_FORMAT_WIDTH ((5 * sizeof (int)) / 2 + 1) 133 134 /* 135 * block device image metadata (in-memory version) 136 */ 137 struct rbd_image_header { 138 /* These six fields never change for a given rbd image */ 139 char *object_prefix; 140 __u8 obj_order; 141 __u8 crypt_type; 142 __u8 comp_type; 143 u64 stripe_unit; 144 u64 stripe_count; 145 u64 features; /* Might be changeable someday? */ 146 147 /* The remaining fields need to be updated occasionally */ 148 u64 image_size; 149 struct ceph_snap_context *snapc; 150 char *snap_names; /* format 1 only */ 151 u64 *snap_sizes; /* format 1 only */ 152 }; 153 154 /* 155 * An rbd image specification. 156 * 157 * The tuple (pool_id, image_id, snap_id) is sufficient to uniquely 158 * identify an image. Each rbd_dev structure includes a pointer to 159 * an rbd_spec structure that encapsulates this identity. 160 * 161 * Each of the id's in an rbd_spec has an associated name. For a 162 * user-mapped image, the names are supplied and the id's associated 163 * with them are looked up. For a layered image, a parent image is 164 * defined by the tuple, and the names are looked up. 165 * 166 * An rbd_dev structure contains a parent_spec pointer which is 167 * non-null if the image it represents is a child in a layered 168 * image. This pointer will refer to the rbd_spec structure used 169 * by the parent rbd_dev for its own identity (i.e., the structure 170 * is shared between the parent and child). 171 * 172 * Since these structures are populated once, during the discovery 173 * phase of image construction, they are effectively immutable so 174 * we make no effort to synchronize access to them. 175 * 176 * Note that code herein does not assume the image name is known (it 177 * could be a null pointer). 178 */ 179 struct rbd_spec { 180 u64 pool_id; 181 const char *pool_name; 182 183 const char *image_id; 184 const char *image_name; 185 186 u64 snap_id; 187 const char *snap_name; 188 189 struct kref kref; 190 }; 191 192 /* 193 * an instance of the client. multiple devices may share an rbd client. 194 */ 195 struct rbd_client { 196 struct ceph_client *client; 197 struct kref kref; 198 struct list_head node; 199 }; 200 201 struct rbd_img_request; 202 typedef void (*rbd_img_callback_t)(struct rbd_img_request *); 203 204 #define BAD_WHICH U32_MAX /* Good which or bad which, which? */ 205 206 struct rbd_obj_request; 207 typedef void (*rbd_obj_callback_t)(struct rbd_obj_request *); 208 209 enum obj_request_type { 210 OBJ_REQUEST_NODATA, OBJ_REQUEST_BIO, OBJ_REQUEST_PAGES 211 }; 212 213 enum obj_operation_type { 214 OBJ_OP_WRITE, 215 OBJ_OP_READ, 216 OBJ_OP_DISCARD, 217 }; 218 219 enum obj_req_flags { 220 OBJ_REQ_DONE, /* completion flag: not done = 0, done = 1 */ 221 OBJ_REQ_IMG_DATA, /* object usage: standalone = 0, image = 1 */ 222 OBJ_REQ_KNOWN, /* EXISTS flag valid: no = 0, yes = 1 */ 223 OBJ_REQ_EXISTS, /* target exists: no = 0, yes = 1 */ 224 }; 225 226 struct rbd_obj_request { 227 const char *object_name; 228 u64 offset; /* object start byte */ 229 u64 length; /* bytes from offset */ 230 unsigned long flags; 231 232 /* 233 * An object request associated with an image will have its 234 * img_data flag set; a standalone object request will not. 235 * 236 * A standalone object request will have which == BAD_WHICH 237 * and a null obj_request pointer. 238 * 239 * An object request initiated in support of a layered image 240 * object (to check for its existence before a write) will 241 * have which == BAD_WHICH and a non-null obj_request pointer. 242 * 243 * Finally, an object request for rbd image data will have 244 * which != BAD_WHICH, and will have a non-null img_request 245 * pointer. The value of which will be in the range 246 * 0..(img_request->obj_request_count-1). 247 */ 248 union { 249 struct rbd_obj_request *obj_request; /* STAT op */ 250 struct { 251 struct rbd_img_request *img_request; 252 u64 img_offset; 253 /* links for img_request->obj_requests list */ 254 struct list_head links; 255 }; 256 }; 257 u32 which; /* posn image request list */ 258 259 enum obj_request_type type; 260 union { 261 struct bio *bio_list; 262 struct { 263 struct page **pages; 264 u32 page_count; 265 }; 266 }; 267 struct page **copyup_pages; 268 u32 copyup_page_count; 269 270 struct ceph_osd_request *osd_req; 271 272 u64 xferred; /* bytes transferred */ 273 int result; 274 275 rbd_obj_callback_t callback; 276 struct completion completion; 277 278 struct kref kref; 279 }; 280 281 enum img_req_flags { 282 IMG_REQ_WRITE, /* I/O direction: read = 0, write = 1 */ 283 IMG_REQ_CHILD, /* initiator: block = 0, child image = 1 */ 284 IMG_REQ_LAYERED, /* ENOENT handling: normal = 0, layered = 1 */ 285 IMG_REQ_DISCARD, /* discard: normal = 0, discard request = 1 */ 286 }; 287 288 struct rbd_img_request { 289 struct rbd_device *rbd_dev; 290 u64 offset; /* starting image byte offset */ 291 u64 length; /* byte count from offset */ 292 unsigned long flags; 293 union { 294 u64 snap_id; /* for reads */ 295 struct ceph_snap_context *snapc; /* for writes */ 296 }; 297 union { 298 struct request *rq; /* block request */ 299 struct rbd_obj_request *obj_request; /* obj req initiator */ 300 }; 301 struct page **copyup_pages; 302 u32 copyup_page_count; 303 spinlock_t completion_lock;/* protects next_completion */ 304 u32 next_completion; 305 rbd_img_callback_t callback; 306 u64 xferred;/* aggregate bytes transferred */ 307 int result; /* first nonzero obj_request result */ 308 309 u32 obj_request_count; 310 struct list_head obj_requests; /* rbd_obj_request structs */ 311 312 struct kref kref; 313 }; 314 315 #define for_each_obj_request(ireq, oreq) \ 316 list_for_each_entry(oreq, &(ireq)->obj_requests, links) 317 #define for_each_obj_request_from(ireq, oreq) \ 318 list_for_each_entry_from(oreq, &(ireq)->obj_requests, links) 319 #define for_each_obj_request_safe(ireq, oreq, n) \ 320 list_for_each_entry_safe_reverse(oreq, n, &(ireq)->obj_requests, links) 321 322 struct rbd_mapping { 323 u64 size; 324 u64 features; 325 bool read_only; 326 }; 327 328 /* 329 * a single device 330 */ 331 struct rbd_device { 332 int dev_id; /* blkdev unique id */ 333 334 int major; /* blkdev assigned major */ 335 int minor; 336 struct gendisk *disk; /* blkdev's gendisk and rq */ 337 338 u32 image_format; /* Either 1 or 2 */ 339 struct rbd_client *rbd_client; 340 341 char name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */ 342 343 struct list_head rq_queue; /* incoming rq queue */ 344 spinlock_t lock; /* queue, flags, open_count */ 345 struct workqueue_struct *rq_wq; 346 struct work_struct rq_work; 347 348 struct rbd_image_header header; 349 unsigned long flags; /* possibly lock protected */ 350 struct rbd_spec *spec; 351 352 char *header_name; 353 354 struct ceph_file_layout layout; 355 356 struct ceph_osd_event *watch_event; 357 struct rbd_obj_request *watch_request; 358 359 struct rbd_spec *parent_spec; 360 u64 parent_overlap; 361 atomic_t parent_ref; 362 struct rbd_device *parent; 363 364 /* protects updating the header */ 365 struct rw_semaphore header_rwsem; 366 367 struct rbd_mapping mapping; 368 369 struct list_head node; 370 371 /* sysfs related */ 372 struct device dev; 373 unsigned long open_count; /* protected by lock */ 374 }; 375 376 /* 377 * Flag bits for rbd_dev->flags. If atomicity is required, 378 * rbd_dev->lock is used to protect access. 379 * 380 * Currently, only the "removing" flag (which is coupled with the 381 * "open_count" field) requires atomic access. 382 */ 383 enum rbd_dev_flags { 384 RBD_DEV_FLAG_EXISTS, /* mapped snapshot has not been deleted */ 385 RBD_DEV_FLAG_REMOVING, /* this mapping is being removed */ 386 }; 387 388 static DEFINE_MUTEX(client_mutex); /* Serialize client creation */ 389 390 static LIST_HEAD(rbd_dev_list); /* devices */ 391 static DEFINE_SPINLOCK(rbd_dev_list_lock); 392 393 static LIST_HEAD(rbd_client_list); /* clients */ 394 static DEFINE_SPINLOCK(rbd_client_list_lock); 395 396 /* Slab caches for frequently-allocated structures */ 397 398 static struct kmem_cache *rbd_img_request_cache; 399 static struct kmem_cache *rbd_obj_request_cache; 400 static struct kmem_cache *rbd_segment_name_cache; 401 402 static int rbd_major; 403 static DEFINE_IDA(rbd_dev_id_ida); 404 405 /* 406 * Default to false for now, as single-major requires >= 0.75 version of 407 * userspace rbd utility. 408 */ 409 static bool single_major = false; 410 module_param(single_major, bool, S_IRUGO); 411 MODULE_PARM_DESC(single_major, "Use a single major number for all rbd devices (default: false)"); 412 413 static int rbd_img_request_submit(struct rbd_img_request *img_request); 414 415 static void rbd_dev_device_release(struct device *dev); 416 417 static ssize_t rbd_add(struct bus_type *bus, const char *buf, 418 size_t count); 419 static ssize_t rbd_remove(struct bus_type *bus, const char *buf, 420 size_t count); 421 static ssize_t rbd_add_single_major(struct bus_type *bus, const char *buf, 422 size_t count); 423 static ssize_t rbd_remove_single_major(struct bus_type *bus, const char *buf, 424 size_t count); 425 static int rbd_dev_image_probe(struct rbd_device *rbd_dev, bool mapping); 426 static void rbd_spec_put(struct rbd_spec *spec); 427 428 static int rbd_dev_id_to_minor(int dev_id) 429 { 430 return dev_id << RBD_SINGLE_MAJOR_PART_SHIFT; 431 } 432 433 static int minor_to_rbd_dev_id(int minor) 434 { 435 return minor >> RBD_SINGLE_MAJOR_PART_SHIFT; 436 } 437 438 static BUS_ATTR(add, S_IWUSR, NULL, rbd_add); 439 static BUS_ATTR(remove, S_IWUSR, NULL, rbd_remove); 440 static BUS_ATTR(add_single_major, S_IWUSR, NULL, rbd_add_single_major); 441 static BUS_ATTR(remove_single_major, S_IWUSR, NULL, rbd_remove_single_major); 442 443 static struct attribute *rbd_bus_attrs[] = { 444 &bus_attr_add.attr, 445 &bus_attr_remove.attr, 446 &bus_attr_add_single_major.attr, 447 &bus_attr_remove_single_major.attr, 448 NULL, 449 }; 450 451 static umode_t rbd_bus_is_visible(struct kobject *kobj, 452 struct attribute *attr, int index) 453 { 454 if (!single_major && 455 (attr == &bus_attr_add_single_major.attr || 456 attr == &bus_attr_remove_single_major.attr)) 457 return 0; 458 459 return attr->mode; 460 } 461 462 static const struct attribute_group rbd_bus_group = { 463 .attrs = rbd_bus_attrs, 464 .is_visible = rbd_bus_is_visible, 465 }; 466 __ATTRIBUTE_GROUPS(rbd_bus); 467 468 static struct bus_type rbd_bus_type = { 469 .name = "rbd", 470 .bus_groups = rbd_bus_groups, 471 }; 472 473 static void rbd_root_dev_release(struct device *dev) 474 { 475 } 476 477 static struct device rbd_root_dev = { 478 .init_name = "rbd", 479 .release = rbd_root_dev_release, 480 }; 481 482 static __printf(2, 3) 483 void rbd_warn(struct rbd_device *rbd_dev, const char *fmt, ...) 484 { 485 struct va_format vaf; 486 va_list args; 487 488 va_start(args, fmt); 489 vaf.fmt = fmt; 490 vaf.va = &args; 491 492 if (!rbd_dev) 493 printk(KERN_WARNING "%s: %pV\n", RBD_DRV_NAME, &vaf); 494 else if (rbd_dev->disk) 495 printk(KERN_WARNING "%s: %s: %pV\n", 496 RBD_DRV_NAME, rbd_dev->disk->disk_name, &vaf); 497 else if (rbd_dev->spec && rbd_dev->spec->image_name) 498 printk(KERN_WARNING "%s: image %s: %pV\n", 499 RBD_DRV_NAME, rbd_dev->spec->image_name, &vaf); 500 else if (rbd_dev->spec && rbd_dev->spec->image_id) 501 printk(KERN_WARNING "%s: id %s: %pV\n", 502 RBD_DRV_NAME, rbd_dev->spec->image_id, &vaf); 503 else /* punt */ 504 printk(KERN_WARNING "%s: rbd_dev %p: %pV\n", 505 RBD_DRV_NAME, rbd_dev, &vaf); 506 va_end(args); 507 } 508 509 #ifdef RBD_DEBUG 510 #define rbd_assert(expr) \ 511 if (unlikely(!(expr))) { \ 512 printk(KERN_ERR "\nAssertion failure in %s() " \ 513 "at line %d:\n\n" \ 514 "\trbd_assert(%s);\n\n", \ 515 __func__, __LINE__, #expr); \ 516 BUG(); \ 517 } 518 #else /* !RBD_DEBUG */ 519 # define rbd_assert(expr) ((void) 0) 520 #endif /* !RBD_DEBUG */ 521 522 static int rbd_img_obj_request_submit(struct rbd_obj_request *obj_request); 523 static void rbd_img_parent_read(struct rbd_obj_request *obj_request); 524 static void rbd_dev_remove_parent(struct rbd_device *rbd_dev); 525 526 static int rbd_dev_refresh(struct rbd_device *rbd_dev); 527 static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev); 528 static int rbd_dev_header_info(struct rbd_device *rbd_dev); 529 static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev); 530 static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev, 531 u64 snap_id); 532 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id, 533 u8 *order, u64 *snap_size); 534 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id, 535 u64 *snap_features); 536 static u64 rbd_snap_id_by_name(struct rbd_device *rbd_dev, const char *name); 537 538 static int rbd_open(struct block_device *bdev, fmode_t mode) 539 { 540 struct rbd_device *rbd_dev = bdev->bd_disk->private_data; 541 bool removing = false; 542 543 if ((mode & FMODE_WRITE) && rbd_dev->mapping.read_only) 544 return -EROFS; 545 546 spin_lock_irq(&rbd_dev->lock); 547 if (test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags)) 548 removing = true; 549 else 550 rbd_dev->open_count++; 551 spin_unlock_irq(&rbd_dev->lock); 552 if (removing) 553 return -ENOENT; 554 555 (void) get_device(&rbd_dev->dev); 556 557 return 0; 558 } 559 560 static void rbd_release(struct gendisk *disk, fmode_t mode) 561 { 562 struct rbd_device *rbd_dev = disk->private_data; 563 unsigned long open_count_before; 564 565 spin_lock_irq(&rbd_dev->lock); 566 open_count_before = rbd_dev->open_count--; 567 spin_unlock_irq(&rbd_dev->lock); 568 rbd_assert(open_count_before > 0); 569 570 put_device(&rbd_dev->dev); 571 } 572 573 static int rbd_ioctl_set_ro(struct rbd_device *rbd_dev, unsigned long arg) 574 { 575 int ret = 0; 576 int val; 577 bool ro; 578 bool ro_changed = false; 579 580 /* get_user() may sleep, so call it before taking rbd_dev->lock */ 581 if (get_user(val, (int __user *)(arg))) 582 return -EFAULT; 583 584 ro = val ? true : false; 585 /* Snapshot doesn't allow to write*/ 586 if (rbd_dev->spec->snap_id != CEPH_NOSNAP && !ro) 587 return -EROFS; 588 589 spin_lock_irq(&rbd_dev->lock); 590 /* prevent others open this device */ 591 if (rbd_dev->open_count > 1) { 592 ret = -EBUSY; 593 goto out; 594 } 595 596 if (rbd_dev->mapping.read_only != ro) { 597 rbd_dev->mapping.read_only = ro; 598 ro_changed = true; 599 } 600 601 out: 602 spin_unlock_irq(&rbd_dev->lock); 603 /* set_disk_ro() may sleep, so call it after releasing rbd_dev->lock */ 604 if (ret == 0 && ro_changed) 605 set_disk_ro(rbd_dev->disk, ro ? 1 : 0); 606 607 return ret; 608 } 609 610 static int rbd_ioctl(struct block_device *bdev, fmode_t mode, 611 unsigned int cmd, unsigned long arg) 612 { 613 struct rbd_device *rbd_dev = bdev->bd_disk->private_data; 614 int ret = 0; 615 616 switch (cmd) { 617 case BLKROSET: 618 ret = rbd_ioctl_set_ro(rbd_dev, arg); 619 break; 620 default: 621 ret = -ENOTTY; 622 } 623 624 return ret; 625 } 626 627 #ifdef CONFIG_COMPAT 628 static int rbd_compat_ioctl(struct block_device *bdev, fmode_t mode, 629 unsigned int cmd, unsigned long arg) 630 { 631 return rbd_ioctl(bdev, mode, cmd, arg); 632 } 633 #endif /* CONFIG_COMPAT */ 634 635 static const struct block_device_operations rbd_bd_ops = { 636 .owner = THIS_MODULE, 637 .open = rbd_open, 638 .release = rbd_release, 639 .ioctl = rbd_ioctl, 640 #ifdef CONFIG_COMPAT 641 .compat_ioctl = rbd_compat_ioctl, 642 #endif 643 }; 644 645 /* 646 * Initialize an rbd client instance. Success or not, this function 647 * consumes ceph_opts. Caller holds client_mutex. 648 */ 649 static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts) 650 { 651 struct rbd_client *rbdc; 652 int ret = -ENOMEM; 653 654 dout("%s:\n", __func__); 655 rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL); 656 if (!rbdc) 657 goto out_opt; 658 659 kref_init(&rbdc->kref); 660 INIT_LIST_HEAD(&rbdc->node); 661 662 rbdc->client = ceph_create_client(ceph_opts, rbdc, 0, 0); 663 if (IS_ERR(rbdc->client)) 664 goto out_rbdc; 665 ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */ 666 667 ret = ceph_open_session(rbdc->client); 668 if (ret < 0) 669 goto out_client; 670 671 spin_lock(&rbd_client_list_lock); 672 list_add_tail(&rbdc->node, &rbd_client_list); 673 spin_unlock(&rbd_client_list_lock); 674 675 dout("%s: rbdc %p\n", __func__, rbdc); 676 677 return rbdc; 678 out_client: 679 ceph_destroy_client(rbdc->client); 680 out_rbdc: 681 kfree(rbdc); 682 out_opt: 683 if (ceph_opts) 684 ceph_destroy_options(ceph_opts); 685 dout("%s: error %d\n", __func__, ret); 686 687 return ERR_PTR(ret); 688 } 689 690 static struct rbd_client *__rbd_get_client(struct rbd_client *rbdc) 691 { 692 kref_get(&rbdc->kref); 693 694 return rbdc; 695 } 696 697 /* 698 * Find a ceph client with specific addr and configuration. If 699 * found, bump its reference count. 700 */ 701 static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts) 702 { 703 struct rbd_client *client_node; 704 bool found = false; 705 706 if (ceph_opts->flags & CEPH_OPT_NOSHARE) 707 return NULL; 708 709 spin_lock(&rbd_client_list_lock); 710 list_for_each_entry(client_node, &rbd_client_list, node) { 711 if (!ceph_compare_options(ceph_opts, client_node->client)) { 712 __rbd_get_client(client_node); 713 714 found = true; 715 break; 716 } 717 } 718 spin_unlock(&rbd_client_list_lock); 719 720 return found ? client_node : NULL; 721 } 722 723 /* 724 * mount options 725 */ 726 enum { 727 Opt_last_int, 728 /* int args above */ 729 Opt_last_string, 730 /* string args above */ 731 Opt_read_only, 732 Opt_read_write, 733 /* Boolean args above */ 734 Opt_last_bool, 735 }; 736 737 static match_table_t rbd_opts_tokens = { 738 /* int args above */ 739 /* string args above */ 740 {Opt_read_only, "read_only"}, 741 {Opt_read_only, "ro"}, /* Alternate spelling */ 742 {Opt_read_write, "read_write"}, 743 {Opt_read_write, "rw"}, /* Alternate spelling */ 744 /* Boolean args above */ 745 {-1, NULL} 746 }; 747 748 struct rbd_options { 749 bool read_only; 750 }; 751 752 #define RBD_READ_ONLY_DEFAULT false 753 754 static int parse_rbd_opts_token(char *c, void *private) 755 { 756 struct rbd_options *rbd_opts = private; 757 substring_t argstr[MAX_OPT_ARGS]; 758 int token, intval, ret; 759 760 token = match_token(c, rbd_opts_tokens, argstr); 761 if (token < 0) 762 return -EINVAL; 763 764 if (token < Opt_last_int) { 765 ret = match_int(&argstr[0], &intval); 766 if (ret < 0) { 767 pr_err("bad mount option arg (not int) " 768 "at '%s'\n", c); 769 return ret; 770 } 771 dout("got int token %d val %d\n", token, intval); 772 } else if (token > Opt_last_int && token < Opt_last_string) { 773 dout("got string token %d val %s\n", token, 774 argstr[0].from); 775 } else if (token > Opt_last_string && token < Opt_last_bool) { 776 dout("got Boolean token %d\n", token); 777 } else { 778 dout("got token %d\n", token); 779 } 780 781 switch (token) { 782 case Opt_read_only: 783 rbd_opts->read_only = true; 784 break; 785 case Opt_read_write: 786 rbd_opts->read_only = false; 787 break; 788 default: 789 rbd_assert(false); 790 break; 791 } 792 return 0; 793 } 794 795 static char* obj_op_name(enum obj_operation_type op_type) 796 { 797 switch (op_type) { 798 case OBJ_OP_READ: 799 return "read"; 800 case OBJ_OP_WRITE: 801 return "write"; 802 case OBJ_OP_DISCARD: 803 return "discard"; 804 default: 805 return "???"; 806 } 807 } 808 809 /* 810 * Get a ceph client with specific addr and configuration, if one does 811 * not exist create it. Either way, ceph_opts is consumed by this 812 * function. 813 */ 814 static struct rbd_client *rbd_get_client(struct ceph_options *ceph_opts) 815 { 816 struct rbd_client *rbdc; 817 818 mutex_lock_nested(&client_mutex, SINGLE_DEPTH_NESTING); 819 rbdc = rbd_client_find(ceph_opts); 820 if (rbdc) /* using an existing client */ 821 ceph_destroy_options(ceph_opts); 822 else 823 rbdc = rbd_client_create(ceph_opts); 824 mutex_unlock(&client_mutex); 825 826 return rbdc; 827 } 828 829 /* 830 * Destroy ceph client 831 * 832 * Caller must hold rbd_client_list_lock. 833 */ 834 static void rbd_client_release(struct kref *kref) 835 { 836 struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref); 837 838 dout("%s: rbdc %p\n", __func__, rbdc); 839 spin_lock(&rbd_client_list_lock); 840 list_del(&rbdc->node); 841 spin_unlock(&rbd_client_list_lock); 842 843 ceph_destroy_client(rbdc->client); 844 kfree(rbdc); 845 } 846 847 /* 848 * Drop reference to ceph client node. If it's not referenced anymore, release 849 * it. 850 */ 851 static void rbd_put_client(struct rbd_client *rbdc) 852 { 853 if (rbdc) 854 kref_put(&rbdc->kref, rbd_client_release); 855 } 856 857 static bool rbd_image_format_valid(u32 image_format) 858 { 859 return image_format == 1 || image_format == 2; 860 } 861 862 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk) 863 { 864 size_t size; 865 u32 snap_count; 866 867 /* The header has to start with the magic rbd header text */ 868 if (memcmp(&ondisk->text, RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT))) 869 return false; 870 871 /* The bio layer requires at least sector-sized I/O */ 872 873 if (ondisk->options.order < SECTOR_SHIFT) 874 return false; 875 876 /* If we use u64 in a few spots we may be able to loosen this */ 877 878 if (ondisk->options.order > 8 * sizeof (int) - 1) 879 return false; 880 881 /* 882 * The size of a snapshot header has to fit in a size_t, and 883 * that limits the number of snapshots. 884 */ 885 snap_count = le32_to_cpu(ondisk->snap_count); 886 size = SIZE_MAX - sizeof (struct ceph_snap_context); 887 if (snap_count > size / sizeof (__le64)) 888 return false; 889 890 /* 891 * Not only that, but the size of the entire the snapshot 892 * header must also be representable in a size_t. 893 */ 894 size -= snap_count * sizeof (__le64); 895 if ((u64) size < le64_to_cpu(ondisk->snap_names_len)) 896 return false; 897 898 return true; 899 } 900 901 /* 902 * Fill an rbd image header with information from the given format 1 903 * on-disk header. 904 */ 905 static int rbd_header_from_disk(struct rbd_device *rbd_dev, 906 struct rbd_image_header_ondisk *ondisk) 907 { 908 struct rbd_image_header *header = &rbd_dev->header; 909 bool first_time = header->object_prefix == NULL; 910 struct ceph_snap_context *snapc; 911 char *object_prefix = NULL; 912 char *snap_names = NULL; 913 u64 *snap_sizes = NULL; 914 u32 snap_count; 915 size_t size; 916 int ret = -ENOMEM; 917 u32 i; 918 919 /* Allocate this now to avoid having to handle failure below */ 920 921 if (first_time) { 922 size_t len; 923 924 len = strnlen(ondisk->object_prefix, 925 sizeof (ondisk->object_prefix)); 926 object_prefix = kmalloc(len + 1, GFP_KERNEL); 927 if (!object_prefix) 928 return -ENOMEM; 929 memcpy(object_prefix, ondisk->object_prefix, len); 930 object_prefix[len] = '\0'; 931 } 932 933 /* Allocate the snapshot context and fill it in */ 934 935 snap_count = le32_to_cpu(ondisk->snap_count); 936 snapc = ceph_create_snap_context(snap_count, GFP_KERNEL); 937 if (!snapc) 938 goto out_err; 939 snapc->seq = le64_to_cpu(ondisk->snap_seq); 940 if (snap_count) { 941 struct rbd_image_snap_ondisk *snaps; 942 u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len); 943 944 /* We'll keep a copy of the snapshot names... */ 945 946 if (snap_names_len > (u64)SIZE_MAX) 947 goto out_2big; 948 snap_names = kmalloc(snap_names_len, GFP_KERNEL); 949 if (!snap_names) 950 goto out_err; 951 952 /* ...as well as the array of their sizes. */ 953 954 size = snap_count * sizeof (*header->snap_sizes); 955 snap_sizes = kmalloc(size, GFP_KERNEL); 956 if (!snap_sizes) 957 goto out_err; 958 959 /* 960 * Copy the names, and fill in each snapshot's id 961 * and size. 962 * 963 * Note that rbd_dev_v1_header_info() guarantees the 964 * ondisk buffer we're working with has 965 * snap_names_len bytes beyond the end of the 966 * snapshot id array, this memcpy() is safe. 967 */ 968 memcpy(snap_names, &ondisk->snaps[snap_count], snap_names_len); 969 snaps = ondisk->snaps; 970 for (i = 0; i < snap_count; i++) { 971 snapc->snaps[i] = le64_to_cpu(snaps[i].id); 972 snap_sizes[i] = le64_to_cpu(snaps[i].image_size); 973 } 974 } 975 976 /* We won't fail any more, fill in the header */ 977 978 if (first_time) { 979 header->object_prefix = object_prefix; 980 header->obj_order = ondisk->options.order; 981 header->crypt_type = ondisk->options.crypt_type; 982 header->comp_type = ondisk->options.comp_type; 983 /* The rest aren't used for format 1 images */ 984 header->stripe_unit = 0; 985 header->stripe_count = 0; 986 header->features = 0; 987 } else { 988 ceph_put_snap_context(header->snapc); 989 kfree(header->snap_names); 990 kfree(header->snap_sizes); 991 } 992 993 /* The remaining fields always get updated (when we refresh) */ 994 995 header->image_size = le64_to_cpu(ondisk->image_size); 996 header->snapc = snapc; 997 header->snap_names = snap_names; 998 header->snap_sizes = snap_sizes; 999 1000 return 0; 1001 out_2big: 1002 ret = -EIO; 1003 out_err: 1004 kfree(snap_sizes); 1005 kfree(snap_names); 1006 ceph_put_snap_context(snapc); 1007 kfree(object_prefix); 1008 1009 return ret; 1010 } 1011 1012 static const char *_rbd_dev_v1_snap_name(struct rbd_device *rbd_dev, u32 which) 1013 { 1014 const char *snap_name; 1015 1016 rbd_assert(which < rbd_dev->header.snapc->num_snaps); 1017 1018 /* Skip over names until we find the one we are looking for */ 1019 1020 snap_name = rbd_dev->header.snap_names; 1021 while (which--) 1022 snap_name += strlen(snap_name) + 1; 1023 1024 return kstrdup(snap_name, GFP_KERNEL); 1025 } 1026 1027 /* 1028 * Snapshot id comparison function for use with qsort()/bsearch(). 1029 * Note that result is for snapshots in *descending* order. 1030 */ 1031 static int snapid_compare_reverse(const void *s1, const void *s2) 1032 { 1033 u64 snap_id1 = *(u64 *)s1; 1034 u64 snap_id2 = *(u64 *)s2; 1035 1036 if (snap_id1 < snap_id2) 1037 return 1; 1038 return snap_id1 == snap_id2 ? 0 : -1; 1039 } 1040 1041 /* 1042 * Search a snapshot context to see if the given snapshot id is 1043 * present. 1044 * 1045 * Returns the position of the snapshot id in the array if it's found, 1046 * or BAD_SNAP_INDEX otherwise. 1047 * 1048 * Note: The snapshot array is in kept sorted (by the osd) in 1049 * reverse order, highest snapshot id first. 1050 */ 1051 static u32 rbd_dev_snap_index(struct rbd_device *rbd_dev, u64 snap_id) 1052 { 1053 struct ceph_snap_context *snapc = rbd_dev->header.snapc; 1054 u64 *found; 1055 1056 found = bsearch(&snap_id, &snapc->snaps, snapc->num_snaps, 1057 sizeof (snap_id), snapid_compare_reverse); 1058 1059 return found ? (u32)(found - &snapc->snaps[0]) : BAD_SNAP_INDEX; 1060 } 1061 1062 static const char *rbd_dev_v1_snap_name(struct rbd_device *rbd_dev, 1063 u64 snap_id) 1064 { 1065 u32 which; 1066 const char *snap_name; 1067 1068 which = rbd_dev_snap_index(rbd_dev, snap_id); 1069 if (which == BAD_SNAP_INDEX) 1070 return ERR_PTR(-ENOENT); 1071 1072 snap_name = _rbd_dev_v1_snap_name(rbd_dev, which); 1073 return snap_name ? snap_name : ERR_PTR(-ENOMEM); 1074 } 1075 1076 static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id) 1077 { 1078 if (snap_id == CEPH_NOSNAP) 1079 return RBD_SNAP_HEAD_NAME; 1080 1081 rbd_assert(rbd_image_format_valid(rbd_dev->image_format)); 1082 if (rbd_dev->image_format == 1) 1083 return rbd_dev_v1_snap_name(rbd_dev, snap_id); 1084 1085 return rbd_dev_v2_snap_name(rbd_dev, snap_id); 1086 } 1087 1088 static int rbd_snap_size(struct rbd_device *rbd_dev, u64 snap_id, 1089 u64 *snap_size) 1090 { 1091 rbd_assert(rbd_image_format_valid(rbd_dev->image_format)); 1092 if (snap_id == CEPH_NOSNAP) { 1093 *snap_size = rbd_dev->header.image_size; 1094 } else if (rbd_dev->image_format == 1) { 1095 u32 which; 1096 1097 which = rbd_dev_snap_index(rbd_dev, snap_id); 1098 if (which == BAD_SNAP_INDEX) 1099 return -ENOENT; 1100 1101 *snap_size = rbd_dev->header.snap_sizes[which]; 1102 } else { 1103 u64 size = 0; 1104 int ret; 1105 1106 ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, NULL, &size); 1107 if (ret) 1108 return ret; 1109 1110 *snap_size = size; 1111 } 1112 return 0; 1113 } 1114 1115 static int rbd_snap_features(struct rbd_device *rbd_dev, u64 snap_id, 1116 u64 *snap_features) 1117 { 1118 rbd_assert(rbd_image_format_valid(rbd_dev->image_format)); 1119 if (snap_id == CEPH_NOSNAP) { 1120 *snap_features = rbd_dev->header.features; 1121 } else if (rbd_dev->image_format == 1) { 1122 *snap_features = 0; /* No features for format 1 */ 1123 } else { 1124 u64 features = 0; 1125 int ret; 1126 1127 ret = _rbd_dev_v2_snap_features(rbd_dev, snap_id, &features); 1128 if (ret) 1129 return ret; 1130 1131 *snap_features = features; 1132 } 1133 return 0; 1134 } 1135 1136 static int rbd_dev_mapping_set(struct rbd_device *rbd_dev) 1137 { 1138 u64 snap_id = rbd_dev->spec->snap_id; 1139 u64 size = 0; 1140 u64 features = 0; 1141 int ret; 1142 1143 ret = rbd_snap_size(rbd_dev, snap_id, &size); 1144 if (ret) 1145 return ret; 1146 ret = rbd_snap_features(rbd_dev, snap_id, &features); 1147 if (ret) 1148 return ret; 1149 1150 rbd_dev->mapping.size = size; 1151 rbd_dev->mapping.features = features; 1152 1153 return 0; 1154 } 1155 1156 static void rbd_dev_mapping_clear(struct rbd_device *rbd_dev) 1157 { 1158 rbd_dev->mapping.size = 0; 1159 rbd_dev->mapping.features = 0; 1160 } 1161 1162 static void rbd_segment_name_free(const char *name) 1163 { 1164 /* The explicit cast here is needed to drop the const qualifier */ 1165 1166 kmem_cache_free(rbd_segment_name_cache, (void *)name); 1167 } 1168 1169 static const char *rbd_segment_name(struct rbd_device *rbd_dev, u64 offset) 1170 { 1171 char *name; 1172 u64 segment; 1173 int ret; 1174 char *name_format; 1175 1176 name = kmem_cache_alloc(rbd_segment_name_cache, GFP_NOIO); 1177 if (!name) 1178 return NULL; 1179 segment = offset >> rbd_dev->header.obj_order; 1180 name_format = "%s.%012llx"; 1181 if (rbd_dev->image_format == 2) 1182 name_format = "%s.%016llx"; 1183 ret = snprintf(name, CEPH_MAX_OID_NAME_LEN + 1, name_format, 1184 rbd_dev->header.object_prefix, segment); 1185 if (ret < 0 || ret > CEPH_MAX_OID_NAME_LEN) { 1186 pr_err("error formatting segment name for #%llu (%d)\n", 1187 segment, ret); 1188 rbd_segment_name_free(name); 1189 name = NULL; 1190 } 1191 1192 return name; 1193 } 1194 1195 static u64 rbd_segment_offset(struct rbd_device *rbd_dev, u64 offset) 1196 { 1197 u64 segment_size = (u64) 1 << rbd_dev->header.obj_order; 1198 1199 return offset & (segment_size - 1); 1200 } 1201 1202 static u64 rbd_segment_length(struct rbd_device *rbd_dev, 1203 u64 offset, u64 length) 1204 { 1205 u64 segment_size = (u64) 1 << rbd_dev->header.obj_order; 1206 1207 offset &= segment_size - 1; 1208 1209 rbd_assert(length <= U64_MAX - offset); 1210 if (offset + length > segment_size) 1211 length = segment_size - offset; 1212 1213 return length; 1214 } 1215 1216 /* 1217 * returns the size of an object in the image 1218 */ 1219 static u64 rbd_obj_bytes(struct rbd_image_header *header) 1220 { 1221 return 1 << header->obj_order; 1222 } 1223 1224 /* 1225 * bio helpers 1226 */ 1227 1228 static void bio_chain_put(struct bio *chain) 1229 { 1230 struct bio *tmp; 1231 1232 while (chain) { 1233 tmp = chain; 1234 chain = chain->bi_next; 1235 bio_put(tmp); 1236 } 1237 } 1238 1239 /* 1240 * zeros a bio chain, starting at specific offset 1241 */ 1242 static void zero_bio_chain(struct bio *chain, int start_ofs) 1243 { 1244 struct bio_vec bv; 1245 struct bvec_iter iter; 1246 unsigned long flags; 1247 void *buf; 1248 int pos = 0; 1249 1250 while (chain) { 1251 bio_for_each_segment(bv, chain, iter) { 1252 if (pos + bv.bv_len > start_ofs) { 1253 int remainder = max(start_ofs - pos, 0); 1254 buf = bvec_kmap_irq(&bv, &flags); 1255 memset(buf + remainder, 0, 1256 bv.bv_len - remainder); 1257 flush_dcache_page(bv.bv_page); 1258 bvec_kunmap_irq(buf, &flags); 1259 } 1260 pos += bv.bv_len; 1261 } 1262 1263 chain = chain->bi_next; 1264 } 1265 } 1266 1267 /* 1268 * similar to zero_bio_chain(), zeros data defined by a page array, 1269 * starting at the given byte offset from the start of the array and 1270 * continuing up to the given end offset. The pages array is 1271 * assumed to be big enough to hold all bytes up to the end. 1272 */ 1273 static void zero_pages(struct page **pages, u64 offset, u64 end) 1274 { 1275 struct page **page = &pages[offset >> PAGE_SHIFT]; 1276 1277 rbd_assert(end > offset); 1278 rbd_assert(end - offset <= (u64)SIZE_MAX); 1279 while (offset < end) { 1280 size_t page_offset; 1281 size_t length; 1282 unsigned long flags; 1283 void *kaddr; 1284 1285 page_offset = offset & ~PAGE_MASK; 1286 length = min_t(size_t, PAGE_SIZE - page_offset, end - offset); 1287 local_irq_save(flags); 1288 kaddr = kmap_atomic(*page); 1289 memset(kaddr + page_offset, 0, length); 1290 flush_dcache_page(*page); 1291 kunmap_atomic(kaddr); 1292 local_irq_restore(flags); 1293 1294 offset += length; 1295 page++; 1296 } 1297 } 1298 1299 /* 1300 * Clone a portion of a bio, starting at the given byte offset 1301 * and continuing for the number of bytes indicated. 1302 */ 1303 static struct bio *bio_clone_range(struct bio *bio_src, 1304 unsigned int offset, 1305 unsigned int len, 1306 gfp_t gfpmask) 1307 { 1308 struct bio *bio; 1309 1310 bio = bio_clone(bio_src, gfpmask); 1311 if (!bio) 1312 return NULL; /* ENOMEM */ 1313 1314 bio_advance(bio, offset); 1315 bio->bi_iter.bi_size = len; 1316 1317 return bio; 1318 } 1319 1320 /* 1321 * Clone a portion of a bio chain, starting at the given byte offset 1322 * into the first bio in the source chain and continuing for the 1323 * number of bytes indicated. The result is another bio chain of 1324 * exactly the given length, or a null pointer on error. 1325 * 1326 * The bio_src and offset parameters are both in-out. On entry they 1327 * refer to the first source bio and the offset into that bio where 1328 * the start of data to be cloned is located. 1329 * 1330 * On return, bio_src is updated to refer to the bio in the source 1331 * chain that contains first un-cloned byte, and *offset will 1332 * contain the offset of that byte within that bio. 1333 */ 1334 static struct bio *bio_chain_clone_range(struct bio **bio_src, 1335 unsigned int *offset, 1336 unsigned int len, 1337 gfp_t gfpmask) 1338 { 1339 struct bio *bi = *bio_src; 1340 unsigned int off = *offset; 1341 struct bio *chain = NULL; 1342 struct bio **end; 1343 1344 /* Build up a chain of clone bios up to the limit */ 1345 1346 if (!bi || off >= bi->bi_iter.bi_size || !len) 1347 return NULL; /* Nothing to clone */ 1348 1349 end = &chain; 1350 while (len) { 1351 unsigned int bi_size; 1352 struct bio *bio; 1353 1354 if (!bi) { 1355 rbd_warn(NULL, "bio_chain exhausted with %u left", len); 1356 goto out_err; /* EINVAL; ran out of bio's */ 1357 } 1358 bi_size = min_t(unsigned int, bi->bi_iter.bi_size - off, len); 1359 bio = bio_clone_range(bi, off, bi_size, gfpmask); 1360 if (!bio) 1361 goto out_err; /* ENOMEM */ 1362 1363 *end = bio; 1364 end = &bio->bi_next; 1365 1366 off += bi_size; 1367 if (off == bi->bi_iter.bi_size) { 1368 bi = bi->bi_next; 1369 off = 0; 1370 } 1371 len -= bi_size; 1372 } 1373 *bio_src = bi; 1374 *offset = off; 1375 1376 return chain; 1377 out_err: 1378 bio_chain_put(chain); 1379 1380 return NULL; 1381 } 1382 1383 /* 1384 * The default/initial value for all object request flags is 0. For 1385 * each flag, once its value is set to 1 it is never reset to 0 1386 * again. 1387 */ 1388 static void obj_request_img_data_set(struct rbd_obj_request *obj_request) 1389 { 1390 if (test_and_set_bit(OBJ_REQ_IMG_DATA, &obj_request->flags)) { 1391 struct rbd_device *rbd_dev; 1392 1393 rbd_dev = obj_request->img_request->rbd_dev; 1394 rbd_warn(rbd_dev, "obj_request %p already marked img_data", 1395 obj_request); 1396 } 1397 } 1398 1399 static bool obj_request_img_data_test(struct rbd_obj_request *obj_request) 1400 { 1401 smp_mb(); 1402 return test_bit(OBJ_REQ_IMG_DATA, &obj_request->flags) != 0; 1403 } 1404 1405 static void obj_request_done_set(struct rbd_obj_request *obj_request) 1406 { 1407 if (test_and_set_bit(OBJ_REQ_DONE, &obj_request->flags)) { 1408 struct rbd_device *rbd_dev = NULL; 1409 1410 if (obj_request_img_data_test(obj_request)) 1411 rbd_dev = obj_request->img_request->rbd_dev; 1412 rbd_warn(rbd_dev, "obj_request %p already marked done", 1413 obj_request); 1414 } 1415 } 1416 1417 static bool obj_request_done_test(struct rbd_obj_request *obj_request) 1418 { 1419 smp_mb(); 1420 return test_bit(OBJ_REQ_DONE, &obj_request->flags) != 0; 1421 } 1422 1423 /* 1424 * This sets the KNOWN flag after (possibly) setting the EXISTS 1425 * flag. The latter is set based on the "exists" value provided. 1426 * 1427 * Note that for our purposes once an object exists it never goes 1428 * away again. It's possible that the response from two existence 1429 * checks are separated by the creation of the target object, and 1430 * the first ("doesn't exist") response arrives *after* the second 1431 * ("does exist"). In that case we ignore the second one. 1432 */ 1433 static void obj_request_existence_set(struct rbd_obj_request *obj_request, 1434 bool exists) 1435 { 1436 if (exists) 1437 set_bit(OBJ_REQ_EXISTS, &obj_request->flags); 1438 set_bit(OBJ_REQ_KNOWN, &obj_request->flags); 1439 smp_mb(); 1440 } 1441 1442 static bool obj_request_known_test(struct rbd_obj_request *obj_request) 1443 { 1444 smp_mb(); 1445 return test_bit(OBJ_REQ_KNOWN, &obj_request->flags) != 0; 1446 } 1447 1448 static bool obj_request_exists_test(struct rbd_obj_request *obj_request) 1449 { 1450 smp_mb(); 1451 return test_bit(OBJ_REQ_EXISTS, &obj_request->flags) != 0; 1452 } 1453 1454 static bool obj_request_overlaps_parent(struct rbd_obj_request *obj_request) 1455 { 1456 struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev; 1457 1458 return obj_request->img_offset < 1459 round_up(rbd_dev->parent_overlap, rbd_obj_bytes(&rbd_dev->header)); 1460 } 1461 1462 static void rbd_obj_request_get(struct rbd_obj_request *obj_request) 1463 { 1464 dout("%s: obj %p (was %d)\n", __func__, obj_request, 1465 atomic_read(&obj_request->kref.refcount)); 1466 kref_get(&obj_request->kref); 1467 } 1468 1469 static void rbd_obj_request_destroy(struct kref *kref); 1470 static void rbd_obj_request_put(struct rbd_obj_request *obj_request) 1471 { 1472 rbd_assert(obj_request != NULL); 1473 dout("%s: obj %p (was %d)\n", __func__, obj_request, 1474 atomic_read(&obj_request->kref.refcount)); 1475 kref_put(&obj_request->kref, rbd_obj_request_destroy); 1476 } 1477 1478 static void rbd_img_request_get(struct rbd_img_request *img_request) 1479 { 1480 dout("%s: img %p (was %d)\n", __func__, img_request, 1481 atomic_read(&img_request->kref.refcount)); 1482 kref_get(&img_request->kref); 1483 } 1484 1485 static bool img_request_child_test(struct rbd_img_request *img_request); 1486 static void rbd_parent_request_destroy(struct kref *kref); 1487 static void rbd_img_request_destroy(struct kref *kref); 1488 static void rbd_img_request_put(struct rbd_img_request *img_request) 1489 { 1490 rbd_assert(img_request != NULL); 1491 dout("%s: img %p (was %d)\n", __func__, img_request, 1492 atomic_read(&img_request->kref.refcount)); 1493 if (img_request_child_test(img_request)) 1494 kref_put(&img_request->kref, rbd_parent_request_destroy); 1495 else 1496 kref_put(&img_request->kref, rbd_img_request_destroy); 1497 } 1498 1499 static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request, 1500 struct rbd_obj_request *obj_request) 1501 { 1502 rbd_assert(obj_request->img_request == NULL); 1503 1504 /* Image request now owns object's original reference */ 1505 obj_request->img_request = img_request; 1506 obj_request->which = img_request->obj_request_count; 1507 rbd_assert(!obj_request_img_data_test(obj_request)); 1508 obj_request_img_data_set(obj_request); 1509 rbd_assert(obj_request->which != BAD_WHICH); 1510 img_request->obj_request_count++; 1511 list_add_tail(&obj_request->links, &img_request->obj_requests); 1512 dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request, 1513 obj_request->which); 1514 } 1515 1516 static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request, 1517 struct rbd_obj_request *obj_request) 1518 { 1519 rbd_assert(obj_request->which != BAD_WHICH); 1520 1521 dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request, 1522 obj_request->which); 1523 list_del(&obj_request->links); 1524 rbd_assert(img_request->obj_request_count > 0); 1525 img_request->obj_request_count--; 1526 rbd_assert(obj_request->which == img_request->obj_request_count); 1527 obj_request->which = BAD_WHICH; 1528 rbd_assert(obj_request_img_data_test(obj_request)); 1529 rbd_assert(obj_request->img_request == img_request); 1530 obj_request->img_request = NULL; 1531 obj_request->callback = NULL; 1532 rbd_obj_request_put(obj_request); 1533 } 1534 1535 static bool obj_request_type_valid(enum obj_request_type type) 1536 { 1537 switch (type) { 1538 case OBJ_REQUEST_NODATA: 1539 case OBJ_REQUEST_BIO: 1540 case OBJ_REQUEST_PAGES: 1541 return true; 1542 default: 1543 return false; 1544 } 1545 } 1546 1547 static int rbd_obj_request_submit(struct ceph_osd_client *osdc, 1548 struct rbd_obj_request *obj_request) 1549 { 1550 dout("%s %p\n", __func__, obj_request); 1551 return ceph_osdc_start_request(osdc, obj_request->osd_req, false); 1552 } 1553 1554 static void rbd_obj_request_end(struct rbd_obj_request *obj_request) 1555 { 1556 dout("%s %p\n", __func__, obj_request); 1557 ceph_osdc_cancel_request(obj_request->osd_req); 1558 } 1559 1560 /* 1561 * Wait for an object request to complete. If interrupted, cancel the 1562 * underlying osd request. 1563 */ 1564 static int rbd_obj_request_wait(struct rbd_obj_request *obj_request) 1565 { 1566 int ret; 1567 1568 dout("%s %p\n", __func__, obj_request); 1569 1570 ret = wait_for_completion_interruptible(&obj_request->completion); 1571 if (ret < 0) { 1572 dout("%s %p interrupted\n", __func__, obj_request); 1573 rbd_obj_request_end(obj_request); 1574 return ret; 1575 } 1576 1577 dout("%s %p done\n", __func__, obj_request); 1578 return 0; 1579 } 1580 1581 static void rbd_img_request_complete(struct rbd_img_request *img_request) 1582 { 1583 1584 dout("%s: img %p\n", __func__, img_request); 1585 1586 /* 1587 * If no error occurred, compute the aggregate transfer 1588 * count for the image request. We could instead use 1589 * atomic64_cmpxchg() to update it as each object request 1590 * completes; not clear which way is better off hand. 1591 */ 1592 if (!img_request->result) { 1593 struct rbd_obj_request *obj_request; 1594 u64 xferred = 0; 1595 1596 for_each_obj_request(img_request, obj_request) 1597 xferred += obj_request->xferred; 1598 img_request->xferred = xferred; 1599 } 1600 1601 if (img_request->callback) 1602 img_request->callback(img_request); 1603 else 1604 rbd_img_request_put(img_request); 1605 } 1606 1607 /* 1608 * The default/initial value for all image request flags is 0. Each 1609 * is conditionally set to 1 at image request initialization time 1610 * and currently never change thereafter. 1611 */ 1612 static void img_request_write_set(struct rbd_img_request *img_request) 1613 { 1614 set_bit(IMG_REQ_WRITE, &img_request->flags); 1615 smp_mb(); 1616 } 1617 1618 static bool img_request_write_test(struct rbd_img_request *img_request) 1619 { 1620 smp_mb(); 1621 return test_bit(IMG_REQ_WRITE, &img_request->flags) != 0; 1622 } 1623 1624 /* 1625 * Set the discard flag when the img_request is an discard request 1626 */ 1627 static void img_request_discard_set(struct rbd_img_request *img_request) 1628 { 1629 set_bit(IMG_REQ_DISCARD, &img_request->flags); 1630 smp_mb(); 1631 } 1632 1633 static bool img_request_discard_test(struct rbd_img_request *img_request) 1634 { 1635 smp_mb(); 1636 return test_bit(IMG_REQ_DISCARD, &img_request->flags) != 0; 1637 } 1638 1639 static void img_request_child_set(struct rbd_img_request *img_request) 1640 { 1641 set_bit(IMG_REQ_CHILD, &img_request->flags); 1642 smp_mb(); 1643 } 1644 1645 static void img_request_child_clear(struct rbd_img_request *img_request) 1646 { 1647 clear_bit(IMG_REQ_CHILD, &img_request->flags); 1648 smp_mb(); 1649 } 1650 1651 static bool img_request_child_test(struct rbd_img_request *img_request) 1652 { 1653 smp_mb(); 1654 return test_bit(IMG_REQ_CHILD, &img_request->flags) != 0; 1655 } 1656 1657 static void img_request_layered_set(struct rbd_img_request *img_request) 1658 { 1659 set_bit(IMG_REQ_LAYERED, &img_request->flags); 1660 smp_mb(); 1661 } 1662 1663 static void img_request_layered_clear(struct rbd_img_request *img_request) 1664 { 1665 clear_bit(IMG_REQ_LAYERED, &img_request->flags); 1666 smp_mb(); 1667 } 1668 1669 static bool img_request_layered_test(struct rbd_img_request *img_request) 1670 { 1671 smp_mb(); 1672 return test_bit(IMG_REQ_LAYERED, &img_request->flags) != 0; 1673 } 1674 1675 static enum obj_operation_type 1676 rbd_img_request_op_type(struct rbd_img_request *img_request) 1677 { 1678 if (img_request_write_test(img_request)) 1679 return OBJ_OP_WRITE; 1680 else if (img_request_discard_test(img_request)) 1681 return OBJ_OP_DISCARD; 1682 else 1683 return OBJ_OP_READ; 1684 } 1685 1686 static void 1687 rbd_img_obj_request_read_callback(struct rbd_obj_request *obj_request) 1688 { 1689 u64 xferred = obj_request->xferred; 1690 u64 length = obj_request->length; 1691 1692 dout("%s: obj %p img %p result %d %llu/%llu\n", __func__, 1693 obj_request, obj_request->img_request, obj_request->result, 1694 xferred, length); 1695 /* 1696 * ENOENT means a hole in the image. We zero-fill the entire 1697 * length of the request. A short read also implies zero-fill 1698 * to the end of the request. An error requires the whole 1699 * length of the request to be reported finished with an error 1700 * to the block layer. In each case we update the xferred 1701 * count to indicate the whole request was satisfied. 1702 */ 1703 rbd_assert(obj_request->type != OBJ_REQUEST_NODATA); 1704 if (obj_request->result == -ENOENT) { 1705 if (obj_request->type == OBJ_REQUEST_BIO) 1706 zero_bio_chain(obj_request->bio_list, 0); 1707 else 1708 zero_pages(obj_request->pages, 0, length); 1709 obj_request->result = 0; 1710 } else if (xferred < length && !obj_request->result) { 1711 if (obj_request->type == OBJ_REQUEST_BIO) 1712 zero_bio_chain(obj_request->bio_list, xferred); 1713 else 1714 zero_pages(obj_request->pages, xferred, length); 1715 } 1716 obj_request->xferred = length; 1717 obj_request_done_set(obj_request); 1718 } 1719 1720 static void rbd_obj_request_complete(struct rbd_obj_request *obj_request) 1721 { 1722 dout("%s: obj %p cb %p\n", __func__, obj_request, 1723 obj_request->callback); 1724 if (obj_request->callback) 1725 obj_request->callback(obj_request); 1726 else 1727 complete_all(&obj_request->completion); 1728 } 1729 1730 static void rbd_osd_trivial_callback(struct rbd_obj_request *obj_request) 1731 { 1732 dout("%s: obj %p\n", __func__, obj_request); 1733 obj_request_done_set(obj_request); 1734 } 1735 1736 static void rbd_osd_read_callback(struct rbd_obj_request *obj_request) 1737 { 1738 struct rbd_img_request *img_request = NULL; 1739 struct rbd_device *rbd_dev = NULL; 1740 bool layered = false; 1741 1742 if (obj_request_img_data_test(obj_request)) { 1743 img_request = obj_request->img_request; 1744 layered = img_request && img_request_layered_test(img_request); 1745 rbd_dev = img_request->rbd_dev; 1746 } 1747 1748 dout("%s: obj %p img %p result %d %llu/%llu\n", __func__, 1749 obj_request, img_request, obj_request->result, 1750 obj_request->xferred, obj_request->length); 1751 if (layered && obj_request->result == -ENOENT && 1752 obj_request->img_offset < rbd_dev->parent_overlap) 1753 rbd_img_parent_read(obj_request); 1754 else if (img_request) 1755 rbd_img_obj_request_read_callback(obj_request); 1756 else 1757 obj_request_done_set(obj_request); 1758 } 1759 1760 static void rbd_osd_write_callback(struct rbd_obj_request *obj_request) 1761 { 1762 dout("%s: obj %p result %d %llu\n", __func__, obj_request, 1763 obj_request->result, obj_request->length); 1764 /* 1765 * There is no such thing as a successful short write. Set 1766 * it to our originally-requested length. 1767 */ 1768 obj_request->xferred = obj_request->length; 1769 obj_request_done_set(obj_request); 1770 } 1771 1772 static void rbd_osd_discard_callback(struct rbd_obj_request *obj_request) 1773 { 1774 dout("%s: obj %p result %d %llu\n", __func__, obj_request, 1775 obj_request->result, obj_request->length); 1776 /* 1777 * There is no such thing as a successful short discard. Set 1778 * it to our originally-requested length. 1779 */ 1780 obj_request->xferred = obj_request->length; 1781 /* discarding a non-existent object is not a problem */ 1782 if (obj_request->result == -ENOENT) 1783 obj_request->result = 0; 1784 obj_request_done_set(obj_request); 1785 } 1786 1787 /* 1788 * For a simple stat call there's nothing to do. We'll do more if 1789 * this is part of a write sequence for a layered image. 1790 */ 1791 static void rbd_osd_stat_callback(struct rbd_obj_request *obj_request) 1792 { 1793 dout("%s: obj %p\n", __func__, obj_request); 1794 obj_request_done_set(obj_request); 1795 } 1796 1797 static void rbd_osd_req_callback(struct ceph_osd_request *osd_req, 1798 struct ceph_msg *msg) 1799 { 1800 struct rbd_obj_request *obj_request = osd_req->r_priv; 1801 u16 opcode; 1802 1803 dout("%s: osd_req %p msg %p\n", __func__, osd_req, msg); 1804 rbd_assert(osd_req == obj_request->osd_req); 1805 if (obj_request_img_data_test(obj_request)) { 1806 rbd_assert(obj_request->img_request); 1807 rbd_assert(obj_request->which != BAD_WHICH); 1808 } else { 1809 rbd_assert(obj_request->which == BAD_WHICH); 1810 } 1811 1812 if (osd_req->r_result < 0) 1813 obj_request->result = osd_req->r_result; 1814 1815 rbd_assert(osd_req->r_num_ops <= CEPH_OSD_MAX_OP); 1816 1817 /* 1818 * We support a 64-bit length, but ultimately it has to be 1819 * passed to blk_end_request(), which takes an unsigned int. 1820 */ 1821 obj_request->xferred = osd_req->r_reply_op_len[0]; 1822 rbd_assert(obj_request->xferred < (u64)UINT_MAX); 1823 1824 opcode = osd_req->r_ops[0].op; 1825 switch (opcode) { 1826 case CEPH_OSD_OP_READ: 1827 rbd_osd_read_callback(obj_request); 1828 break; 1829 case CEPH_OSD_OP_SETALLOCHINT: 1830 rbd_assert(osd_req->r_ops[1].op == CEPH_OSD_OP_WRITE); 1831 /* fall through */ 1832 case CEPH_OSD_OP_WRITE: 1833 rbd_osd_write_callback(obj_request); 1834 break; 1835 case CEPH_OSD_OP_STAT: 1836 rbd_osd_stat_callback(obj_request); 1837 break; 1838 case CEPH_OSD_OP_DELETE: 1839 case CEPH_OSD_OP_TRUNCATE: 1840 case CEPH_OSD_OP_ZERO: 1841 rbd_osd_discard_callback(obj_request); 1842 break; 1843 case CEPH_OSD_OP_CALL: 1844 case CEPH_OSD_OP_NOTIFY_ACK: 1845 case CEPH_OSD_OP_WATCH: 1846 rbd_osd_trivial_callback(obj_request); 1847 break; 1848 default: 1849 rbd_warn(NULL, "%s: unsupported op %hu", 1850 obj_request->object_name, (unsigned short) opcode); 1851 break; 1852 } 1853 1854 if (obj_request_done_test(obj_request)) 1855 rbd_obj_request_complete(obj_request); 1856 } 1857 1858 static void rbd_osd_req_format_read(struct rbd_obj_request *obj_request) 1859 { 1860 struct rbd_img_request *img_request = obj_request->img_request; 1861 struct ceph_osd_request *osd_req = obj_request->osd_req; 1862 u64 snap_id; 1863 1864 rbd_assert(osd_req != NULL); 1865 1866 snap_id = img_request ? img_request->snap_id : CEPH_NOSNAP; 1867 ceph_osdc_build_request(osd_req, obj_request->offset, 1868 NULL, snap_id, NULL); 1869 } 1870 1871 static void rbd_osd_req_format_write(struct rbd_obj_request *obj_request) 1872 { 1873 struct rbd_img_request *img_request = obj_request->img_request; 1874 struct ceph_osd_request *osd_req = obj_request->osd_req; 1875 struct ceph_snap_context *snapc; 1876 struct timespec mtime = CURRENT_TIME; 1877 1878 rbd_assert(osd_req != NULL); 1879 1880 snapc = img_request ? img_request->snapc : NULL; 1881 ceph_osdc_build_request(osd_req, obj_request->offset, 1882 snapc, CEPH_NOSNAP, &mtime); 1883 } 1884 1885 /* 1886 * Create an osd request. A read request has one osd op (read). 1887 * A write request has either one (watch) or two (hint+write) osd ops. 1888 * (All rbd data writes are prefixed with an allocation hint op, but 1889 * technically osd watch is a write request, hence this distinction.) 1890 */ 1891 static struct ceph_osd_request *rbd_osd_req_create( 1892 struct rbd_device *rbd_dev, 1893 enum obj_operation_type op_type, 1894 unsigned int num_ops, 1895 struct rbd_obj_request *obj_request) 1896 { 1897 struct ceph_snap_context *snapc = NULL; 1898 struct ceph_osd_client *osdc; 1899 struct ceph_osd_request *osd_req; 1900 1901 if (obj_request_img_data_test(obj_request) && 1902 (op_type == OBJ_OP_DISCARD || op_type == OBJ_OP_WRITE)) { 1903 struct rbd_img_request *img_request = obj_request->img_request; 1904 if (op_type == OBJ_OP_WRITE) { 1905 rbd_assert(img_request_write_test(img_request)); 1906 } else { 1907 rbd_assert(img_request_discard_test(img_request)); 1908 } 1909 snapc = img_request->snapc; 1910 } 1911 1912 rbd_assert(num_ops == 1 || ((op_type == OBJ_OP_WRITE) && num_ops == 2)); 1913 1914 /* Allocate and initialize the request, for the num_ops ops */ 1915 1916 osdc = &rbd_dev->rbd_client->client->osdc; 1917 osd_req = ceph_osdc_alloc_request(osdc, snapc, num_ops, false, 1918 GFP_ATOMIC); 1919 if (!osd_req) 1920 return NULL; /* ENOMEM */ 1921 1922 if (op_type == OBJ_OP_WRITE || op_type == OBJ_OP_DISCARD) 1923 osd_req->r_flags = CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK; 1924 else 1925 osd_req->r_flags = CEPH_OSD_FLAG_READ; 1926 1927 osd_req->r_callback = rbd_osd_req_callback; 1928 osd_req->r_priv = obj_request; 1929 1930 osd_req->r_base_oloc.pool = ceph_file_layout_pg_pool(rbd_dev->layout); 1931 ceph_oid_set_name(&osd_req->r_base_oid, obj_request->object_name); 1932 1933 return osd_req; 1934 } 1935 1936 /* 1937 * Create a copyup osd request based on the information in the object 1938 * request supplied. A copyup request has two or three osd ops, a 1939 * copyup method call, potentially a hint op, and a write or truncate 1940 * or zero op. 1941 */ 1942 static struct ceph_osd_request * 1943 rbd_osd_req_create_copyup(struct rbd_obj_request *obj_request) 1944 { 1945 struct rbd_img_request *img_request; 1946 struct ceph_snap_context *snapc; 1947 struct rbd_device *rbd_dev; 1948 struct ceph_osd_client *osdc; 1949 struct ceph_osd_request *osd_req; 1950 int num_osd_ops = 3; 1951 1952 rbd_assert(obj_request_img_data_test(obj_request)); 1953 img_request = obj_request->img_request; 1954 rbd_assert(img_request); 1955 rbd_assert(img_request_write_test(img_request) || 1956 img_request_discard_test(img_request)); 1957 1958 if (img_request_discard_test(img_request)) 1959 num_osd_ops = 2; 1960 1961 /* Allocate and initialize the request, for all the ops */ 1962 1963 snapc = img_request->snapc; 1964 rbd_dev = img_request->rbd_dev; 1965 osdc = &rbd_dev->rbd_client->client->osdc; 1966 osd_req = ceph_osdc_alloc_request(osdc, snapc, num_osd_ops, 1967 false, GFP_ATOMIC); 1968 if (!osd_req) 1969 return NULL; /* ENOMEM */ 1970 1971 osd_req->r_flags = CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK; 1972 osd_req->r_callback = rbd_osd_req_callback; 1973 osd_req->r_priv = obj_request; 1974 1975 osd_req->r_base_oloc.pool = ceph_file_layout_pg_pool(rbd_dev->layout); 1976 ceph_oid_set_name(&osd_req->r_base_oid, obj_request->object_name); 1977 1978 return osd_req; 1979 } 1980 1981 1982 static void rbd_osd_req_destroy(struct ceph_osd_request *osd_req) 1983 { 1984 ceph_osdc_put_request(osd_req); 1985 } 1986 1987 /* object_name is assumed to be a non-null pointer and NUL-terminated */ 1988 1989 static struct rbd_obj_request *rbd_obj_request_create(const char *object_name, 1990 u64 offset, u64 length, 1991 enum obj_request_type type) 1992 { 1993 struct rbd_obj_request *obj_request; 1994 size_t size; 1995 char *name; 1996 1997 rbd_assert(obj_request_type_valid(type)); 1998 1999 size = strlen(object_name) + 1; 2000 name = kmalloc(size, GFP_KERNEL); 2001 if (!name) 2002 return NULL; 2003 2004 obj_request = kmem_cache_zalloc(rbd_obj_request_cache, GFP_KERNEL); 2005 if (!obj_request) { 2006 kfree(name); 2007 return NULL; 2008 } 2009 2010 obj_request->object_name = memcpy(name, object_name, size); 2011 obj_request->offset = offset; 2012 obj_request->length = length; 2013 obj_request->flags = 0; 2014 obj_request->which = BAD_WHICH; 2015 obj_request->type = type; 2016 INIT_LIST_HEAD(&obj_request->links); 2017 init_completion(&obj_request->completion); 2018 kref_init(&obj_request->kref); 2019 2020 dout("%s: \"%s\" %llu/%llu %d -> obj %p\n", __func__, object_name, 2021 offset, length, (int)type, obj_request); 2022 2023 return obj_request; 2024 } 2025 2026 static void rbd_obj_request_destroy(struct kref *kref) 2027 { 2028 struct rbd_obj_request *obj_request; 2029 2030 obj_request = container_of(kref, struct rbd_obj_request, kref); 2031 2032 dout("%s: obj %p\n", __func__, obj_request); 2033 2034 rbd_assert(obj_request->img_request == NULL); 2035 rbd_assert(obj_request->which == BAD_WHICH); 2036 2037 if (obj_request->osd_req) 2038 rbd_osd_req_destroy(obj_request->osd_req); 2039 2040 rbd_assert(obj_request_type_valid(obj_request->type)); 2041 switch (obj_request->type) { 2042 case OBJ_REQUEST_NODATA: 2043 break; /* Nothing to do */ 2044 case OBJ_REQUEST_BIO: 2045 if (obj_request->bio_list) 2046 bio_chain_put(obj_request->bio_list); 2047 break; 2048 case OBJ_REQUEST_PAGES: 2049 if (obj_request->pages) 2050 ceph_release_page_vector(obj_request->pages, 2051 obj_request->page_count); 2052 break; 2053 } 2054 2055 kfree(obj_request->object_name); 2056 obj_request->object_name = NULL; 2057 kmem_cache_free(rbd_obj_request_cache, obj_request); 2058 } 2059 2060 /* It's OK to call this for a device with no parent */ 2061 2062 static void rbd_spec_put(struct rbd_spec *spec); 2063 static void rbd_dev_unparent(struct rbd_device *rbd_dev) 2064 { 2065 rbd_dev_remove_parent(rbd_dev); 2066 rbd_spec_put(rbd_dev->parent_spec); 2067 rbd_dev->parent_spec = NULL; 2068 rbd_dev->parent_overlap = 0; 2069 } 2070 2071 /* 2072 * Parent image reference counting is used to determine when an 2073 * image's parent fields can be safely torn down--after there are no 2074 * more in-flight requests to the parent image. When the last 2075 * reference is dropped, cleaning them up is safe. 2076 */ 2077 static void rbd_dev_parent_put(struct rbd_device *rbd_dev) 2078 { 2079 int counter; 2080 2081 if (!rbd_dev->parent_spec) 2082 return; 2083 2084 counter = atomic_dec_return_safe(&rbd_dev->parent_ref); 2085 if (counter > 0) 2086 return; 2087 2088 /* Last reference; clean up parent data structures */ 2089 2090 if (!counter) 2091 rbd_dev_unparent(rbd_dev); 2092 else 2093 rbd_warn(rbd_dev, "parent reference underflow"); 2094 } 2095 2096 /* 2097 * If an image has a non-zero parent overlap, get a reference to its 2098 * parent. 2099 * 2100 * We must get the reference before checking for the overlap to 2101 * coordinate properly with zeroing the parent overlap in 2102 * rbd_dev_v2_parent_info() when an image gets flattened. We 2103 * drop it again if there is no overlap. 2104 * 2105 * Returns true if the rbd device has a parent with a non-zero 2106 * overlap and a reference for it was successfully taken, or 2107 * false otherwise. 2108 */ 2109 static bool rbd_dev_parent_get(struct rbd_device *rbd_dev) 2110 { 2111 int counter; 2112 2113 if (!rbd_dev->parent_spec) 2114 return false; 2115 2116 counter = atomic_inc_return_safe(&rbd_dev->parent_ref); 2117 if (counter > 0 && rbd_dev->parent_overlap) 2118 return true; 2119 2120 /* Image was flattened, but parent is not yet torn down */ 2121 2122 if (counter < 0) 2123 rbd_warn(rbd_dev, "parent reference overflow"); 2124 2125 return false; 2126 } 2127 2128 /* 2129 * Caller is responsible for filling in the list of object requests 2130 * that comprises the image request, and the Linux request pointer 2131 * (if there is one). 2132 */ 2133 static struct rbd_img_request *rbd_img_request_create( 2134 struct rbd_device *rbd_dev, 2135 u64 offset, u64 length, 2136 enum obj_operation_type op_type, 2137 struct ceph_snap_context *snapc) 2138 { 2139 struct rbd_img_request *img_request; 2140 2141 img_request = kmem_cache_alloc(rbd_img_request_cache, GFP_NOIO); 2142 if (!img_request) 2143 return NULL; 2144 2145 img_request->rq = NULL; 2146 img_request->rbd_dev = rbd_dev; 2147 img_request->offset = offset; 2148 img_request->length = length; 2149 img_request->flags = 0; 2150 if (op_type == OBJ_OP_DISCARD) { 2151 img_request_discard_set(img_request); 2152 img_request->snapc = snapc; 2153 } else if (op_type == OBJ_OP_WRITE) { 2154 img_request_write_set(img_request); 2155 img_request->snapc = snapc; 2156 } else { 2157 img_request->snap_id = rbd_dev->spec->snap_id; 2158 } 2159 if (rbd_dev_parent_get(rbd_dev)) 2160 img_request_layered_set(img_request); 2161 spin_lock_init(&img_request->completion_lock); 2162 img_request->next_completion = 0; 2163 img_request->callback = NULL; 2164 img_request->result = 0; 2165 img_request->obj_request_count = 0; 2166 INIT_LIST_HEAD(&img_request->obj_requests); 2167 kref_init(&img_request->kref); 2168 2169 dout("%s: rbd_dev %p %s %llu/%llu -> img %p\n", __func__, rbd_dev, 2170 obj_op_name(op_type), offset, length, img_request); 2171 2172 return img_request; 2173 } 2174 2175 static void rbd_img_request_destroy(struct kref *kref) 2176 { 2177 struct rbd_img_request *img_request; 2178 struct rbd_obj_request *obj_request; 2179 struct rbd_obj_request *next_obj_request; 2180 2181 img_request = container_of(kref, struct rbd_img_request, kref); 2182 2183 dout("%s: img %p\n", __func__, img_request); 2184 2185 for_each_obj_request_safe(img_request, obj_request, next_obj_request) 2186 rbd_img_obj_request_del(img_request, obj_request); 2187 rbd_assert(img_request->obj_request_count == 0); 2188 2189 if (img_request_layered_test(img_request)) { 2190 img_request_layered_clear(img_request); 2191 rbd_dev_parent_put(img_request->rbd_dev); 2192 } 2193 2194 if (img_request_write_test(img_request) || 2195 img_request_discard_test(img_request)) 2196 ceph_put_snap_context(img_request->snapc); 2197 2198 kmem_cache_free(rbd_img_request_cache, img_request); 2199 } 2200 2201 static struct rbd_img_request *rbd_parent_request_create( 2202 struct rbd_obj_request *obj_request, 2203 u64 img_offset, u64 length) 2204 { 2205 struct rbd_img_request *parent_request; 2206 struct rbd_device *rbd_dev; 2207 2208 rbd_assert(obj_request->img_request); 2209 rbd_dev = obj_request->img_request->rbd_dev; 2210 2211 parent_request = rbd_img_request_create(rbd_dev->parent, img_offset, 2212 length, OBJ_OP_READ, NULL); 2213 if (!parent_request) 2214 return NULL; 2215 2216 img_request_child_set(parent_request); 2217 rbd_obj_request_get(obj_request); 2218 parent_request->obj_request = obj_request; 2219 2220 return parent_request; 2221 } 2222 2223 static void rbd_parent_request_destroy(struct kref *kref) 2224 { 2225 struct rbd_img_request *parent_request; 2226 struct rbd_obj_request *orig_request; 2227 2228 parent_request = container_of(kref, struct rbd_img_request, kref); 2229 orig_request = parent_request->obj_request; 2230 2231 parent_request->obj_request = NULL; 2232 rbd_obj_request_put(orig_request); 2233 img_request_child_clear(parent_request); 2234 2235 rbd_img_request_destroy(kref); 2236 } 2237 2238 static bool rbd_img_obj_end_request(struct rbd_obj_request *obj_request) 2239 { 2240 struct rbd_img_request *img_request; 2241 unsigned int xferred; 2242 int result; 2243 bool more; 2244 2245 rbd_assert(obj_request_img_data_test(obj_request)); 2246 img_request = obj_request->img_request; 2247 2248 rbd_assert(obj_request->xferred <= (u64)UINT_MAX); 2249 xferred = (unsigned int)obj_request->xferred; 2250 result = obj_request->result; 2251 if (result) { 2252 struct rbd_device *rbd_dev = img_request->rbd_dev; 2253 enum obj_operation_type op_type; 2254 2255 if (img_request_discard_test(img_request)) 2256 op_type = OBJ_OP_DISCARD; 2257 else if (img_request_write_test(img_request)) 2258 op_type = OBJ_OP_WRITE; 2259 else 2260 op_type = OBJ_OP_READ; 2261 2262 rbd_warn(rbd_dev, "%s %llx at %llx (%llx)", 2263 obj_op_name(op_type), obj_request->length, 2264 obj_request->img_offset, obj_request->offset); 2265 rbd_warn(rbd_dev, " result %d xferred %x", 2266 result, xferred); 2267 if (!img_request->result) 2268 img_request->result = result; 2269 } 2270 2271 /* Image object requests don't own their page array */ 2272 2273 if (obj_request->type == OBJ_REQUEST_PAGES) { 2274 obj_request->pages = NULL; 2275 obj_request->page_count = 0; 2276 } 2277 2278 if (img_request_child_test(img_request)) { 2279 rbd_assert(img_request->obj_request != NULL); 2280 more = obj_request->which < img_request->obj_request_count - 1; 2281 } else { 2282 rbd_assert(img_request->rq != NULL); 2283 more = blk_end_request(img_request->rq, result, xferred); 2284 } 2285 2286 return more; 2287 } 2288 2289 static void rbd_img_obj_callback(struct rbd_obj_request *obj_request) 2290 { 2291 struct rbd_img_request *img_request; 2292 u32 which = obj_request->which; 2293 bool more = true; 2294 2295 rbd_assert(obj_request_img_data_test(obj_request)); 2296 img_request = obj_request->img_request; 2297 2298 dout("%s: img %p obj %p\n", __func__, img_request, obj_request); 2299 rbd_assert(img_request != NULL); 2300 rbd_assert(img_request->obj_request_count > 0); 2301 rbd_assert(which != BAD_WHICH); 2302 rbd_assert(which < img_request->obj_request_count); 2303 2304 spin_lock_irq(&img_request->completion_lock); 2305 if (which != img_request->next_completion) 2306 goto out; 2307 2308 for_each_obj_request_from(img_request, obj_request) { 2309 rbd_assert(more); 2310 rbd_assert(which < img_request->obj_request_count); 2311 2312 if (!obj_request_done_test(obj_request)) 2313 break; 2314 more = rbd_img_obj_end_request(obj_request); 2315 which++; 2316 } 2317 2318 rbd_assert(more ^ (which == img_request->obj_request_count)); 2319 img_request->next_completion = which; 2320 out: 2321 spin_unlock_irq(&img_request->completion_lock); 2322 rbd_img_request_put(img_request); 2323 2324 if (!more) 2325 rbd_img_request_complete(img_request); 2326 } 2327 2328 /* 2329 * Add individual osd ops to the given ceph_osd_request and prepare 2330 * them for submission. num_ops is the current number of 2331 * osd operations already to the object request. 2332 */ 2333 static void rbd_img_obj_request_fill(struct rbd_obj_request *obj_request, 2334 struct ceph_osd_request *osd_request, 2335 enum obj_operation_type op_type, 2336 unsigned int num_ops) 2337 { 2338 struct rbd_img_request *img_request = obj_request->img_request; 2339 struct rbd_device *rbd_dev = img_request->rbd_dev; 2340 u64 object_size = rbd_obj_bytes(&rbd_dev->header); 2341 u64 offset = obj_request->offset; 2342 u64 length = obj_request->length; 2343 u64 img_end; 2344 u16 opcode; 2345 2346 if (op_type == OBJ_OP_DISCARD) { 2347 if (!offset && length == object_size && 2348 (!img_request_layered_test(img_request) || 2349 !obj_request_overlaps_parent(obj_request))) { 2350 opcode = CEPH_OSD_OP_DELETE; 2351 } else if ((offset + length == object_size)) { 2352 opcode = CEPH_OSD_OP_TRUNCATE; 2353 } else { 2354 down_read(&rbd_dev->header_rwsem); 2355 img_end = rbd_dev->header.image_size; 2356 up_read(&rbd_dev->header_rwsem); 2357 2358 if (obj_request->img_offset + length == img_end) 2359 opcode = CEPH_OSD_OP_TRUNCATE; 2360 else 2361 opcode = CEPH_OSD_OP_ZERO; 2362 } 2363 } else if (op_type == OBJ_OP_WRITE) { 2364 opcode = CEPH_OSD_OP_WRITE; 2365 osd_req_op_alloc_hint_init(osd_request, num_ops, 2366 object_size, object_size); 2367 num_ops++; 2368 } else { 2369 opcode = CEPH_OSD_OP_READ; 2370 } 2371 2372 osd_req_op_extent_init(osd_request, num_ops, opcode, offset, length, 2373 0, 0); 2374 if (obj_request->type == OBJ_REQUEST_BIO) 2375 osd_req_op_extent_osd_data_bio(osd_request, num_ops, 2376 obj_request->bio_list, length); 2377 else if (obj_request->type == OBJ_REQUEST_PAGES) 2378 osd_req_op_extent_osd_data_pages(osd_request, num_ops, 2379 obj_request->pages, length, 2380 offset & ~PAGE_MASK, false, false); 2381 2382 /* Discards are also writes */ 2383 if (op_type == OBJ_OP_WRITE || op_type == OBJ_OP_DISCARD) 2384 rbd_osd_req_format_write(obj_request); 2385 else 2386 rbd_osd_req_format_read(obj_request); 2387 } 2388 2389 /* 2390 * Split up an image request into one or more object requests, each 2391 * to a different object. The "type" parameter indicates whether 2392 * "data_desc" is the pointer to the head of a list of bio 2393 * structures, or the base of a page array. In either case this 2394 * function assumes data_desc describes memory sufficient to hold 2395 * all data described by the image request. 2396 */ 2397 static int rbd_img_request_fill(struct rbd_img_request *img_request, 2398 enum obj_request_type type, 2399 void *data_desc) 2400 { 2401 struct rbd_device *rbd_dev = img_request->rbd_dev; 2402 struct rbd_obj_request *obj_request = NULL; 2403 struct rbd_obj_request *next_obj_request; 2404 struct bio *bio_list = NULL; 2405 unsigned int bio_offset = 0; 2406 struct page **pages = NULL; 2407 enum obj_operation_type op_type; 2408 u64 img_offset; 2409 u64 resid; 2410 2411 dout("%s: img %p type %d data_desc %p\n", __func__, img_request, 2412 (int)type, data_desc); 2413 2414 img_offset = img_request->offset; 2415 resid = img_request->length; 2416 rbd_assert(resid > 0); 2417 op_type = rbd_img_request_op_type(img_request); 2418 2419 if (type == OBJ_REQUEST_BIO) { 2420 bio_list = data_desc; 2421 rbd_assert(img_offset == 2422 bio_list->bi_iter.bi_sector << SECTOR_SHIFT); 2423 } else if (type == OBJ_REQUEST_PAGES) { 2424 pages = data_desc; 2425 } 2426 2427 while (resid) { 2428 struct ceph_osd_request *osd_req; 2429 const char *object_name; 2430 u64 offset; 2431 u64 length; 2432 2433 object_name = rbd_segment_name(rbd_dev, img_offset); 2434 if (!object_name) 2435 goto out_unwind; 2436 offset = rbd_segment_offset(rbd_dev, img_offset); 2437 length = rbd_segment_length(rbd_dev, img_offset, resid); 2438 obj_request = rbd_obj_request_create(object_name, 2439 offset, length, type); 2440 /* object request has its own copy of the object name */ 2441 rbd_segment_name_free(object_name); 2442 if (!obj_request) 2443 goto out_unwind; 2444 2445 /* 2446 * set obj_request->img_request before creating the 2447 * osd_request so that it gets the right snapc 2448 */ 2449 rbd_img_obj_request_add(img_request, obj_request); 2450 2451 if (type == OBJ_REQUEST_BIO) { 2452 unsigned int clone_size; 2453 2454 rbd_assert(length <= (u64)UINT_MAX); 2455 clone_size = (unsigned int)length; 2456 obj_request->bio_list = 2457 bio_chain_clone_range(&bio_list, 2458 &bio_offset, 2459 clone_size, 2460 GFP_ATOMIC); 2461 if (!obj_request->bio_list) 2462 goto out_unwind; 2463 } else if (type == OBJ_REQUEST_PAGES) { 2464 unsigned int page_count; 2465 2466 obj_request->pages = pages; 2467 page_count = (u32)calc_pages_for(offset, length); 2468 obj_request->page_count = page_count; 2469 if ((offset + length) & ~PAGE_MASK) 2470 page_count--; /* more on last page */ 2471 pages += page_count; 2472 } 2473 2474 osd_req = rbd_osd_req_create(rbd_dev, op_type, 2475 (op_type == OBJ_OP_WRITE) ? 2 : 1, 2476 obj_request); 2477 if (!osd_req) 2478 goto out_unwind; 2479 2480 obj_request->osd_req = osd_req; 2481 obj_request->callback = rbd_img_obj_callback; 2482 obj_request->img_offset = img_offset; 2483 2484 rbd_img_obj_request_fill(obj_request, osd_req, op_type, 0); 2485 2486 rbd_img_request_get(img_request); 2487 2488 img_offset += length; 2489 resid -= length; 2490 } 2491 2492 return 0; 2493 2494 out_unwind: 2495 for_each_obj_request_safe(img_request, obj_request, next_obj_request) 2496 rbd_img_obj_request_del(img_request, obj_request); 2497 2498 return -ENOMEM; 2499 } 2500 2501 static void 2502 rbd_img_obj_copyup_callback(struct rbd_obj_request *obj_request) 2503 { 2504 struct rbd_img_request *img_request; 2505 struct rbd_device *rbd_dev; 2506 struct page **pages; 2507 u32 page_count; 2508 2509 rbd_assert(obj_request->type == OBJ_REQUEST_BIO || 2510 obj_request->type == OBJ_REQUEST_NODATA); 2511 rbd_assert(obj_request_img_data_test(obj_request)); 2512 img_request = obj_request->img_request; 2513 rbd_assert(img_request); 2514 2515 rbd_dev = img_request->rbd_dev; 2516 rbd_assert(rbd_dev); 2517 2518 pages = obj_request->copyup_pages; 2519 rbd_assert(pages != NULL); 2520 obj_request->copyup_pages = NULL; 2521 page_count = obj_request->copyup_page_count; 2522 rbd_assert(page_count); 2523 obj_request->copyup_page_count = 0; 2524 ceph_release_page_vector(pages, page_count); 2525 2526 /* 2527 * We want the transfer count to reflect the size of the 2528 * original write request. There is no such thing as a 2529 * successful short write, so if the request was successful 2530 * we can just set it to the originally-requested length. 2531 */ 2532 if (!obj_request->result) 2533 obj_request->xferred = obj_request->length; 2534 2535 /* Finish up with the normal image object callback */ 2536 2537 rbd_img_obj_callback(obj_request); 2538 } 2539 2540 static void 2541 rbd_img_obj_parent_read_full_callback(struct rbd_img_request *img_request) 2542 { 2543 struct rbd_obj_request *orig_request; 2544 struct ceph_osd_request *osd_req; 2545 struct ceph_osd_client *osdc; 2546 struct rbd_device *rbd_dev; 2547 struct page **pages; 2548 enum obj_operation_type op_type; 2549 u32 page_count; 2550 int img_result; 2551 u64 parent_length; 2552 2553 rbd_assert(img_request_child_test(img_request)); 2554 2555 /* First get what we need from the image request */ 2556 2557 pages = img_request->copyup_pages; 2558 rbd_assert(pages != NULL); 2559 img_request->copyup_pages = NULL; 2560 page_count = img_request->copyup_page_count; 2561 rbd_assert(page_count); 2562 img_request->copyup_page_count = 0; 2563 2564 orig_request = img_request->obj_request; 2565 rbd_assert(orig_request != NULL); 2566 rbd_assert(obj_request_type_valid(orig_request->type)); 2567 img_result = img_request->result; 2568 parent_length = img_request->length; 2569 rbd_assert(parent_length == img_request->xferred); 2570 rbd_img_request_put(img_request); 2571 2572 rbd_assert(orig_request->img_request); 2573 rbd_dev = orig_request->img_request->rbd_dev; 2574 rbd_assert(rbd_dev); 2575 2576 /* 2577 * If the overlap has become 0 (most likely because the 2578 * image has been flattened) we need to free the pages 2579 * and re-submit the original write request. 2580 */ 2581 if (!rbd_dev->parent_overlap) { 2582 struct ceph_osd_client *osdc; 2583 2584 ceph_release_page_vector(pages, page_count); 2585 osdc = &rbd_dev->rbd_client->client->osdc; 2586 img_result = rbd_obj_request_submit(osdc, orig_request); 2587 if (!img_result) 2588 return; 2589 } 2590 2591 if (img_result) 2592 goto out_err; 2593 2594 /* 2595 * The original osd request is of no use to use any more. 2596 * We need a new one that can hold the three ops in a copyup 2597 * request. Allocate the new copyup osd request for the 2598 * original request, and release the old one. 2599 */ 2600 img_result = -ENOMEM; 2601 osd_req = rbd_osd_req_create_copyup(orig_request); 2602 if (!osd_req) 2603 goto out_err; 2604 rbd_osd_req_destroy(orig_request->osd_req); 2605 orig_request->osd_req = osd_req; 2606 orig_request->copyup_pages = pages; 2607 orig_request->copyup_page_count = page_count; 2608 2609 /* Initialize the copyup op */ 2610 2611 osd_req_op_cls_init(osd_req, 0, CEPH_OSD_OP_CALL, "rbd", "copyup"); 2612 osd_req_op_cls_request_data_pages(osd_req, 0, pages, parent_length, 0, 2613 false, false); 2614 2615 /* Add the other op(s) */ 2616 2617 op_type = rbd_img_request_op_type(orig_request->img_request); 2618 rbd_img_obj_request_fill(orig_request, osd_req, op_type, 1); 2619 2620 /* All set, send it off. */ 2621 2622 orig_request->callback = rbd_img_obj_copyup_callback; 2623 osdc = &rbd_dev->rbd_client->client->osdc; 2624 img_result = rbd_obj_request_submit(osdc, orig_request); 2625 if (!img_result) 2626 return; 2627 out_err: 2628 /* Record the error code and complete the request */ 2629 2630 orig_request->result = img_result; 2631 orig_request->xferred = 0; 2632 obj_request_done_set(orig_request); 2633 rbd_obj_request_complete(orig_request); 2634 } 2635 2636 /* 2637 * Read from the parent image the range of data that covers the 2638 * entire target of the given object request. This is used for 2639 * satisfying a layered image write request when the target of an 2640 * object request from the image request does not exist. 2641 * 2642 * A page array big enough to hold the returned data is allocated 2643 * and supplied to rbd_img_request_fill() as the "data descriptor." 2644 * When the read completes, this page array will be transferred to 2645 * the original object request for the copyup operation. 2646 * 2647 * If an error occurs, record it as the result of the original 2648 * object request and mark it done so it gets completed. 2649 */ 2650 static int rbd_img_obj_parent_read_full(struct rbd_obj_request *obj_request) 2651 { 2652 struct rbd_img_request *img_request = NULL; 2653 struct rbd_img_request *parent_request = NULL; 2654 struct rbd_device *rbd_dev; 2655 u64 img_offset; 2656 u64 length; 2657 struct page **pages = NULL; 2658 u32 page_count; 2659 int result; 2660 2661 rbd_assert(obj_request_img_data_test(obj_request)); 2662 rbd_assert(obj_request_type_valid(obj_request->type)); 2663 2664 img_request = obj_request->img_request; 2665 rbd_assert(img_request != NULL); 2666 rbd_dev = img_request->rbd_dev; 2667 rbd_assert(rbd_dev->parent != NULL); 2668 2669 /* 2670 * Determine the byte range covered by the object in the 2671 * child image to which the original request was to be sent. 2672 */ 2673 img_offset = obj_request->img_offset - obj_request->offset; 2674 length = (u64)1 << rbd_dev->header.obj_order; 2675 2676 /* 2677 * There is no defined parent data beyond the parent 2678 * overlap, so limit what we read at that boundary if 2679 * necessary. 2680 */ 2681 if (img_offset + length > rbd_dev->parent_overlap) { 2682 rbd_assert(img_offset < rbd_dev->parent_overlap); 2683 length = rbd_dev->parent_overlap - img_offset; 2684 } 2685 2686 /* 2687 * Allocate a page array big enough to receive the data read 2688 * from the parent. 2689 */ 2690 page_count = (u32)calc_pages_for(0, length); 2691 pages = ceph_alloc_page_vector(page_count, GFP_KERNEL); 2692 if (IS_ERR(pages)) { 2693 result = PTR_ERR(pages); 2694 pages = NULL; 2695 goto out_err; 2696 } 2697 2698 result = -ENOMEM; 2699 parent_request = rbd_parent_request_create(obj_request, 2700 img_offset, length); 2701 if (!parent_request) 2702 goto out_err; 2703 2704 result = rbd_img_request_fill(parent_request, OBJ_REQUEST_PAGES, pages); 2705 if (result) 2706 goto out_err; 2707 parent_request->copyup_pages = pages; 2708 parent_request->copyup_page_count = page_count; 2709 2710 parent_request->callback = rbd_img_obj_parent_read_full_callback; 2711 result = rbd_img_request_submit(parent_request); 2712 if (!result) 2713 return 0; 2714 2715 parent_request->copyup_pages = NULL; 2716 parent_request->copyup_page_count = 0; 2717 parent_request->obj_request = NULL; 2718 rbd_obj_request_put(obj_request); 2719 out_err: 2720 if (pages) 2721 ceph_release_page_vector(pages, page_count); 2722 if (parent_request) 2723 rbd_img_request_put(parent_request); 2724 obj_request->result = result; 2725 obj_request->xferred = 0; 2726 obj_request_done_set(obj_request); 2727 2728 return result; 2729 } 2730 2731 static void rbd_img_obj_exists_callback(struct rbd_obj_request *obj_request) 2732 { 2733 struct rbd_obj_request *orig_request; 2734 struct rbd_device *rbd_dev; 2735 int result; 2736 2737 rbd_assert(!obj_request_img_data_test(obj_request)); 2738 2739 /* 2740 * All we need from the object request is the original 2741 * request and the result of the STAT op. Grab those, then 2742 * we're done with the request. 2743 */ 2744 orig_request = obj_request->obj_request; 2745 obj_request->obj_request = NULL; 2746 rbd_obj_request_put(orig_request); 2747 rbd_assert(orig_request); 2748 rbd_assert(orig_request->img_request); 2749 2750 result = obj_request->result; 2751 obj_request->result = 0; 2752 2753 dout("%s: obj %p for obj %p result %d %llu/%llu\n", __func__, 2754 obj_request, orig_request, result, 2755 obj_request->xferred, obj_request->length); 2756 rbd_obj_request_put(obj_request); 2757 2758 /* 2759 * If the overlap has become 0 (most likely because the 2760 * image has been flattened) we need to free the pages 2761 * and re-submit the original write request. 2762 */ 2763 rbd_dev = orig_request->img_request->rbd_dev; 2764 if (!rbd_dev->parent_overlap) { 2765 struct ceph_osd_client *osdc; 2766 2767 osdc = &rbd_dev->rbd_client->client->osdc; 2768 result = rbd_obj_request_submit(osdc, orig_request); 2769 if (!result) 2770 return; 2771 } 2772 2773 /* 2774 * Our only purpose here is to determine whether the object 2775 * exists, and we don't want to treat the non-existence as 2776 * an error. If something else comes back, transfer the 2777 * error to the original request and complete it now. 2778 */ 2779 if (!result) { 2780 obj_request_existence_set(orig_request, true); 2781 } else if (result == -ENOENT) { 2782 obj_request_existence_set(orig_request, false); 2783 } else if (result) { 2784 orig_request->result = result; 2785 goto out; 2786 } 2787 2788 /* 2789 * Resubmit the original request now that we have recorded 2790 * whether the target object exists. 2791 */ 2792 orig_request->result = rbd_img_obj_request_submit(orig_request); 2793 out: 2794 if (orig_request->result) 2795 rbd_obj_request_complete(orig_request); 2796 } 2797 2798 static int rbd_img_obj_exists_submit(struct rbd_obj_request *obj_request) 2799 { 2800 struct rbd_obj_request *stat_request; 2801 struct rbd_device *rbd_dev; 2802 struct ceph_osd_client *osdc; 2803 struct page **pages = NULL; 2804 u32 page_count; 2805 size_t size; 2806 int ret; 2807 2808 /* 2809 * The response data for a STAT call consists of: 2810 * le64 length; 2811 * struct { 2812 * le32 tv_sec; 2813 * le32 tv_nsec; 2814 * } mtime; 2815 */ 2816 size = sizeof (__le64) + sizeof (__le32) + sizeof (__le32); 2817 page_count = (u32)calc_pages_for(0, size); 2818 pages = ceph_alloc_page_vector(page_count, GFP_KERNEL); 2819 if (IS_ERR(pages)) 2820 return PTR_ERR(pages); 2821 2822 ret = -ENOMEM; 2823 stat_request = rbd_obj_request_create(obj_request->object_name, 0, 0, 2824 OBJ_REQUEST_PAGES); 2825 if (!stat_request) 2826 goto out; 2827 2828 rbd_obj_request_get(obj_request); 2829 stat_request->obj_request = obj_request; 2830 stat_request->pages = pages; 2831 stat_request->page_count = page_count; 2832 2833 rbd_assert(obj_request->img_request); 2834 rbd_dev = obj_request->img_request->rbd_dev; 2835 stat_request->osd_req = rbd_osd_req_create(rbd_dev, OBJ_OP_READ, 1, 2836 stat_request); 2837 if (!stat_request->osd_req) 2838 goto out; 2839 stat_request->callback = rbd_img_obj_exists_callback; 2840 2841 osd_req_op_init(stat_request->osd_req, 0, CEPH_OSD_OP_STAT); 2842 osd_req_op_raw_data_in_pages(stat_request->osd_req, 0, pages, size, 0, 2843 false, false); 2844 rbd_osd_req_format_read(stat_request); 2845 2846 osdc = &rbd_dev->rbd_client->client->osdc; 2847 ret = rbd_obj_request_submit(osdc, stat_request); 2848 out: 2849 if (ret) 2850 rbd_obj_request_put(obj_request); 2851 2852 return ret; 2853 } 2854 2855 static bool img_obj_request_simple(struct rbd_obj_request *obj_request) 2856 { 2857 struct rbd_img_request *img_request; 2858 struct rbd_device *rbd_dev; 2859 2860 rbd_assert(obj_request_img_data_test(obj_request)); 2861 2862 img_request = obj_request->img_request; 2863 rbd_assert(img_request); 2864 rbd_dev = img_request->rbd_dev; 2865 2866 /* Reads */ 2867 if (!img_request_write_test(img_request) && 2868 !img_request_discard_test(img_request)) 2869 return true; 2870 2871 /* Non-layered writes */ 2872 if (!img_request_layered_test(img_request)) 2873 return true; 2874 2875 /* 2876 * Layered writes outside of the parent overlap range don't 2877 * share any data with the parent. 2878 */ 2879 if (!obj_request_overlaps_parent(obj_request)) 2880 return true; 2881 2882 /* 2883 * Entire-object layered writes - we will overwrite whatever 2884 * parent data there is anyway. 2885 */ 2886 if (!obj_request->offset && 2887 obj_request->length == rbd_obj_bytes(&rbd_dev->header)) 2888 return true; 2889 2890 /* 2891 * If the object is known to already exist, its parent data has 2892 * already been copied. 2893 */ 2894 if (obj_request_known_test(obj_request) && 2895 obj_request_exists_test(obj_request)) 2896 return true; 2897 2898 return false; 2899 } 2900 2901 static int rbd_img_obj_request_submit(struct rbd_obj_request *obj_request) 2902 { 2903 if (img_obj_request_simple(obj_request)) { 2904 struct rbd_device *rbd_dev; 2905 struct ceph_osd_client *osdc; 2906 2907 rbd_dev = obj_request->img_request->rbd_dev; 2908 osdc = &rbd_dev->rbd_client->client->osdc; 2909 2910 return rbd_obj_request_submit(osdc, obj_request); 2911 } 2912 2913 /* 2914 * It's a layered write. The target object might exist but 2915 * we may not know that yet. If we know it doesn't exist, 2916 * start by reading the data for the full target object from 2917 * the parent so we can use it for a copyup to the target. 2918 */ 2919 if (obj_request_known_test(obj_request)) 2920 return rbd_img_obj_parent_read_full(obj_request); 2921 2922 /* We don't know whether the target exists. Go find out. */ 2923 2924 return rbd_img_obj_exists_submit(obj_request); 2925 } 2926 2927 static int rbd_img_request_submit(struct rbd_img_request *img_request) 2928 { 2929 struct rbd_obj_request *obj_request; 2930 struct rbd_obj_request *next_obj_request; 2931 2932 dout("%s: img %p\n", __func__, img_request); 2933 for_each_obj_request_safe(img_request, obj_request, next_obj_request) { 2934 int ret; 2935 2936 ret = rbd_img_obj_request_submit(obj_request); 2937 if (ret) 2938 return ret; 2939 } 2940 2941 return 0; 2942 } 2943 2944 static void rbd_img_parent_read_callback(struct rbd_img_request *img_request) 2945 { 2946 struct rbd_obj_request *obj_request; 2947 struct rbd_device *rbd_dev; 2948 u64 obj_end; 2949 u64 img_xferred; 2950 int img_result; 2951 2952 rbd_assert(img_request_child_test(img_request)); 2953 2954 /* First get what we need from the image request and release it */ 2955 2956 obj_request = img_request->obj_request; 2957 img_xferred = img_request->xferred; 2958 img_result = img_request->result; 2959 rbd_img_request_put(img_request); 2960 2961 /* 2962 * If the overlap has become 0 (most likely because the 2963 * image has been flattened) we need to re-submit the 2964 * original request. 2965 */ 2966 rbd_assert(obj_request); 2967 rbd_assert(obj_request->img_request); 2968 rbd_dev = obj_request->img_request->rbd_dev; 2969 if (!rbd_dev->parent_overlap) { 2970 struct ceph_osd_client *osdc; 2971 2972 osdc = &rbd_dev->rbd_client->client->osdc; 2973 img_result = rbd_obj_request_submit(osdc, obj_request); 2974 if (!img_result) 2975 return; 2976 } 2977 2978 obj_request->result = img_result; 2979 if (obj_request->result) 2980 goto out; 2981 2982 /* 2983 * We need to zero anything beyond the parent overlap 2984 * boundary. Since rbd_img_obj_request_read_callback() 2985 * will zero anything beyond the end of a short read, an 2986 * easy way to do this is to pretend the data from the 2987 * parent came up short--ending at the overlap boundary. 2988 */ 2989 rbd_assert(obj_request->img_offset < U64_MAX - obj_request->length); 2990 obj_end = obj_request->img_offset + obj_request->length; 2991 if (obj_end > rbd_dev->parent_overlap) { 2992 u64 xferred = 0; 2993 2994 if (obj_request->img_offset < rbd_dev->parent_overlap) 2995 xferred = rbd_dev->parent_overlap - 2996 obj_request->img_offset; 2997 2998 obj_request->xferred = min(img_xferred, xferred); 2999 } else { 3000 obj_request->xferred = img_xferred; 3001 } 3002 out: 3003 rbd_img_obj_request_read_callback(obj_request); 3004 rbd_obj_request_complete(obj_request); 3005 } 3006 3007 static void rbd_img_parent_read(struct rbd_obj_request *obj_request) 3008 { 3009 struct rbd_img_request *img_request; 3010 int result; 3011 3012 rbd_assert(obj_request_img_data_test(obj_request)); 3013 rbd_assert(obj_request->img_request != NULL); 3014 rbd_assert(obj_request->result == (s32) -ENOENT); 3015 rbd_assert(obj_request_type_valid(obj_request->type)); 3016 3017 /* rbd_read_finish(obj_request, obj_request->length); */ 3018 img_request = rbd_parent_request_create(obj_request, 3019 obj_request->img_offset, 3020 obj_request->length); 3021 result = -ENOMEM; 3022 if (!img_request) 3023 goto out_err; 3024 3025 if (obj_request->type == OBJ_REQUEST_BIO) 3026 result = rbd_img_request_fill(img_request, OBJ_REQUEST_BIO, 3027 obj_request->bio_list); 3028 else 3029 result = rbd_img_request_fill(img_request, OBJ_REQUEST_PAGES, 3030 obj_request->pages); 3031 if (result) 3032 goto out_err; 3033 3034 img_request->callback = rbd_img_parent_read_callback; 3035 result = rbd_img_request_submit(img_request); 3036 if (result) 3037 goto out_err; 3038 3039 return; 3040 out_err: 3041 if (img_request) 3042 rbd_img_request_put(img_request); 3043 obj_request->result = result; 3044 obj_request->xferred = 0; 3045 obj_request_done_set(obj_request); 3046 } 3047 3048 static int rbd_obj_notify_ack_sync(struct rbd_device *rbd_dev, u64 notify_id) 3049 { 3050 struct rbd_obj_request *obj_request; 3051 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc; 3052 int ret; 3053 3054 obj_request = rbd_obj_request_create(rbd_dev->header_name, 0, 0, 3055 OBJ_REQUEST_NODATA); 3056 if (!obj_request) 3057 return -ENOMEM; 3058 3059 ret = -ENOMEM; 3060 obj_request->osd_req = rbd_osd_req_create(rbd_dev, OBJ_OP_READ, 1, 3061 obj_request); 3062 if (!obj_request->osd_req) 3063 goto out; 3064 3065 osd_req_op_watch_init(obj_request->osd_req, 0, CEPH_OSD_OP_NOTIFY_ACK, 3066 notify_id, 0, 0); 3067 rbd_osd_req_format_read(obj_request); 3068 3069 ret = rbd_obj_request_submit(osdc, obj_request); 3070 if (ret) 3071 goto out; 3072 ret = rbd_obj_request_wait(obj_request); 3073 out: 3074 rbd_obj_request_put(obj_request); 3075 3076 return ret; 3077 } 3078 3079 static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data) 3080 { 3081 struct rbd_device *rbd_dev = (struct rbd_device *)data; 3082 int ret; 3083 3084 if (!rbd_dev) 3085 return; 3086 3087 dout("%s: \"%s\" notify_id %llu opcode %u\n", __func__, 3088 rbd_dev->header_name, (unsigned long long)notify_id, 3089 (unsigned int)opcode); 3090 3091 /* 3092 * Until adequate refresh error handling is in place, there is 3093 * not much we can do here, except warn. 3094 * 3095 * See http://tracker.ceph.com/issues/5040 3096 */ 3097 ret = rbd_dev_refresh(rbd_dev); 3098 if (ret) 3099 rbd_warn(rbd_dev, "refresh failed: %d", ret); 3100 3101 ret = rbd_obj_notify_ack_sync(rbd_dev, notify_id); 3102 if (ret) 3103 rbd_warn(rbd_dev, "notify_ack ret %d", ret); 3104 } 3105 3106 /* 3107 * Send a (un)watch request and wait for the ack. Return a request 3108 * with a ref held on success or error. 3109 */ 3110 static struct rbd_obj_request *rbd_obj_watch_request_helper( 3111 struct rbd_device *rbd_dev, 3112 bool watch) 3113 { 3114 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc; 3115 struct rbd_obj_request *obj_request; 3116 int ret; 3117 3118 obj_request = rbd_obj_request_create(rbd_dev->header_name, 0, 0, 3119 OBJ_REQUEST_NODATA); 3120 if (!obj_request) 3121 return ERR_PTR(-ENOMEM); 3122 3123 obj_request->osd_req = rbd_osd_req_create(rbd_dev, OBJ_OP_WRITE, 1, 3124 obj_request); 3125 if (!obj_request->osd_req) { 3126 ret = -ENOMEM; 3127 goto out; 3128 } 3129 3130 osd_req_op_watch_init(obj_request->osd_req, 0, CEPH_OSD_OP_WATCH, 3131 rbd_dev->watch_event->cookie, 0, watch); 3132 rbd_osd_req_format_write(obj_request); 3133 3134 if (watch) 3135 ceph_osdc_set_request_linger(osdc, obj_request->osd_req); 3136 3137 ret = rbd_obj_request_submit(osdc, obj_request); 3138 if (ret) 3139 goto out; 3140 3141 ret = rbd_obj_request_wait(obj_request); 3142 if (ret) 3143 goto out; 3144 3145 ret = obj_request->result; 3146 if (ret) { 3147 if (watch) 3148 rbd_obj_request_end(obj_request); 3149 goto out; 3150 } 3151 3152 return obj_request; 3153 3154 out: 3155 rbd_obj_request_put(obj_request); 3156 return ERR_PTR(ret); 3157 } 3158 3159 /* 3160 * Initiate a watch request, synchronously. 3161 */ 3162 static int rbd_dev_header_watch_sync(struct rbd_device *rbd_dev) 3163 { 3164 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc; 3165 struct rbd_obj_request *obj_request; 3166 int ret; 3167 3168 rbd_assert(!rbd_dev->watch_event); 3169 rbd_assert(!rbd_dev->watch_request); 3170 3171 ret = ceph_osdc_create_event(osdc, rbd_watch_cb, rbd_dev, 3172 &rbd_dev->watch_event); 3173 if (ret < 0) 3174 return ret; 3175 3176 obj_request = rbd_obj_watch_request_helper(rbd_dev, true); 3177 if (IS_ERR(obj_request)) { 3178 ceph_osdc_cancel_event(rbd_dev->watch_event); 3179 rbd_dev->watch_event = NULL; 3180 return PTR_ERR(obj_request); 3181 } 3182 3183 /* 3184 * A watch request is set to linger, so the underlying osd 3185 * request won't go away until we unregister it. We retain 3186 * a pointer to the object request during that time (in 3187 * rbd_dev->watch_request), so we'll keep a reference to it. 3188 * We'll drop that reference after we've unregistered it in 3189 * rbd_dev_header_unwatch_sync(). 3190 */ 3191 rbd_dev->watch_request = obj_request; 3192 3193 return 0; 3194 } 3195 3196 /* 3197 * Tear down a watch request, synchronously. 3198 */ 3199 static void rbd_dev_header_unwatch_sync(struct rbd_device *rbd_dev) 3200 { 3201 struct rbd_obj_request *obj_request; 3202 3203 rbd_assert(rbd_dev->watch_event); 3204 rbd_assert(rbd_dev->watch_request); 3205 3206 rbd_obj_request_end(rbd_dev->watch_request); 3207 rbd_obj_request_put(rbd_dev->watch_request); 3208 rbd_dev->watch_request = NULL; 3209 3210 obj_request = rbd_obj_watch_request_helper(rbd_dev, false); 3211 if (!IS_ERR(obj_request)) 3212 rbd_obj_request_put(obj_request); 3213 else 3214 rbd_warn(rbd_dev, "unable to tear down watch request (%ld)", 3215 PTR_ERR(obj_request)); 3216 3217 ceph_osdc_cancel_event(rbd_dev->watch_event); 3218 rbd_dev->watch_event = NULL; 3219 } 3220 3221 /* 3222 * Synchronous osd object method call. Returns the number of bytes 3223 * returned in the outbound buffer, or a negative error code. 3224 */ 3225 static int rbd_obj_method_sync(struct rbd_device *rbd_dev, 3226 const char *object_name, 3227 const char *class_name, 3228 const char *method_name, 3229 const void *outbound, 3230 size_t outbound_size, 3231 void *inbound, 3232 size_t inbound_size) 3233 { 3234 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc; 3235 struct rbd_obj_request *obj_request; 3236 struct page **pages; 3237 u32 page_count; 3238 int ret; 3239 3240 /* 3241 * Method calls are ultimately read operations. The result 3242 * should placed into the inbound buffer provided. They 3243 * also supply outbound data--parameters for the object 3244 * method. Currently if this is present it will be a 3245 * snapshot id. 3246 */ 3247 page_count = (u32)calc_pages_for(0, inbound_size); 3248 pages = ceph_alloc_page_vector(page_count, GFP_KERNEL); 3249 if (IS_ERR(pages)) 3250 return PTR_ERR(pages); 3251 3252 ret = -ENOMEM; 3253 obj_request = rbd_obj_request_create(object_name, 0, inbound_size, 3254 OBJ_REQUEST_PAGES); 3255 if (!obj_request) 3256 goto out; 3257 3258 obj_request->pages = pages; 3259 obj_request->page_count = page_count; 3260 3261 obj_request->osd_req = rbd_osd_req_create(rbd_dev, OBJ_OP_READ, 1, 3262 obj_request); 3263 if (!obj_request->osd_req) 3264 goto out; 3265 3266 osd_req_op_cls_init(obj_request->osd_req, 0, CEPH_OSD_OP_CALL, 3267 class_name, method_name); 3268 if (outbound_size) { 3269 struct ceph_pagelist *pagelist; 3270 3271 pagelist = kmalloc(sizeof (*pagelist), GFP_NOFS); 3272 if (!pagelist) 3273 goto out; 3274 3275 ceph_pagelist_init(pagelist); 3276 ceph_pagelist_append(pagelist, outbound, outbound_size); 3277 osd_req_op_cls_request_data_pagelist(obj_request->osd_req, 0, 3278 pagelist); 3279 } 3280 osd_req_op_cls_response_data_pages(obj_request->osd_req, 0, 3281 obj_request->pages, inbound_size, 3282 0, false, false); 3283 rbd_osd_req_format_read(obj_request); 3284 3285 ret = rbd_obj_request_submit(osdc, obj_request); 3286 if (ret) 3287 goto out; 3288 ret = rbd_obj_request_wait(obj_request); 3289 if (ret) 3290 goto out; 3291 3292 ret = obj_request->result; 3293 if (ret < 0) 3294 goto out; 3295 3296 rbd_assert(obj_request->xferred < (u64)INT_MAX); 3297 ret = (int)obj_request->xferred; 3298 ceph_copy_from_page_vector(pages, inbound, 0, obj_request->xferred); 3299 out: 3300 if (obj_request) 3301 rbd_obj_request_put(obj_request); 3302 else 3303 ceph_release_page_vector(pages, page_count); 3304 3305 return ret; 3306 } 3307 3308 static void rbd_handle_request(struct rbd_device *rbd_dev, struct request *rq) 3309 { 3310 struct rbd_img_request *img_request; 3311 struct ceph_snap_context *snapc = NULL; 3312 u64 offset = (u64)blk_rq_pos(rq) << SECTOR_SHIFT; 3313 u64 length = blk_rq_bytes(rq); 3314 enum obj_operation_type op_type; 3315 u64 mapping_size; 3316 int result; 3317 3318 if (rq->cmd_flags & REQ_DISCARD) 3319 op_type = OBJ_OP_DISCARD; 3320 else if (rq->cmd_flags & REQ_WRITE) 3321 op_type = OBJ_OP_WRITE; 3322 else 3323 op_type = OBJ_OP_READ; 3324 3325 /* Ignore/skip any zero-length requests */ 3326 3327 if (!length) { 3328 dout("%s: zero-length request\n", __func__); 3329 result = 0; 3330 goto err_rq; 3331 } 3332 3333 /* Only reads are allowed to a read-only device */ 3334 3335 if (op_type != OBJ_OP_READ) { 3336 if (rbd_dev->mapping.read_only) { 3337 result = -EROFS; 3338 goto err_rq; 3339 } 3340 rbd_assert(rbd_dev->spec->snap_id == CEPH_NOSNAP); 3341 } 3342 3343 /* 3344 * Quit early if the mapped snapshot no longer exists. It's 3345 * still possible the snapshot will have disappeared by the 3346 * time our request arrives at the osd, but there's no sense in 3347 * sending it if we already know. 3348 */ 3349 if (!test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags)) { 3350 dout("request for non-existent snapshot"); 3351 rbd_assert(rbd_dev->spec->snap_id != CEPH_NOSNAP); 3352 result = -ENXIO; 3353 goto err_rq; 3354 } 3355 3356 if (offset && length > U64_MAX - offset + 1) { 3357 rbd_warn(rbd_dev, "bad request range (%llu~%llu)", offset, 3358 length); 3359 result = -EINVAL; 3360 goto err_rq; /* Shouldn't happen */ 3361 } 3362 3363 down_read(&rbd_dev->header_rwsem); 3364 mapping_size = rbd_dev->mapping.size; 3365 if (op_type != OBJ_OP_READ) { 3366 snapc = rbd_dev->header.snapc; 3367 ceph_get_snap_context(snapc); 3368 } 3369 up_read(&rbd_dev->header_rwsem); 3370 3371 if (offset + length > mapping_size) { 3372 rbd_warn(rbd_dev, "beyond EOD (%llu~%llu > %llu)", offset, 3373 length, mapping_size); 3374 result = -EIO; 3375 goto err_rq; 3376 } 3377 3378 img_request = rbd_img_request_create(rbd_dev, offset, length, op_type, 3379 snapc); 3380 if (!img_request) { 3381 result = -ENOMEM; 3382 goto err_rq; 3383 } 3384 img_request->rq = rq; 3385 3386 if (op_type == OBJ_OP_DISCARD) 3387 result = rbd_img_request_fill(img_request, OBJ_REQUEST_NODATA, 3388 NULL); 3389 else 3390 result = rbd_img_request_fill(img_request, OBJ_REQUEST_BIO, 3391 rq->bio); 3392 if (result) 3393 goto err_img_request; 3394 3395 result = rbd_img_request_submit(img_request); 3396 if (result) 3397 goto err_img_request; 3398 3399 return; 3400 3401 err_img_request: 3402 rbd_img_request_put(img_request); 3403 err_rq: 3404 if (result) 3405 rbd_warn(rbd_dev, "%s %llx at %llx result %d", 3406 obj_op_name(op_type), length, offset, result); 3407 if (snapc) 3408 ceph_put_snap_context(snapc); 3409 blk_end_request_all(rq, result); 3410 } 3411 3412 static void rbd_request_workfn(struct work_struct *work) 3413 { 3414 struct rbd_device *rbd_dev = 3415 container_of(work, struct rbd_device, rq_work); 3416 struct request *rq, *next; 3417 LIST_HEAD(requests); 3418 3419 spin_lock_irq(&rbd_dev->lock); /* rq->q->queue_lock */ 3420 list_splice_init(&rbd_dev->rq_queue, &requests); 3421 spin_unlock_irq(&rbd_dev->lock); 3422 3423 list_for_each_entry_safe(rq, next, &requests, queuelist) { 3424 list_del_init(&rq->queuelist); 3425 rbd_handle_request(rbd_dev, rq); 3426 } 3427 } 3428 3429 /* 3430 * Called with q->queue_lock held and interrupts disabled, possibly on 3431 * the way to schedule(). Do not sleep here! 3432 */ 3433 static void rbd_request_fn(struct request_queue *q) 3434 { 3435 struct rbd_device *rbd_dev = q->queuedata; 3436 struct request *rq; 3437 int queued = 0; 3438 3439 rbd_assert(rbd_dev); 3440 3441 while ((rq = blk_fetch_request(q))) { 3442 /* Ignore any non-FS requests that filter through. */ 3443 if (rq->cmd_type != REQ_TYPE_FS) { 3444 dout("%s: non-fs request type %d\n", __func__, 3445 (int) rq->cmd_type); 3446 __blk_end_request_all(rq, 0); 3447 continue; 3448 } 3449 3450 list_add_tail(&rq->queuelist, &rbd_dev->rq_queue); 3451 queued++; 3452 } 3453 3454 if (queued) 3455 queue_work(rbd_dev->rq_wq, &rbd_dev->rq_work); 3456 } 3457 3458 /* 3459 * a queue callback. Makes sure that we don't create a bio that spans across 3460 * multiple osd objects. One exception would be with a single page bios, 3461 * which we handle later at bio_chain_clone_range() 3462 */ 3463 static int rbd_merge_bvec(struct request_queue *q, struct bvec_merge_data *bmd, 3464 struct bio_vec *bvec) 3465 { 3466 struct rbd_device *rbd_dev = q->queuedata; 3467 sector_t sector_offset; 3468 sector_t sectors_per_obj; 3469 sector_t obj_sector_offset; 3470 int ret; 3471 3472 /* 3473 * Find how far into its rbd object the partition-relative 3474 * bio start sector is to offset relative to the enclosing 3475 * device. 3476 */ 3477 sector_offset = get_start_sect(bmd->bi_bdev) + bmd->bi_sector; 3478 sectors_per_obj = 1 << (rbd_dev->header.obj_order - SECTOR_SHIFT); 3479 obj_sector_offset = sector_offset & (sectors_per_obj - 1); 3480 3481 /* 3482 * Compute the number of bytes from that offset to the end 3483 * of the object. Account for what's already used by the bio. 3484 */ 3485 ret = (int) (sectors_per_obj - obj_sector_offset) << SECTOR_SHIFT; 3486 if (ret > bmd->bi_size) 3487 ret -= bmd->bi_size; 3488 else 3489 ret = 0; 3490 3491 /* 3492 * Don't send back more than was asked for. And if the bio 3493 * was empty, let the whole thing through because: "Note 3494 * that a block device *must* allow a single page to be 3495 * added to an empty bio." 3496 */ 3497 rbd_assert(bvec->bv_len <= PAGE_SIZE); 3498 if (ret > (int) bvec->bv_len || !bmd->bi_size) 3499 ret = (int) bvec->bv_len; 3500 3501 return ret; 3502 } 3503 3504 static void rbd_free_disk(struct rbd_device *rbd_dev) 3505 { 3506 struct gendisk *disk = rbd_dev->disk; 3507 3508 if (!disk) 3509 return; 3510 3511 rbd_dev->disk = NULL; 3512 if (disk->flags & GENHD_FL_UP) { 3513 del_gendisk(disk); 3514 if (disk->queue) 3515 blk_cleanup_queue(disk->queue); 3516 } 3517 put_disk(disk); 3518 } 3519 3520 static int rbd_obj_read_sync(struct rbd_device *rbd_dev, 3521 const char *object_name, 3522 u64 offset, u64 length, void *buf) 3523 3524 { 3525 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc; 3526 struct rbd_obj_request *obj_request; 3527 struct page **pages = NULL; 3528 u32 page_count; 3529 size_t size; 3530 int ret; 3531 3532 page_count = (u32) calc_pages_for(offset, length); 3533 pages = ceph_alloc_page_vector(page_count, GFP_KERNEL); 3534 if (IS_ERR(pages)) 3535 ret = PTR_ERR(pages); 3536 3537 ret = -ENOMEM; 3538 obj_request = rbd_obj_request_create(object_name, offset, length, 3539 OBJ_REQUEST_PAGES); 3540 if (!obj_request) 3541 goto out; 3542 3543 obj_request->pages = pages; 3544 obj_request->page_count = page_count; 3545 3546 obj_request->osd_req = rbd_osd_req_create(rbd_dev, OBJ_OP_READ, 1, 3547 obj_request); 3548 if (!obj_request->osd_req) 3549 goto out; 3550 3551 osd_req_op_extent_init(obj_request->osd_req, 0, CEPH_OSD_OP_READ, 3552 offset, length, 0, 0); 3553 osd_req_op_extent_osd_data_pages(obj_request->osd_req, 0, 3554 obj_request->pages, 3555 obj_request->length, 3556 obj_request->offset & ~PAGE_MASK, 3557 false, false); 3558 rbd_osd_req_format_read(obj_request); 3559 3560 ret = rbd_obj_request_submit(osdc, obj_request); 3561 if (ret) 3562 goto out; 3563 ret = rbd_obj_request_wait(obj_request); 3564 if (ret) 3565 goto out; 3566 3567 ret = obj_request->result; 3568 if (ret < 0) 3569 goto out; 3570 3571 rbd_assert(obj_request->xferred <= (u64) SIZE_MAX); 3572 size = (size_t) obj_request->xferred; 3573 ceph_copy_from_page_vector(pages, buf, 0, size); 3574 rbd_assert(size <= (size_t)INT_MAX); 3575 ret = (int)size; 3576 out: 3577 if (obj_request) 3578 rbd_obj_request_put(obj_request); 3579 else 3580 ceph_release_page_vector(pages, page_count); 3581 3582 return ret; 3583 } 3584 3585 /* 3586 * Read the complete header for the given rbd device. On successful 3587 * return, the rbd_dev->header field will contain up-to-date 3588 * information about the image. 3589 */ 3590 static int rbd_dev_v1_header_info(struct rbd_device *rbd_dev) 3591 { 3592 struct rbd_image_header_ondisk *ondisk = NULL; 3593 u32 snap_count = 0; 3594 u64 names_size = 0; 3595 u32 want_count; 3596 int ret; 3597 3598 /* 3599 * The complete header will include an array of its 64-bit 3600 * snapshot ids, followed by the names of those snapshots as 3601 * a contiguous block of NUL-terminated strings. Note that 3602 * the number of snapshots could change by the time we read 3603 * it in, in which case we re-read it. 3604 */ 3605 do { 3606 size_t size; 3607 3608 kfree(ondisk); 3609 3610 size = sizeof (*ondisk); 3611 size += snap_count * sizeof (struct rbd_image_snap_ondisk); 3612 size += names_size; 3613 ondisk = kmalloc(size, GFP_KERNEL); 3614 if (!ondisk) 3615 return -ENOMEM; 3616 3617 ret = rbd_obj_read_sync(rbd_dev, rbd_dev->header_name, 3618 0, size, ondisk); 3619 if (ret < 0) 3620 goto out; 3621 if ((size_t)ret < size) { 3622 ret = -ENXIO; 3623 rbd_warn(rbd_dev, "short header read (want %zd got %d)", 3624 size, ret); 3625 goto out; 3626 } 3627 if (!rbd_dev_ondisk_valid(ondisk)) { 3628 ret = -ENXIO; 3629 rbd_warn(rbd_dev, "invalid header"); 3630 goto out; 3631 } 3632 3633 names_size = le64_to_cpu(ondisk->snap_names_len); 3634 want_count = snap_count; 3635 snap_count = le32_to_cpu(ondisk->snap_count); 3636 } while (snap_count != want_count); 3637 3638 ret = rbd_header_from_disk(rbd_dev, ondisk); 3639 out: 3640 kfree(ondisk); 3641 3642 return ret; 3643 } 3644 3645 /* 3646 * Clear the rbd device's EXISTS flag if the snapshot it's mapped to 3647 * has disappeared from the (just updated) snapshot context. 3648 */ 3649 static void rbd_exists_validate(struct rbd_device *rbd_dev) 3650 { 3651 u64 snap_id; 3652 3653 if (!test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags)) 3654 return; 3655 3656 snap_id = rbd_dev->spec->snap_id; 3657 if (snap_id == CEPH_NOSNAP) 3658 return; 3659 3660 if (rbd_dev_snap_index(rbd_dev, snap_id) == BAD_SNAP_INDEX) 3661 clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags); 3662 } 3663 3664 static void rbd_dev_update_size(struct rbd_device *rbd_dev) 3665 { 3666 sector_t size; 3667 bool removing; 3668 3669 /* 3670 * Don't hold the lock while doing disk operations, 3671 * or lock ordering will conflict with the bdev mutex via: 3672 * rbd_add() -> blkdev_get() -> rbd_open() 3673 */ 3674 spin_lock_irq(&rbd_dev->lock); 3675 removing = test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags); 3676 spin_unlock_irq(&rbd_dev->lock); 3677 /* 3678 * If the device is being removed, rbd_dev->disk has 3679 * been destroyed, so don't try to update its size 3680 */ 3681 if (!removing) { 3682 size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE; 3683 dout("setting size to %llu sectors", (unsigned long long)size); 3684 set_capacity(rbd_dev->disk, size); 3685 revalidate_disk(rbd_dev->disk); 3686 } 3687 } 3688 3689 static int rbd_dev_refresh(struct rbd_device *rbd_dev) 3690 { 3691 u64 mapping_size; 3692 int ret; 3693 3694 down_write(&rbd_dev->header_rwsem); 3695 mapping_size = rbd_dev->mapping.size; 3696 3697 ret = rbd_dev_header_info(rbd_dev); 3698 if (ret) 3699 return ret; 3700 3701 /* 3702 * If there is a parent, see if it has disappeared due to the 3703 * mapped image getting flattened. 3704 */ 3705 if (rbd_dev->parent) { 3706 ret = rbd_dev_v2_parent_info(rbd_dev); 3707 if (ret) 3708 return ret; 3709 } 3710 3711 if (rbd_dev->spec->snap_id == CEPH_NOSNAP) { 3712 if (rbd_dev->mapping.size != rbd_dev->header.image_size) 3713 rbd_dev->mapping.size = rbd_dev->header.image_size; 3714 } else { 3715 /* validate mapped snapshot's EXISTS flag */ 3716 rbd_exists_validate(rbd_dev); 3717 } 3718 3719 up_write(&rbd_dev->header_rwsem); 3720 3721 if (mapping_size != rbd_dev->mapping.size) 3722 rbd_dev_update_size(rbd_dev); 3723 3724 return 0; 3725 } 3726 3727 static int rbd_init_disk(struct rbd_device *rbd_dev) 3728 { 3729 struct gendisk *disk; 3730 struct request_queue *q; 3731 u64 segment_size; 3732 3733 /* create gendisk info */ 3734 disk = alloc_disk(single_major ? 3735 (1 << RBD_SINGLE_MAJOR_PART_SHIFT) : 3736 RBD_MINORS_PER_MAJOR); 3737 if (!disk) 3738 return -ENOMEM; 3739 3740 snprintf(disk->disk_name, sizeof(disk->disk_name), RBD_DRV_NAME "%d", 3741 rbd_dev->dev_id); 3742 disk->major = rbd_dev->major; 3743 disk->first_minor = rbd_dev->minor; 3744 if (single_major) 3745 disk->flags |= GENHD_FL_EXT_DEVT; 3746 disk->fops = &rbd_bd_ops; 3747 disk->private_data = rbd_dev; 3748 3749 q = blk_init_queue(rbd_request_fn, &rbd_dev->lock); 3750 if (!q) 3751 goto out_disk; 3752 3753 /* We use the default size, but let's be explicit about it. */ 3754 blk_queue_physical_block_size(q, SECTOR_SIZE); 3755 3756 /* set io sizes to object size */ 3757 segment_size = rbd_obj_bytes(&rbd_dev->header); 3758 blk_queue_max_hw_sectors(q, segment_size / SECTOR_SIZE); 3759 blk_queue_max_segment_size(q, segment_size); 3760 blk_queue_io_min(q, segment_size); 3761 blk_queue_io_opt(q, segment_size); 3762 3763 /* enable the discard support */ 3764 queue_flag_set_unlocked(QUEUE_FLAG_DISCARD, q); 3765 q->limits.discard_granularity = segment_size; 3766 q->limits.discard_alignment = segment_size; 3767 q->limits.max_discard_sectors = segment_size / SECTOR_SIZE; 3768 q->limits.discard_zeroes_data = 1; 3769 3770 blk_queue_merge_bvec(q, rbd_merge_bvec); 3771 disk->queue = q; 3772 3773 q->queuedata = rbd_dev; 3774 3775 rbd_dev->disk = disk; 3776 3777 return 0; 3778 out_disk: 3779 put_disk(disk); 3780 3781 return -ENOMEM; 3782 } 3783 3784 /* 3785 sysfs 3786 */ 3787 3788 static struct rbd_device *dev_to_rbd_dev(struct device *dev) 3789 { 3790 return container_of(dev, struct rbd_device, dev); 3791 } 3792 3793 static ssize_t rbd_size_show(struct device *dev, 3794 struct device_attribute *attr, char *buf) 3795 { 3796 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev); 3797 3798 return sprintf(buf, "%llu\n", 3799 (unsigned long long)rbd_dev->mapping.size); 3800 } 3801 3802 /* 3803 * Note this shows the features for whatever's mapped, which is not 3804 * necessarily the base image. 3805 */ 3806 static ssize_t rbd_features_show(struct device *dev, 3807 struct device_attribute *attr, char *buf) 3808 { 3809 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev); 3810 3811 return sprintf(buf, "0x%016llx\n", 3812 (unsigned long long)rbd_dev->mapping.features); 3813 } 3814 3815 static ssize_t rbd_major_show(struct device *dev, 3816 struct device_attribute *attr, char *buf) 3817 { 3818 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev); 3819 3820 if (rbd_dev->major) 3821 return sprintf(buf, "%d\n", rbd_dev->major); 3822 3823 return sprintf(buf, "(none)\n"); 3824 } 3825 3826 static ssize_t rbd_minor_show(struct device *dev, 3827 struct device_attribute *attr, char *buf) 3828 { 3829 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev); 3830 3831 return sprintf(buf, "%d\n", rbd_dev->minor); 3832 } 3833 3834 static ssize_t rbd_client_id_show(struct device *dev, 3835 struct device_attribute *attr, char *buf) 3836 { 3837 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev); 3838 3839 return sprintf(buf, "client%lld\n", 3840 ceph_client_id(rbd_dev->rbd_client->client)); 3841 } 3842 3843 static ssize_t rbd_pool_show(struct device *dev, 3844 struct device_attribute *attr, char *buf) 3845 { 3846 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev); 3847 3848 return sprintf(buf, "%s\n", rbd_dev->spec->pool_name); 3849 } 3850 3851 static ssize_t rbd_pool_id_show(struct device *dev, 3852 struct device_attribute *attr, char *buf) 3853 { 3854 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev); 3855 3856 return sprintf(buf, "%llu\n", 3857 (unsigned long long) rbd_dev->spec->pool_id); 3858 } 3859 3860 static ssize_t rbd_name_show(struct device *dev, 3861 struct device_attribute *attr, char *buf) 3862 { 3863 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev); 3864 3865 if (rbd_dev->spec->image_name) 3866 return sprintf(buf, "%s\n", rbd_dev->spec->image_name); 3867 3868 return sprintf(buf, "(unknown)\n"); 3869 } 3870 3871 static ssize_t rbd_image_id_show(struct device *dev, 3872 struct device_attribute *attr, char *buf) 3873 { 3874 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev); 3875 3876 return sprintf(buf, "%s\n", rbd_dev->spec->image_id); 3877 } 3878 3879 /* 3880 * Shows the name of the currently-mapped snapshot (or 3881 * RBD_SNAP_HEAD_NAME for the base image). 3882 */ 3883 static ssize_t rbd_snap_show(struct device *dev, 3884 struct device_attribute *attr, 3885 char *buf) 3886 { 3887 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev); 3888 3889 return sprintf(buf, "%s\n", rbd_dev->spec->snap_name); 3890 } 3891 3892 /* 3893 * For a v2 image, shows the chain of parent images, separated by empty 3894 * lines. For v1 images or if there is no parent, shows "(no parent 3895 * image)". 3896 */ 3897 static ssize_t rbd_parent_show(struct device *dev, 3898 struct device_attribute *attr, 3899 char *buf) 3900 { 3901 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev); 3902 ssize_t count = 0; 3903 3904 if (!rbd_dev->parent) 3905 return sprintf(buf, "(no parent image)\n"); 3906 3907 for ( ; rbd_dev->parent; rbd_dev = rbd_dev->parent) { 3908 struct rbd_spec *spec = rbd_dev->parent_spec; 3909 3910 count += sprintf(&buf[count], "%s" 3911 "pool_id %llu\npool_name %s\n" 3912 "image_id %s\nimage_name %s\n" 3913 "snap_id %llu\nsnap_name %s\n" 3914 "overlap %llu\n", 3915 !count ? "" : "\n", /* first? */ 3916 spec->pool_id, spec->pool_name, 3917 spec->image_id, spec->image_name ?: "(unknown)", 3918 spec->snap_id, spec->snap_name, 3919 rbd_dev->parent_overlap); 3920 } 3921 3922 return count; 3923 } 3924 3925 static ssize_t rbd_image_refresh(struct device *dev, 3926 struct device_attribute *attr, 3927 const char *buf, 3928 size_t size) 3929 { 3930 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev); 3931 int ret; 3932 3933 ret = rbd_dev_refresh(rbd_dev); 3934 if (ret) 3935 return ret; 3936 3937 return size; 3938 } 3939 3940 static DEVICE_ATTR(size, S_IRUGO, rbd_size_show, NULL); 3941 static DEVICE_ATTR(features, S_IRUGO, rbd_features_show, NULL); 3942 static DEVICE_ATTR(major, S_IRUGO, rbd_major_show, NULL); 3943 static DEVICE_ATTR(minor, S_IRUGO, rbd_minor_show, NULL); 3944 static DEVICE_ATTR(client_id, S_IRUGO, rbd_client_id_show, NULL); 3945 static DEVICE_ATTR(pool, S_IRUGO, rbd_pool_show, NULL); 3946 static DEVICE_ATTR(pool_id, S_IRUGO, rbd_pool_id_show, NULL); 3947 static DEVICE_ATTR(name, S_IRUGO, rbd_name_show, NULL); 3948 static DEVICE_ATTR(image_id, S_IRUGO, rbd_image_id_show, NULL); 3949 static DEVICE_ATTR(refresh, S_IWUSR, NULL, rbd_image_refresh); 3950 static DEVICE_ATTR(current_snap, S_IRUGO, rbd_snap_show, NULL); 3951 static DEVICE_ATTR(parent, S_IRUGO, rbd_parent_show, NULL); 3952 3953 static struct attribute *rbd_attrs[] = { 3954 &dev_attr_size.attr, 3955 &dev_attr_features.attr, 3956 &dev_attr_major.attr, 3957 &dev_attr_minor.attr, 3958 &dev_attr_client_id.attr, 3959 &dev_attr_pool.attr, 3960 &dev_attr_pool_id.attr, 3961 &dev_attr_name.attr, 3962 &dev_attr_image_id.attr, 3963 &dev_attr_current_snap.attr, 3964 &dev_attr_parent.attr, 3965 &dev_attr_refresh.attr, 3966 NULL 3967 }; 3968 3969 static struct attribute_group rbd_attr_group = { 3970 .attrs = rbd_attrs, 3971 }; 3972 3973 static const struct attribute_group *rbd_attr_groups[] = { 3974 &rbd_attr_group, 3975 NULL 3976 }; 3977 3978 static void rbd_sysfs_dev_release(struct device *dev) 3979 { 3980 } 3981 3982 static struct device_type rbd_device_type = { 3983 .name = "rbd", 3984 .groups = rbd_attr_groups, 3985 .release = rbd_sysfs_dev_release, 3986 }; 3987 3988 static struct rbd_spec *rbd_spec_get(struct rbd_spec *spec) 3989 { 3990 kref_get(&spec->kref); 3991 3992 return spec; 3993 } 3994 3995 static void rbd_spec_free(struct kref *kref); 3996 static void rbd_spec_put(struct rbd_spec *spec) 3997 { 3998 if (spec) 3999 kref_put(&spec->kref, rbd_spec_free); 4000 } 4001 4002 static struct rbd_spec *rbd_spec_alloc(void) 4003 { 4004 struct rbd_spec *spec; 4005 4006 spec = kzalloc(sizeof (*spec), GFP_KERNEL); 4007 if (!spec) 4008 return NULL; 4009 4010 spec->pool_id = CEPH_NOPOOL; 4011 spec->snap_id = CEPH_NOSNAP; 4012 kref_init(&spec->kref); 4013 4014 return spec; 4015 } 4016 4017 static void rbd_spec_free(struct kref *kref) 4018 { 4019 struct rbd_spec *spec = container_of(kref, struct rbd_spec, kref); 4020 4021 kfree(spec->pool_name); 4022 kfree(spec->image_id); 4023 kfree(spec->image_name); 4024 kfree(spec->snap_name); 4025 kfree(spec); 4026 } 4027 4028 static struct rbd_device *rbd_dev_create(struct rbd_client *rbdc, 4029 struct rbd_spec *spec) 4030 { 4031 struct rbd_device *rbd_dev; 4032 4033 rbd_dev = kzalloc(sizeof (*rbd_dev), GFP_KERNEL); 4034 if (!rbd_dev) 4035 return NULL; 4036 4037 spin_lock_init(&rbd_dev->lock); 4038 INIT_LIST_HEAD(&rbd_dev->rq_queue); 4039 INIT_WORK(&rbd_dev->rq_work, rbd_request_workfn); 4040 rbd_dev->flags = 0; 4041 atomic_set(&rbd_dev->parent_ref, 0); 4042 INIT_LIST_HEAD(&rbd_dev->node); 4043 init_rwsem(&rbd_dev->header_rwsem); 4044 4045 rbd_dev->spec = spec; 4046 rbd_dev->rbd_client = rbdc; 4047 4048 /* Initialize the layout used for all rbd requests */ 4049 4050 rbd_dev->layout.fl_stripe_unit = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER); 4051 rbd_dev->layout.fl_stripe_count = cpu_to_le32(1); 4052 rbd_dev->layout.fl_object_size = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER); 4053 rbd_dev->layout.fl_pg_pool = cpu_to_le32((u32) spec->pool_id); 4054 4055 return rbd_dev; 4056 } 4057 4058 static void rbd_dev_destroy(struct rbd_device *rbd_dev) 4059 { 4060 rbd_put_client(rbd_dev->rbd_client); 4061 rbd_spec_put(rbd_dev->spec); 4062 kfree(rbd_dev); 4063 } 4064 4065 /* 4066 * Get the size and object order for an image snapshot, or if 4067 * snap_id is CEPH_NOSNAP, gets this information for the base 4068 * image. 4069 */ 4070 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id, 4071 u8 *order, u64 *snap_size) 4072 { 4073 __le64 snapid = cpu_to_le64(snap_id); 4074 int ret; 4075 struct { 4076 u8 order; 4077 __le64 size; 4078 } __attribute__ ((packed)) size_buf = { 0 }; 4079 4080 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name, 4081 "rbd", "get_size", 4082 &snapid, sizeof (snapid), 4083 &size_buf, sizeof (size_buf)); 4084 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret); 4085 if (ret < 0) 4086 return ret; 4087 if (ret < sizeof (size_buf)) 4088 return -ERANGE; 4089 4090 if (order) { 4091 *order = size_buf.order; 4092 dout(" order %u", (unsigned int)*order); 4093 } 4094 *snap_size = le64_to_cpu(size_buf.size); 4095 4096 dout(" snap_id 0x%016llx snap_size = %llu\n", 4097 (unsigned long long)snap_id, 4098 (unsigned long long)*snap_size); 4099 4100 return 0; 4101 } 4102 4103 static int rbd_dev_v2_image_size(struct rbd_device *rbd_dev) 4104 { 4105 return _rbd_dev_v2_snap_size(rbd_dev, CEPH_NOSNAP, 4106 &rbd_dev->header.obj_order, 4107 &rbd_dev->header.image_size); 4108 } 4109 4110 static int rbd_dev_v2_object_prefix(struct rbd_device *rbd_dev) 4111 { 4112 void *reply_buf; 4113 int ret; 4114 void *p; 4115 4116 reply_buf = kzalloc(RBD_OBJ_PREFIX_LEN_MAX, GFP_KERNEL); 4117 if (!reply_buf) 4118 return -ENOMEM; 4119 4120 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name, 4121 "rbd", "get_object_prefix", NULL, 0, 4122 reply_buf, RBD_OBJ_PREFIX_LEN_MAX); 4123 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret); 4124 if (ret < 0) 4125 goto out; 4126 4127 p = reply_buf; 4128 rbd_dev->header.object_prefix = ceph_extract_encoded_string(&p, 4129 p + ret, NULL, GFP_NOIO); 4130 ret = 0; 4131 4132 if (IS_ERR(rbd_dev->header.object_prefix)) { 4133 ret = PTR_ERR(rbd_dev->header.object_prefix); 4134 rbd_dev->header.object_prefix = NULL; 4135 } else { 4136 dout(" object_prefix = %s\n", rbd_dev->header.object_prefix); 4137 } 4138 out: 4139 kfree(reply_buf); 4140 4141 return ret; 4142 } 4143 4144 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id, 4145 u64 *snap_features) 4146 { 4147 __le64 snapid = cpu_to_le64(snap_id); 4148 struct { 4149 __le64 features; 4150 __le64 incompat; 4151 } __attribute__ ((packed)) features_buf = { 0 }; 4152 u64 incompat; 4153 int ret; 4154 4155 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name, 4156 "rbd", "get_features", 4157 &snapid, sizeof (snapid), 4158 &features_buf, sizeof (features_buf)); 4159 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret); 4160 if (ret < 0) 4161 return ret; 4162 if (ret < sizeof (features_buf)) 4163 return -ERANGE; 4164 4165 incompat = le64_to_cpu(features_buf.incompat); 4166 if (incompat & ~RBD_FEATURES_SUPPORTED) 4167 return -ENXIO; 4168 4169 *snap_features = le64_to_cpu(features_buf.features); 4170 4171 dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n", 4172 (unsigned long long)snap_id, 4173 (unsigned long long)*snap_features, 4174 (unsigned long long)le64_to_cpu(features_buf.incompat)); 4175 4176 return 0; 4177 } 4178 4179 static int rbd_dev_v2_features(struct rbd_device *rbd_dev) 4180 { 4181 return _rbd_dev_v2_snap_features(rbd_dev, CEPH_NOSNAP, 4182 &rbd_dev->header.features); 4183 } 4184 4185 static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev) 4186 { 4187 struct rbd_spec *parent_spec; 4188 size_t size; 4189 void *reply_buf = NULL; 4190 __le64 snapid; 4191 void *p; 4192 void *end; 4193 u64 pool_id; 4194 char *image_id; 4195 u64 snap_id; 4196 u64 overlap; 4197 int ret; 4198 4199 parent_spec = rbd_spec_alloc(); 4200 if (!parent_spec) 4201 return -ENOMEM; 4202 4203 size = sizeof (__le64) + /* pool_id */ 4204 sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX + /* image_id */ 4205 sizeof (__le64) + /* snap_id */ 4206 sizeof (__le64); /* overlap */ 4207 reply_buf = kmalloc(size, GFP_KERNEL); 4208 if (!reply_buf) { 4209 ret = -ENOMEM; 4210 goto out_err; 4211 } 4212 4213 snapid = cpu_to_le64(rbd_dev->spec->snap_id); 4214 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name, 4215 "rbd", "get_parent", 4216 &snapid, sizeof (snapid), 4217 reply_buf, size); 4218 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret); 4219 if (ret < 0) 4220 goto out_err; 4221 4222 p = reply_buf; 4223 end = reply_buf + ret; 4224 ret = -ERANGE; 4225 ceph_decode_64_safe(&p, end, pool_id, out_err); 4226 if (pool_id == CEPH_NOPOOL) { 4227 /* 4228 * Either the parent never existed, or we have 4229 * record of it but the image got flattened so it no 4230 * longer has a parent. When the parent of a 4231 * layered image disappears we immediately set the 4232 * overlap to 0. The effect of this is that all new 4233 * requests will be treated as if the image had no 4234 * parent. 4235 */ 4236 if (rbd_dev->parent_overlap) { 4237 rbd_dev->parent_overlap = 0; 4238 smp_mb(); 4239 rbd_dev_parent_put(rbd_dev); 4240 pr_info("%s: clone image has been flattened\n", 4241 rbd_dev->disk->disk_name); 4242 } 4243 4244 goto out; /* No parent? No problem. */ 4245 } 4246 4247 /* The ceph file layout needs to fit pool id in 32 bits */ 4248 4249 ret = -EIO; 4250 if (pool_id > (u64)U32_MAX) { 4251 rbd_warn(NULL, "parent pool id too large (%llu > %u)", 4252 (unsigned long long)pool_id, U32_MAX); 4253 goto out_err; 4254 } 4255 4256 image_id = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL); 4257 if (IS_ERR(image_id)) { 4258 ret = PTR_ERR(image_id); 4259 goto out_err; 4260 } 4261 ceph_decode_64_safe(&p, end, snap_id, out_err); 4262 ceph_decode_64_safe(&p, end, overlap, out_err); 4263 4264 /* 4265 * The parent won't change (except when the clone is 4266 * flattened, already handled that). So we only need to 4267 * record the parent spec we have not already done so. 4268 */ 4269 if (!rbd_dev->parent_spec) { 4270 parent_spec->pool_id = pool_id; 4271 parent_spec->image_id = image_id; 4272 parent_spec->snap_id = snap_id; 4273 rbd_dev->parent_spec = parent_spec; 4274 parent_spec = NULL; /* rbd_dev now owns this */ 4275 } else { 4276 kfree(image_id); 4277 } 4278 4279 /* 4280 * We always update the parent overlap. If it's zero we 4281 * treat it specially. 4282 */ 4283 rbd_dev->parent_overlap = overlap; 4284 smp_mb(); 4285 if (!overlap) { 4286 4287 /* A null parent_spec indicates it's the initial probe */ 4288 4289 if (parent_spec) { 4290 /* 4291 * The overlap has become zero, so the clone 4292 * must have been resized down to 0 at some 4293 * point. Treat this the same as a flatten. 4294 */ 4295 rbd_dev_parent_put(rbd_dev); 4296 pr_info("%s: clone image now standalone\n", 4297 rbd_dev->disk->disk_name); 4298 } else { 4299 /* 4300 * For the initial probe, if we find the 4301 * overlap is zero we just pretend there was 4302 * no parent image. 4303 */ 4304 rbd_warn(rbd_dev, "ignoring parent with overlap 0"); 4305 } 4306 } 4307 out: 4308 ret = 0; 4309 out_err: 4310 kfree(reply_buf); 4311 rbd_spec_put(parent_spec); 4312 4313 return ret; 4314 } 4315 4316 static int rbd_dev_v2_striping_info(struct rbd_device *rbd_dev) 4317 { 4318 struct { 4319 __le64 stripe_unit; 4320 __le64 stripe_count; 4321 } __attribute__ ((packed)) striping_info_buf = { 0 }; 4322 size_t size = sizeof (striping_info_buf); 4323 void *p; 4324 u64 obj_size; 4325 u64 stripe_unit; 4326 u64 stripe_count; 4327 int ret; 4328 4329 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name, 4330 "rbd", "get_stripe_unit_count", NULL, 0, 4331 (char *)&striping_info_buf, size); 4332 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret); 4333 if (ret < 0) 4334 return ret; 4335 if (ret < size) 4336 return -ERANGE; 4337 4338 /* 4339 * We don't actually support the "fancy striping" feature 4340 * (STRIPINGV2) yet, but if the striping sizes are the 4341 * defaults the behavior is the same as before. So find 4342 * out, and only fail if the image has non-default values. 4343 */ 4344 ret = -EINVAL; 4345 obj_size = (u64)1 << rbd_dev->header.obj_order; 4346 p = &striping_info_buf; 4347 stripe_unit = ceph_decode_64(&p); 4348 if (stripe_unit != obj_size) { 4349 rbd_warn(rbd_dev, "unsupported stripe unit " 4350 "(got %llu want %llu)", 4351 stripe_unit, obj_size); 4352 return -EINVAL; 4353 } 4354 stripe_count = ceph_decode_64(&p); 4355 if (stripe_count != 1) { 4356 rbd_warn(rbd_dev, "unsupported stripe count " 4357 "(got %llu want 1)", stripe_count); 4358 return -EINVAL; 4359 } 4360 rbd_dev->header.stripe_unit = stripe_unit; 4361 rbd_dev->header.stripe_count = stripe_count; 4362 4363 return 0; 4364 } 4365 4366 static char *rbd_dev_image_name(struct rbd_device *rbd_dev) 4367 { 4368 size_t image_id_size; 4369 char *image_id; 4370 void *p; 4371 void *end; 4372 size_t size; 4373 void *reply_buf = NULL; 4374 size_t len = 0; 4375 char *image_name = NULL; 4376 int ret; 4377 4378 rbd_assert(!rbd_dev->spec->image_name); 4379 4380 len = strlen(rbd_dev->spec->image_id); 4381 image_id_size = sizeof (__le32) + len; 4382 image_id = kmalloc(image_id_size, GFP_KERNEL); 4383 if (!image_id) 4384 return NULL; 4385 4386 p = image_id; 4387 end = image_id + image_id_size; 4388 ceph_encode_string(&p, end, rbd_dev->spec->image_id, (u32)len); 4389 4390 size = sizeof (__le32) + RBD_IMAGE_NAME_LEN_MAX; 4391 reply_buf = kmalloc(size, GFP_KERNEL); 4392 if (!reply_buf) 4393 goto out; 4394 4395 ret = rbd_obj_method_sync(rbd_dev, RBD_DIRECTORY, 4396 "rbd", "dir_get_name", 4397 image_id, image_id_size, 4398 reply_buf, size); 4399 if (ret < 0) 4400 goto out; 4401 p = reply_buf; 4402 end = reply_buf + ret; 4403 4404 image_name = ceph_extract_encoded_string(&p, end, &len, GFP_KERNEL); 4405 if (IS_ERR(image_name)) 4406 image_name = NULL; 4407 else 4408 dout("%s: name is %s len is %zd\n", __func__, image_name, len); 4409 out: 4410 kfree(reply_buf); 4411 kfree(image_id); 4412 4413 return image_name; 4414 } 4415 4416 static u64 rbd_v1_snap_id_by_name(struct rbd_device *rbd_dev, const char *name) 4417 { 4418 struct ceph_snap_context *snapc = rbd_dev->header.snapc; 4419 const char *snap_name; 4420 u32 which = 0; 4421 4422 /* Skip over names until we find the one we are looking for */ 4423 4424 snap_name = rbd_dev->header.snap_names; 4425 while (which < snapc->num_snaps) { 4426 if (!strcmp(name, snap_name)) 4427 return snapc->snaps[which]; 4428 snap_name += strlen(snap_name) + 1; 4429 which++; 4430 } 4431 return CEPH_NOSNAP; 4432 } 4433 4434 static u64 rbd_v2_snap_id_by_name(struct rbd_device *rbd_dev, const char *name) 4435 { 4436 struct ceph_snap_context *snapc = rbd_dev->header.snapc; 4437 u32 which; 4438 bool found = false; 4439 u64 snap_id; 4440 4441 for (which = 0; !found && which < snapc->num_snaps; which++) { 4442 const char *snap_name; 4443 4444 snap_id = snapc->snaps[which]; 4445 snap_name = rbd_dev_v2_snap_name(rbd_dev, snap_id); 4446 if (IS_ERR(snap_name)) { 4447 /* ignore no-longer existing snapshots */ 4448 if (PTR_ERR(snap_name) == -ENOENT) 4449 continue; 4450 else 4451 break; 4452 } 4453 found = !strcmp(name, snap_name); 4454 kfree(snap_name); 4455 } 4456 return found ? snap_id : CEPH_NOSNAP; 4457 } 4458 4459 /* 4460 * Assumes name is never RBD_SNAP_HEAD_NAME; returns CEPH_NOSNAP if 4461 * no snapshot by that name is found, or if an error occurs. 4462 */ 4463 static u64 rbd_snap_id_by_name(struct rbd_device *rbd_dev, const char *name) 4464 { 4465 if (rbd_dev->image_format == 1) 4466 return rbd_v1_snap_id_by_name(rbd_dev, name); 4467 4468 return rbd_v2_snap_id_by_name(rbd_dev, name); 4469 } 4470 4471 /* 4472 * An image being mapped will have everything but the snap id. 4473 */ 4474 static int rbd_spec_fill_snap_id(struct rbd_device *rbd_dev) 4475 { 4476 struct rbd_spec *spec = rbd_dev->spec; 4477 4478 rbd_assert(spec->pool_id != CEPH_NOPOOL && spec->pool_name); 4479 rbd_assert(spec->image_id && spec->image_name); 4480 rbd_assert(spec->snap_name); 4481 4482 if (strcmp(spec->snap_name, RBD_SNAP_HEAD_NAME)) { 4483 u64 snap_id; 4484 4485 snap_id = rbd_snap_id_by_name(rbd_dev, spec->snap_name); 4486 if (snap_id == CEPH_NOSNAP) 4487 return -ENOENT; 4488 4489 spec->snap_id = snap_id; 4490 } else { 4491 spec->snap_id = CEPH_NOSNAP; 4492 } 4493 4494 return 0; 4495 } 4496 4497 /* 4498 * A parent image will have all ids but none of the names. 4499 * 4500 * All names in an rbd spec are dynamically allocated. It's OK if we 4501 * can't figure out the name for an image id. 4502 */ 4503 static int rbd_spec_fill_names(struct rbd_device *rbd_dev) 4504 { 4505 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc; 4506 struct rbd_spec *spec = rbd_dev->spec; 4507 const char *pool_name; 4508 const char *image_name; 4509 const char *snap_name; 4510 int ret; 4511 4512 rbd_assert(spec->pool_id != CEPH_NOPOOL); 4513 rbd_assert(spec->image_id); 4514 rbd_assert(spec->snap_id != CEPH_NOSNAP); 4515 4516 /* Get the pool name; we have to make our own copy of this */ 4517 4518 pool_name = ceph_pg_pool_name_by_id(osdc->osdmap, spec->pool_id); 4519 if (!pool_name) { 4520 rbd_warn(rbd_dev, "no pool with id %llu", spec->pool_id); 4521 return -EIO; 4522 } 4523 pool_name = kstrdup(pool_name, GFP_KERNEL); 4524 if (!pool_name) 4525 return -ENOMEM; 4526 4527 /* Fetch the image name; tolerate failure here */ 4528 4529 image_name = rbd_dev_image_name(rbd_dev); 4530 if (!image_name) 4531 rbd_warn(rbd_dev, "unable to get image name"); 4532 4533 /* Fetch the snapshot name */ 4534 4535 snap_name = rbd_snap_name(rbd_dev, spec->snap_id); 4536 if (IS_ERR(snap_name)) { 4537 ret = PTR_ERR(snap_name); 4538 goto out_err; 4539 } 4540 4541 spec->pool_name = pool_name; 4542 spec->image_name = image_name; 4543 spec->snap_name = snap_name; 4544 4545 return 0; 4546 4547 out_err: 4548 kfree(image_name); 4549 kfree(pool_name); 4550 return ret; 4551 } 4552 4553 static int rbd_dev_v2_snap_context(struct rbd_device *rbd_dev) 4554 { 4555 size_t size; 4556 int ret; 4557 void *reply_buf; 4558 void *p; 4559 void *end; 4560 u64 seq; 4561 u32 snap_count; 4562 struct ceph_snap_context *snapc; 4563 u32 i; 4564 4565 /* 4566 * We'll need room for the seq value (maximum snapshot id), 4567 * snapshot count, and array of that many snapshot ids. 4568 * For now we have a fixed upper limit on the number we're 4569 * prepared to receive. 4570 */ 4571 size = sizeof (__le64) + sizeof (__le32) + 4572 RBD_MAX_SNAP_COUNT * sizeof (__le64); 4573 reply_buf = kzalloc(size, GFP_KERNEL); 4574 if (!reply_buf) 4575 return -ENOMEM; 4576 4577 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name, 4578 "rbd", "get_snapcontext", NULL, 0, 4579 reply_buf, size); 4580 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret); 4581 if (ret < 0) 4582 goto out; 4583 4584 p = reply_buf; 4585 end = reply_buf + ret; 4586 ret = -ERANGE; 4587 ceph_decode_64_safe(&p, end, seq, out); 4588 ceph_decode_32_safe(&p, end, snap_count, out); 4589 4590 /* 4591 * Make sure the reported number of snapshot ids wouldn't go 4592 * beyond the end of our buffer. But before checking that, 4593 * make sure the computed size of the snapshot context we 4594 * allocate is representable in a size_t. 4595 */ 4596 if (snap_count > (SIZE_MAX - sizeof (struct ceph_snap_context)) 4597 / sizeof (u64)) { 4598 ret = -EINVAL; 4599 goto out; 4600 } 4601 if (!ceph_has_room(&p, end, snap_count * sizeof (__le64))) 4602 goto out; 4603 ret = 0; 4604 4605 snapc = ceph_create_snap_context(snap_count, GFP_KERNEL); 4606 if (!snapc) { 4607 ret = -ENOMEM; 4608 goto out; 4609 } 4610 snapc->seq = seq; 4611 for (i = 0; i < snap_count; i++) 4612 snapc->snaps[i] = ceph_decode_64(&p); 4613 4614 ceph_put_snap_context(rbd_dev->header.snapc); 4615 rbd_dev->header.snapc = snapc; 4616 4617 dout(" snap context seq = %llu, snap_count = %u\n", 4618 (unsigned long long)seq, (unsigned int)snap_count); 4619 out: 4620 kfree(reply_buf); 4621 4622 return ret; 4623 } 4624 4625 static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev, 4626 u64 snap_id) 4627 { 4628 size_t size; 4629 void *reply_buf; 4630 __le64 snapid; 4631 int ret; 4632 void *p; 4633 void *end; 4634 char *snap_name; 4635 4636 size = sizeof (__le32) + RBD_MAX_SNAP_NAME_LEN; 4637 reply_buf = kmalloc(size, GFP_KERNEL); 4638 if (!reply_buf) 4639 return ERR_PTR(-ENOMEM); 4640 4641 snapid = cpu_to_le64(snap_id); 4642 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name, 4643 "rbd", "get_snapshot_name", 4644 &snapid, sizeof (snapid), 4645 reply_buf, size); 4646 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret); 4647 if (ret < 0) { 4648 snap_name = ERR_PTR(ret); 4649 goto out; 4650 } 4651 4652 p = reply_buf; 4653 end = reply_buf + ret; 4654 snap_name = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL); 4655 if (IS_ERR(snap_name)) 4656 goto out; 4657 4658 dout(" snap_id 0x%016llx snap_name = %s\n", 4659 (unsigned long long)snap_id, snap_name); 4660 out: 4661 kfree(reply_buf); 4662 4663 return snap_name; 4664 } 4665 4666 static int rbd_dev_v2_header_info(struct rbd_device *rbd_dev) 4667 { 4668 bool first_time = rbd_dev->header.object_prefix == NULL; 4669 int ret; 4670 4671 ret = rbd_dev_v2_image_size(rbd_dev); 4672 if (ret) 4673 return ret; 4674 4675 if (first_time) { 4676 ret = rbd_dev_v2_header_onetime(rbd_dev); 4677 if (ret) 4678 return ret; 4679 } 4680 4681 ret = rbd_dev_v2_snap_context(rbd_dev); 4682 dout("rbd_dev_v2_snap_context returned %d\n", ret); 4683 4684 return ret; 4685 } 4686 4687 static int rbd_dev_header_info(struct rbd_device *rbd_dev) 4688 { 4689 rbd_assert(rbd_image_format_valid(rbd_dev->image_format)); 4690 4691 if (rbd_dev->image_format == 1) 4692 return rbd_dev_v1_header_info(rbd_dev); 4693 4694 return rbd_dev_v2_header_info(rbd_dev); 4695 } 4696 4697 static int rbd_bus_add_dev(struct rbd_device *rbd_dev) 4698 { 4699 struct device *dev; 4700 int ret; 4701 4702 dev = &rbd_dev->dev; 4703 dev->bus = &rbd_bus_type; 4704 dev->type = &rbd_device_type; 4705 dev->parent = &rbd_root_dev; 4706 dev->release = rbd_dev_device_release; 4707 dev_set_name(dev, "%d", rbd_dev->dev_id); 4708 ret = device_register(dev); 4709 4710 return ret; 4711 } 4712 4713 static void rbd_bus_del_dev(struct rbd_device *rbd_dev) 4714 { 4715 device_unregister(&rbd_dev->dev); 4716 } 4717 4718 /* 4719 * Get a unique rbd identifier for the given new rbd_dev, and add 4720 * the rbd_dev to the global list. 4721 */ 4722 static int rbd_dev_id_get(struct rbd_device *rbd_dev) 4723 { 4724 int new_dev_id; 4725 4726 new_dev_id = ida_simple_get(&rbd_dev_id_ida, 4727 0, minor_to_rbd_dev_id(1 << MINORBITS), 4728 GFP_KERNEL); 4729 if (new_dev_id < 0) 4730 return new_dev_id; 4731 4732 rbd_dev->dev_id = new_dev_id; 4733 4734 spin_lock(&rbd_dev_list_lock); 4735 list_add_tail(&rbd_dev->node, &rbd_dev_list); 4736 spin_unlock(&rbd_dev_list_lock); 4737 4738 dout("rbd_dev %p given dev id %d\n", rbd_dev, rbd_dev->dev_id); 4739 4740 return 0; 4741 } 4742 4743 /* 4744 * Remove an rbd_dev from the global list, and record that its 4745 * identifier is no longer in use. 4746 */ 4747 static void rbd_dev_id_put(struct rbd_device *rbd_dev) 4748 { 4749 spin_lock(&rbd_dev_list_lock); 4750 list_del_init(&rbd_dev->node); 4751 spin_unlock(&rbd_dev_list_lock); 4752 4753 ida_simple_remove(&rbd_dev_id_ida, rbd_dev->dev_id); 4754 4755 dout("rbd_dev %p released dev id %d\n", rbd_dev, rbd_dev->dev_id); 4756 } 4757 4758 /* 4759 * Skips over white space at *buf, and updates *buf to point to the 4760 * first found non-space character (if any). Returns the length of 4761 * the token (string of non-white space characters) found. Note 4762 * that *buf must be terminated with '\0'. 4763 */ 4764 static inline size_t next_token(const char **buf) 4765 { 4766 /* 4767 * These are the characters that produce nonzero for 4768 * isspace() in the "C" and "POSIX" locales. 4769 */ 4770 const char *spaces = " \f\n\r\t\v"; 4771 4772 *buf += strspn(*buf, spaces); /* Find start of token */ 4773 4774 return strcspn(*buf, spaces); /* Return token length */ 4775 } 4776 4777 /* 4778 * Finds the next token in *buf, and if the provided token buffer is 4779 * big enough, copies the found token into it. The result, if 4780 * copied, is guaranteed to be terminated with '\0'. Note that *buf 4781 * must be terminated with '\0' on entry. 4782 * 4783 * Returns the length of the token found (not including the '\0'). 4784 * Return value will be 0 if no token is found, and it will be >= 4785 * token_size if the token would not fit. 4786 * 4787 * The *buf pointer will be updated to point beyond the end of the 4788 * found token. Note that this occurs even if the token buffer is 4789 * too small to hold it. 4790 */ 4791 static inline size_t copy_token(const char **buf, 4792 char *token, 4793 size_t token_size) 4794 { 4795 size_t len; 4796 4797 len = next_token(buf); 4798 if (len < token_size) { 4799 memcpy(token, *buf, len); 4800 *(token + len) = '\0'; 4801 } 4802 *buf += len; 4803 4804 return len; 4805 } 4806 4807 /* 4808 * Finds the next token in *buf, dynamically allocates a buffer big 4809 * enough to hold a copy of it, and copies the token into the new 4810 * buffer. The copy is guaranteed to be terminated with '\0'. Note 4811 * that a duplicate buffer is created even for a zero-length token. 4812 * 4813 * Returns a pointer to the newly-allocated duplicate, or a null 4814 * pointer if memory for the duplicate was not available. If 4815 * the lenp argument is a non-null pointer, the length of the token 4816 * (not including the '\0') is returned in *lenp. 4817 * 4818 * If successful, the *buf pointer will be updated to point beyond 4819 * the end of the found token. 4820 * 4821 * Note: uses GFP_KERNEL for allocation. 4822 */ 4823 static inline char *dup_token(const char **buf, size_t *lenp) 4824 { 4825 char *dup; 4826 size_t len; 4827 4828 len = next_token(buf); 4829 dup = kmemdup(*buf, len + 1, GFP_KERNEL); 4830 if (!dup) 4831 return NULL; 4832 *(dup + len) = '\0'; 4833 *buf += len; 4834 4835 if (lenp) 4836 *lenp = len; 4837 4838 return dup; 4839 } 4840 4841 /* 4842 * Parse the options provided for an "rbd add" (i.e., rbd image 4843 * mapping) request. These arrive via a write to /sys/bus/rbd/add, 4844 * and the data written is passed here via a NUL-terminated buffer. 4845 * Returns 0 if successful or an error code otherwise. 4846 * 4847 * The information extracted from these options is recorded in 4848 * the other parameters which return dynamically-allocated 4849 * structures: 4850 * ceph_opts 4851 * The address of a pointer that will refer to a ceph options 4852 * structure. Caller must release the returned pointer using 4853 * ceph_destroy_options() when it is no longer needed. 4854 * rbd_opts 4855 * Address of an rbd options pointer. Fully initialized by 4856 * this function; caller must release with kfree(). 4857 * spec 4858 * Address of an rbd image specification pointer. Fully 4859 * initialized by this function based on parsed options. 4860 * Caller must release with rbd_spec_put(). 4861 * 4862 * The options passed take this form: 4863 * <mon_addrs> <options> <pool_name> <image_name> [<snap_id>] 4864 * where: 4865 * <mon_addrs> 4866 * A comma-separated list of one or more monitor addresses. 4867 * A monitor address is an ip address, optionally followed 4868 * by a port number (separated by a colon). 4869 * I.e.: ip1[:port1][,ip2[:port2]...] 4870 * <options> 4871 * A comma-separated list of ceph and/or rbd options. 4872 * <pool_name> 4873 * The name of the rados pool containing the rbd image. 4874 * <image_name> 4875 * The name of the image in that pool to map. 4876 * <snap_id> 4877 * An optional snapshot id. If provided, the mapping will 4878 * present data from the image at the time that snapshot was 4879 * created. The image head is used if no snapshot id is 4880 * provided. Snapshot mappings are always read-only. 4881 */ 4882 static int rbd_add_parse_args(const char *buf, 4883 struct ceph_options **ceph_opts, 4884 struct rbd_options **opts, 4885 struct rbd_spec **rbd_spec) 4886 { 4887 size_t len; 4888 char *options; 4889 const char *mon_addrs; 4890 char *snap_name; 4891 size_t mon_addrs_size; 4892 struct rbd_spec *spec = NULL; 4893 struct rbd_options *rbd_opts = NULL; 4894 struct ceph_options *copts; 4895 int ret; 4896 4897 /* The first four tokens are required */ 4898 4899 len = next_token(&buf); 4900 if (!len) { 4901 rbd_warn(NULL, "no monitor address(es) provided"); 4902 return -EINVAL; 4903 } 4904 mon_addrs = buf; 4905 mon_addrs_size = len + 1; 4906 buf += len; 4907 4908 ret = -EINVAL; 4909 options = dup_token(&buf, NULL); 4910 if (!options) 4911 return -ENOMEM; 4912 if (!*options) { 4913 rbd_warn(NULL, "no options provided"); 4914 goto out_err; 4915 } 4916 4917 spec = rbd_spec_alloc(); 4918 if (!spec) 4919 goto out_mem; 4920 4921 spec->pool_name = dup_token(&buf, NULL); 4922 if (!spec->pool_name) 4923 goto out_mem; 4924 if (!*spec->pool_name) { 4925 rbd_warn(NULL, "no pool name provided"); 4926 goto out_err; 4927 } 4928 4929 spec->image_name = dup_token(&buf, NULL); 4930 if (!spec->image_name) 4931 goto out_mem; 4932 if (!*spec->image_name) { 4933 rbd_warn(NULL, "no image name provided"); 4934 goto out_err; 4935 } 4936 4937 /* 4938 * Snapshot name is optional; default is to use "-" 4939 * (indicating the head/no snapshot). 4940 */ 4941 len = next_token(&buf); 4942 if (!len) { 4943 buf = RBD_SNAP_HEAD_NAME; /* No snapshot supplied */ 4944 len = sizeof (RBD_SNAP_HEAD_NAME) - 1; 4945 } else if (len > RBD_MAX_SNAP_NAME_LEN) { 4946 ret = -ENAMETOOLONG; 4947 goto out_err; 4948 } 4949 snap_name = kmemdup(buf, len + 1, GFP_KERNEL); 4950 if (!snap_name) 4951 goto out_mem; 4952 *(snap_name + len) = '\0'; 4953 spec->snap_name = snap_name; 4954 4955 /* Initialize all rbd options to the defaults */ 4956 4957 rbd_opts = kzalloc(sizeof (*rbd_opts), GFP_KERNEL); 4958 if (!rbd_opts) 4959 goto out_mem; 4960 4961 rbd_opts->read_only = RBD_READ_ONLY_DEFAULT; 4962 4963 copts = ceph_parse_options(options, mon_addrs, 4964 mon_addrs + mon_addrs_size - 1, 4965 parse_rbd_opts_token, rbd_opts); 4966 if (IS_ERR(copts)) { 4967 ret = PTR_ERR(copts); 4968 goto out_err; 4969 } 4970 kfree(options); 4971 4972 *ceph_opts = copts; 4973 *opts = rbd_opts; 4974 *rbd_spec = spec; 4975 4976 return 0; 4977 out_mem: 4978 ret = -ENOMEM; 4979 out_err: 4980 kfree(rbd_opts); 4981 rbd_spec_put(spec); 4982 kfree(options); 4983 4984 return ret; 4985 } 4986 4987 /* 4988 * Return pool id (>= 0) or a negative error code. 4989 */ 4990 static int rbd_add_get_pool_id(struct rbd_client *rbdc, const char *pool_name) 4991 { 4992 u64 newest_epoch; 4993 unsigned long timeout = rbdc->client->options->mount_timeout * HZ; 4994 int tries = 0; 4995 int ret; 4996 4997 again: 4998 ret = ceph_pg_poolid_by_name(rbdc->client->osdc.osdmap, pool_name); 4999 if (ret == -ENOENT && tries++ < 1) { 5000 ret = ceph_monc_do_get_version(&rbdc->client->monc, "osdmap", 5001 &newest_epoch); 5002 if (ret < 0) 5003 return ret; 5004 5005 if (rbdc->client->osdc.osdmap->epoch < newest_epoch) { 5006 ceph_monc_request_next_osdmap(&rbdc->client->monc); 5007 (void) ceph_monc_wait_osdmap(&rbdc->client->monc, 5008 newest_epoch, timeout); 5009 goto again; 5010 } else { 5011 /* the osdmap we have is new enough */ 5012 return -ENOENT; 5013 } 5014 } 5015 5016 return ret; 5017 } 5018 5019 /* 5020 * An rbd format 2 image has a unique identifier, distinct from the 5021 * name given to it by the user. Internally, that identifier is 5022 * what's used to specify the names of objects related to the image. 5023 * 5024 * A special "rbd id" object is used to map an rbd image name to its 5025 * id. If that object doesn't exist, then there is no v2 rbd image 5026 * with the supplied name. 5027 * 5028 * This function will record the given rbd_dev's image_id field if 5029 * it can be determined, and in that case will return 0. If any 5030 * errors occur a negative errno will be returned and the rbd_dev's 5031 * image_id field will be unchanged (and should be NULL). 5032 */ 5033 static int rbd_dev_image_id(struct rbd_device *rbd_dev) 5034 { 5035 int ret; 5036 size_t size; 5037 char *object_name; 5038 void *response; 5039 char *image_id; 5040 5041 /* 5042 * When probing a parent image, the image id is already 5043 * known (and the image name likely is not). There's no 5044 * need to fetch the image id again in this case. We 5045 * do still need to set the image format though. 5046 */ 5047 if (rbd_dev->spec->image_id) { 5048 rbd_dev->image_format = *rbd_dev->spec->image_id ? 2 : 1; 5049 5050 return 0; 5051 } 5052 5053 /* 5054 * First, see if the format 2 image id file exists, and if 5055 * so, get the image's persistent id from it. 5056 */ 5057 size = sizeof (RBD_ID_PREFIX) + strlen(rbd_dev->spec->image_name); 5058 object_name = kmalloc(size, GFP_NOIO); 5059 if (!object_name) 5060 return -ENOMEM; 5061 sprintf(object_name, "%s%s", RBD_ID_PREFIX, rbd_dev->spec->image_name); 5062 dout("rbd id object name is %s\n", object_name); 5063 5064 /* Response will be an encoded string, which includes a length */ 5065 5066 size = sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX; 5067 response = kzalloc(size, GFP_NOIO); 5068 if (!response) { 5069 ret = -ENOMEM; 5070 goto out; 5071 } 5072 5073 /* If it doesn't exist we'll assume it's a format 1 image */ 5074 5075 ret = rbd_obj_method_sync(rbd_dev, object_name, 5076 "rbd", "get_id", NULL, 0, 5077 response, RBD_IMAGE_ID_LEN_MAX); 5078 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret); 5079 if (ret == -ENOENT) { 5080 image_id = kstrdup("", GFP_KERNEL); 5081 ret = image_id ? 0 : -ENOMEM; 5082 if (!ret) 5083 rbd_dev->image_format = 1; 5084 } else if (ret >= 0) { 5085 void *p = response; 5086 5087 image_id = ceph_extract_encoded_string(&p, p + ret, 5088 NULL, GFP_NOIO); 5089 ret = PTR_ERR_OR_ZERO(image_id); 5090 if (!ret) 5091 rbd_dev->image_format = 2; 5092 } 5093 5094 if (!ret) { 5095 rbd_dev->spec->image_id = image_id; 5096 dout("image_id is %s\n", image_id); 5097 } 5098 out: 5099 kfree(response); 5100 kfree(object_name); 5101 5102 return ret; 5103 } 5104 5105 /* 5106 * Undo whatever state changes are made by v1 or v2 header info 5107 * call. 5108 */ 5109 static void rbd_dev_unprobe(struct rbd_device *rbd_dev) 5110 { 5111 struct rbd_image_header *header; 5112 5113 /* Drop parent reference unless it's already been done (or none) */ 5114 5115 if (rbd_dev->parent_overlap) 5116 rbd_dev_parent_put(rbd_dev); 5117 5118 /* Free dynamic fields from the header, then zero it out */ 5119 5120 header = &rbd_dev->header; 5121 ceph_put_snap_context(header->snapc); 5122 kfree(header->snap_sizes); 5123 kfree(header->snap_names); 5124 kfree(header->object_prefix); 5125 memset(header, 0, sizeof (*header)); 5126 } 5127 5128 static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev) 5129 { 5130 int ret; 5131 5132 ret = rbd_dev_v2_object_prefix(rbd_dev); 5133 if (ret) 5134 goto out_err; 5135 5136 /* 5137 * Get the and check features for the image. Currently the 5138 * features are assumed to never change. 5139 */ 5140 ret = rbd_dev_v2_features(rbd_dev); 5141 if (ret) 5142 goto out_err; 5143 5144 /* If the image supports fancy striping, get its parameters */ 5145 5146 if (rbd_dev->header.features & RBD_FEATURE_STRIPINGV2) { 5147 ret = rbd_dev_v2_striping_info(rbd_dev); 5148 if (ret < 0) 5149 goto out_err; 5150 } 5151 /* No support for crypto and compression type format 2 images */ 5152 5153 return 0; 5154 out_err: 5155 rbd_dev->header.features = 0; 5156 kfree(rbd_dev->header.object_prefix); 5157 rbd_dev->header.object_prefix = NULL; 5158 5159 return ret; 5160 } 5161 5162 static int rbd_dev_probe_parent(struct rbd_device *rbd_dev) 5163 { 5164 struct rbd_device *parent = NULL; 5165 struct rbd_spec *parent_spec; 5166 struct rbd_client *rbdc; 5167 int ret; 5168 5169 if (!rbd_dev->parent_spec) 5170 return 0; 5171 /* 5172 * We need to pass a reference to the client and the parent 5173 * spec when creating the parent rbd_dev. Images related by 5174 * parent/child relationships always share both. 5175 */ 5176 parent_spec = rbd_spec_get(rbd_dev->parent_spec); 5177 rbdc = __rbd_get_client(rbd_dev->rbd_client); 5178 5179 ret = -ENOMEM; 5180 parent = rbd_dev_create(rbdc, parent_spec); 5181 if (!parent) 5182 goto out_err; 5183 5184 ret = rbd_dev_image_probe(parent, false); 5185 if (ret < 0) 5186 goto out_err; 5187 rbd_dev->parent = parent; 5188 atomic_set(&rbd_dev->parent_ref, 1); 5189 5190 return 0; 5191 out_err: 5192 if (parent) { 5193 rbd_dev_unparent(rbd_dev); 5194 kfree(rbd_dev->header_name); 5195 rbd_dev_destroy(parent); 5196 } else { 5197 rbd_put_client(rbdc); 5198 rbd_spec_put(parent_spec); 5199 } 5200 5201 return ret; 5202 } 5203 5204 static int rbd_dev_device_setup(struct rbd_device *rbd_dev) 5205 { 5206 int ret; 5207 5208 /* Get an id and fill in device name. */ 5209 5210 ret = rbd_dev_id_get(rbd_dev); 5211 if (ret) 5212 return ret; 5213 5214 BUILD_BUG_ON(DEV_NAME_LEN 5215 < sizeof (RBD_DRV_NAME) + MAX_INT_FORMAT_WIDTH); 5216 sprintf(rbd_dev->name, "%s%d", RBD_DRV_NAME, rbd_dev->dev_id); 5217 5218 /* Record our major and minor device numbers. */ 5219 5220 if (!single_major) { 5221 ret = register_blkdev(0, rbd_dev->name); 5222 if (ret < 0) 5223 goto err_out_id; 5224 5225 rbd_dev->major = ret; 5226 rbd_dev->minor = 0; 5227 } else { 5228 rbd_dev->major = rbd_major; 5229 rbd_dev->minor = rbd_dev_id_to_minor(rbd_dev->dev_id); 5230 } 5231 5232 /* Set up the blkdev mapping. */ 5233 5234 ret = rbd_init_disk(rbd_dev); 5235 if (ret) 5236 goto err_out_blkdev; 5237 5238 ret = rbd_dev_mapping_set(rbd_dev); 5239 if (ret) 5240 goto err_out_disk; 5241 5242 set_capacity(rbd_dev->disk, rbd_dev->mapping.size / SECTOR_SIZE); 5243 set_disk_ro(rbd_dev->disk, rbd_dev->mapping.read_only); 5244 5245 rbd_dev->rq_wq = alloc_workqueue("%s", WQ_MEM_RECLAIM, 0, 5246 rbd_dev->disk->disk_name); 5247 if (!rbd_dev->rq_wq) { 5248 ret = -ENOMEM; 5249 goto err_out_mapping; 5250 } 5251 5252 ret = rbd_bus_add_dev(rbd_dev); 5253 if (ret) 5254 goto err_out_workqueue; 5255 5256 /* Everything's ready. Announce the disk to the world. */ 5257 5258 set_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags); 5259 add_disk(rbd_dev->disk); 5260 5261 pr_info("%s: added with size 0x%llx\n", rbd_dev->disk->disk_name, 5262 (unsigned long long) rbd_dev->mapping.size); 5263 5264 return ret; 5265 5266 err_out_workqueue: 5267 destroy_workqueue(rbd_dev->rq_wq); 5268 rbd_dev->rq_wq = NULL; 5269 err_out_mapping: 5270 rbd_dev_mapping_clear(rbd_dev); 5271 err_out_disk: 5272 rbd_free_disk(rbd_dev); 5273 err_out_blkdev: 5274 if (!single_major) 5275 unregister_blkdev(rbd_dev->major, rbd_dev->name); 5276 err_out_id: 5277 rbd_dev_id_put(rbd_dev); 5278 rbd_dev_mapping_clear(rbd_dev); 5279 5280 return ret; 5281 } 5282 5283 static int rbd_dev_header_name(struct rbd_device *rbd_dev) 5284 { 5285 struct rbd_spec *spec = rbd_dev->spec; 5286 size_t size; 5287 5288 /* Record the header object name for this rbd image. */ 5289 5290 rbd_assert(rbd_image_format_valid(rbd_dev->image_format)); 5291 5292 if (rbd_dev->image_format == 1) 5293 size = strlen(spec->image_name) + sizeof (RBD_SUFFIX); 5294 else 5295 size = sizeof (RBD_HEADER_PREFIX) + strlen(spec->image_id); 5296 5297 rbd_dev->header_name = kmalloc(size, GFP_KERNEL); 5298 if (!rbd_dev->header_name) 5299 return -ENOMEM; 5300 5301 if (rbd_dev->image_format == 1) 5302 sprintf(rbd_dev->header_name, "%s%s", 5303 spec->image_name, RBD_SUFFIX); 5304 else 5305 sprintf(rbd_dev->header_name, "%s%s", 5306 RBD_HEADER_PREFIX, spec->image_id); 5307 return 0; 5308 } 5309 5310 static void rbd_dev_image_release(struct rbd_device *rbd_dev) 5311 { 5312 rbd_dev_unprobe(rbd_dev); 5313 kfree(rbd_dev->header_name); 5314 rbd_dev->header_name = NULL; 5315 rbd_dev->image_format = 0; 5316 kfree(rbd_dev->spec->image_id); 5317 rbd_dev->spec->image_id = NULL; 5318 5319 rbd_dev_destroy(rbd_dev); 5320 } 5321 5322 /* 5323 * Probe for the existence of the header object for the given rbd 5324 * device. If this image is the one being mapped (i.e., not a 5325 * parent), initiate a watch on its header object before using that 5326 * object to get detailed information about the rbd image. 5327 */ 5328 static int rbd_dev_image_probe(struct rbd_device *rbd_dev, bool mapping) 5329 { 5330 int ret; 5331 5332 /* 5333 * Get the id from the image id object. Unless there's an 5334 * error, rbd_dev->spec->image_id will be filled in with 5335 * a dynamically-allocated string, and rbd_dev->image_format 5336 * will be set to either 1 or 2. 5337 */ 5338 ret = rbd_dev_image_id(rbd_dev); 5339 if (ret) 5340 return ret; 5341 5342 ret = rbd_dev_header_name(rbd_dev); 5343 if (ret) 5344 goto err_out_format; 5345 5346 if (mapping) { 5347 ret = rbd_dev_header_watch_sync(rbd_dev); 5348 if (ret) 5349 goto out_header_name; 5350 } 5351 5352 ret = rbd_dev_header_info(rbd_dev); 5353 if (ret) 5354 goto err_out_watch; 5355 5356 /* 5357 * If this image is the one being mapped, we have pool name and 5358 * id, image name and id, and snap name - need to fill snap id. 5359 * Otherwise this is a parent image, identified by pool, image 5360 * and snap ids - need to fill in names for those ids. 5361 */ 5362 if (mapping) 5363 ret = rbd_spec_fill_snap_id(rbd_dev); 5364 else 5365 ret = rbd_spec_fill_names(rbd_dev); 5366 if (ret) 5367 goto err_out_probe; 5368 5369 if (rbd_dev->header.features & RBD_FEATURE_LAYERING) { 5370 ret = rbd_dev_v2_parent_info(rbd_dev); 5371 if (ret) 5372 goto err_out_probe; 5373 5374 /* 5375 * Need to warn users if this image is the one being 5376 * mapped and has a parent. 5377 */ 5378 if (mapping && rbd_dev->parent_spec) 5379 rbd_warn(rbd_dev, 5380 "WARNING: kernel layering is EXPERIMENTAL!"); 5381 } 5382 5383 ret = rbd_dev_probe_parent(rbd_dev); 5384 if (ret) 5385 goto err_out_probe; 5386 5387 dout("discovered format %u image, header name is %s\n", 5388 rbd_dev->image_format, rbd_dev->header_name); 5389 return 0; 5390 5391 err_out_probe: 5392 rbd_dev_unprobe(rbd_dev); 5393 err_out_watch: 5394 if (mapping) 5395 rbd_dev_header_unwatch_sync(rbd_dev); 5396 out_header_name: 5397 kfree(rbd_dev->header_name); 5398 rbd_dev->header_name = NULL; 5399 err_out_format: 5400 rbd_dev->image_format = 0; 5401 kfree(rbd_dev->spec->image_id); 5402 rbd_dev->spec->image_id = NULL; 5403 return ret; 5404 } 5405 5406 static ssize_t do_rbd_add(struct bus_type *bus, 5407 const char *buf, 5408 size_t count) 5409 { 5410 struct rbd_device *rbd_dev = NULL; 5411 struct ceph_options *ceph_opts = NULL; 5412 struct rbd_options *rbd_opts = NULL; 5413 struct rbd_spec *spec = NULL; 5414 struct rbd_client *rbdc; 5415 bool read_only; 5416 int rc = -ENOMEM; 5417 5418 if (!try_module_get(THIS_MODULE)) 5419 return -ENODEV; 5420 5421 /* parse add command */ 5422 rc = rbd_add_parse_args(buf, &ceph_opts, &rbd_opts, &spec); 5423 if (rc < 0) 5424 goto err_out_module; 5425 read_only = rbd_opts->read_only; 5426 kfree(rbd_opts); 5427 rbd_opts = NULL; /* done with this */ 5428 5429 rbdc = rbd_get_client(ceph_opts); 5430 if (IS_ERR(rbdc)) { 5431 rc = PTR_ERR(rbdc); 5432 goto err_out_args; 5433 } 5434 5435 /* pick the pool */ 5436 rc = rbd_add_get_pool_id(rbdc, spec->pool_name); 5437 if (rc < 0) 5438 goto err_out_client; 5439 spec->pool_id = (u64)rc; 5440 5441 /* The ceph file layout needs to fit pool id in 32 bits */ 5442 5443 if (spec->pool_id > (u64)U32_MAX) { 5444 rbd_warn(NULL, "pool id too large (%llu > %u)", 5445 (unsigned long long)spec->pool_id, U32_MAX); 5446 rc = -EIO; 5447 goto err_out_client; 5448 } 5449 5450 rbd_dev = rbd_dev_create(rbdc, spec); 5451 if (!rbd_dev) 5452 goto err_out_client; 5453 rbdc = NULL; /* rbd_dev now owns this */ 5454 spec = NULL; /* rbd_dev now owns this */ 5455 5456 rc = rbd_dev_image_probe(rbd_dev, true); 5457 if (rc < 0) 5458 goto err_out_rbd_dev; 5459 5460 /* If we are mapping a snapshot it must be marked read-only */ 5461 5462 if (rbd_dev->spec->snap_id != CEPH_NOSNAP) 5463 read_only = true; 5464 rbd_dev->mapping.read_only = read_only; 5465 5466 rc = rbd_dev_device_setup(rbd_dev); 5467 if (rc) { 5468 /* 5469 * rbd_dev_header_unwatch_sync() can't be moved into 5470 * rbd_dev_image_release() without refactoring, see 5471 * commit 1f3ef78861ac. 5472 */ 5473 rbd_dev_header_unwatch_sync(rbd_dev); 5474 rbd_dev_image_release(rbd_dev); 5475 goto err_out_module; 5476 } 5477 5478 return count; 5479 5480 err_out_rbd_dev: 5481 rbd_dev_destroy(rbd_dev); 5482 err_out_client: 5483 rbd_put_client(rbdc); 5484 err_out_args: 5485 rbd_spec_put(spec); 5486 err_out_module: 5487 module_put(THIS_MODULE); 5488 5489 dout("Error adding device %s\n", buf); 5490 5491 return (ssize_t)rc; 5492 } 5493 5494 static ssize_t rbd_add(struct bus_type *bus, 5495 const char *buf, 5496 size_t count) 5497 { 5498 if (single_major) 5499 return -EINVAL; 5500 5501 return do_rbd_add(bus, buf, count); 5502 } 5503 5504 static ssize_t rbd_add_single_major(struct bus_type *bus, 5505 const char *buf, 5506 size_t count) 5507 { 5508 return do_rbd_add(bus, buf, count); 5509 } 5510 5511 static void rbd_dev_device_release(struct device *dev) 5512 { 5513 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev); 5514 5515 destroy_workqueue(rbd_dev->rq_wq); 5516 rbd_free_disk(rbd_dev); 5517 clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags); 5518 rbd_dev_mapping_clear(rbd_dev); 5519 if (!single_major) 5520 unregister_blkdev(rbd_dev->major, rbd_dev->name); 5521 rbd_dev_id_put(rbd_dev); 5522 rbd_dev_mapping_clear(rbd_dev); 5523 } 5524 5525 static void rbd_dev_remove_parent(struct rbd_device *rbd_dev) 5526 { 5527 while (rbd_dev->parent) { 5528 struct rbd_device *first = rbd_dev; 5529 struct rbd_device *second = first->parent; 5530 struct rbd_device *third; 5531 5532 /* 5533 * Follow to the parent with no grandparent and 5534 * remove it. 5535 */ 5536 while (second && (third = second->parent)) { 5537 first = second; 5538 second = third; 5539 } 5540 rbd_assert(second); 5541 rbd_dev_image_release(second); 5542 first->parent = NULL; 5543 first->parent_overlap = 0; 5544 5545 rbd_assert(first->parent_spec); 5546 rbd_spec_put(first->parent_spec); 5547 first->parent_spec = NULL; 5548 } 5549 } 5550 5551 static ssize_t do_rbd_remove(struct bus_type *bus, 5552 const char *buf, 5553 size_t count) 5554 { 5555 struct rbd_device *rbd_dev = NULL; 5556 struct list_head *tmp; 5557 int dev_id; 5558 unsigned long ul; 5559 bool already = false; 5560 int ret; 5561 5562 ret = kstrtoul(buf, 10, &ul); 5563 if (ret) 5564 return ret; 5565 5566 /* convert to int; abort if we lost anything in the conversion */ 5567 dev_id = (int)ul; 5568 if (dev_id != ul) 5569 return -EINVAL; 5570 5571 ret = -ENOENT; 5572 spin_lock(&rbd_dev_list_lock); 5573 list_for_each(tmp, &rbd_dev_list) { 5574 rbd_dev = list_entry(tmp, struct rbd_device, node); 5575 if (rbd_dev->dev_id == dev_id) { 5576 ret = 0; 5577 break; 5578 } 5579 } 5580 if (!ret) { 5581 spin_lock_irq(&rbd_dev->lock); 5582 if (rbd_dev->open_count) 5583 ret = -EBUSY; 5584 else 5585 already = test_and_set_bit(RBD_DEV_FLAG_REMOVING, 5586 &rbd_dev->flags); 5587 spin_unlock_irq(&rbd_dev->lock); 5588 } 5589 spin_unlock(&rbd_dev_list_lock); 5590 if (ret < 0 || already) 5591 return ret; 5592 5593 rbd_dev_header_unwatch_sync(rbd_dev); 5594 /* 5595 * flush remaining watch callbacks - these must be complete 5596 * before the osd_client is shutdown 5597 */ 5598 dout("%s: flushing notifies", __func__); 5599 ceph_osdc_flush_notifies(&rbd_dev->rbd_client->client->osdc); 5600 5601 /* 5602 * Don't free anything from rbd_dev->disk until after all 5603 * notifies are completely processed. Otherwise 5604 * rbd_bus_del_dev() will race with rbd_watch_cb(), resulting 5605 * in a potential use after free of rbd_dev->disk or rbd_dev. 5606 */ 5607 rbd_bus_del_dev(rbd_dev); 5608 rbd_dev_image_release(rbd_dev); 5609 module_put(THIS_MODULE); 5610 5611 return count; 5612 } 5613 5614 static ssize_t rbd_remove(struct bus_type *bus, 5615 const char *buf, 5616 size_t count) 5617 { 5618 if (single_major) 5619 return -EINVAL; 5620 5621 return do_rbd_remove(bus, buf, count); 5622 } 5623 5624 static ssize_t rbd_remove_single_major(struct bus_type *bus, 5625 const char *buf, 5626 size_t count) 5627 { 5628 return do_rbd_remove(bus, buf, count); 5629 } 5630 5631 /* 5632 * create control files in sysfs 5633 * /sys/bus/rbd/... 5634 */ 5635 static int rbd_sysfs_init(void) 5636 { 5637 int ret; 5638 5639 ret = device_register(&rbd_root_dev); 5640 if (ret < 0) 5641 return ret; 5642 5643 ret = bus_register(&rbd_bus_type); 5644 if (ret < 0) 5645 device_unregister(&rbd_root_dev); 5646 5647 return ret; 5648 } 5649 5650 static void rbd_sysfs_cleanup(void) 5651 { 5652 bus_unregister(&rbd_bus_type); 5653 device_unregister(&rbd_root_dev); 5654 } 5655 5656 static int rbd_slab_init(void) 5657 { 5658 rbd_assert(!rbd_img_request_cache); 5659 rbd_img_request_cache = kmem_cache_create("rbd_img_request", 5660 sizeof (struct rbd_img_request), 5661 __alignof__(struct rbd_img_request), 5662 0, NULL); 5663 if (!rbd_img_request_cache) 5664 return -ENOMEM; 5665 5666 rbd_assert(!rbd_obj_request_cache); 5667 rbd_obj_request_cache = kmem_cache_create("rbd_obj_request", 5668 sizeof (struct rbd_obj_request), 5669 __alignof__(struct rbd_obj_request), 5670 0, NULL); 5671 if (!rbd_obj_request_cache) 5672 goto out_err; 5673 5674 rbd_assert(!rbd_segment_name_cache); 5675 rbd_segment_name_cache = kmem_cache_create("rbd_segment_name", 5676 CEPH_MAX_OID_NAME_LEN + 1, 1, 0, NULL); 5677 if (rbd_segment_name_cache) 5678 return 0; 5679 out_err: 5680 if (rbd_obj_request_cache) { 5681 kmem_cache_destroy(rbd_obj_request_cache); 5682 rbd_obj_request_cache = NULL; 5683 } 5684 5685 kmem_cache_destroy(rbd_img_request_cache); 5686 rbd_img_request_cache = NULL; 5687 5688 return -ENOMEM; 5689 } 5690 5691 static void rbd_slab_exit(void) 5692 { 5693 rbd_assert(rbd_segment_name_cache); 5694 kmem_cache_destroy(rbd_segment_name_cache); 5695 rbd_segment_name_cache = NULL; 5696 5697 rbd_assert(rbd_obj_request_cache); 5698 kmem_cache_destroy(rbd_obj_request_cache); 5699 rbd_obj_request_cache = NULL; 5700 5701 rbd_assert(rbd_img_request_cache); 5702 kmem_cache_destroy(rbd_img_request_cache); 5703 rbd_img_request_cache = NULL; 5704 } 5705 5706 static int __init rbd_init(void) 5707 { 5708 int rc; 5709 5710 if (!libceph_compatible(NULL)) { 5711 rbd_warn(NULL, "libceph incompatibility (quitting)"); 5712 return -EINVAL; 5713 } 5714 5715 rc = rbd_slab_init(); 5716 if (rc) 5717 return rc; 5718 5719 if (single_major) { 5720 rbd_major = register_blkdev(0, RBD_DRV_NAME); 5721 if (rbd_major < 0) { 5722 rc = rbd_major; 5723 goto err_out_slab; 5724 } 5725 } 5726 5727 rc = rbd_sysfs_init(); 5728 if (rc) 5729 goto err_out_blkdev; 5730 5731 if (single_major) 5732 pr_info("loaded (major %d)\n", rbd_major); 5733 else 5734 pr_info("loaded\n"); 5735 5736 return 0; 5737 5738 err_out_blkdev: 5739 if (single_major) 5740 unregister_blkdev(rbd_major, RBD_DRV_NAME); 5741 err_out_slab: 5742 rbd_slab_exit(); 5743 return rc; 5744 } 5745 5746 static void __exit rbd_exit(void) 5747 { 5748 ida_destroy(&rbd_dev_id_ida); 5749 rbd_sysfs_cleanup(); 5750 if (single_major) 5751 unregister_blkdev(rbd_major, RBD_DRV_NAME); 5752 rbd_slab_exit(); 5753 } 5754 5755 module_init(rbd_init); 5756 module_exit(rbd_exit); 5757 5758 MODULE_AUTHOR("Alex Elder <elder@inktank.com>"); 5759 MODULE_AUTHOR("Sage Weil <sage@newdream.net>"); 5760 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>"); 5761 /* following authorship retained from original osdblk.c */ 5762 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>"); 5763 5764 MODULE_DESCRIPTION("RADOS Block Device (RBD) driver"); 5765 MODULE_LICENSE("GPL"); 5766