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