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