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