libceph: keep source rather than message osd op array
[deliverable/linux.git] / drivers / block / rbd.c
1 /*
2 rbd.c -- Export ceph rados objects as a Linux block device
3
4
5 based on drivers/block/osdblk.c:
6
7 Copyright 2009 Red Hat, Inc.
8
9 This program is free software; you can redistribute it and/or modify
10 it under the terms of the GNU General Public License as published by
11 the Free Software Foundation.
12
13 This program is distributed in the hope that it will be useful,
14 but WITHOUT ANY WARRANTY; without even the implied warranty of
15 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 GNU General Public License for more details.
17
18 You should have received a copy of the GNU General Public License
19 along with this program; see the file COPYING. If not, write to
20 the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
21
22
23
24 For usage instructions, please refer to:
25
26 Documentation/ABI/testing/sysfs-bus-rbd
27
28 */
29
30 #include <linux/ceph/libceph.h>
31 #include <linux/ceph/osd_client.h>
32 #include <linux/ceph/mon_client.h>
33 #include <linux/ceph/decode.h>
34 #include <linux/parser.h>
35
36 #include <linux/kernel.h>
37 #include <linux/device.h>
38 #include <linux/module.h>
39 #include <linux/fs.h>
40 #include <linux/blkdev.h>
41
42 #include "rbd_types.h"
43
44 #define RBD_DEBUG /* Activate rbd_assert() calls */
45
46 /*
47 * The basic unit of block I/O is a sector. It is interpreted in a
48 * number of contexts in Linux (blk, bio, genhd), but the default is
49 * universally 512 bytes. These symbols are just slightly more
50 * meaningful than the bare numbers they represent.
51 */
52 #define SECTOR_SHIFT 9
53 #define SECTOR_SIZE (1ULL << SECTOR_SHIFT)
54
55 #define RBD_DRV_NAME "rbd"
56 #define RBD_DRV_NAME_LONG "rbd (rados block device)"
57
58 #define RBD_MINORS_PER_MAJOR 256 /* max minors per blkdev */
59
60 #define RBD_SNAP_DEV_NAME_PREFIX "snap_"
61 #define RBD_MAX_SNAP_NAME_LEN \
62 (NAME_MAX - (sizeof (RBD_SNAP_DEV_NAME_PREFIX) - 1))
63
64 #define RBD_MAX_SNAP_COUNT 510 /* allows max snapc to fit in 4KB */
65
66 #define RBD_SNAP_HEAD_NAME "-"
67
68 /* This allows a single page to hold an image name sent by OSD */
69 #define RBD_IMAGE_NAME_LEN_MAX (PAGE_SIZE - sizeof (__le32) - 1)
70 #define RBD_IMAGE_ID_LEN_MAX 64
71
72 #define RBD_OBJ_PREFIX_LEN_MAX 64
73
74 /* Feature bits */
75
76 #define RBD_FEATURE_LAYERING 1
77
78 /* Features supported by this (client software) implementation. */
79
80 #define RBD_FEATURES_ALL (0)
81
82 /*
83 * An RBD device name will be "rbd#", where the "rbd" comes from
84 * RBD_DRV_NAME above, and # is a unique integer identifier.
85 * MAX_INT_FORMAT_WIDTH is used in ensuring DEV_NAME_LEN is big
86 * enough to hold all possible device names.
87 */
88 #define DEV_NAME_LEN 32
89 #define MAX_INT_FORMAT_WIDTH ((5 * sizeof (int)) / 2 + 1)
90
91 /*
92 * block device image metadata (in-memory version)
93 */
94 struct rbd_image_header {
95 /* These four fields never change for a given rbd image */
96 char *object_prefix;
97 u64 features;
98 __u8 obj_order;
99 __u8 crypt_type;
100 __u8 comp_type;
101
102 /* The remaining fields need to be updated occasionally */
103 u64 image_size;
104 struct ceph_snap_context *snapc;
105 char *snap_names;
106 u64 *snap_sizes;
107
108 u64 obj_version;
109 };
110
111 /*
112 * An rbd image specification.
113 *
114 * The tuple (pool_id, image_id, snap_id) is sufficient to uniquely
115 * identify an image. Each rbd_dev structure includes a pointer to
116 * an rbd_spec structure that encapsulates this identity.
117 *
118 * Each of the id's in an rbd_spec has an associated name. For a
119 * user-mapped image, the names are supplied and the id's associated
120 * with them are looked up. For a layered image, a parent image is
121 * defined by the tuple, and the names are looked up.
122 *
123 * An rbd_dev structure contains a parent_spec pointer which is
124 * non-null if the image it represents is a child in a layered
125 * image. This pointer will refer to the rbd_spec structure used
126 * by the parent rbd_dev for its own identity (i.e., the structure
127 * is shared between the parent and child).
128 *
129 * Since these structures are populated once, during the discovery
130 * phase of image construction, they are effectively immutable so
131 * we make no effort to synchronize access to them.
132 *
133 * Note that code herein does not assume the image name is known (it
134 * could be a null pointer).
135 */
136 struct rbd_spec {
137 u64 pool_id;
138 char *pool_name;
139
140 char *image_id;
141 char *image_name;
142
143 u64 snap_id;
144 char *snap_name;
145
146 struct kref kref;
147 };
148
149 /*
150 * an instance of the client. multiple devices may share an rbd client.
151 */
152 struct rbd_client {
153 struct ceph_client *client;
154 struct kref kref;
155 struct list_head node;
156 };
157
158 struct rbd_img_request;
159 typedef void (*rbd_img_callback_t)(struct rbd_img_request *);
160
161 #define BAD_WHICH U32_MAX /* Good which or bad which, which? */
162
163 struct rbd_obj_request;
164 typedef void (*rbd_obj_callback_t)(struct rbd_obj_request *);
165
166 enum obj_request_type {
167 OBJ_REQUEST_NODATA, OBJ_REQUEST_BIO, OBJ_REQUEST_PAGES
168 };
169
170 struct rbd_obj_request {
171 const char *object_name;
172 u64 offset; /* object start byte */
173 u64 length; /* bytes from offset */
174
175 struct rbd_img_request *img_request;
176 struct list_head links; /* img_request->obj_requests */
177 u32 which; /* posn image request list */
178
179 enum obj_request_type type;
180 union {
181 struct bio *bio_list;
182 struct {
183 struct page **pages;
184 u32 page_count;
185 };
186 };
187
188 struct ceph_osd_request *osd_req;
189
190 u64 xferred; /* bytes transferred */
191 u64 version;
192 int result;
193 atomic_t done;
194
195 rbd_obj_callback_t callback;
196 struct completion completion;
197
198 struct kref kref;
199 };
200
201 struct rbd_img_request {
202 struct request *rq;
203 struct rbd_device *rbd_dev;
204 u64 offset; /* starting image byte offset */
205 u64 length; /* byte count from offset */
206 bool write_request; /* false for read */
207 union {
208 struct ceph_snap_context *snapc; /* for writes */
209 u64 snap_id; /* for reads */
210 };
211 spinlock_t completion_lock;/* protects next_completion */
212 u32 next_completion;
213 rbd_img_callback_t callback;
214
215 u32 obj_request_count;
216 struct list_head obj_requests; /* rbd_obj_request structs */
217
218 struct kref kref;
219 };
220
221 #define for_each_obj_request(ireq, oreq) \
222 list_for_each_entry(oreq, &(ireq)->obj_requests, links)
223 #define for_each_obj_request_from(ireq, oreq) \
224 list_for_each_entry_from(oreq, &(ireq)->obj_requests, links)
225 #define for_each_obj_request_safe(ireq, oreq, n) \
226 list_for_each_entry_safe_reverse(oreq, n, &(ireq)->obj_requests, links)
227
228 struct rbd_snap {
229 struct device dev;
230 const char *name;
231 u64 size;
232 struct list_head node;
233 u64 id;
234 u64 features;
235 };
236
237 struct rbd_mapping {
238 u64 size;
239 u64 features;
240 bool read_only;
241 };
242
243 /*
244 * a single device
245 */
246 struct rbd_device {
247 int dev_id; /* blkdev unique id */
248
249 int major; /* blkdev assigned major */
250 struct gendisk *disk; /* blkdev's gendisk and rq */
251
252 u32 image_format; /* Either 1 or 2 */
253 struct rbd_client *rbd_client;
254
255 char name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */
256
257 spinlock_t lock; /* queue, flags, open_count */
258
259 struct rbd_image_header header;
260 unsigned long flags; /* possibly lock protected */
261 struct rbd_spec *spec;
262
263 char *header_name;
264
265 struct ceph_file_layout layout;
266
267 struct ceph_osd_event *watch_event;
268 struct rbd_obj_request *watch_request;
269
270 struct rbd_spec *parent_spec;
271 u64 parent_overlap;
272
273 /* protects updating the header */
274 struct rw_semaphore header_rwsem;
275
276 struct rbd_mapping mapping;
277
278 struct list_head node;
279
280 /* list of snapshots */
281 struct list_head snaps;
282
283 /* sysfs related */
284 struct device dev;
285 unsigned long open_count; /* protected by lock */
286 };
287
288 /*
289 * Flag bits for rbd_dev->flags. If atomicity is required,
290 * rbd_dev->lock is used to protect access.
291 *
292 * Currently, only the "removing" flag (which is coupled with the
293 * "open_count" field) requires atomic access.
294 */
295 enum rbd_dev_flags {
296 RBD_DEV_FLAG_EXISTS, /* mapped snapshot has not been deleted */
297 RBD_DEV_FLAG_REMOVING, /* this mapping is being removed */
298 };
299
300 static DEFINE_MUTEX(ctl_mutex); /* Serialize open/close/setup/teardown */
301
302 static LIST_HEAD(rbd_dev_list); /* devices */
303 static DEFINE_SPINLOCK(rbd_dev_list_lock);
304
305 static LIST_HEAD(rbd_client_list); /* clients */
306 static DEFINE_SPINLOCK(rbd_client_list_lock);
307
308 static int rbd_dev_snaps_update(struct rbd_device *rbd_dev);
309 static int rbd_dev_snaps_register(struct rbd_device *rbd_dev);
310
311 static void rbd_dev_release(struct device *dev);
312 static void rbd_remove_snap_dev(struct rbd_snap *snap);
313
314 static ssize_t rbd_add(struct bus_type *bus, const char *buf,
315 size_t count);
316 static ssize_t rbd_remove(struct bus_type *bus, const char *buf,
317 size_t count);
318
319 static struct bus_attribute rbd_bus_attrs[] = {
320 __ATTR(add, S_IWUSR, NULL, rbd_add),
321 __ATTR(remove, S_IWUSR, NULL, rbd_remove),
322 __ATTR_NULL
323 };
324
325 static struct bus_type rbd_bus_type = {
326 .name = "rbd",
327 .bus_attrs = rbd_bus_attrs,
328 };
329
330 static void rbd_root_dev_release(struct device *dev)
331 {
332 }
333
334 static struct device rbd_root_dev = {
335 .init_name = "rbd",
336 .release = rbd_root_dev_release,
337 };
338
339 static __printf(2, 3)
340 void rbd_warn(struct rbd_device *rbd_dev, const char *fmt, ...)
341 {
342 struct va_format vaf;
343 va_list args;
344
345 va_start(args, fmt);
346 vaf.fmt = fmt;
347 vaf.va = &args;
348
349 if (!rbd_dev)
350 printk(KERN_WARNING "%s: %pV\n", RBD_DRV_NAME, &vaf);
351 else if (rbd_dev->disk)
352 printk(KERN_WARNING "%s: %s: %pV\n",
353 RBD_DRV_NAME, rbd_dev->disk->disk_name, &vaf);
354 else if (rbd_dev->spec && rbd_dev->spec->image_name)
355 printk(KERN_WARNING "%s: image %s: %pV\n",
356 RBD_DRV_NAME, rbd_dev->spec->image_name, &vaf);
357 else if (rbd_dev->spec && rbd_dev->spec->image_id)
358 printk(KERN_WARNING "%s: id %s: %pV\n",
359 RBD_DRV_NAME, rbd_dev->spec->image_id, &vaf);
360 else /* punt */
361 printk(KERN_WARNING "%s: rbd_dev %p: %pV\n",
362 RBD_DRV_NAME, rbd_dev, &vaf);
363 va_end(args);
364 }
365
366 #ifdef RBD_DEBUG
367 #define rbd_assert(expr) \
368 if (unlikely(!(expr))) { \
369 printk(KERN_ERR "\nAssertion failure in %s() " \
370 "at line %d:\n\n" \
371 "\trbd_assert(%s);\n\n", \
372 __func__, __LINE__, #expr); \
373 BUG(); \
374 }
375 #else /* !RBD_DEBUG */
376 # define rbd_assert(expr) ((void) 0)
377 #endif /* !RBD_DEBUG */
378
379 static int rbd_dev_refresh(struct rbd_device *rbd_dev, u64 *hver);
380 static int rbd_dev_v2_refresh(struct rbd_device *rbd_dev, u64 *hver);
381
382 static int rbd_open(struct block_device *bdev, fmode_t mode)
383 {
384 struct rbd_device *rbd_dev = bdev->bd_disk->private_data;
385 bool removing = false;
386
387 if ((mode & FMODE_WRITE) && rbd_dev->mapping.read_only)
388 return -EROFS;
389
390 spin_lock_irq(&rbd_dev->lock);
391 if (test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
392 removing = true;
393 else
394 rbd_dev->open_count++;
395 spin_unlock_irq(&rbd_dev->lock);
396 if (removing)
397 return -ENOENT;
398
399 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
400 (void) get_device(&rbd_dev->dev);
401 set_device_ro(bdev, rbd_dev->mapping.read_only);
402 mutex_unlock(&ctl_mutex);
403
404 return 0;
405 }
406
407 static int rbd_release(struct gendisk *disk, fmode_t mode)
408 {
409 struct rbd_device *rbd_dev = disk->private_data;
410 unsigned long open_count_before;
411
412 spin_lock_irq(&rbd_dev->lock);
413 open_count_before = rbd_dev->open_count--;
414 spin_unlock_irq(&rbd_dev->lock);
415 rbd_assert(open_count_before > 0);
416
417 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
418 put_device(&rbd_dev->dev);
419 mutex_unlock(&ctl_mutex);
420
421 return 0;
422 }
423
424 static const struct block_device_operations rbd_bd_ops = {
425 .owner = THIS_MODULE,
426 .open = rbd_open,
427 .release = rbd_release,
428 };
429
430 /*
431 * Initialize an rbd client instance.
432 * We own *ceph_opts.
433 */
434 static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
435 {
436 struct rbd_client *rbdc;
437 int ret = -ENOMEM;
438
439 dout("%s:\n", __func__);
440 rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
441 if (!rbdc)
442 goto out_opt;
443
444 kref_init(&rbdc->kref);
445 INIT_LIST_HEAD(&rbdc->node);
446
447 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
448
449 rbdc->client = ceph_create_client(ceph_opts, rbdc, 0, 0);
450 if (IS_ERR(rbdc->client))
451 goto out_mutex;
452 ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
453
454 ret = ceph_open_session(rbdc->client);
455 if (ret < 0)
456 goto out_err;
457
458 spin_lock(&rbd_client_list_lock);
459 list_add_tail(&rbdc->node, &rbd_client_list);
460 spin_unlock(&rbd_client_list_lock);
461
462 mutex_unlock(&ctl_mutex);
463 dout("%s: rbdc %p\n", __func__, rbdc);
464
465 return rbdc;
466
467 out_err:
468 ceph_destroy_client(rbdc->client);
469 out_mutex:
470 mutex_unlock(&ctl_mutex);
471 kfree(rbdc);
472 out_opt:
473 if (ceph_opts)
474 ceph_destroy_options(ceph_opts);
475 dout("%s: error %d\n", __func__, ret);
476
477 return ERR_PTR(ret);
478 }
479
480 /*
481 * Find a ceph client with specific addr and configuration. If
482 * found, bump its reference count.
483 */
484 static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
485 {
486 struct rbd_client *client_node;
487 bool found = false;
488
489 if (ceph_opts->flags & CEPH_OPT_NOSHARE)
490 return NULL;
491
492 spin_lock(&rbd_client_list_lock);
493 list_for_each_entry(client_node, &rbd_client_list, node) {
494 if (!ceph_compare_options(ceph_opts, client_node->client)) {
495 kref_get(&client_node->kref);
496 found = true;
497 break;
498 }
499 }
500 spin_unlock(&rbd_client_list_lock);
501
502 return found ? client_node : NULL;
503 }
504
505 /*
506 * mount options
507 */
508 enum {
509 Opt_last_int,
510 /* int args above */
511 Opt_last_string,
512 /* string args above */
513 Opt_read_only,
514 Opt_read_write,
515 /* Boolean args above */
516 Opt_last_bool,
517 };
518
519 static match_table_t rbd_opts_tokens = {
520 /* int args above */
521 /* string args above */
522 {Opt_read_only, "read_only"},
523 {Opt_read_only, "ro"}, /* Alternate spelling */
524 {Opt_read_write, "read_write"},
525 {Opt_read_write, "rw"}, /* Alternate spelling */
526 /* Boolean args above */
527 {-1, NULL}
528 };
529
530 struct rbd_options {
531 bool read_only;
532 };
533
534 #define RBD_READ_ONLY_DEFAULT false
535
536 static int parse_rbd_opts_token(char *c, void *private)
537 {
538 struct rbd_options *rbd_opts = private;
539 substring_t argstr[MAX_OPT_ARGS];
540 int token, intval, ret;
541
542 token = match_token(c, rbd_opts_tokens, argstr);
543 if (token < 0)
544 return -EINVAL;
545
546 if (token < Opt_last_int) {
547 ret = match_int(&argstr[0], &intval);
548 if (ret < 0) {
549 pr_err("bad mount option arg (not int) "
550 "at '%s'\n", c);
551 return ret;
552 }
553 dout("got int token %d val %d\n", token, intval);
554 } else if (token > Opt_last_int && token < Opt_last_string) {
555 dout("got string token %d val %s\n", token,
556 argstr[0].from);
557 } else if (token > Opt_last_string && token < Opt_last_bool) {
558 dout("got Boolean token %d\n", token);
559 } else {
560 dout("got token %d\n", token);
561 }
562
563 switch (token) {
564 case Opt_read_only:
565 rbd_opts->read_only = true;
566 break;
567 case Opt_read_write:
568 rbd_opts->read_only = false;
569 break;
570 default:
571 rbd_assert(false);
572 break;
573 }
574 return 0;
575 }
576
577 /*
578 * Get a ceph client with specific addr and configuration, if one does
579 * not exist create it.
580 */
581 static struct rbd_client *rbd_get_client(struct ceph_options *ceph_opts)
582 {
583 struct rbd_client *rbdc;
584
585 rbdc = rbd_client_find(ceph_opts);
586 if (rbdc) /* using an existing client */
587 ceph_destroy_options(ceph_opts);
588 else
589 rbdc = rbd_client_create(ceph_opts);
590
591 return rbdc;
592 }
593
594 /*
595 * Destroy ceph client
596 *
597 * Caller must hold rbd_client_list_lock.
598 */
599 static void rbd_client_release(struct kref *kref)
600 {
601 struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
602
603 dout("%s: rbdc %p\n", __func__, rbdc);
604 spin_lock(&rbd_client_list_lock);
605 list_del(&rbdc->node);
606 spin_unlock(&rbd_client_list_lock);
607
608 ceph_destroy_client(rbdc->client);
609 kfree(rbdc);
610 }
611
612 /*
613 * Drop reference to ceph client node. If it's not referenced anymore, release
614 * it.
615 */
616 static void rbd_put_client(struct rbd_client *rbdc)
617 {
618 if (rbdc)
619 kref_put(&rbdc->kref, rbd_client_release);
620 }
621
622 static bool rbd_image_format_valid(u32 image_format)
623 {
624 return image_format == 1 || image_format == 2;
625 }
626
627 static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
628 {
629 size_t size;
630 u32 snap_count;
631
632 /* The header has to start with the magic rbd header text */
633 if (memcmp(&ondisk->text, RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT)))
634 return false;
635
636 /* The bio layer requires at least sector-sized I/O */
637
638 if (ondisk->options.order < SECTOR_SHIFT)
639 return false;
640
641 /* If we use u64 in a few spots we may be able to loosen this */
642
643 if (ondisk->options.order > 8 * sizeof (int) - 1)
644 return false;
645
646 /*
647 * The size of a snapshot header has to fit in a size_t, and
648 * that limits the number of snapshots.
649 */
650 snap_count = le32_to_cpu(ondisk->snap_count);
651 size = SIZE_MAX - sizeof (struct ceph_snap_context);
652 if (snap_count > size / sizeof (__le64))
653 return false;
654
655 /*
656 * Not only that, but the size of the entire the snapshot
657 * header must also be representable in a size_t.
658 */
659 size -= snap_count * sizeof (__le64);
660 if ((u64) size < le64_to_cpu(ondisk->snap_names_len))
661 return false;
662
663 return true;
664 }
665
666 /*
667 * Create a new header structure, translate header format from the on-disk
668 * header.
669 */
670 static int rbd_header_from_disk(struct rbd_image_header *header,
671 struct rbd_image_header_ondisk *ondisk)
672 {
673 u32 snap_count;
674 size_t len;
675 size_t size;
676 u32 i;
677
678 memset(header, 0, sizeof (*header));
679
680 snap_count = le32_to_cpu(ondisk->snap_count);
681
682 len = strnlen(ondisk->object_prefix, sizeof (ondisk->object_prefix));
683 header->object_prefix = kmalloc(len + 1, GFP_KERNEL);
684 if (!header->object_prefix)
685 return -ENOMEM;
686 memcpy(header->object_prefix, ondisk->object_prefix, len);
687 header->object_prefix[len] = '\0';
688
689 if (snap_count) {
690 u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len);
691
692 /* Save a copy of the snapshot names */
693
694 if (snap_names_len > (u64) SIZE_MAX)
695 return -EIO;
696 header->snap_names = kmalloc(snap_names_len, GFP_KERNEL);
697 if (!header->snap_names)
698 goto out_err;
699 /*
700 * Note that rbd_dev_v1_header_read() guarantees
701 * the ondisk buffer we're working with has
702 * snap_names_len bytes beyond the end of the
703 * snapshot id array, this memcpy() is safe.
704 */
705 memcpy(header->snap_names, &ondisk->snaps[snap_count],
706 snap_names_len);
707
708 /* Record each snapshot's size */
709
710 size = snap_count * sizeof (*header->snap_sizes);
711 header->snap_sizes = kmalloc(size, GFP_KERNEL);
712 if (!header->snap_sizes)
713 goto out_err;
714 for (i = 0; i < snap_count; i++)
715 header->snap_sizes[i] =
716 le64_to_cpu(ondisk->snaps[i].image_size);
717 } else {
718 WARN_ON(ondisk->snap_names_len);
719 header->snap_names = NULL;
720 header->snap_sizes = NULL;
721 }
722
723 header->features = 0; /* No features support in v1 images */
724 header->obj_order = ondisk->options.order;
725 header->crypt_type = ondisk->options.crypt_type;
726 header->comp_type = ondisk->options.comp_type;
727
728 /* Allocate and fill in the snapshot context */
729
730 header->image_size = le64_to_cpu(ondisk->image_size);
731 size = sizeof (struct ceph_snap_context);
732 size += snap_count * sizeof (header->snapc->snaps[0]);
733 header->snapc = kzalloc(size, GFP_KERNEL);
734 if (!header->snapc)
735 goto out_err;
736
737 atomic_set(&header->snapc->nref, 1);
738 header->snapc->seq = le64_to_cpu(ondisk->snap_seq);
739 header->snapc->num_snaps = snap_count;
740 for (i = 0; i < snap_count; i++)
741 header->snapc->snaps[i] =
742 le64_to_cpu(ondisk->snaps[i].id);
743
744 return 0;
745
746 out_err:
747 kfree(header->snap_sizes);
748 header->snap_sizes = NULL;
749 kfree(header->snap_names);
750 header->snap_names = NULL;
751 kfree(header->object_prefix);
752 header->object_prefix = NULL;
753
754 return -ENOMEM;
755 }
756
757 static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id)
758 {
759 struct rbd_snap *snap;
760
761 if (snap_id == CEPH_NOSNAP)
762 return RBD_SNAP_HEAD_NAME;
763
764 list_for_each_entry(snap, &rbd_dev->snaps, node)
765 if (snap_id == snap->id)
766 return snap->name;
767
768 return NULL;
769 }
770
771 static int snap_by_name(struct rbd_device *rbd_dev, const char *snap_name)
772 {
773
774 struct rbd_snap *snap;
775
776 list_for_each_entry(snap, &rbd_dev->snaps, node) {
777 if (!strcmp(snap_name, snap->name)) {
778 rbd_dev->spec->snap_id = snap->id;
779 rbd_dev->mapping.size = snap->size;
780 rbd_dev->mapping.features = snap->features;
781
782 return 0;
783 }
784 }
785
786 return -ENOENT;
787 }
788
789 static int rbd_dev_set_mapping(struct rbd_device *rbd_dev)
790 {
791 int ret;
792
793 if (!memcmp(rbd_dev->spec->snap_name, RBD_SNAP_HEAD_NAME,
794 sizeof (RBD_SNAP_HEAD_NAME))) {
795 rbd_dev->spec->snap_id = CEPH_NOSNAP;
796 rbd_dev->mapping.size = rbd_dev->header.image_size;
797 rbd_dev->mapping.features = rbd_dev->header.features;
798 ret = 0;
799 } else {
800 ret = snap_by_name(rbd_dev, rbd_dev->spec->snap_name);
801 if (ret < 0)
802 goto done;
803 rbd_dev->mapping.read_only = true;
804 }
805 set_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
806
807 done:
808 return ret;
809 }
810
811 static void rbd_header_free(struct rbd_image_header *header)
812 {
813 kfree(header->object_prefix);
814 header->object_prefix = NULL;
815 kfree(header->snap_sizes);
816 header->snap_sizes = NULL;
817 kfree(header->snap_names);
818 header->snap_names = NULL;
819 ceph_put_snap_context(header->snapc);
820 header->snapc = NULL;
821 }
822
823 static const char *rbd_segment_name(struct rbd_device *rbd_dev, u64 offset)
824 {
825 char *name;
826 u64 segment;
827 int ret;
828
829 name = kmalloc(MAX_OBJ_NAME_SIZE + 1, GFP_NOIO);
830 if (!name)
831 return NULL;
832 segment = offset >> rbd_dev->header.obj_order;
833 ret = snprintf(name, MAX_OBJ_NAME_SIZE + 1, "%s.%012llx",
834 rbd_dev->header.object_prefix, segment);
835 if (ret < 0 || ret > MAX_OBJ_NAME_SIZE) {
836 pr_err("error formatting segment name for #%llu (%d)\n",
837 segment, ret);
838 kfree(name);
839 name = NULL;
840 }
841
842 return name;
843 }
844
845 static u64 rbd_segment_offset(struct rbd_device *rbd_dev, u64 offset)
846 {
847 u64 segment_size = (u64) 1 << rbd_dev->header.obj_order;
848
849 return offset & (segment_size - 1);
850 }
851
852 static u64 rbd_segment_length(struct rbd_device *rbd_dev,
853 u64 offset, u64 length)
854 {
855 u64 segment_size = (u64) 1 << rbd_dev->header.obj_order;
856
857 offset &= segment_size - 1;
858
859 rbd_assert(length <= U64_MAX - offset);
860 if (offset + length > segment_size)
861 length = segment_size - offset;
862
863 return length;
864 }
865
866 /*
867 * returns the size of an object in the image
868 */
869 static u64 rbd_obj_bytes(struct rbd_image_header *header)
870 {
871 return 1 << header->obj_order;
872 }
873
874 /*
875 * bio helpers
876 */
877
878 static void bio_chain_put(struct bio *chain)
879 {
880 struct bio *tmp;
881
882 while (chain) {
883 tmp = chain;
884 chain = chain->bi_next;
885 bio_put(tmp);
886 }
887 }
888
889 /*
890 * zeros a bio chain, starting at specific offset
891 */
892 static void zero_bio_chain(struct bio *chain, int start_ofs)
893 {
894 struct bio_vec *bv;
895 unsigned long flags;
896 void *buf;
897 int i;
898 int pos = 0;
899
900 while (chain) {
901 bio_for_each_segment(bv, chain, i) {
902 if (pos + bv->bv_len > start_ofs) {
903 int remainder = max(start_ofs - pos, 0);
904 buf = bvec_kmap_irq(bv, &flags);
905 memset(buf + remainder, 0,
906 bv->bv_len - remainder);
907 bvec_kunmap_irq(buf, &flags);
908 }
909 pos += bv->bv_len;
910 }
911
912 chain = chain->bi_next;
913 }
914 }
915
916 /*
917 * Clone a portion of a bio, starting at the given byte offset
918 * and continuing for the number of bytes indicated.
919 */
920 static struct bio *bio_clone_range(struct bio *bio_src,
921 unsigned int offset,
922 unsigned int len,
923 gfp_t gfpmask)
924 {
925 struct bio_vec *bv;
926 unsigned int resid;
927 unsigned short idx;
928 unsigned int voff;
929 unsigned short end_idx;
930 unsigned short vcnt;
931 struct bio *bio;
932
933 /* Handle the easy case for the caller */
934
935 if (!offset && len == bio_src->bi_size)
936 return bio_clone(bio_src, gfpmask);
937
938 if (WARN_ON_ONCE(!len))
939 return NULL;
940 if (WARN_ON_ONCE(len > bio_src->bi_size))
941 return NULL;
942 if (WARN_ON_ONCE(offset > bio_src->bi_size - len))
943 return NULL;
944
945 /* Find first affected segment... */
946
947 resid = offset;
948 __bio_for_each_segment(bv, bio_src, idx, 0) {
949 if (resid < bv->bv_len)
950 break;
951 resid -= bv->bv_len;
952 }
953 voff = resid;
954
955 /* ...and the last affected segment */
956
957 resid += len;
958 __bio_for_each_segment(bv, bio_src, end_idx, idx) {
959 if (resid <= bv->bv_len)
960 break;
961 resid -= bv->bv_len;
962 }
963 vcnt = end_idx - idx + 1;
964
965 /* Build the clone */
966
967 bio = bio_alloc(gfpmask, (unsigned int) vcnt);
968 if (!bio)
969 return NULL; /* ENOMEM */
970
971 bio->bi_bdev = bio_src->bi_bdev;
972 bio->bi_sector = bio_src->bi_sector + (offset >> SECTOR_SHIFT);
973 bio->bi_rw = bio_src->bi_rw;
974 bio->bi_flags |= 1 << BIO_CLONED;
975
976 /*
977 * Copy over our part of the bio_vec, then update the first
978 * and last (or only) entries.
979 */
980 memcpy(&bio->bi_io_vec[0], &bio_src->bi_io_vec[idx],
981 vcnt * sizeof (struct bio_vec));
982 bio->bi_io_vec[0].bv_offset += voff;
983 if (vcnt > 1) {
984 bio->bi_io_vec[0].bv_len -= voff;
985 bio->bi_io_vec[vcnt - 1].bv_len = resid;
986 } else {
987 bio->bi_io_vec[0].bv_len = len;
988 }
989
990 bio->bi_vcnt = vcnt;
991 bio->bi_size = len;
992 bio->bi_idx = 0;
993
994 return bio;
995 }
996
997 /*
998 * Clone a portion of a bio chain, starting at the given byte offset
999 * into the first bio in the source chain and continuing for the
1000 * number of bytes indicated. The result is another bio chain of
1001 * exactly the given length, or a null pointer on error.
1002 *
1003 * The bio_src and offset parameters are both in-out. On entry they
1004 * refer to the first source bio and the offset into that bio where
1005 * the start of data to be cloned is located.
1006 *
1007 * On return, bio_src is updated to refer to the bio in the source
1008 * chain that contains first un-cloned byte, and *offset will
1009 * contain the offset of that byte within that bio.
1010 */
1011 static struct bio *bio_chain_clone_range(struct bio **bio_src,
1012 unsigned int *offset,
1013 unsigned int len,
1014 gfp_t gfpmask)
1015 {
1016 struct bio *bi = *bio_src;
1017 unsigned int off = *offset;
1018 struct bio *chain = NULL;
1019 struct bio **end;
1020
1021 /* Build up a chain of clone bios up to the limit */
1022
1023 if (!bi || off >= bi->bi_size || !len)
1024 return NULL; /* Nothing to clone */
1025
1026 end = &chain;
1027 while (len) {
1028 unsigned int bi_size;
1029 struct bio *bio;
1030
1031 if (!bi) {
1032 rbd_warn(NULL, "bio_chain exhausted with %u left", len);
1033 goto out_err; /* EINVAL; ran out of bio's */
1034 }
1035 bi_size = min_t(unsigned int, bi->bi_size - off, len);
1036 bio = bio_clone_range(bi, off, bi_size, gfpmask);
1037 if (!bio)
1038 goto out_err; /* ENOMEM */
1039
1040 *end = bio;
1041 end = &bio->bi_next;
1042
1043 off += bi_size;
1044 if (off == bi->bi_size) {
1045 bi = bi->bi_next;
1046 off = 0;
1047 }
1048 len -= bi_size;
1049 }
1050 *bio_src = bi;
1051 *offset = off;
1052
1053 return chain;
1054 out_err:
1055 bio_chain_put(chain);
1056
1057 return NULL;
1058 }
1059
1060 static void rbd_obj_request_get(struct rbd_obj_request *obj_request)
1061 {
1062 dout("%s: obj %p (was %d)\n", __func__, obj_request,
1063 atomic_read(&obj_request->kref.refcount));
1064 kref_get(&obj_request->kref);
1065 }
1066
1067 static void rbd_obj_request_destroy(struct kref *kref);
1068 static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
1069 {
1070 rbd_assert(obj_request != NULL);
1071 dout("%s: obj %p (was %d)\n", __func__, obj_request,
1072 atomic_read(&obj_request->kref.refcount));
1073 kref_put(&obj_request->kref, rbd_obj_request_destroy);
1074 }
1075
1076 static void rbd_img_request_get(struct rbd_img_request *img_request)
1077 {
1078 dout("%s: img %p (was %d)\n", __func__, img_request,
1079 atomic_read(&img_request->kref.refcount));
1080 kref_get(&img_request->kref);
1081 }
1082
1083 static void rbd_img_request_destroy(struct kref *kref);
1084 static void rbd_img_request_put(struct rbd_img_request *img_request)
1085 {
1086 rbd_assert(img_request != NULL);
1087 dout("%s: img %p (was %d)\n", __func__, img_request,
1088 atomic_read(&img_request->kref.refcount));
1089 kref_put(&img_request->kref, rbd_img_request_destroy);
1090 }
1091
1092 static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request,
1093 struct rbd_obj_request *obj_request)
1094 {
1095 rbd_assert(obj_request->img_request == NULL);
1096
1097 rbd_obj_request_get(obj_request);
1098 obj_request->img_request = img_request;
1099 obj_request->which = img_request->obj_request_count;
1100 rbd_assert(obj_request->which != BAD_WHICH);
1101 img_request->obj_request_count++;
1102 list_add_tail(&obj_request->links, &img_request->obj_requests);
1103 dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request,
1104 obj_request->which);
1105 }
1106
1107 static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
1108 struct rbd_obj_request *obj_request)
1109 {
1110 rbd_assert(obj_request->which != BAD_WHICH);
1111
1112 dout("%s: img %p obj %p w=%u\n", __func__, img_request, obj_request,
1113 obj_request->which);
1114 list_del(&obj_request->links);
1115 rbd_assert(img_request->obj_request_count > 0);
1116 img_request->obj_request_count--;
1117 rbd_assert(obj_request->which == img_request->obj_request_count);
1118 obj_request->which = BAD_WHICH;
1119 rbd_assert(obj_request->img_request == img_request);
1120 obj_request->img_request = NULL;
1121 obj_request->callback = NULL;
1122 rbd_obj_request_put(obj_request);
1123 }
1124
1125 static bool obj_request_type_valid(enum obj_request_type type)
1126 {
1127 switch (type) {
1128 case OBJ_REQUEST_NODATA:
1129 case OBJ_REQUEST_BIO:
1130 case OBJ_REQUEST_PAGES:
1131 return true;
1132 default:
1133 return false;
1134 }
1135 }
1136
1137 static int rbd_obj_request_submit(struct ceph_osd_client *osdc,
1138 struct rbd_obj_request *obj_request)
1139 {
1140 dout("%s: osdc %p obj %p\n", __func__, osdc, obj_request);
1141
1142 return ceph_osdc_start_request(osdc, obj_request->osd_req, false);
1143 }
1144
1145 static void rbd_img_request_complete(struct rbd_img_request *img_request)
1146 {
1147 dout("%s: img %p\n", __func__, img_request);
1148 if (img_request->callback)
1149 img_request->callback(img_request);
1150 else
1151 rbd_img_request_put(img_request);
1152 }
1153
1154 /* Caller is responsible for rbd_obj_request_destroy(obj_request) */
1155
1156 static int rbd_obj_request_wait(struct rbd_obj_request *obj_request)
1157 {
1158 dout("%s: obj %p\n", __func__, obj_request);
1159
1160 return wait_for_completion_interruptible(&obj_request->completion);
1161 }
1162
1163 static void obj_request_done_init(struct rbd_obj_request *obj_request)
1164 {
1165 atomic_set(&obj_request->done, 0);
1166 smp_wmb();
1167 }
1168
1169 static void obj_request_done_set(struct rbd_obj_request *obj_request)
1170 {
1171 int done;
1172
1173 done = atomic_inc_return(&obj_request->done);
1174 if (done > 1) {
1175 struct rbd_img_request *img_request = obj_request->img_request;
1176 struct rbd_device *rbd_dev;
1177
1178 rbd_dev = img_request ? img_request->rbd_dev : NULL;
1179 rbd_warn(rbd_dev, "obj_request %p was already done\n",
1180 obj_request);
1181 }
1182 }
1183
1184 static bool obj_request_done_test(struct rbd_obj_request *obj_request)
1185 {
1186 smp_mb();
1187 return atomic_read(&obj_request->done) != 0;
1188 }
1189
1190 static void
1191 rbd_img_obj_request_read_callback(struct rbd_obj_request *obj_request)
1192 {
1193 dout("%s: obj %p img %p result %d %llu/%llu\n", __func__,
1194 obj_request, obj_request->img_request, obj_request->result,
1195 obj_request->xferred, obj_request->length);
1196 /*
1197 * ENOENT means a hole in the image. We zero-fill the
1198 * entire length of the request. A short read also implies
1199 * zero-fill to the end of the request. Either way we
1200 * update the xferred count to indicate the whole request
1201 * was satisfied.
1202 */
1203 BUG_ON(obj_request->type != OBJ_REQUEST_BIO);
1204 if (obj_request->result == -ENOENT) {
1205 zero_bio_chain(obj_request->bio_list, 0);
1206 obj_request->result = 0;
1207 obj_request->xferred = obj_request->length;
1208 } else if (obj_request->xferred < obj_request->length &&
1209 !obj_request->result) {
1210 zero_bio_chain(obj_request->bio_list, obj_request->xferred);
1211 obj_request->xferred = obj_request->length;
1212 }
1213 obj_request_done_set(obj_request);
1214 }
1215
1216 static void rbd_obj_request_complete(struct rbd_obj_request *obj_request)
1217 {
1218 dout("%s: obj %p cb %p\n", __func__, obj_request,
1219 obj_request->callback);
1220 if (obj_request->callback)
1221 obj_request->callback(obj_request);
1222 else
1223 complete_all(&obj_request->completion);
1224 }
1225
1226 static void rbd_osd_trivial_callback(struct rbd_obj_request *obj_request)
1227 {
1228 dout("%s: obj %p\n", __func__, obj_request);
1229 obj_request_done_set(obj_request);
1230 }
1231
1232 static void rbd_osd_read_callback(struct rbd_obj_request *obj_request)
1233 {
1234 dout("%s: obj %p result %d %llu/%llu\n", __func__, obj_request,
1235 obj_request->result, obj_request->xferred, obj_request->length);
1236 if (obj_request->img_request)
1237 rbd_img_obj_request_read_callback(obj_request);
1238 else
1239 obj_request_done_set(obj_request);
1240 }
1241
1242 static void rbd_osd_write_callback(struct rbd_obj_request *obj_request)
1243 {
1244 dout("%s: obj %p result %d %llu\n", __func__, obj_request,
1245 obj_request->result, obj_request->length);
1246 /*
1247 * There is no such thing as a successful short write.
1248 * Our xferred value is the number of bytes transferred
1249 * back. Set it to our originally-requested length.
1250 */
1251 obj_request->xferred = obj_request->length;
1252 obj_request_done_set(obj_request);
1253 }
1254
1255 /*
1256 * For a simple stat call there's nothing to do. We'll do more if
1257 * this is part of a write sequence for a layered image.
1258 */
1259 static void rbd_osd_stat_callback(struct rbd_obj_request *obj_request)
1260 {
1261 dout("%s: obj %p\n", __func__, obj_request);
1262 obj_request_done_set(obj_request);
1263 }
1264
1265 static void rbd_osd_req_callback(struct ceph_osd_request *osd_req,
1266 struct ceph_msg *msg)
1267 {
1268 struct rbd_obj_request *obj_request = osd_req->r_priv;
1269 u16 opcode;
1270
1271 dout("%s: osd_req %p msg %p\n", __func__, osd_req, msg);
1272 rbd_assert(osd_req == obj_request->osd_req);
1273 rbd_assert(!!obj_request->img_request ^
1274 (obj_request->which == BAD_WHICH));
1275
1276 if (osd_req->r_result < 0)
1277 obj_request->result = osd_req->r_result;
1278 obj_request->version = le64_to_cpu(osd_req->r_reassert_version.version);
1279
1280 WARN_ON(osd_req->r_num_ops != 1); /* For now */
1281
1282 /*
1283 * We support a 64-bit length, but ultimately it has to be
1284 * passed to blk_end_request(), which takes an unsigned int.
1285 */
1286 obj_request->xferred = osd_req->r_reply_op_len[0];
1287 rbd_assert(obj_request->xferred < (u64) UINT_MAX);
1288 opcode = osd_req->r_ops[0].op;
1289 switch (opcode) {
1290 case CEPH_OSD_OP_READ:
1291 rbd_osd_read_callback(obj_request);
1292 break;
1293 case CEPH_OSD_OP_WRITE:
1294 rbd_osd_write_callback(obj_request);
1295 break;
1296 case CEPH_OSD_OP_STAT:
1297 rbd_osd_stat_callback(obj_request);
1298 break;
1299 case CEPH_OSD_OP_CALL:
1300 case CEPH_OSD_OP_NOTIFY_ACK:
1301 case CEPH_OSD_OP_WATCH:
1302 rbd_osd_trivial_callback(obj_request);
1303 break;
1304 default:
1305 rbd_warn(NULL, "%s: unsupported op %hu\n",
1306 obj_request->object_name, (unsigned short) opcode);
1307 break;
1308 }
1309
1310 if (obj_request_done_test(obj_request))
1311 rbd_obj_request_complete(obj_request);
1312 }
1313
1314 static void rbd_osd_req_format_op(struct rbd_obj_request *obj_request,
1315 bool write_request)
1316 {
1317 struct rbd_img_request *img_request = obj_request->img_request;
1318 struct ceph_snap_context *snapc = NULL;
1319 u64 snap_id = CEPH_NOSNAP;
1320 struct timespec *mtime = NULL;
1321 struct timespec now;
1322
1323 rbd_assert(obj_request->osd_req != NULL);
1324
1325 if (write_request) {
1326 now = CURRENT_TIME;
1327 mtime = &now;
1328 if (img_request)
1329 snapc = img_request->snapc;
1330 } else if (img_request) {
1331 snap_id = img_request->snap_id;
1332 }
1333
1334 ceph_osdc_build_request(obj_request->osd_req, obj_request->offset,
1335 snapc, snap_id, mtime);
1336 }
1337
1338 static struct ceph_osd_request *rbd_osd_req_create(
1339 struct rbd_device *rbd_dev,
1340 bool write_request,
1341 struct rbd_obj_request *obj_request)
1342 {
1343 struct rbd_img_request *img_request = obj_request->img_request;
1344 struct ceph_snap_context *snapc = NULL;
1345 struct ceph_osd_client *osdc;
1346 struct ceph_osd_request *osd_req;
1347 struct ceph_osd_data *osd_data;
1348 u64 offset = obj_request->offset;
1349
1350 if (img_request) {
1351 rbd_assert(img_request->write_request == write_request);
1352 if (img_request->write_request)
1353 snapc = img_request->snapc;
1354 }
1355
1356 /* Allocate and initialize the request, for the single op */
1357
1358 osdc = &rbd_dev->rbd_client->client->osdc;
1359 osd_req = ceph_osdc_alloc_request(osdc, snapc, 1, false, GFP_ATOMIC);
1360 if (!osd_req)
1361 return NULL; /* ENOMEM */
1362 osd_data = write_request ? &osd_req->r_data_out : &osd_req->r_data_in;
1363
1364 rbd_assert(obj_request_type_valid(obj_request->type));
1365 switch (obj_request->type) {
1366 case OBJ_REQUEST_NODATA:
1367 break; /* Nothing to do */
1368 case OBJ_REQUEST_BIO:
1369 rbd_assert(obj_request->bio_list != NULL);
1370 ceph_osd_data_bio_init(osd_data, obj_request->bio_list,
1371 obj_request->length);
1372 break;
1373 case OBJ_REQUEST_PAGES:
1374 ceph_osd_data_pages_init(osd_data, obj_request->pages,
1375 obj_request->length, offset & ~PAGE_MASK,
1376 false, false);
1377 break;
1378 }
1379
1380 if (write_request)
1381 osd_req->r_flags = CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK;
1382 else
1383 osd_req->r_flags = CEPH_OSD_FLAG_READ;
1384
1385 osd_req->r_callback = rbd_osd_req_callback;
1386 osd_req->r_priv = obj_request;
1387
1388 osd_req->r_oid_len = strlen(obj_request->object_name);
1389 rbd_assert(osd_req->r_oid_len < sizeof (osd_req->r_oid));
1390 memcpy(osd_req->r_oid, obj_request->object_name, osd_req->r_oid_len);
1391
1392 osd_req->r_file_layout = rbd_dev->layout; /* struct */
1393
1394 return osd_req;
1395 }
1396
1397 static void rbd_osd_req_destroy(struct ceph_osd_request *osd_req)
1398 {
1399 ceph_osdc_put_request(osd_req);
1400 }
1401
1402 /* object_name is assumed to be a non-null pointer and NUL-terminated */
1403
1404 static struct rbd_obj_request *rbd_obj_request_create(const char *object_name,
1405 u64 offset, u64 length,
1406 enum obj_request_type type)
1407 {
1408 struct rbd_obj_request *obj_request;
1409 size_t size;
1410 char *name;
1411
1412 rbd_assert(obj_request_type_valid(type));
1413
1414 size = strlen(object_name) + 1;
1415 obj_request = kzalloc(sizeof (*obj_request) + size, GFP_KERNEL);
1416 if (!obj_request)
1417 return NULL;
1418
1419 name = (char *)(obj_request + 1);
1420 obj_request->object_name = memcpy(name, object_name, size);
1421 obj_request->offset = offset;
1422 obj_request->length = length;
1423 obj_request->which = BAD_WHICH;
1424 obj_request->type = type;
1425 INIT_LIST_HEAD(&obj_request->links);
1426 obj_request_done_init(obj_request);
1427 init_completion(&obj_request->completion);
1428 kref_init(&obj_request->kref);
1429
1430 dout("%s: \"%s\" %llu/%llu %d -> obj %p\n", __func__, object_name,
1431 offset, length, (int)type, obj_request);
1432
1433 return obj_request;
1434 }
1435
1436 static void rbd_obj_request_destroy(struct kref *kref)
1437 {
1438 struct rbd_obj_request *obj_request;
1439
1440 obj_request = container_of(kref, struct rbd_obj_request, kref);
1441
1442 dout("%s: obj %p\n", __func__, obj_request);
1443
1444 rbd_assert(obj_request->img_request == NULL);
1445 rbd_assert(obj_request->which == BAD_WHICH);
1446
1447 if (obj_request->osd_req)
1448 rbd_osd_req_destroy(obj_request->osd_req);
1449
1450 rbd_assert(obj_request_type_valid(obj_request->type));
1451 switch (obj_request->type) {
1452 case OBJ_REQUEST_NODATA:
1453 break; /* Nothing to do */
1454 case OBJ_REQUEST_BIO:
1455 if (obj_request->bio_list)
1456 bio_chain_put(obj_request->bio_list);
1457 break;
1458 case OBJ_REQUEST_PAGES:
1459 if (obj_request->pages)
1460 ceph_release_page_vector(obj_request->pages,
1461 obj_request->page_count);
1462 break;
1463 }
1464
1465 kfree(obj_request);
1466 }
1467
1468 /*
1469 * Caller is responsible for filling in the list of object requests
1470 * that comprises the image request, and the Linux request pointer
1471 * (if there is one).
1472 */
1473 static struct rbd_img_request *rbd_img_request_create(
1474 struct rbd_device *rbd_dev,
1475 u64 offset, u64 length,
1476 bool write_request)
1477 {
1478 struct rbd_img_request *img_request;
1479 struct ceph_snap_context *snapc = NULL;
1480
1481 img_request = kmalloc(sizeof (*img_request), GFP_ATOMIC);
1482 if (!img_request)
1483 return NULL;
1484
1485 if (write_request) {
1486 down_read(&rbd_dev->header_rwsem);
1487 snapc = ceph_get_snap_context(rbd_dev->header.snapc);
1488 up_read(&rbd_dev->header_rwsem);
1489 if (WARN_ON(!snapc)) {
1490 kfree(img_request);
1491 return NULL; /* Shouldn't happen */
1492 }
1493 }
1494
1495 img_request->rq = NULL;
1496 img_request->rbd_dev = rbd_dev;
1497 img_request->offset = offset;
1498 img_request->length = length;
1499 img_request->write_request = write_request;
1500 if (write_request)
1501 img_request->snapc = snapc;
1502 else
1503 img_request->snap_id = rbd_dev->spec->snap_id;
1504 spin_lock_init(&img_request->completion_lock);
1505 img_request->next_completion = 0;
1506 img_request->callback = NULL;
1507 img_request->obj_request_count = 0;
1508 INIT_LIST_HEAD(&img_request->obj_requests);
1509 kref_init(&img_request->kref);
1510
1511 rbd_img_request_get(img_request); /* Avoid a warning */
1512 rbd_img_request_put(img_request); /* TEMPORARY */
1513
1514 dout("%s: rbd_dev %p %s %llu/%llu -> img %p\n", __func__, rbd_dev,
1515 write_request ? "write" : "read", offset, length,
1516 img_request);
1517
1518 return img_request;
1519 }
1520
1521 static void rbd_img_request_destroy(struct kref *kref)
1522 {
1523 struct rbd_img_request *img_request;
1524 struct rbd_obj_request *obj_request;
1525 struct rbd_obj_request *next_obj_request;
1526
1527 img_request = container_of(kref, struct rbd_img_request, kref);
1528
1529 dout("%s: img %p\n", __func__, img_request);
1530
1531 for_each_obj_request_safe(img_request, obj_request, next_obj_request)
1532 rbd_img_obj_request_del(img_request, obj_request);
1533 rbd_assert(img_request->obj_request_count == 0);
1534
1535 if (img_request->write_request)
1536 ceph_put_snap_context(img_request->snapc);
1537
1538 kfree(img_request);
1539 }
1540
1541 static int rbd_img_request_fill_bio(struct rbd_img_request *img_request,
1542 struct bio *bio_list)
1543 {
1544 struct rbd_device *rbd_dev = img_request->rbd_dev;
1545 struct rbd_obj_request *obj_request = NULL;
1546 struct rbd_obj_request *next_obj_request;
1547 bool write_request = img_request->write_request;
1548 unsigned int bio_offset;
1549 u64 image_offset;
1550 u64 resid;
1551 u16 opcode;
1552
1553 dout("%s: img %p bio %p\n", __func__, img_request, bio_list);
1554
1555 opcode = write_request ? CEPH_OSD_OP_WRITE : CEPH_OSD_OP_READ;
1556 bio_offset = 0;
1557 image_offset = img_request->offset;
1558 rbd_assert(image_offset == bio_list->bi_sector << SECTOR_SHIFT);
1559 resid = img_request->length;
1560 rbd_assert(resid > 0);
1561 while (resid) {
1562 const char *object_name;
1563 unsigned int clone_size;
1564 struct ceph_osd_req_op *op;
1565 u64 offset;
1566 u64 length;
1567
1568 object_name = rbd_segment_name(rbd_dev, image_offset);
1569 if (!object_name)
1570 goto out_unwind;
1571 offset = rbd_segment_offset(rbd_dev, image_offset);
1572 length = rbd_segment_length(rbd_dev, image_offset, resid);
1573 obj_request = rbd_obj_request_create(object_name,
1574 offset, length,
1575 OBJ_REQUEST_BIO);
1576 kfree(object_name); /* object request has its own copy */
1577 if (!obj_request)
1578 goto out_unwind;
1579
1580 rbd_assert(length <= (u64) UINT_MAX);
1581 clone_size = (unsigned int) length;
1582 obj_request->bio_list = bio_chain_clone_range(&bio_list,
1583 &bio_offset, clone_size,
1584 GFP_ATOMIC);
1585 if (!obj_request->bio_list)
1586 goto out_partial;
1587
1588 obj_request->osd_req = rbd_osd_req_create(rbd_dev,
1589 write_request, obj_request);
1590 if (!obj_request->osd_req)
1591 goto out_partial;
1592
1593 op = &obj_request->osd_req->r_ops[0];
1594 osd_req_op_extent_init(op, opcode, offset, length, 0, 0);
1595 rbd_osd_req_format_op(obj_request, write_request);
1596
1597 /* status and version are initially zero-filled */
1598
1599 rbd_img_obj_request_add(img_request, obj_request);
1600
1601 image_offset += length;
1602 resid -= length;
1603 }
1604
1605 return 0;
1606
1607 out_partial:
1608 rbd_obj_request_put(obj_request);
1609 out_unwind:
1610 for_each_obj_request_safe(img_request, obj_request, next_obj_request)
1611 rbd_obj_request_put(obj_request);
1612
1613 return -ENOMEM;
1614 }
1615
1616 static void rbd_img_obj_callback(struct rbd_obj_request *obj_request)
1617 {
1618 struct rbd_img_request *img_request;
1619 u32 which = obj_request->which;
1620 bool more = true;
1621
1622 img_request = obj_request->img_request;
1623
1624 dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
1625 rbd_assert(img_request != NULL);
1626 rbd_assert(img_request->rq != NULL);
1627 rbd_assert(img_request->obj_request_count > 0);
1628 rbd_assert(which != BAD_WHICH);
1629 rbd_assert(which < img_request->obj_request_count);
1630 rbd_assert(which >= img_request->next_completion);
1631
1632 spin_lock_irq(&img_request->completion_lock);
1633 if (which != img_request->next_completion)
1634 goto out;
1635
1636 for_each_obj_request_from(img_request, obj_request) {
1637 unsigned int xferred;
1638 int result;
1639
1640 rbd_assert(more);
1641 rbd_assert(which < img_request->obj_request_count);
1642
1643 if (!obj_request_done_test(obj_request))
1644 break;
1645
1646 rbd_assert(obj_request->xferred <= (u64) UINT_MAX);
1647 xferred = (unsigned int) obj_request->xferred;
1648 result = (int) obj_request->result;
1649 if (result)
1650 rbd_warn(NULL, "obj_request %s result %d xferred %u\n",
1651 img_request->write_request ? "write" : "read",
1652 result, xferred);
1653
1654 more = blk_end_request(img_request->rq, result, xferred);
1655 which++;
1656 }
1657
1658 rbd_assert(more ^ (which == img_request->obj_request_count));
1659 img_request->next_completion = which;
1660 out:
1661 spin_unlock_irq(&img_request->completion_lock);
1662
1663 if (!more)
1664 rbd_img_request_complete(img_request);
1665 }
1666
1667 static int rbd_img_request_submit(struct rbd_img_request *img_request)
1668 {
1669 struct rbd_device *rbd_dev = img_request->rbd_dev;
1670 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1671 struct rbd_obj_request *obj_request;
1672 struct rbd_obj_request *next_obj_request;
1673
1674 dout("%s: img %p\n", __func__, img_request);
1675 for_each_obj_request_safe(img_request, obj_request, next_obj_request) {
1676 int ret;
1677
1678 obj_request->callback = rbd_img_obj_callback;
1679 ret = rbd_obj_request_submit(osdc, obj_request);
1680 if (ret)
1681 return ret;
1682 /*
1683 * The image request has its own reference to each
1684 * of its object requests, so we can safely drop the
1685 * initial one here.
1686 */
1687 rbd_obj_request_put(obj_request);
1688 }
1689
1690 return 0;
1691 }
1692
1693 static int rbd_obj_notify_ack(struct rbd_device *rbd_dev,
1694 u64 ver, u64 notify_id)
1695 {
1696 struct rbd_obj_request *obj_request;
1697 struct ceph_osd_req_op *op;
1698 struct ceph_osd_client *osdc;
1699 int ret;
1700
1701 obj_request = rbd_obj_request_create(rbd_dev->header_name, 0, 0,
1702 OBJ_REQUEST_NODATA);
1703 if (!obj_request)
1704 return -ENOMEM;
1705
1706 ret = -ENOMEM;
1707 obj_request->osd_req = rbd_osd_req_create(rbd_dev, false, obj_request);
1708 if (!obj_request->osd_req)
1709 goto out;
1710
1711 op = &obj_request->osd_req->r_ops[0];
1712 osd_req_op_watch_init(op, CEPH_OSD_OP_NOTIFY_ACK, notify_id, ver, 0);
1713 rbd_osd_req_format_op(obj_request, false);
1714
1715 osdc = &rbd_dev->rbd_client->client->osdc;
1716 obj_request->callback = rbd_obj_request_put;
1717 ret = rbd_obj_request_submit(osdc, obj_request);
1718 out:
1719 if (ret)
1720 rbd_obj_request_put(obj_request);
1721
1722 return ret;
1723 }
1724
1725 static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
1726 {
1727 struct rbd_device *rbd_dev = (struct rbd_device *)data;
1728 u64 hver;
1729 int rc;
1730
1731 if (!rbd_dev)
1732 return;
1733
1734 dout("%s: \"%s\" notify_id %llu opcode %u\n", __func__,
1735 rbd_dev->header_name, (unsigned long long) notify_id,
1736 (unsigned int) opcode);
1737 rc = rbd_dev_refresh(rbd_dev, &hver);
1738 if (rc)
1739 rbd_warn(rbd_dev, "got notification but failed to "
1740 " update snaps: %d\n", rc);
1741
1742 rbd_obj_notify_ack(rbd_dev, hver, notify_id);
1743 }
1744
1745 /*
1746 * Request sync osd watch/unwatch. The value of "start" determines
1747 * whether a watch request is being initiated or torn down.
1748 */
1749 static int rbd_dev_header_watch_sync(struct rbd_device *rbd_dev, int start)
1750 {
1751 struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1752 struct rbd_obj_request *obj_request;
1753 struct ceph_osd_req_op *op;
1754 int ret;
1755
1756 rbd_assert(start ^ !!rbd_dev->watch_event);
1757 rbd_assert(start ^ !!rbd_dev->watch_request);
1758
1759 if (start) {
1760 ret = ceph_osdc_create_event(osdc, rbd_watch_cb, rbd_dev,
1761 &rbd_dev->watch_event);
1762 if (ret < 0)
1763 return ret;
1764 rbd_assert(rbd_dev->watch_event != NULL);
1765 }
1766
1767 ret = -ENOMEM;
1768 obj_request = rbd_obj_request_create(rbd_dev->header_name, 0, 0,
1769 OBJ_REQUEST_NODATA);
1770 if (!obj_request)
1771 goto out_cancel;
1772
1773 obj_request->osd_req = rbd_osd_req_create(rbd_dev, true, obj_request);
1774 if (!obj_request->osd_req)
1775 goto out_cancel;
1776
1777 op = &obj_request->osd_req->r_ops[0];
1778 osd_req_op_watch_init(op, CEPH_OSD_OP_WATCH,
1779 rbd_dev->watch_event->cookie,
1780 rbd_dev->header.obj_version, start);
1781 rbd_osd_req_format_op(obj_request, true);
1782
1783 if (start)
1784 ceph_osdc_set_request_linger(osdc, obj_request->osd_req);
1785 else
1786 ceph_osdc_unregister_linger_request(osdc,
1787 rbd_dev->watch_request->osd_req);
1788 ret = rbd_obj_request_submit(osdc, obj_request);
1789 if (ret)
1790 goto out_cancel;
1791 ret = rbd_obj_request_wait(obj_request);
1792 if (ret)
1793 goto out_cancel;
1794 ret = obj_request->result;
1795 if (ret)
1796 goto out_cancel;
1797
1798 /*
1799 * A watch request is set to linger, so the underlying osd
1800 * request won't go away until we unregister it. We retain
1801 * a pointer to the object request during that time (in
1802 * rbd_dev->watch_request), so we'll keep a reference to
1803 * it. We'll drop that reference (below) after we've
1804 * unregistered it.
1805 */
1806 if (start) {
1807 rbd_dev->watch_request = obj_request;
1808
1809 return 0;
1810 }
1811
1812 /* We have successfully torn down the watch request */
1813
1814 rbd_obj_request_put(rbd_dev->watch_request);
1815 rbd_dev->watch_request = NULL;
1816 out_cancel:
1817 /* Cancel the event if we're tearing down, or on error */
1818 ceph_osdc_cancel_event(rbd_dev->watch_event);
1819 rbd_dev->watch_event = NULL;
1820 if (obj_request)
1821 rbd_obj_request_put(obj_request);
1822
1823 return ret;
1824 }
1825
1826 /*
1827 * Synchronous osd object method call
1828 */
1829 static int rbd_obj_method_sync(struct rbd_device *rbd_dev,
1830 const char *object_name,
1831 const char *class_name,
1832 const char *method_name,
1833 const char *outbound,
1834 size_t outbound_size,
1835 char *inbound,
1836 size_t inbound_size,
1837 u64 *version)
1838 {
1839 struct rbd_obj_request *obj_request;
1840 struct ceph_osd_client *osdc;
1841 struct ceph_osd_req_op *op;
1842 struct page **pages;
1843 u32 page_count;
1844 int ret;
1845
1846 /*
1847 * Method calls are ultimately read operations. The result
1848 * should placed into the inbound buffer provided. They
1849 * also supply outbound data--parameters for the object
1850 * method. Currently if this is present it will be a
1851 * snapshot id.
1852 */
1853 page_count = (u32) calc_pages_for(0, inbound_size);
1854 pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
1855 if (IS_ERR(pages))
1856 return PTR_ERR(pages);
1857
1858 ret = -ENOMEM;
1859 obj_request = rbd_obj_request_create(object_name, 0, inbound_size,
1860 OBJ_REQUEST_PAGES);
1861 if (!obj_request)
1862 goto out;
1863
1864 obj_request->pages = pages;
1865 obj_request->page_count = page_count;
1866
1867 obj_request->osd_req = rbd_osd_req_create(rbd_dev, false, obj_request);
1868 if (!obj_request->osd_req)
1869 goto out;
1870
1871 op = &obj_request->osd_req->r_ops[0];
1872 osd_req_op_cls_init(op, CEPH_OSD_OP_CALL, class_name, method_name,
1873 outbound, outbound_size);
1874 rbd_osd_req_format_op(obj_request, false);
1875
1876 osdc = &rbd_dev->rbd_client->client->osdc;
1877 ret = rbd_obj_request_submit(osdc, obj_request);
1878 if (ret)
1879 goto out;
1880 ret = rbd_obj_request_wait(obj_request);
1881 if (ret)
1882 goto out;
1883
1884 ret = obj_request->result;
1885 if (ret < 0)
1886 goto out;
1887 ret = 0;
1888 ceph_copy_from_page_vector(pages, inbound, 0, obj_request->xferred);
1889 if (version)
1890 *version = obj_request->version;
1891 out:
1892 if (obj_request)
1893 rbd_obj_request_put(obj_request);
1894 else
1895 ceph_release_page_vector(pages, page_count);
1896
1897 return ret;
1898 }
1899
1900 static void rbd_request_fn(struct request_queue *q)
1901 __releases(q->queue_lock) __acquires(q->queue_lock)
1902 {
1903 struct rbd_device *rbd_dev = q->queuedata;
1904 bool read_only = rbd_dev->mapping.read_only;
1905 struct request *rq;
1906 int result;
1907
1908 while ((rq = blk_fetch_request(q))) {
1909 bool write_request = rq_data_dir(rq) == WRITE;
1910 struct rbd_img_request *img_request;
1911 u64 offset;
1912 u64 length;
1913
1914 /* Ignore any non-FS requests that filter through. */
1915
1916 if (rq->cmd_type != REQ_TYPE_FS) {
1917 dout("%s: non-fs request type %d\n", __func__,
1918 (int) rq->cmd_type);
1919 __blk_end_request_all(rq, 0);
1920 continue;
1921 }
1922
1923 /* Ignore/skip any zero-length requests */
1924
1925 offset = (u64) blk_rq_pos(rq) << SECTOR_SHIFT;
1926 length = (u64) blk_rq_bytes(rq);
1927
1928 if (!length) {
1929 dout("%s: zero-length request\n", __func__);
1930 __blk_end_request_all(rq, 0);
1931 continue;
1932 }
1933
1934 spin_unlock_irq(q->queue_lock);
1935
1936 /* Disallow writes to a read-only device */
1937
1938 if (write_request) {
1939 result = -EROFS;
1940 if (read_only)
1941 goto end_request;
1942 rbd_assert(rbd_dev->spec->snap_id == CEPH_NOSNAP);
1943 }
1944
1945 /*
1946 * Quit early if the mapped snapshot no longer
1947 * exists. It's still possible the snapshot will
1948 * have disappeared by the time our request arrives
1949 * at the osd, but there's no sense in sending it if
1950 * we already know.
1951 */
1952 if (!test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags)) {
1953 dout("request for non-existent snapshot");
1954 rbd_assert(rbd_dev->spec->snap_id != CEPH_NOSNAP);
1955 result = -ENXIO;
1956 goto end_request;
1957 }
1958
1959 result = -EINVAL;
1960 if (WARN_ON(offset && length > U64_MAX - offset + 1))
1961 goto end_request; /* Shouldn't happen */
1962
1963 result = -ENOMEM;
1964 img_request = rbd_img_request_create(rbd_dev, offset, length,
1965 write_request);
1966 if (!img_request)
1967 goto end_request;
1968
1969 img_request->rq = rq;
1970
1971 result = rbd_img_request_fill_bio(img_request, rq->bio);
1972 if (!result)
1973 result = rbd_img_request_submit(img_request);
1974 if (result)
1975 rbd_img_request_put(img_request);
1976 end_request:
1977 spin_lock_irq(q->queue_lock);
1978 if (result < 0) {
1979 rbd_warn(rbd_dev, "obj_request %s result %d\n",
1980 write_request ? "write" : "read", result);
1981 __blk_end_request_all(rq, result);
1982 }
1983 }
1984 }
1985
1986 /*
1987 * a queue callback. Makes sure that we don't create a bio that spans across
1988 * multiple osd objects. One exception would be with a single page bios,
1989 * which we handle later at bio_chain_clone_range()
1990 */
1991 static int rbd_merge_bvec(struct request_queue *q, struct bvec_merge_data *bmd,
1992 struct bio_vec *bvec)
1993 {
1994 struct rbd_device *rbd_dev = q->queuedata;
1995 sector_t sector_offset;
1996 sector_t sectors_per_obj;
1997 sector_t obj_sector_offset;
1998 int ret;
1999
2000 /*
2001 * Find how far into its rbd object the partition-relative
2002 * bio start sector is to offset relative to the enclosing
2003 * device.
2004 */
2005 sector_offset = get_start_sect(bmd->bi_bdev) + bmd->bi_sector;
2006 sectors_per_obj = 1 << (rbd_dev->header.obj_order - SECTOR_SHIFT);
2007 obj_sector_offset = sector_offset & (sectors_per_obj - 1);
2008
2009 /*
2010 * Compute the number of bytes from that offset to the end
2011 * of the object. Account for what's already used by the bio.
2012 */
2013 ret = (int) (sectors_per_obj - obj_sector_offset) << SECTOR_SHIFT;
2014 if (ret > bmd->bi_size)
2015 ret -= bmd->bi_size;
2016 else
2017 ret = 0;
2018
2019 /*
2020 * Don't send back more than was asked for. And if the bio
2021 * was empty, let the whole thing through because: "Note
2022 * that a block device *must* allow a single page to be
2023 * added to an empty bio."
2024 */
2025 rbd_assert(bvec->bv_len <= PAGE_SIZE);
2026 if (ret > (int) bvec->bv_len || !bmd->bi_size)
2027 ret = (int) bvec->bv_len;
2028
2029 return ret;
2030 }
2031
2032 static void rbd_free_disk(struct rbd_device *rbd_dev)
2033 {
2034 struct gendisk *disk = rbd_dev->disk;
2035
2036 if (!disk)
2037 return;
2038
2039 if (disk->flags & GENHD_FL_UP)
2040 del_gendisk(disk);
2041 if (disk->queue)
2042 blk_cleanup_queue(disk->queue);
2043 put_disk(disk);
2044 }
2045
2046 static int rbd_obj_read_sync(struct rbd_device *rbd_dev,
2047 const char *object_name,
2048 u64 offset, u64 length,
2049 char *buf, u64 *version)
2050
2051 {
2052 struct rbd_obj_request *obj_request;
2053 struct ceph_osd_req_op *op;
2054 struct ceph_osd_client *osdc;
2055 struct page **pages = NULL;
2056 u32 page_count;
2057 size_t size;
2058 int ret;
2059
2060 page_count = (u32) calc_pages_for(offset, length);
2061 pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
2062 if (IS_ERR(pages))
2063 ret = PTR_ERR(pages);
2064
2065 ret = -ENOMEM;
2066 obj_request = rbd_obj_request_create(object_name, offset, length,
2067 OBJ_REQUEST_PAGES);
2068 if (!obj_request)
2069 goto out;
2070
2071 obj_request->pages = pages;
2072 obj_request->page_count = page_count;
2073
2074 obj_request->osd_req = rbd_osd_req_create(rbd_dev, false, obj_request);
2075 if (!obj_request->osd_req)
2076 goto out;
2077
2078 op = &obj_request->osd_req->r_ops[0];
2079 osd_req_op_extent_init(op, CEPH_OSD_OP_READ, offset, length, 0, 0);
2080 rbd_osd_req_format_op(obj_request, false);
2081
2082 osdc = &rbd_dev->rbd_client->client->osdc;
2083 ret = rbd_obj_request_submit(osdc, obj_request);
2084 if (ret)
2085 goto out;
2086 ret = rbd_obj_request_wait(obj_request);
2087 if (ret)
2088 goto out;
2089
2090 ret = obj_request->result;
2091 if (ret < 0)
2092 goto out;
2093
2094 rbd_assert(obj_request->xferred <= (u64) SIZE_MAX);
2095 size = (size_t) obj_request->xferred;
2096 ceph_copy_from_page_vector(pages, buf, 0, size);
2097 rbd_assert(size <= (size_t) INT_MAX);
2098 ret = (int) size;
2099 if (version)
2100 *version = obj_request->version;
2101 out:
2102 if (obj_request)
2103 rbd_obj_request_put(obj_request);
2104 else
2105 ceph_release_page_vector(pages, page_count);
2106
2107 return ret;
2108 }
2109
2110 /*
2111 * Read the complete header for the given rbd device.
2112 *
2113 * Returns a pointer to a dynamically-allocated buffer containing
2114 * the complete and validated header. Caller can pass the address
2115 * of a variable that will be filled in with the version of the
2116 * header object at the time it was read.
2117 *
2118 * Returns a pointer-coded errno if a failure occurs.
2119 */
2120 static struct rbd_image_header_ondisk *
2121 rbd_dev_v1_header_read(struct rbd_device *rbd_dev, u64 *version)
2122 {
2123 struct rbd_image_header_ondisk *ondisk = NULL;
2124 u32 snap_count = 0;
2125 u64 names_size = 0;
2126 u32 want_count;
2127 int ret;
2128
2129 /*
2130 * The complete header will include an array of its 64-bit
2131 * snapshot ids, followed by the names of those snapshots as
2132 * a contiguous block of NUL-terminated strings. Note that
2133 * the number of snapshots could change by the time we read
2134 * it in, in which case we re-read it.
2135 */
2136 do {
2137 size_t size;
2138
2139 kfree(ondisk);
2140
2141 size = sizeof (*ondisk);
2142 size += snap_count * sizeof (struct rbd_image_snap_ondisk);
2143 size += names_size;
2144 ondisk = kmalloc(size, GFP_KERNEL);
2145 if (!ondisk)
2146 return ERR_PTR(-ENOMEM);
2147
2148 ret = rbd_obj_read_sync(rbd_dev, rbd_dev->header_name,
2149 0, size,
2150 (char *) ondisk, version);
2151 if (ret < 0)
2152 goto out_err;
2153 if (WARN_ON((size_t) ret < size)) {
2154 ret = -ENXIO;
2155 rbd_warn(rbd_dev, "short header read (want %zd got %d)",
2156 size, ret);
2157 goto out_err;
2158 }
2159 if (!rbd_dev_ondisk_valid(ondisk)) {
2160 ret = -ENXIO;
2161 rbd_warn(rbd_dev, "invalid header");
2162 goto out_err;
2163 }
2164
2165 names_size = le64_to_cpu(ondisk->snap_names_len);
2166 want_count = snap_count;
2167 snap_count = le32_to_cpu(ondisk->snap_count);
2168 } while (snap_count != want_count);
2169
2170 return ondisk;
2171
2172 out_err:
2173 kfree(ondisk);
2174
2175 return ERR_PTR(ret);
2176 }
2177
2178 /*
2179 * reload the ondisk the header
2180 */
2181 static int rbd_read_header(struct rbd_device *rbd_dev,
2182 struct rbd_image_header *header)
2183 {
2184 struct rbd_image_header_ondisk *ondisk;
2185 u64 ver = 0;
2186 int ret;
2187
2188 ondisk = rbd_dev_v1_header_read(rbd_dev, &ver);
2189 if (IS_ERR(ondisk))
2190 return PTR_ERR(ondisk);
2191 ret = rbd_header_from_disk(header, ondisk);
2192 if (ret >= 0)
2193 header->obj_version = ver;
2194 kfree(ondisk);
2195
2196 return ret;
2197 }
2198
2199 static void rbd_remove_all_snaps(struct rbd_device *rbd_dev)
2200 {
2201 struct rbd_snap *snap;
2202 struct rbd_snap *next;
2203
2204 list_for_each_entry_safe(snap, next, &rbd_dev->snaps, node)
2205 rbd_remove_snap_dev(snap);
2206 }
2207
2208 static void rbd_update_mapping_size(struct rbd_device *rbd_dev)
2209 {
2210 sector_t size;
2211
2212 if (rbd_dev->spec->snap_id != CEPH_NOSNAP)
2213 return;
2214
2215 size = (sector_t) rbd_dev->header.image_size / SECTOR_SIZE;
2216 dout("setting size to %llu sectors", (unsigned long long) size);
2217 rbd_dev->mapping.size = (u64) size;
2218 set_capacity(rbd_dev->disk, size);
2219 }
2220
2221 /*
2222 * only read the first part of the ondisk header, without the snaps info
2223 */
2224 static int rbd_dev_v1_refresh(struct rbd_device *rbd_dev, u64 *hver)
2225 {
2226 int ret;
2227 struct rbd_image_header h;
2228
2229 ret = rbd_read_header(rbd_dev, &h);
2230 if (ret < 0)
2231 return ret;
2232
2233 down_write(&rbd_dev->header_rwsem);
2234
2235 /* Update image size, and check for resize of mapped image */
2236 rbd_dev->header.image_size = h.image_size;
2237 rbd_update_mapping_size(rbd_dev);
2238
2239 /* rbd_dev->header.object_prefix shouldn't change */
2240 kfree(rbd_dev->header.snap_sizes);
2241 kfree(rbd_dev->header.snap_names);
2242 /* osd requests may still refer to snapc */
2243 ceph_put_snap_context(rbd_dev->header.snapc);
2244
2245 if (hver)
2246 *hver = h.obj_version;
2247 rbd_dev->header.obj_version = h.obj_version;
2248 rbd_dev->header.image_size = h.image_size;
2249 rbd_dev->header.snapc = h.snapc;
2250 rbd_dev->header.snap_names = h.snap_names;
2251 rbd_dev->header.snap_sizes = h.snap_sizes;
2252 /* Free the extra copy of the object prefix */
2253 WARN_ON(strcmp(rbd_dev->header.object_prefix, h.object_prefix));
2254 kfree(h.object_prefix);
2255
2256 ret = rbd_dev_snaps_update(rbd_dev);
2257 if (!ret)
2258 ret = rbd_dev_snaps_register(rbd_dev);
2259
2260 up_write(&rbd_dev->header_rwsem);
2261
2262 return ret;
2263 }
2264
2265 static int rbd_dev_refresh(struct rbd_device *rbd_dev, u64 *hver)
2266 {
2267 int ret;
2268
2269 rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
2270 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2271 if (rbd_dev->image_format == 1)
2272 ret = rbd_dev_v1_refresh(rbd_dev, hver);
2273 else
2274 ret = rbd_dev_v2_refresh(rbd_dev, hver);
2275 mutex_unlock(&ctl_mutex);
2276
2277 return ret;
2278 }
2279
2280 static int rbd_init_disk(struct rbd_device *rbd_dev)
2281 {
2282 struct gendisk *disk;
2283 struct request_queue *q;
2284 u64 segment_size;
2285
2286 /* create gendisk info */
2287 disk = alloc_disk(RBD_MINORS_PER_MAJOR);
2288 if (!disk)
2289 return -ENOMEM;
2290
2291 snprintf(disk->disk_name, sizeof(disk->disk_name), RBD_DRV_NAME "%d",
2292 rbd_dev->dev_id);
2293 disk->major = rbd_dev->major;
2294 disk->first_minor = 0;
2295 disk->fops = &rbd_bd_ops;
2296 disk->private_data = rbd_dev;
2297
2298 q = blk_init_queue(rbd_request_fn, &rbd_dev->lock);
2299 if (!q)
2300 goto out_disk;
2301
2302 /* We use the default size, but let's be explicit about it. */
2303 blk_queue_physical_block_size(q, SECTOR_SIZE);
2304
2305 /* set io sizes to object size */
2306 segment_size = rbd_obj_bytes(&rbd_dev->header);
2307 blk_queue_max_hw_sectors(q, segment_size / SECTOR_SIZE);
2308 blk_queue_max_segment_size(q, segment_size);
2309 blk_queue_io_min(q, segment_size);
2310 blk_queue_io_opt(q, segment_size);
2311
2312 blk_queue_merge_bvec(q, rbd_merge_bvec);
2313 disk->queue = q;
2314
2315 q->queuedata = rbd_dev;
2316
2317 rbd_dev->disk = disk;
2318
2319 set_capacity(rbd_dev->disk, rbd_dev->mapping.size / SECTOR_SIZE);
2320
2321 return 0;
2322 out_disk:
2323 put_disk(disk);
2324
2325 return -ENOMEM;
2326 }
2327
2328 /*
2329 sysfs
2330 */
2331
2332 static struct rbd_device *dev_to_rbd_dev(struct device *dev)
2333 {
2334 return container_of(dev, struct rbd_device, dev);
2335 }
2336
2337 static ssize_t rbd_size_show(struct device *dev,
2338 struct device_attribute *attr, char *buf)
2339 {
2340 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2341 sector_t size;
2342
2343 down_read(&rbd_dev->header_rwsem);
2344 size = get_capacity(rbd_dev->disk);
2345 up_read(&rbd_dev->header_rwsem);
2346
2347 return sprintf(buf, "%llu\n", (unsigned long long) size * SECTOR_SIZE);
2348 }
2349
2350 /*
2351 * Note this shows the features for whatever's mapped, which is not
2352 * necessarily the base image.
2353 */
2354 static ssize_t rbd_features_show(struct device *dev,
2355 struct device_attribute *attr, char *buf)
2356 {
2357 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2358
2359 return sprintf(buf, "0x%016llx\n",
2360 (unsigned long long) rbd_dev->mapping.features);
2361 }
2362
2363 static ssize_t rbd_major_show(struct device *dev,
2364 struct device_attribute *attr, char *buf)
2365 {
2366 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2367
2368 return sprintf(buf, "%d\n", rbd_dev->major);
2369 }
2370
2371 static ssize_t rbd_client_id_show(struct device *dev,
2372 struct device_attribute *attr, char *buf)
2373 {
2374 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2375
2376 return sprintf(buf, "client%lld\n",
2377 ceph_client_id(rbd_dev->rbd_client->client));
2378 }
2379
2380 static ssize_t rbd_pool_show(struct device *dev,
2381 struct device_attribute *attr, char *buf)
2382 {
2383 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2384
2385 return sprintf(buf, "%s\n", rbd_dev->spec->pool_name);
2386 }
2387
2388 static ssize_t rbd_pool_id_show(struct device *dev,
2389 struct device_attribute *attr, char *buf)
2390 {
2391 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2392
2393 return sprintf(buf, "%llu\n",
2394 (unsigned long long) rbd_dev->spec->pool_id);
2395 }
2396
2397 static ssize_t rbd_name_show(struct device *dev,
2398 struct device_attribute *attr, char *buf)
2399 {
2400 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2401
2402 if (rbd_dev->spec->image_name)
2403 return sprintf(buf, "%s\n", rbd_dev->spec->image_name);
2404
2405 return sprintf(buf, "(unknown)\n");
2406 }
2407
2408 static ssize_t rbd_image_id_show(struct device *dev,
2409 struct device_attribute *attr, char *buf)
2410 {
2411 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2412
2413 return sprintf(buf, "%s\n", rbd_dev->spec->image_id);
2414 }
2415
2416 /*
2417 * Shows the name of the currently-mapped snapshot (or
2418 * RBD_SNAP_HEAD_NAME for the base image).
2419 */
2420 static ssize_t rbd_snap_show(struct device *dev,
2421 struct device_attribute *attr,
2422 char *buf)
2423 {
2424 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2425
2426 return sprintf(buf, "%s\n", rbd_dev->spec->snap_name);
2427 }
2428
2429 /*
2430 * For an rbd v2 image, shows the pool id, image id, and snapshot id
2431 * for the parent image. If there is no parent, simply shows
2432 * "(no parent image)".
2433 */
2434 static ssize_t rbd_parent_show(struct device *dev,
2435 struct device_attribute *attr,
2436 char *buf)
2437 {
2438 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2439 struct rbd_spec *spec = rbd_dev->parent_spec;
2440 int count;
2441 char *bufp = buf;
2442
2443 if (!spec)
2444 return sprintf(buf, "(no parent image)\n");
2445
2446 count = sprintf(bufp, "pool_id %llu\npool_name %s\n",
2447 (unsigned long long) spec->pool_id, spec->pool_name);
2448 if (count < 0)
2449 return count;
2450 bufp += count;
2451
2452 count = sprintf(bufp, "image_id %s\nimage_name %s\n", spec->image_id,
2453 spec->image_name ? spec->image_name : "(unknown)");
2454 if (count < 0)
2455 return count;
2456 bufp += count;
2457
2458 count = sprintf(bufp, "snap_id %llu\nsnap_name %s\n",
2459 (unsigned long long) spec->snap_id, spec->snap_name);
2460 if (count < 0)
2461 return count;
2462 bufp += count;
2463
2464 count = sprintf(bufp, "overlap %llu\n", rbd_dev->parent_overlap);
2465 if (count < 0)
2466 return count;
2467 bufp += count;
2468
2469 return (ssize_t) (bufp - buf);
2470 }
2471
2472 static ssize_t rbd_image_refresh(struct device *dev,
2473 struct device_attribute *attr,
2474 const char *buf,
2475 size_t size)
2476 {
2477 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
2478 int ret;
2479
2480 ret = rbd_dev_refresh(rbd_dev, NULL);
2481
2482 return ret < 0 ? ret : size;
2483 }
2484
2485 static DEVICE_ATTR(size, S_IRUGO, rbd_size_show, NULL);
2486 static DEVICE_ATTR(features, S_IRUGO, rbd_features_show, NULL);
2487 static DEVICE_ATTR(major, S_IRUGO, rbd_major_show, NULL);
2488 static DEVICE_ATTR(client_id, S_IRUGO, rbd_client_id_show, NULL);
2489 static DEVICE_ATTR(pool, S_IRUGO, rbd_pool_show, NULL);
2490 static DEVICE_ATTR(pool_id, S_IRUGO, rbd_pool_id_show, NULL);
2491 static DEVICE_ATTR(name, S_IRUGO, rbd_name_show, NULL);
2492 static DEVICE_ATTR(image_id, S_IRUGO, rbd_image_id_show, NULL);
2493 static DEVICE_ATTR(refresh, S_IWUSR, NULL, rbd_image_refresh);
2494 static DEVICE_ATTR(current_snap, S_IRUGO, rbd_snap_show, NULL);
2495 static DEVICE_ATTR(parent, S_IRUGO, rbd_parent_show, NULL);
2496
2497 static struct attribute *rbd_attrs[] = {
2498 &dev_attr_size.attr,
2499 &dev_attr_features.attr,
2500 &dev_attr_major.attr,
2501 &dev_attr_client_id.attr,
2502 &dev_attr_pool.attr,
2503 &dev_attr_pool_id.attr,
2504 &dev_attr_name.attr,
2505 &dev_attr_image_id.attr,
2506 &dev_attr_current_snap.attr,
2507 &dev_attr_parent.attr,
2508 &dev_attr_refresh.attr,
2509 NULL
2510 };
2511
2512 static struct attribute_group rbd_attr_group = {
2513 .attrs = rbd_attrs,
2514 };
2515
2516 static const struct attribute_group *rbd_attr_groups[] = {
2517 &rbd_attr_group,
2518 NULL
2519 };
2520
2521 static void rbd_sysfs_dev_release(struct device *dev)
2522 {
2523 }
2524
2525 static struct device_type rbd_device_type = {
2526 .name = "rbd",
2527 .groups = rbd_attr_groups,
2528 .release = rbd_sysfs_dev_release,
2529 };
2530
2531
2532 /*
2533 sysfs - snapshots
2534 */
2535
2536 static ssize_t rbd_snap_size_show(struct device *dev,
2537 struct device_attribute *attr,
2538 char *buf)
2539 {
2540 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
2541
2542 return sprintf(buf, "%llu\n", (unsigned long long)snap->size);
2543 }
2544
2545 static ssize_t rbd_snap_id_show(struct device *dev,
2546 struct device_attribute *attr,
2547 char *buf)
2548 {
2549 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
2550
2551 return sprintf(buf, "%llu\n", (unsigned long long)snap->id);
2552 }
2553
2554 static ssize_t rbd_snap_features_show(struct device *dev,
2555 struct device_attribute *attr,
2556 char *buf)
2557 {
2558 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
2559
2560 return sprintf(buf, "0x%016llx\n",
2561 (unsigned long long) snap->features);
2562 }
2563
2564 static DEVICE_ATTR(snap_size, S_IRUGO, rbd_snap_size_show, NULL);
2565 static DEVICE_ATTR(snap_id, S_IRUGO, rbd_snap_id_show, NULL);
2566 static DEVICE_ATTR(snap_features, S_IRUGO, rbd_snap_features_show, NULL);
2567
2568 static struct attribute *rbd_snap_attrs[] = {
2569 &dev_attr_snap_size.attr,
2570 &dev_attr_snap_id.attr,
2571 &dev_attr_snap_features.attr,
2572 NULL,
2573 };
2574
2575 static struct attribute_group rbd_snap_attr_group = {
2576 .attrs = rbd_snap_attrs,
2577 };
2578
2579 static void rbd_snap_dev_release(struct device *dev)
2580 {
2581 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
2582 kfree(snap->name);
2583 kfree(snap);
2584 }
2585
2586 static const struct attribute_group *rbd_snap_attr_groups[] = {
2587 &rbd_snap_attr_group,
2588 NULL
2589 };
2590
2591 static struct device_type rbd_snap_device_type = {
2592 .groups = rbd_snap_attr_groups,
2593 .release = rbd_snap_dev_release,
2594 };
2595
2596 static struct rbd_spec *rbd_spec_get(struct rbd_spec *spec)
2597 {
2598 kref_get(&spec->kref);
2599
2600 return spec;
2601 }
2602
2603 static void rbd_spec_free(struct kref *kref);
2604 static void rbd_spec_put(struct rbd_spec *spec)
2605 {
2606 if (spec)
2607 kref_put(&spec->kref, rbd_spec_free);
2608 }
2609
2610 static struct rbd_spec *rbd_spec_alloc(void)
2611 {
2612 struct rbd_spec *spec;
2613
2614 spec = kzalloc(sizeof (*spec), GFP_KERNEL);
2615 if (!spec)
2616 return NULL;
2617 kref_init(&spec->kref);
2618
2619 rbd_spec_put(rbd_spec_get(spec)); /* TEMPORARY */
2620
2621 return spec;
2622 }
2623
2624 static void rbd_spec_free(struct kref *kref)
2625 {
2626 struct rbd_spec *spec = container_of(kref, struct rbd_spec, kref);
2627
2628 kfree(spec->pool_name);
2629 kfree(spec->image_id);
2630 kfree(spec->image_name);
2631 kfree(spec->snap_name);
2632 kfree(spec);
2633 }
2634
2635 static struct rbd_device *rbd_dev_create(struct rbd_client *rbdc,
2636 struct rbd_spec *spec)
2637 {
2638 struct rbd_device *rbd_dev;
2639
2640 rbd_dev = kzalloc(sizeof (*rbd_dev), GFP_KERNEL);
2641 if (!rbd_dev)
2642 return NULL;
2643
2644 spin_lock_init(&rbd_dev->lock);
2645 rbd_dev->flags = 0;
2646 INIT_LIST_HEAD(&rbd_dev->node);
2647 INIT_LIST_HEAD(&rbd_dev->snaps);
2648 init_rwsem(&rbd_dev->header_rwsem);
2649
2650 rbd_dev->spec = spec;
2651 rbd_dev->rbd_client = rbdc;
2652
2653 /* Initialize the layout used for all rbd requests */
2654
2655 rbd_dev->layout.fl_stripe_unit = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
2656 rbd_dev->layout.fl_stripe_count = cpu_to_le32(1);
2657 rbd_dev->layout.fl_object_size = cpu_to_le32(1 << RBD_MAX_OBJ_ORDER);
2658 rbd_dev->layout.fl_pg_pool = cpu_to_le32((u32) spec->pool_id);
2659
2660 return rbd_dev;
2661 }
2662
2663 static void rbd_dev_destroy(struct rbd_device *rbd_dev)
2664 {
2665 rbd_spec_put(rbd_dev->parent_spec);
2666 kfree(rbd_dev->header_name);
2667 rbd_put_client(rbd_dev->rbd_client);
2668 rbd_spec_put(rbd_dev->spec);
2669 kfree(rbd_dev);
2670 }
2671
2672 static bool rbd_snap_registered(struct rbd_snap *snap)
2673 {
2674 bool ret = snap->dev.type == &rbd_snap_device_type;
2675 bool reg = device_is_registered(&snap->dev);
2676
2677 rbd_assert(!ret ^ reg);
2678
2679 return ret;
2680 }
2681
2682 static void rbd_remove_snap_dev(struct rbd_snap *snap)
2683 {
2684 list_del(&snap->node);
2685 if (device_is_registered(&snap->dev))
2686 device_unregister(&snap->dev);
2687 }
2688
2689 static int rbd_register_snap_dev(struct rbd_snap *snap,
2690 struct device *parent)
2691 {
2692 struct device *dev = &snap->dev;
2693 int ret;
2694
2695 dev->type = &rbd_snap_device_type;
2696 dev->parent = parent;
2697 dev->release = rbd_snap_dev_release;
2698 dev_set_name(dev, "%s%s", RBD_SNAP_DEV_NAME_PREFIX, snap->name);
2699 dout("%s: registering device for snapshot %s\n", __func__, snap->name);
2700
2701 ret = device_register(dev);
2702
2703 return ret;
2704 }
2705
2706 static struct rbd_snap *__rbd_add_snap_dev(struct rbd_device *rbd_dev,
2707 const char *snap_name,
2708 u64 snap_id, u64 snap_size,
2709 u64 snap_features)
2710 {
2711 struct rbd_snap *snap;
2712 int ret;
2713
2714 snap = kzalloc(sizeof (*snap), GFP_KERNEL);
2715 if (!snap)
2716 return ERR_PTR(-ENOMEM);
2717
2718 ret = -ENOMEM;
2719 snap->name = kstrdup(snap_name, GFP_KERNEL);
2720 if (!snap->name)
2721 goto err;
2722
2723 snap->id = snap_id;
2724 snap->size = snap_size;
2725 snap->features = snap_features;
2726
2727 return snap;
2728
2729 err:
2730 kfree(snap->name);
2731 kfree(snap);
2732
2733 return ERR_PTR(ret);
2734 }
2735
2736 static char *rbd_dev_v1_snap_info(struct rbd_device *rbd_dev, u32 which,
2737 u64 *snap_size, u64 *snap_features)
2738 {
2739 char *snap_name;
2740
2741 rbd_assert(which < rbd_dev->header.snapc->num_snaps);
2742
2743 *snap_size = rbd_dev->header.snap_sizes[which];
2744 *snap_features = 0; /* No features for v1 */
2745
2746 /* Skip over names until we find the one we are looking for */
2747
2748 snap_name = rbd_dev->header.snap_names;
2749 while (which--)
2750 snap_name += strlen(snap_name) + 1;
2751
2752 return snap_name;
2753 }
2754
2755 /*
2756 * Get the size and object order for an image snapshot, or if
2757 * snap_id is CEPH_NOSNAP, gets this information for the base
2758 * image.
2759 */
2760 static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
2761 u8 *order, u64 *snap_size)
2762 {
2763 __le64 snapid = cpu_to_le64(snap_id);
2764 int ret;
2765 struct {
2766 u8 order;
2767 __le64 size;
2768 } __attribute__ ((packed)) size_buf = { 0 };
2769
2770 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
2771 "rbd", "get_size",
2772 (char *) &snapid, sizeof (snapid),
2773 (char *) &size_buf, sizeof (size_buf), NULL);
2774 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
2775 if (ret < 0)
2776 return ret;
2777
2778 *order = size_buf.order;
2779 *snap_size = le64_to_cpu(size_buf.size);
2780
2781 dout(" snap_id 0x%016llx order = %u, snap_size = %llu\n",
2782 (unsigned long long) snap_id, (unsigned int) *order,
2783 (unsigned long long) *snap_size);
2784
2785 return 0;
2786 }
2787
2788 static int rbd_dev_v2_image_size(struct rbd_device *rbd_dev)
2789 {
2790 return _rbd_dev_v2_snap_size(rbd_dev, CEPH_NOSNAP,
2791 &rbd_dev->header.obj_order,
2792 &rbd_dev->header.image_size);
2793 }
2794
2795 static int rbd_dev_v2_object_prefix(struct rbd_device *rbd_dev)
2796 {
2797 void *reply_buf;
2798 int ret;
2799 void *p;
2800
2801 reply_buf = kzalloc(RBD_OBJ_PREFIX_LEN_MAX, GFP_KERNEL);
2802 if (!reply_buf)
2803 return -ENOMEM;
2804
2805 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
2806 "rbd", "get_object_prefix",
2807 NULL, 0,
2808 reply_buf, RBD_OBJ_PREFIX_LEN_MAX, NULL);
2809 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
2810 if (ret < 0)
2811 goto out;
2812
2813 p = reply_buf;
2814 rbd_dev->header.object_prefix = ceph_extract_encoded_string(&p,
2815 p + RBD_OBJ_PREFIX_LEN_MAX,
2816 NULL, GFP_NOIO);
2817
2818 if (IS_ERR(rbd_dev->header.object_prefix)) {
2819 ret = PTR_ERR(rbd_dev->header.object_prefix);
2820 rbd_dev->header.object_prefix = NULL;
2821 } else {
2822 dout(" object_prefix = %s\n", rbd_dev->header.object_prefix);
2823 }
2824
2825 out:
2826 kfree(reply_buf);
2827
2828 return ret;
2829 }
2830
2831 static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
2832 u64 *snap_features)
2833 {
2834 __le64 snapid = cpu_to_le64(snap_id);
2835 struct {
2836 __le64 features;
2837 __le64 incompat;
2838 } features_buf = { 0 };
2839 u64 incompat;
2840 int ret;
2841
2842 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
2843 "rbd", "get_features",
2844 (char *) &snapid, sizeof (snapid),
2845 (char *) &features_buf, sizeof (features_buf),
2846 NULL);
2847 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
2848 if (ret < 0)
2849 return ret;
2850
2851 incompat = le64_to_cpu(features_buf.incompat);
2852 if (incompat & ~RBD_FEATURES_ALL)
2853 return -ENXIO;
2854
2855 *snap_features = le64_to_cpu(features_buf.features);
2856
2857 dout(" snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
2858 (unsigned long long) snap_id,
2859 (unsigned long long) *snap_features,
2860 (unsigned long long) le64_to_cpu(features_buf.incompat));
2861
2862 return 0;
2863 }
2864
2865 static int rbd_dev_v2_features(struct rbd_device *rbd_dev)
2866 {
2867 return _rbd_dev_v2_snap_features(rbd_dev, CEPH_NOSNAP,
2868 &rbd_dev->header.features);
2869 }
2870
2871 static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev)
2872 {
2873 struct rbd_spec *parent_spec;
2874 size_t size;
2875 void *reply_buf = NULL;
2876 __le64 snapid;
2877 void *p;
2878 void *end;
2879 char *image_id;
2880 u64 overlap;
2881 int ret;
2882
2883 parent_spec = rbd_spec_alloc();
2884 if (!parent_spec)
2885 return -ENOMEM;
2886
2887 size = sizeof (__le64) + /* pool_id */
2888 sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX + /* image_id */
2889 sizeof (__le64) + /* snap_id */
2890 sizeof (__le64); /* overlap */
2891 reply_buf = kmalloc(size, GFP_KERNEL);
2892 if (!reply_buf) {
2893 ret = -ENOMEM;
2894 goto out_err;
2895 }
2896
2897 snapid = cpu_to_le64(CEPH_NOSNAP);
2898 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
2899 "rbd", "get_parent",
2900 (char *) &snapid, sizeof (snapid),
2901 (char *) reply_buf, size, NULL);
2902 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
2903 if (ret < 0)
2904 goto out_err;
2905
2906 ret = -ERANGE;
2907 p = reply_buf;
2908 end = (char *) reply_buf + size;
2909 ceph_decode_64_safe(&p, end, parent_spec->pool_id, out_err);
2910 if (parent_spec->pool_id == CEPH_NOPOOL)
2911 goto out; /* No parent? No problem. */
2912
2913 /* The ceph file layout needs to fit pool id in 32 bits */
2914
2915 ret = -EIO;
2916 if (WARN_ON(parent_spec->pool_id > (u64) U32_MAX))
2917 goto out;
2918
2919 image_id = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
2920 if (IS_ERR(image_id)) {
2921 ret = PTR_ERR(image_id);
2922 goto out_err;
2923 }
2924 parent_spec->image_id = image_id;
2925 ceph_decode_64_safe(&p, end, parent_spec->snap_id, out_err);
2926 ceph_decode_64_safe(&p, end, overlap, out_err);
2927
2928 rbd_dev->parent_overlap = overlap;
2929 rbd_dev->parent_spec = parent_spec;
2930 parent_spec = NULL; /* rbd_dev now owns this */
2931 out:
2932 ret = 0;
2933 out_err:
2934 kfree(reply_buf);
2935 rbd_spec_put(parent_spec);
2936
2937 return ret;
2938 }
2939
2940 static char *rbd_dev_image_name(struct rbd_device *rbd_dev)
2941 {
2942 size_t image_id_size;
2943 char *image_id;
2944 void *p;
2945 void *end;
2946 size_t size;
2947 void *reply_buf = NULL;
2948 size_t len = 0;
2949 char *image_name = NULL;
2950 int ret;
2951
2952 rbd_assert(!rbd_dev->spec->image_name);
2953
2954 len = strlen(rbd_dev->spec->image_id);
2955 image_id_size = sizeof (__le32) + len;
2956 image_id = kmalloc(image_id_size, GFP_KERNEL);
2957 if (!image_id)
2958 return NULL;
2959
2960 p = image_id;
2961 end = (char *) image_id + image_id_size;
2962 ceph_encode_string(&p, end, rbd_dev->spec->image_id, (u32) len);
2963
2964 size = sizeof (__le32) + RBD_IMAGE_NAME_LEN_MAX;
2965 reply_buf = kmalloc(size, GFP_KERNEL);
2966 if (!reply_buf)
2967 goto out;
2968
2969 ret = rbd_obj_method_sync(rbd_dev, RBD_DIRECTORY,
2970 "rbd", "dir_get_name",
2971 image_id, image_id_size,
2972 (char *) reply_buf, size, NULL);
2973 if (ret < 0)
2974 goto out;
2975 p = reply_buf;
2976 end = (char *) reply_buf + size;
2977 image_name = ceph_extract_encoded_string(&p, end, &len, GFP_KERNEL);
2978 if (IS_ERR(image_name))
2979 image_name = NULL;
2980 else
2981 dout("%s: name is %s len is %zd\n", __func__, image_name, len);
2982 out:
2983 kfree(reply_buf);
2984 kfree(image_id);
2985
2986 return image_name;
2987 }
2988
2989 /*
2990 * When a parent image gets probed, we only have the pool, image,
2991 * and snapshot ids but not the names of any of them. This call
2992 * is made later to fill in those names. It has to be done after
2993 * rbd_dev_snaps_update() has completed because some of the
2994 * information (in particular, snapshot name) is not available
2995 * until then.
2996 */
2997 static int rbd_dev_probe_update_spec(struct rbd_device *rbd_dev)
2998 {
2999 struct ceph_osd_client *osdc;
3000 const char *name;
3001 void *reply_buf = NULL;
3002 int ret;
3003
3004 if (rbd_dev->spec->pool_name)
3005 return 0; /* Already have the names */
3006
3007 /* Look up the pool name */
3008
3009 osdc = &rbd_dev->rbd_client->client->osdc;
3010 name = ceph_pg_pool_name_by_id(osdc->osdmap, rbd_dev->spec->pool_id);
3011 if (!name) {
3012 rbd_warn(rbd_dev, "there is no pool with id %llu",
3013 rbd_dev->spec->pool_id); /* Really a BUG() */
3014 return -EIO;
3015 }
3016
3017 rbd_dev->spec->pool_name = kstrdup(name, GFP_KERNEL);
3018 if (!rbd_dev->spec->pool_name)
3019 return -ENOMEM;
3020
3021 /* Fetch the image name; tolerate failure here */
3022
3023 name = rbd_dev_image_name(rbd_dev);
3024 if (name)
3025 rbd_dev->spec->image_name = (char *) name;
3026 else
3027 rbd_warn(rbd_dev, "unable to get image name");
3028
3029 /* Look up the snapshot name. */
3030
3031 name = rbd_snap_name(rbd_dev, rbd_dev->spec->snap_id);
3032 if (!name) {
3033 rbd_warn(rbd_dev, "no snapshot with id %llu",
3034 rbd_dev->spec->snap_id); /* Really a BUG() */
3035 ret = -EIO;
3036 goto out_err;
3037 }
3038 rbd_dev->spec->snap_name = kstrdup(name, GFP_KERNEL);
3039 if(!rbd_dev->spec->snap_name)
3040 goto out_err;
3041
3042 return 0;
3043 out_err:
3044 kfree(reply_buf);
3045 kfree(rbd_dev->spec->pool_name);
3046 rbd_dev->spec->pool_name = NULL;
3047
3048 return ret;
3049 }
3050
3051 static int rbd_dev_v2_snap_context(struct rbd_device *rbd_dev, u64 *ver)
3052 {
3053 size_t size;
3054 int ret;
3055 void *reply_buf;
3056 void *p;
3057 void *end;
3058 u64 seq;
3059 u32 snap_count;
3060 struct ceph_snap_context *snapc;
3061 u32 i;
3062
3063 /*
3064 * We'll need room for the seq value (maximum snapshot id),
3065 * snapshot count, and array of that many snapshot ids.
3066 * For now we have a fixed upper limit on the number we're
3067 * prepared to receive.
3068 */
3069 size = sizeof (__le64) + sizeof (__le32) +
3070 RBD_MAX_SNAP_COUNT * sizeof (__le64);
3071 reply_buf = kzalloc(size, GFP_KERNEL);
3072 if (!reply_buf)
3073 return -ENOMEM;
3074
3075 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
3076 "rbd", "get_snapcontext",
3077 NULL, 0,
3078 reply_buf, size, ver);
3079 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
3080 if (ret < 0)
3081 goto out;
3082
3083 ret = -ERANGE;
3084 p = reply_buf;
3085 end = (char *) reply_buf + size;
3086 ceph_decode_64_safe(&p, end, seq, out);
3087 ceph_decode_32_safe(&p, end, snap_count, out);
3088
3089 /*
3090 * Make sure the reported number of snapshot ids wouldn't go
3091 * beyond the end of our buffer. But before checking that,
3092 * make sure the computed size of the snapshot context we
3093 * allocate is representable in a size_t.
3094 */
3095 if (snap_count > (SIZE_MAX - sizeof (struct ceph_snap_context))
3096 / sizeof (u64)) {
3097 ret = -EINVAL;
3098 goto out;
3099 }
3100 if (!ceph_has_room(&p, end, snap_count * sizeof (__le64)))
3101 goto out;
3102
3103 size = sizeof (struct ceph_snap_context) +
3104 snap_count * sizeof (snapc->snaps[0]);
3105 snapc = kmalloc(size, GFP_KERNEL);
3106 if (!snapc) {
3107 ret = -ENOMEM;
3108 goto out;
3109 }
3110
3111 atomic_set(&snapc->nref, 1);
3112 snapc->seq = seq;
3113 snapc->num_snaps = snap_count;
3114 for (i = 0; i < snap_count; i++)
3115 snapc->snaps[i] = ceph_decode_64(&p);
3116
3117 rbd_dev->header.snapc = snapc;
3118
3119 dout(" snap context seq = %llu, snap_count = %u\n",
3120 (unsigned long long) seq, (unsigned int) snap_count);
3121
3122 out:
3123 kfree(reply_buf);
3124
3125 return 0;
3126 }
3127
3128 static char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev, u32 which)
3129 {
3130 size_t size;
3131 void *reply_buf;
3132 __le64 snap_id;
3133 int ret;
3134 void *p;
3135 void *end;
3136 char *snap_name;
3137
3138 size = sizeof (__le32) + RBD_MAX_SNAP_NAME_LEN;
3139 reply_buf = kmalloc(size, GFP_KERNEL);
3140 if (!reply_buf)
3141 return ERR_PTR(-ENOMEM);
3142
3143 snap_id = cpu_to_le64(rbd_dev->header.snapc->snaps[which]);
3144 ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
3145 "rbd", "get_snapshot_name",
3146 (char *) &snap_id, sizeof (snap_id),
3147 reply_buf, size, NULL);
3148 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
3149 if (ret < 0)
3150 goto out;
3151
3152 p = reply_buf;
3153 end = (char *) reply_buf + size;
3154 snap_name = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
3155 if (IS_ERR(snap_name)) {
3156 ret = PTR_ERR(snap_name);
3157 goto out;
3158 } else {
3159 dout(" snap_id 0x%016llx snap_name = %s\n",
3160 (unsigned long long) le64_to_cpu(snap_id), snap_name);
3161 }
3162 kfree(reply_buf);
3163
3164 return snap_name;
3165 out:
3166 kfree(reply_buf);
3167
3168 return ERR_PTR(ret);
3169 }
3170
3171 static char *rbd_dev_v2_snap_info(struct rbd_device *rbd_dev, u32 which,
3172 u64 *snap_size, u64 *snap_features)
3173 {
3174 u64 snap_id;
3175 u8 order;
3176 int ret;
3177
3178 snap_id = rbd_dev->header.snapc->snaps[which];
3179 ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, &order, snap_size);
3180 if (ret)
3181 return ERR_PTR(ret);
3182 ret = _rbd_dev_v2_snap_features(rbd_dev, snap_id, snap_features);
3183 if (ret)
3184 return ERR_PTR(ret);
3185
3186 return rbd_dev_v2_snap_name(rbd_dev, which);
3187 }
3188
3189 static char *rbd_dev_snap_info(struct rbd_device *rbd_dev, u32 which,
3190 u64 *snap_size, u64 *snap_features)
3191 {
3192 if (rbd_dev->image_format == 1)
3193 return rbd_dev_v1_snap_info(rbd_dev, which,
3194 snap_size, snap_features);
3195 if (rbd_dev->image_format == 2)
3196 return rbd_dev_v2_snap_info(rbd_dev, which,
3197 snap_size, snap_features);
3198 return ERR_PTR(-EINVAL);
3199 }
3200
3201 static int rbd_dev_v2_refresh(struct rbd_device *rbd_dev, u64 *hver)
3202 {
3203 int ret;
3204 __u8 obj_order;
3205
3206 down_write(&rbd_dev->header_rwsem);
3207
3208 /* Grab old order first, to see if it changes */
3209
3210 obj_order = rbd_dev->header.obj_order,
3211 ret = rbd_dev_v2_image_size(rbd_dev);
3212 if (ret)
3213 goto out;
3214 if (rbd_dev->header.obj_order != obj_order) {
3215 ret = -EIO;
3216 goto out;
3217 }
3218 rbd_update_mapping_size(rbd_dev);
3219
3220 ret = rbd_dev_v2_snap_context(rbd_dev, hver);
3221 dout("rbd_dev_v2_snap_context returned %d\n", ret);
3222 if (ret)
3223 goto out;
3224 ret = rbd_dev_snaps_update(rbd_dev);
3225 dout("rbd_dev_snaps_update returned %d\n", ret);
3226 if (ret)
3227 goto out;
3228 ret = rbd_dev_snaps_register(rbd_dev);
3229 dout("rbd_dev_snaps_register returned %d\n", ret);
3230 out:
3231 up_write(&rbd_dev->header_rwsem);
3232
3233 return ret;
3234 }
3235
3236 /*
3237 * Scan the rbd device's current snapshot list and compare it to the
3238 * newly-received snapshot context. Remove any existing snapshots
3239 * not present in the new snapshot context. Add a new snapshot for
3240 * any snaphots in the snapshot context not in the current list.
3241 * And verify there are no changes to snapshots we already know
3242 * about.
3243 *
3244 * Assumes the snapshots in the snapshot context are sorted by
3245 * snapshot id, highest id first. (Snapshots in the rbd_dev's list
3246 * are also maintained in that order.)
3247 */
3248 static int rbd_dev_snaps_update(struct rbd_device *rbd_dev)
3249 {
3250 struct ceph_snap_context *snapc = rbd_dev->header.snapc;
3251 const u32 snap_count = snapc->num_snaps;
3252 struct list_head *head = &rbd_dev->snaps;
3253 struct list_head *links = head->next;
3254 u32 index = 0;
3255
3256 dout("%s: snap count is %u\n", __func__, (unsigned int) snap_count);
3257 while (index < snap_count || links != head) {
3258 u64 snap_id;
3259 struct rbd_snap *snap;
3260 char *snap_name;
3261 u64 snap_size = 0;
3262 u64 snap_features = 0;
3263
3264 snap_id = index < snap_count ? snapc->snaps[index]
3265 : CEPH_NOSNAP;
3266 snap = links != head ? list_entry(links, struct rbd_snap, node)
3267 : NULL;
3268 rbd_assert(!snap || snap->id != CEPH_NOSNAP);
3269
3270 if (snap_id == CEPH_NOSNAP || (snap && snap->id > snap_id)) {
3271 struct list_head *next = links->next;
3272
3273 /*
3274 * A previously-existing snapshot is not in
3275 * the new snap context.
3276 *
3277 * If the now missing snapshot is the one the
3278 * image is mapped to, clear its exists flag
3279 * so we can avoid sending any more requests
3280 * to it.
3281 */
3282 if (rbd_dev->spec->snap_id == snap->id)
3283 clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
3284 rbd_remove_snap_dev(snap);
3285 dout("%ssnap id %llu has been removed\n",
3286 rbd_dev->spec->snap_id == snap->id ?
3287 "mapped " : "",
3288 (unsigned long long) snap->id);
3289
3290 /* Done with this list entry; advance */
3291
3292 links = next;
3293 continue;
3294 }
3295
3296 snap_name = rbd_dev_snap_info(rbd_dev, index,
3297 &snap_size, &snap_features);
3298 if (IS_ERR(snap_name))
3299 return PTR_ERR(snap_name);
3300
3301 dout("entry %u: snap_id = %llu\n", (unsigned int) snap_count,
3302 (unsigned long long) snap_id);
3303 if (!snap || (snap_id != CEPH_NOSNAP && snap->id < snap_id)) {
3304 struct rbd_snap *new_snap;
3305
3306 /* We haven't seen this snapshot before */
3307
3308 new_snap = __rbd_add_snap_dev(rbd_dev, snap_name,
3309 snap_id, snap_size, snap_features);
3310 if (IS_ERR(new_snap)) {
3311 int err = PTR_ERR(new_snap);
3312
3313 dout(" failed to add dev, error %d\n", err);
3314
3315 return err;
3316 }
3317
3318 /* New goes before existing, or at end of list */
3319
3320 dout(" added dev%s\n", snap ? "" : " at end\n");
3321 if (snap)
3322 list_add_tail(&new_snap->node, &snap->node);
3323 else
3324 list_add_tail(&new_snap->node, head);
3325 } else {
3326 /* Already have this one */
3327
3328 dout(" already present\n");
3329
3330 rbd_assert(snap->size == snap_size);
3331 rbd_assert(!strcmp(snap->name, snap_name));
3332 rbd_assert(snap->features == snap_features);
3333
3334 /* Done with this list entry; advance */
3335
3336 links = links->next;
3337 }
3338
3339 /* Advance to the next entry in the snapshot context */
3340
3341 index++;
3342 }
3343 dout("%s: done\n", __func__);
3344
3345 return 0;
3346 }
3347
3348 /*
3349 * Scan the list of snapshots and register the devices for any that
3350 * have not already been registered.
3351 */
3352 static int rbd_dev_snaps_register(struct rbd_device *rbd_dev)
3353 {
3354 struct rbd_snap *snap;
3355 int ret = 0;
3356
3357 dout("%s:\n", __func__);
3358 if (WARN_ON(!device_is_registered(&rbd_dev->dev)))
3359 return -EIO;
3360
3361 list_for_each_entry(snap, &rbd_dev->snaps, node) {
3362 if (!rbd_snap_registered(snap)) {
3363 ret = rbd_register_snap_dev(snap, &rbd_dev->dev);
3364 if (ret < 0)
3365 break;
3366 }
3367 }
3368 dout("%s: returning %d\n", __func__, ret);
3369
3370 return ret;
3371 }
3372
3373 static int rbd_bus_add_dev(struct rbd_device *rbd_dev)
3374 {
3375 struct device *dev;
3376 int ret;
3377
3378 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
3379
3380 dev = &rbd_dev->dev;
3381 dev->bus = &rbd_bus_type;
3382 dev->type = &rbd_device_type;
3383 dev->parent = &rbd_root_dev;
3384 dev->release = rbd_dev_release;
3385 dev_set_name(dev, "%d", rbd_dev->dev_id);
3386 ret = device_register(dev);
3387
3388 mutex_unlock(&ctl_mutex);
3389
3390 return ret;
3391 }
3392
3393 static void rbd_bus_del_dev(struct rbd_device *rbd_dev)
3394 {
3395 device_unregister(&rbd_dev->dev);
3396 }
3397
3398 static atomic64_t rbd_dev_id_max = ATOMIC64_INIT(0);
3399
3400 /*
3401 * Get a unique rbd identifier for the given new rbd_dev, and add
3402 * the rbd_dev to the global list. The minimum rbd id is 1.
3403 */
3404 static void rbd_dev_id_get(struct rbd_device *rbd_dev)
3405 {
3406 rbd_dev->dev_id = atomic64_inc_return(&rbd_dev_id_max);
3407
3408 spin_lock(&rbd_dev_list_lock);
3409 list_add_tail(&rbd_dev->node, &rbd_dev_list);
3410 spin_unlock(&rbd_dev_list_lock);
3411 dout("rbd_dev %p given dev id %llu\n", rbd_dev,
3412 (unsigned long long) rbd_dev->dev_id);
3413 }
3414
3415 /*
3416 * Remove an rbd_dev from the global list, and record that its
3417 * identifier is no longer in use.
3418 */
3419 static void rbd_dev_id_put(struct rbd_device *rbd_dev)
3420 {
3421 struct list_head *tmp;
3422 int rbd_id = rbd_dev->dev_id;
3423 int max_id;
3424
3425 rbd_assert(rbd_id > 0);
3426
3427 dout("rbd_dev %p released dev id %llu\n", rbd_dev,
3428 (unsigned long long) rbd_dev->dev_id);
3429 spin_lock(&rbd_dev_list_lock);
3430 list_del_init(&rbd_dev->node);
3431
3432 /*
3433 * If the id being "put" is not the current maximum, there
3434 * is nothing special we need to do.
3435 */
3436 if (rbd_id != atomic64_read(&rbd_dev_id_max)) {
3437 spin_unlock(&rbd_dev_list_lock);
3438 return;
3439 }
3440
3441 /*
3442 * We need to update the current maximum id. Search the
3443 * list to find out what it is. We're more likely to find
3444 * the maximum at the end, so search the list backward.
3445 */
3446 max_id = 0;
3447 list_for_each_prev(tmp, &rbd_dev_list) {
3448 struct rbd_device *rbd_dev;
3449
3450 rbd_dev = list_entry(tmp, struct rbd_device, node);
3451 if (rbd_dev->dev_id > max_id)
3452 max_id = rbd_dev->dev_id;
3453 }
3454 spin_unlock(&rbd_dev_list_lock);
3455
3456 /*
3457 * The max id could have been updated by rbd_dev_id_get(), in
3458 * which case it now accurately reflects the new maximum.
3459 * Be careful not to overwrite the maximum value in that
3460 * case.
3461 */
3462 atomic64_cmpxchg(&rbd_dev_id_max, rbd_id, max_id);
3463 dout(" max dev id has been reset\n");
3464 }
3465
3466 /*
3467 * Skips over white space at *buf, and updates *buf to point to the
3468 * first found non-space character (if any). Returns the length of
3469 * the token (string of non-white space characters) found. Note
3470 * that *buf must be terminated with '\0'.
3471 */
3472 static inline size_t next_token(const char **buf)
3473 {
3474 /*
3475 * These are the characters that produce nonzero for
3476 * isspace() in the "C" and "POSIX" locales.
3477 */
3478 const char *spaces = " \f\n\r\t\v";
3479
3480 *buf += strspn(*buf, spaces); /* Find start of token */
3481
3482 return strcspn(*buf, spaces); /* Return token length */
3483 }
3484
3485 /*
3486 * Finds the next token in *buf, and if the provided token buffer is
3487 * big enough, copies the found token into it. The result, if
3488 * copied, is guaranteed to be terminated with '\0'. Note that *buf
3489 * must be terminated with '\0' on entry.
3490 *
3491 * Returns the length of the token found (not including the '\0').
3492 * Return value will be 0 if no token is found, and it will be >=
3493 * token_size if the token would not fit.
3494 *
3495 * The *buf pointer will be updated to point beyond the end of the
3496 * found token. Note that this occurs even if the token buffer is
3497 * too small to hold it.
3498 */
3499 static inline size_t copy_token(const char **buf,
3500 char *token,
3501 size_t token_size)
3502 {
3503 size_t len;
3504
3505 len = next_token(buf);
3506 if (len < token_size) {
3507 memcpy(token, *buf, len);
3508 *(token + len) = '\0';
3509 }
3510 *buf += len;
3511
3512 return len;
3513 }
3514
3515 /*
3516 * Finds the next token in *buf, dynamically allocates a buffer big
3517 * enough to hold a copy of it, and copies the token into the new
3518 * buffer. The copy is guaranteed to be terminated with '\0'. Note
3519 * that a duplicate buffer is created even for a zero-length token.
3520 *
3521 * Returns a pointer to the newly-allocated duplicate, or a null
3522 * pointer if memory for the duplicate was not available. If
3523 * the lenp argument is a non-null pointer, the length of the token
3524 * (not including the '\0') is returned in *lenp.
3525 *
3526 * If successful, the *buf pointer will be updated to point beyond
3527 * the end of the found token.
3528 *
3529 * Note: uses GFP_KERNEL for allocation.
3530 */
3531 static inline char *dup_token(const char **buf, size_t *lenp)
3532 {
3533 char *dup;
3534 size_t len;
3535
3536 len = next_token(buf);
3537 dup = kmemdup(*buf, len + 1, GFP_KERNEL);
3538 if (!dup)
3539 return NULL;
3540 *(dup + len) = '\0';
3541 *buf += len;
3542
3543 if (lenp)
3544 *lenp = len;
3545
3546 return dup;
3547 }
3548
3549 /*
3550 * Parse the options provided for an "rbd add" (i.e., rbd image
3551 * mapping) request. These arrive via a write to /sys/bus/rbd/add,
3552 * and the data written is passed here via a NUL-terminated buffer.
3553 * Returns 0 if successful or an error code otherwise.
3554 *
3555 * The information extracted from these options is recorded in
3556 * the other parameters which return dynamically-allocated
3557 * structures:
3558 * ceph_opts
3559 * The address of a pointer that will refer to a ceph options
3560 * structure. Caller must release the returned pointer using
3561 * ceph_destroy_options() when it is no longer needed.
3562 * rbd_opts
3563 * Address of an rbd options pointer. Fully initialized by
3564 * this function; caller must release with kfree().
3565 * spec
3566 * Address of an rbd image specification pointer. Fully
3567 * initialized by this function based on parsed options.
3568 * Caller must release with rbd_spec_put().
3569 *
3570 * The options passed take this form:
3571 * <mon_addrs> <options> <pool_name> <image_name> [<snap_id>]
3572 * where:
3573 * <mon_addrs>
3574 * A comma-separated list of one or more monitor addresses.
3575 * A monitor address is an ip address, optionally followed
3576 * by a port number (separated by a colon).
3577 * I.e.: ip1[:port1][,ip2[:port2]...]
3578 * <options>
3579 * A comma-separated list of ceph and/or rbd options.
3580 * <pool_name>
3581 * The name of the rados pool containing the rbd image.
3582 * <image_name>
3583 * The name of the image in that pool to map.
3584 * <snap_id>
3585 * An optional snapshot id. If provided, the mapping will
3586 * present data from the image at the time that snapshot was
3587 * created. The image head is used if no snapshot id is
3588 * provided. Snapshot mappings are always read-only.
3589 */
3590 static int rbd_add_parse_args(const char *buf,
3591 struct ceph_options **ceph_opts,
3592 struct rbd_options **opts,
3593 struct rbd_spec **rbd_spec)
3594 {
3595 size_t len;
3596 char *options;
3597 const char *mon_addrs;
3598 size_t mon_addrs_size;
3599 struct rbd_spec *spec = NULL;
3600 struct rbd_options *rbd_opts = NULL;
3601 struct ceph_options *copts;
3602 int ret;
3603
3604 /* The first four tokens are required */
3605
3606 len = next_token(&buf);
3607 if (!len) {
3608 rbd_warn(NULL, "no monitor address(es) provided");
3609 return -EINVAL;
3610 }
3611 mon_addrs = buf;
3612 mon_addrs_size = len + 1;
3613 buf += len;
3614
3615 ret = -EINVAL;
3616 options = dup_token(&buf, NULL);
3617 if (!options)
3618 return -ENOMEM;
3619 if (!*options) {
3620 rbd_warn(NULL, "no options provided");
3621 goto out_err;
3622 }
3623
3624 spec = rbd_spec_alloc();
3625 if (!spec)
3626 goto out_mem;
3627
3628 spec->pool_name = dup_token(&buf, NULL);
3629 if (!spec->pool_name)
3630 goto out_mem;
3631 if (!*spec->pool_name) {
3632 rbd_warn(NULL, "no pool name provided");
3633 goto out_err;
3634 }
3635
3636 spec->image_name = dup_token(&buf, NULL);
3637 if (!spec->image_name)
3638 goto out_mem;
3639 if (!*spec->image_name) {
3640 rbd_warn(NULL, "no image name provided");
3641 goto out_err;
3642 }
3643
3644 /*
3645 * Snapshot name is optional; default is to use "-"
3646 * (indicating the head/no snapshot).
3647 */
3648 len = next_token(&buf);
3649 if (!len) {
3650 buf = RBD_SNAP_HEAD_NAME; /* No snapshot supplied */
3651 len = sizeof (RBD_SNAP_HEAD_NAME) - 1;
3652 } else if (len > RBD_MAX_SNAP_NAME_LEN) {
3653 ret = -ENAMETOOLONG;
3654 goto out_err;
3655 }
3656 spec->snap_name = kmemdup(buf, len + 1, GFP_KERNEL);
3657 if (!spec->snap_name)
3658 goto out_mem;
3659 *(spec->snap_name + len) = '\0';
3660
3661 /* Initialize all rbd options to the defaults */
3662
3663 rbd_opts = kzalloc(sizeof (*rbd_opts), GFP_KERNEL);
3664 if (!rbd_opts)
3665 goto out_mem;
3666
3667 rbd_opts->read_only = RBD_READ_ONLY_DEFAULT;
3668
3669 copts = ceph_parse_options(options, mon_addrs,
3670 mon_addrs + mon_addrs_size - 1,
3671 parse_rbd_opts_token, rbd_opts);
3672 if (IS_ERR(copts)) {
3673 ret = PTR_ERR(copts);
3674 goto out_err;
3675 }
3676 kfree(options);
3677
3678 *ceph_opts = copts;
3679 *opts = rbd_opts;
3680 *rbd_spec = spec;
3681
3682 return 0;
3683 out_mem:
3684 ret = -ENOMEM;
3685 out_err:
3686 kfree(rbd_opts);
3687 rbd_spec_put(spec);
3688 kfree(options);
3689
3690 return ret;
3691 }
3692
3693 /*
3694 * An rbd format 2 image has a unique identifier, distinct from the
3695 * name given to it by the user. Internally, that identifier is
3696 * what's used to specify the names of objects related to the image.
3697 *
3698 * A special "rbd id" object is used to map an rbd image name to its
3699 * id. If that object doesn't exist, then there is no v2 rbd image
3700 * with the supplied name.
3701 *
3702 * This function will record the given rbd_dev's image_id field if
3703 * it can be determined, and in that case will return 0. If any
3704 * errors occur a negative errno will be returned and the rbd_dev's
3705 * image_id field will be unchanged (and should be NULL).
3706 */
3707 static int rbd_dev_image_id(struct rbd_device *rbd_dev)
3708 {
3709 int ret;
3710 size_t size;
3711 char *object_name;
3712 void *response;
3713 void *p;
3714
3715 /*
3716 * When probing a parent image, the image id is already
3717 * known (and the image name likely is not). There's no
3718 * need to fetch the image id again in this case.
3719 */
3720 if (rbd_dev->spec->image_id)
3721 return 0;
3722
3723 /*
3724 * First, see if the format 2 image id file exists, and if
3725 * so, get the image's persistent id from it.
3726 */
3727 size = sizeof (RBD_ID_PREFIX) + strlen(rbd_dev->spec->image_name);
3728 object_name = kmalloc(size, GFP_NOIO);
3729 if (!object_name)
3730 return -ENOMEM;
3731 sprintf(object_name, "%s%s", RBD_ID_PREFIX, rbd_dev->spec->image_name);
3732 dout("rbd id object name is %s\n", object_name);
3733
3734 /* Response will be an encoded string, which includes a length */
3735
3736 size = sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX;
3737 response = kzalloc(size, GFP_NOIO);
3738 if (!response) {
3739 ret = -ENOMEM;
3740 goto out;
3741 }
3742
3743 ret = rbd_obj_method_sync(rbd_dev, object_name,
3744 "rbd", "get_id",
3745 NULL, 0,
3746 response, RBD_IMAGE_ID_LEN_MAX, NULL);
3747 dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
3748 if (ret < 0)
3749 goto out;
3750
3751 p = response;
3752 rbd_dev->spec->image_id = ceph_extract_encoded_string(&p,
3753 p + RBD_IMAGE_ID_LEN_MAX,
3754 NULL, GFP_NOIO);
3755 if (IS_ERR(rbd_dev->spec->image_id)) {
3756 ret = PTR_ERR(rbd_dev->spec->image_id);
3757 rbd_dev->spec->image_id = NULL;
3758 } else {
3759 dout("image_id is %s\n", rbd_dev->spec->image_id);
3760 }
3761 out:
3762 kfree(response);
3763 kfree(object_name);
3764
3765 return ret;
3766 }
3767
3768 static int rbd_dev_v1_probe(struct rbd_device *rbd_dev)
3769 {
3770 int ret;
3771 size_t size;
3772
3773 /* Version 1 images have no id; empty string is used */
3774
3775 rbd_dev->spec->image_id = kstrdup("", GFP_KERNEL);
3776 if (!rbd_dev->spec->image_id)
3777 return -ENOMEM;
3778
3779 /* Record the header object name for this rbd image. */
3780
3781 size = strlen(rbd_dev->spec->image_name) + sizeof (RBD_SUFFIX);
3782 rbd_dev->header_name = kmalloc(size, GFP_KERNEL);
3783 if (!rbd_dev->header_name) {
3784 ret = -ENOMEM;
3785 goto out_err;
3786 }
3787 sprintf(rbd_dev->header_name, "%s%s",
3788 rbd_dev->spec->image_name, RBD_SUFFIX);
3789
3790 /* Populate rbd image metadata */
3791
3792 ret = rbd_read_header(rbd_dev, &rbd_dev->header);
3793 if (ret < 0)
3794 goto out_err;
3795
3796 /* Version 1 images have no parent (no layering) */
3797
3798 rbd_dev->parent_spec = NULL;
3799 rbd_dev->parent_overlap = 0;
3800
3801 rbd_dev->image_format = 1;
3802
3803 dout("discovered version 1 image, header name is %s\n",
3804 rbd_dev->header_name);
3805
3806 return 0;
3807
3808 out_err:
3809 kfree(rbd_dev->header_name);
3810 rbd_dev->header_name = NULL;
3811 kfree(rbd_dev->spec->image_id);
3812 rbd_dev->spec->image_id = NULL;
3813
3814 return ret;
3815 }
3816
3817 static int rbd_dev_v2_probe(struct rbd_device *rbd_dev)
3818 {
3819 size_t size;
3820 int ret;
3821 u64 ver = 0;
3822
3823 /*
3824 * Image id was filled in by the caller. Record the header
3825 * object name for this rbd image.
3826 */
3827 size = sizeof (RBD_HEADER_PREFIX) + strlen(rbd_dev->spec->image_id);
3828 rbd_dev->header_name = kmalloc(size, GFP_KERNEL);
3829 if (!rbd_dev->header_name)
3830 return -ENOMEM;
3831 sprintf(rbd_dev->header_name, "%s%s",
3832 RBD_HEADER_PREFIX, rbd_dev->spec->image_id);
3833
3834 /* Get the size and object order for the image */
3835
3836 ret = rbd_dev_v2_image_size(rbd_dev);
3837 if (ret < 0)
3838 goto out_err;
3839
3840 /* Get the object prefix (a.k.a. block_name) for the image */
3841
3842 ret = rbd_dev_v2_object_prefix(rbd_dev);
3843 if (ret < 0)
3844 goto out_err;
3845
3846 /* Get the and check features for the image */
3847
3848 ret = rbd_dev_v2_features(rbd_dev);
3849 if (ret < 0)
3850 goto out_err;
3851
3852 /* If the image supports layering, get the parent info */
3853
3854 if (rbd_dev->header.features & RBD_FEATURE_LAYERING) {
3855 ret = rbd_dev_v2_parent_info(rbd_dev);
3856 if (ret < 0)
3857 goto out_err;
3858 }
3859
3860 /* crypto and compression type aren't (yet) supported for v2 images */
3861
3862 rbd_dev->header.crypt_type = 0;
3863 rbd_dev->header.comp_type = 0;
3864
3865 /* Get the snapshot context, plus the header version */
3866
3867 ret = rbd_dev_v2_snap_context(rbd_dev, &ver);
3868 if (ret)
3869 goto out_err;
3870 rbd_dev->header.obj_version = ver;
3871
3872 rbd_dev->image_format = 2;
3873
3874 dout("discovered version 2 image, header name is %s\n",
3875 rbd_dev->header_name);
3876
3877 return 0;
3878 out_err:
3879 rbd_dev->parent_overlap = 0;
3880 rbd_spec_put(rbd_dev->parent_spec);
3881 rbd_dev->parent_spec = NULL;
3882 kfree(rbd_dev->header_name);
3883 rbd_dev->header_name = NULL;
3884 kfree(rbd_dev->header.object_prefix);
3885 rbd_dev->header.object_prefix = NULL;
3886
3887 return ret;
3888 }
3889
3890 static int rbd_dev_probe_finish(struct rbd_device *rbd_dev)
3891 {
3892 int ret;
3893
3894 /* no need to lock here, as rbd_dev is not registered yet */
3895 ret = rbd_dev_snaps_update(rbd_dev);
3896 if (ret)
3897 return ret;
3898
3899 ret = rbd_dev_probe_update_spec(rbd_dev);
3900 if (ret)
3901 goto err_out_snaps;
3902
3903 ret = rbd_dev_set_mapping(rbd_dev);
3904 if (ret)
3905 goto err_out_snaps;
3906
3907 /* generate unique id: find highest unique id, add one */
3908 rbd_dev_id_get(rbd_dev);
3909
3910 /* Fill in the device name, now that we have its id. */
3911 BUILD_BUG_ON(DEV_NAME_LEN
3912 < sizeof (RBD_DRV_NAME) + MAX_INT_FORMAT_WIDTH);
3913 sprintf(rbd_dev->name, "%s%d", RBD_DRV_NAME, rbd_dev->dev_id);
3914
3915 /* Get our block major device number. */
3916
3917 ret = register_blkdev(0, rbd_dev->name);
3918 if (ret < 0)
3919 goto err_out_id;
3920 rbd_dev->major = ret;
3921
3922 /* Set up the blkdev mapping. */
3923
3924 ret = rbd_init_disk(rbd_dev);
3925 if (ret)
3926 goto err_out_blkdev;
3927
3928 ret = rbd_bus_add_dev(rbd_dev);
3929 if (ret)
3930 goto err_out_disk;
3931
3932 /*
3933 * At this point cleanup in the event of an error is the job
3934 * of the sysfs code (initiated by rbd_bus_del_dev()).
3935 */
3936 down_write(&rbd_dev->header_rwsem);
3937 ret = rbd_dev_snaps_register(rbd_dev);
3938 up_write(&rbd_dev->header_rwsem);
3939 if (ret)
3940 goto err_out_bus;
3941
3942 ret = rbd_dev_header_watch_sync(rbd_dev, 1);
3943 if (ret)
3944 goto err_out_bus;
3945
3946 /* Everything's ready. Announce the disk to the world. */
3947
3948 add_disk(rbd_dev->disk);
3949
3950 pr_info("%s: added with size 0x%llx\n", rbd_dev->disk->disk_name,
3951 (unsigned long long) rbd_dev->mapping.size);
3952
3953 return ret;
3954 err_out_bus:
3955 /* this will also clean up rest of rbd_dev stuff */
3956
3957 rbd_bus_del_dev(rbd_dev);
3958
3959 return ret;
3960 err_out_disk:
3961 rbd_free_disk(rbd_dev);
3962 err_out_blkdev:
3963 unregister_blkdev(rbd_dev->major, rbd_dev->name);
3964 err_out_id:
3965 rbd_dev_id_put(rbd_dev);
3966 err_out_snaps:
3967 rbd_remove_all_snaps(rbd_dev);
3968
3969 return ret;
3970 }
3971
3972 /*
3973 * Probe for the existence of the header object for the given rbd
3974 * device. For format 2 images this includes determining the image
3975 * id.
3976 */
3977 static int rbd_dev_probe(struct rbd_device *rbd_dev)
3978 {
3979 int ret;
3980
3981 /*
3982 * Get the id from the image id object. If it's not a
3983 * format 2 image, we'll get ENOENT back, and we'll assume
3984 * it's a format 1 image.
3985 */
3986 ret = rbd_dev_image_id(rbd_dev);
3987 if (ret)
3988 ret = rbd_dev_v1_probe(rbd_dev);
3989 else
3990 ret = rbd_dev_v2_probe(rbd_dev);
3991 if (ret) {
3992 dout("probe failed, returning %d\n", ret);
3993
3994 return ret;
3995 }
3996
3997 ret = rbd_dev_probe_finish(rbd_dev);
3998 if (ret)
3999 rbd_header_free(&rbd_dev->header);
4000
4001 return ret;
4002 }
4003
4004 static ssize_t rbd_add(struct bus_type *bus,
4005 const char *buf,
4006 size_t count)
4007 {
4008 struct rbd_device *rbd_dev = NULL;
4009 struct ceph_options *ceph_opts = NULL;
4010 struct rbd_options *rbd_opts = NULL;
4011 struct rbd_spec *spec = NULL;
4012 struct rbd_client *rbdc;
4013 struct ceph_osd_client *osdc;
4014 int rc = -ENOMEM;
4015
4016 if (!try_module_get(THIS_MODULE))
4017 return -ENODEV;
4018
4019 /* parse add command */
4020 rc = rbd_add_parse_args(buf, &ceph_opts, &rbd_opts, &spec);
4021 if (rc < 0)
4022 goto err_out_module;
4023
4024 rbdc = rbd_get_client(ceph_opts);
4025 if (IS_ERR(rbdc)) {
4026 rc = PTR_ERR(rbdc);
4027 goto err_out_args;
4028 }
4029 ceph_opts = NULL; /* rbd_dev client now owns this */
4030
4031 /* pick the pool */
4032 osdc = &rbdc->client->osdc;
4033 rc = ceph_pg_poolid_by_name(osdc->osdmap, spec->pool_name);
4034 if (rc < 0)
4035 goto err_out_client;
4036 spec->pool_id = (u64) rc;
4037
4038 /* The ceph file layout needs to fit pool id in 32 bits */
4039
4040 if (WARN_ON(spec->pool_id > (u64) U32_MAX)) {
4041 rc = -EIO;
4042 goto err_out_client;
4043 }
4044
4045 rbd_dev = rbd_dev_create(rbdc, spec);
4046 if (!rbd_dev)
4047 goto err_out_client;
4048 rbdc = NULL; /* rbd_dev now owns this */
4049 spec = NULL; /* rbd_dev now owns this */
4050
4051 rbd_dev->mapping.read_only = rbd_opts->read_only;
4052 kfree(rbd_opts);
4053 rbd_opts = NULL; /* done with this */
4054
4055 rc = rbd_dev_probe(rbd_dev);
4056 if (rc < 0)
4057 goto err_out_rbd_dev;
4058
4059 return count;
4060 err_out_rbd_dev:
4061 rbd_dev_destroy(rbd_dev);
4062 err_out_client:
4063 rbd_put_client(rbdc);
4064 err_out_args:
4065 if (ceph_opts)
4066 ceph_destroy_options(ceph_opts);
4067 kfree(rbd_opts);
4068 rbd_spec_put(spec);
4069 err_out_module:
4070 module_put(THIS_MODULE);
4071
4072 dout("Error adding device %s\n", buf);
4073
4074 return (ssize_t) rc;
4075 }
4076
4077 static struct rbd_device *__rbd_get_dev(unsigned long dev_id)
4078 {
4079 struct list_head *tmp;
4080 struct rbd_device *rbd_dev;
4081
4082 spin_lock(&rbd_dev_list_lock);
4083 list_for_each(tmp, &rbd_dev_list) {
4084 rbd_dev = list_entry(tmp, struct rbd_device, node);
4085 if (rbd_dev->dev_id == dev_id) {
4086 spin_unlock(&rbd_dev_list_lock);
4087 return rbd_dev;
4088 }
4089 }
4090 spin_unlock(&rbd_dev_list_lock);
4091 return NULL;
4092 }
4093
4094 static void rbd_dev_release(struct device *dev)
4095 {
4096 struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
4097
4098 if (rbd_dev->watch_event)
4099 rbd_dev_header_watch_sync(rbd_dev, 0);
4100
4101 /* clean up and free blkdev */
4102 rbd_free_disk(rbd_dev);
4103 unregister_blkdev(rbd_dev->major, rbd_dev->name);
4104
4105 /* release allocated disk header fields */
4106 rbd_header_free(&rbd_dev->header);
4107
4108 /* done with the id, and with the rbd_dev */
4109 rbd_dev_id_put(rbd_dev);
4110 rbd_assert(rbd_dev->rbd_client != NULL);
4111 rbd_dev_destroy(rbd_dev);
4112
4113 /* release module ref */
4114 module_put(THIS_MODULE);
4115 }
4116
4117 static ssize_t rbd_remove(struct bus_type *bus,
4118 const char *buf,
4119 size_t count)
4120 {
4121 struct rbd_device *rbd_dev = NULL;
4122 int target_id, rc;
4123 unsigned long ul;
4124 int ret = count;
4125
4126 rc = strict_strtoul(buf, 10, &ul);
4127 if (rc)
4128 return rc;
4129
4130 /* convert to int; abort if we lost anything in the conversion */
4131 target_id = (int) ul;
4132 if (target_id != ul)
4133 return -EINVAL;
4134
4135 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
4136
4137 rbd_dev = __rbd_get_dev(target_id);
4138 if (!rbd_dev) {
4139 ret = -ENOENT;
4140 goto done;
4141 }
4142
4143 spin_lock_irq(&rbd_dev->lock);
4144 if (rbd_dev->open_count)
4145 ret = -EBUSY;
4146 else
4147 set_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags);
4148 spin_unlock_irq(&rbd_dev->lock);
4149 if (ret < 0)
4150 goto done;
4151
4152 rbd_remove_all_snaps(rbd_dev);
4153 rbd_bus_del_dev(rbd_dev);
4154
4155 done:
4156 mutex_unlock(&ctl_mutex);
4157
4158 return ret;
4159 }
4160
4161 /*
4162 * create control files in sysfs
4163 * /sys/bus/rbd/...
4164 */
4165 static int rbd_sysfs_init(void)
4166 {
4167 int ret;
4168
4169 ret = device_register(&rbd_root_dev);
4170 if (ret < 0)
4171 return ret;
4172
4173 ret = bus_register(&rbd_bus_type);
4174 if (ret < 0)
4175 device_unregister(&rbd_root_dev);
4176
4177 return ret;
4178 }
4179
4180 static void rbd_sysfs_cleanup(void)
4181 {
4182 bus_unregister(&rbd_bus_type);
4183 device_unregister(&rbd_root_dev);
4184 }
4185
4186 static int __init rbd_init(void)
4187 {
4188 int rc;
4189
4190 if (!libceph_compatible(NULL)) {
4191 rbd_warn(NULL, "libceph incompatibility (quitting)");
4192
4193 return -EINVAL;
4194 }
4195 rc = rbd_sysfs_init();
4196 if (rc)
4197 return rc;
4198 pr_info("loaded " RBD_DRV_NAME_LONG "\n");
4199 return 0;
4200 }
4201
4202 static void __exit rbd_exit(void)
4203 {
4204 rbd_sysfs_cleanup();
4205 }
4206
4207 module_init(rbd_init);
4208 module_exit(rbd_exit);
4209
4210 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
4211 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
4212 MODULE_DESCRIPTION("rados block device");
4213
4214 /* following authorship retained from original osdblk.c */
4215 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
4216
4217 MODULE_LICENSE("GPL");
This page took 0.1348 seconds and 5 git commands to generate.