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