?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
include: replace linux/module.h with "struct module" wherever possible
[linux-2.6/next.git] / drivers / block / rbd.c
blob15f65b5f3fc7156c50300a6ab4d8efbe8a054e0e
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_rollback(struct device *dev,
187 struct device_attribute *attr,
188 const char *buf,
189 size_t size);
190 static ssize_t rbd_snap_add(struct device *dev,
191 struct device_attribute *attr,
192 const char *buf,
193 size_t count);
194 static void __rbd_remove_snap_dev(struct rbd_device *rbd_dev,
195 struct rbd_snap *snap);;
196
197
198 static struct rbd_device *dev_to_rbd(struct device *dev)
199 {
200 return container_of(dev, struct rbd_device, dev);
201 }
202
203 static struct device *rbd_get_dev(struct rbd_device *rbd_dev)
204 {
205 return get_device(&rbd_dev->dev);
206 }
207
208 static void rbd_put_dev(struct rbd_device *rbd_dev)
209 {
210 put_device(&rbd_dev->dev);
211 }
212
213 static int __rbd_update_snaps(struct rbd_device *rbd_dev);
214
215 static int rbd_open(struct block_device *bdev, fmode_t mode)
216 {
217 struct gendisk *disk = bdev->bd_disk;
218 struct rbd_device *rbd_dev = disk->private_data;
219
220 rbd_get_dev(rbd_dev);
221
222 set_device_ro(bdev, rbd_dev->read_only);
223
224 if ((mode & FMODE_WRITE) && rbd_dev->read_only)
225 return -EROFS;
226
227 return 0;
228 }
229
230 static int rbd_release(struct gendisk *disk, fmode_t mode)
231 {
232 struct rbd_device *rbd_dev = disk->private_data;
233
234 rbd_put_dev(rbd_dev);
235
236 return 0;
237 }
238
239 static const struct block_device_operations rbd_bd_ops = {
240 .owner = THIS_MODULE,
241 .open = rbd_open,
242 .release = rbd_release,
243 };
244
245 /*
246 * Initialize an rbd client instance.
247 * We own *opt.
248 */
249 static struct rbd_client *rbd_client_create(struct ceph_options *opt,
250 struct rbd_options *rbd_opts)
251 {
252 struct rbd_client *rbdc;
253 int ret = -ENOMEM;
254
255 dout("rbd_client_create\n");
256 rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
257 if (!rbdc)
258 goto out_opt;
259
260 kref_init(&rbdc->kref);
261 INIT_LIST_HEAD(&rbdc->node);
262
263 rbdc->client = ceph_create_client(opt, rbdc);
264 if (IS_ERR(rbdc->client))
265 goto out_rbdc;
266 opt = NULL; /* Now rbdc->client is responsible for opt */
267
268 ret = ceph_open_session(rbdc->client);
269 if (ret < 0)
270 goto out_err;
271
272 rbdc->rbd_opts = rbd_opts;
273
274 spin_lock(&node_lock);
275 list_add_tail(&rbdc->node, &rbd_client_list);
276 spin_unlock(&node_lock);
277
278 dout("rbd_client_create created %p\n", rbdc);
279 return rbdc;
280
281 out_err:
282 ceph_destroy_client(rbdc->client);
283 out_rbdc:
284 kfree(rbdc);
285 out_opt:
286 if (opt)
287 ceph_destroy_options(opt);
288 return ERR_PTR(ret);
289 }
290
291 /*
292 * Find a ceph client with specific addr and configuration.
293 */
294 static struct rbd_client *__rbd_client_find(struct ceph_options *opt)
295 {
296 struct rbd_client *client_node;
297
298 if (opt->flags & CEPH_OPT_NOSHARE)
299 return NULL;
300
301 list_for_each_entry(client_node, &rbd_client_list, node)
302 if (ceph_compare_options(opt, client_node->client) == 0)
303 return client_node;
304 return NULL;
305 }
306
307 /*
308 * mount options
309 */
310 enum {
311 Opt_notify_timeout,
312 Opt_last_int,
313 /* int args above */
314 Opt_last_string,
315 /* string args above */
316 };
317
318 static match_table_t rbdopt_tokens = {
319 {Opt_notify_timeout, "notify_timeout=%d"},
320 /* int args above */
321 /* string args above */
322 {-1, NULL}
323 };
324
325 static int parse_rbd_opts_token(char *c, void *private)
326 {
327 struct rbd_options *rbdopt = private;
328 substring_t argstr[MAX_OPT_ARGS];
329 int token, intval, ret;
330
331 token = match_token((char *)c, rbdopt_tokens, argstr);
332 if (token < 0)
333 return -EINVAL;
334
335 if (token < Opt_last_int) {
336 ret = match_int(&argstr[0], &intval);
337 if (ret < 0) {
338 pr_err("bad mount option arg (not int) "
339 "at '%s'\n", c);
340 return ret;
341 }
342 dout("got int token %d val %d\n", token, intval);
343 } else if (token > Opt_last_int && token < Opt_last_string) {
344 dout("got string token %d val %s\n", token,
345 argstr[0].from);
346 } else {
347 dout("got token %d\n", token);
348 }
349
350 switch (token) {
351 case Opt_notify_timeout:
352 rbdopt->notify_timeout = intval;
353 break;
354 default:
355 BUG_ON(token);
356 }
357 return 0;
358 }
359
360 /*
361 * Get a ceph client with specific addr and configuration, if one does
362 * not exist create it.
363 */
364 static int rbd_get_client(struct rbd_device *rbd_dev, const char *mon_addr,
365 char *options)
366 {
367 struct rbd_client *rbdc;
368 struct ceph_options *opt;
369 int ret;
370 struct rbd_options *rbd_opts;
371
372 rbd_opts = kzalloc(sizeof(*rbd_opts), GFP_KERNEL);
373 if (!rbd_opts)
374 return -ENOMEM;
375
376 rbd_opts->notify_timeout = RBD_NOTIFY_TIMEOUT_DEFAULT;
377
378 ret = ceph_parse_options(&opt, options, mon_addr,
379 mon_addr + strlen(mon_addr), parse_rbd_opts_token, rbd_opts);
380 if (ret < 0)
381 goto done_err;
382
383 spin_lock(&node_lock);
384 rbdc = __rbd_client_find(opt);
385 if (rbdc) {
386 ceph_destroy_options(opt);
387
388 /* using an existing client */
389 kref_get(&rbdc->kref);
390 rbd_dev->rbd_client = rbdc;
391 rbd_dev->client = rbdc->client;
392 spin_unlock(&node_lock);
393 return 0;
394 }
395 spin_unlock(&node_lock);
396
397 rbdc = rbd_client_create(opt, rbd_opts);
398 if (IS_ERR(rbdc)) {
399 ret = PTR_ERR(rbdc);
400 goto done_err;
401 }
402
403 rbd_dev->rbd_client = rbdc;
404 rbd_dev->client = rbdc->client;
405 return 0;
406 done_err:
407 kfree(rbd_opts);
408 return ret;
409 }
410
411 /*
412 * Destroy ceph client
413 */
414 static void rbd_client_release(struct kref *kref)
415 {
416 struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
417
418 dout("rbd_release_client %p\n", rbdc);
419 spin_lock(&node_lock);
420 list_del(&rbdc->node);
421 spin_unlock(&node_lock);
422
423 ceph_destroy_client(rbdc->client);
424 kfree(rbdc->rbd_opts);
425 kfree(rbdc);
426 }
427
428 /*
429 * Drop reference to ceph client node. If it's not referenced anymore, release
430 * it.
431 */
432 static void rbd_put_client(struct rbd_device *rbd_dev)
433 {
434 kref_put(&rbd_dev->rbd_client->kref, rbd_client_release);
435 rbd_dev->rbd_client = NULL;
436 rbd_dev->client = NULL;
437 }
438
439 /*
440 * Destroy requests collection
441 */
442 static void rbd_coll_release(struct kref *kref)
443 {
444 struct rbd_req_coll *coll =
445 container_of(kref, struct rbd_req_coll, kref);
446
447 dout("rbd_coll_release %p\n", coll);
448 kfree(coll);
449 }
450
451 /*
452 * Create a new header structure, translate header format from the on-disk
453 * header.
454 */
455 static int rbd_header_from_disk(struct rbd_image_header *header,
456 struct rbd_image_header_ondisk *ondisk,
457 int allocated_snaps,
458 gfp_t gfp_flags)
459 {
460 int i;
461 u32 snap_count = le32_to_cpu(ondisk->snap_count);
462 int ret = -ENOMEM;
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 rollback
1360 */
1361 static int rbd_req_sync_rollback_obj(struct rbd_device *dev,
1362 u64 snapid,
1363 const char *obj)
1364 {
1365 struct ceph_osd_req_op *ops;
1366 int ret = rbd_create_rw_ops(&ops, 1, CEPH_OSD_OP_ROLLBACK, 0);
1367 if (ret < 0)
1368 return ret;
1369
1370 ops[0].snap.snapid = snapid;
1371
1372 ret = rbd_req_sync_op(dev, NULL,
1373 CEPH_NOSNAP,
1374 0,
1375 CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1376 ops,
1377 1, obj, 0, 0, NULL, NULL, NULL);
1378
1379 rbd_destroy_ops(ops);
1380
1381 return ret;
1382 }
1383
1384 /*
1385 * Request sync osd read
1386 */
1387 static int rbd_req_sync_exec(struct rbd_device *dev,
1388 const char *obj,
1389 const char *cls,
1390 const char *method,
1391 const char *data,
1392 int len,
1393 u64 *ver)
1394 {
1395 struct ceph_osd_req_op *ops;
1396 int cls_len = strlen(cls);
1397 int method_len = strlen(method);
1398 int ret = rbd_create_rw_ops(&ops, 1, CEPH_OSD_OP_CALL,
1399 cls_len + method_len + len);
1400 if (ret < 0)
1401 return ret;
1402
1403 ops[0].cls.class_name = cls;
1404 ops[0].cls.class_len = (__u8)cls_len;
1405 ops[0].cls.method_name = method;
1406 ops[0].cls.method_len = (__u8)method_len;
1407 ops[0].cls.argc = 0;
1408 ops[0].cls.indata = data;
1409 ops[0].cls.indata_len = len;
1410
1411 ret = rbd_req_sync_op(dev, NULL,
1412 CEPH_NOSNAP,
1413 0,
1414 CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
1415 ops,
1416 1, obj, 0, 0, NULL, NULL, ver);
1417
1418 rbd_destroy_ops(ops);
1419
1420 dout("cls_exec returned %d\n", ret);
1421 return ret;
1422 }
1423
1424 static struct rbd_req_coll *rbd_alloc_coll(int num_reqs)
1425 {
1426 struct rbd_req_coll *coll =
1427 kzalloc(sizeof(struct rbd_req_coll) +
1428 sizeof(struct rbd_req_status) * num_reqs,
1429 GFP_ATOMIC);
1430
1431 if (!coll)
1432 return NULL;
1433 coll->total = num_reqs;
1434 kref_init(&coll->kref);
1435 return coll;
1436 }
1437
1438 /*
1439 * block device queue callback
1440 */
1441 static void rbd_rq_fn(struct request_queue *q)
1442 {
1443 struct rbd_device *rbd_dev = q->queuedata;
1444 struct request *rq;
1445 struct bio_pair *bp = NULL;
1446
1447 rq = blk_fetch_request(q);
1448
1449 while (1) {
1450 struct bio *bio;
1451 struct bio *rq_bio, *next_bio = NULL;
1452 bool do_write;
1453 int size, op_size = 0;
1454 u64 ofs;
1455 int num_segs, cur_seg = 0;
1456 struct rbd_req_coll *coll;
1457
1458 /* peek at request from block layer */
1459 if (!rq)
1460 break;
1461
1462 dout("fetched request\n");
1463
1464 /* filter out block requests we don't understand */
1465 if ((rq->cmd_type != REQ_TYPE_FS)) {
1466 __blk_end_request_all(rq, 0);
1467 goto next;
1468 }
1469
1470 /* deduce our operation (read, write) */
1471 do_write = (rq_data_dir(rq) == WRITE);
1472
1473 size = blk_rq_bytes(rq);
1474 ofs = blk_rq_pos(rq) * 512ULL;
1475 rq_bio = rq->bio;
1476 if (do_write && rbd_dev->read_only) {
1477 __blk_end_request_all(rq, -EROFS);
1478 goto next;
1479 }
1480
1481 spin_unlock_irq(q->queue_lock);
1482
1483 dout("%s 0x%x bytes at 0x%llx\n",
1484 do_write ? "write" : "read",
1485 size, blk_rq_pos(rq) * 512ULL);
1486
1487 num_segs = rbd_get_num_segments(&rbd_dev->header, ofs, size);
1488 coll = rbd_alloc_coll(num_segs);
1489 if (!coll) {
1490 spin_lock_irq(q->queue_lock);
1491 __blk_end_request_all(rq, -ENOMEM);
1492 goto next;
1493 }
1494
1495 do {
1496 /* a bio clone to be passed down to OSD req */
1497 dout("rq->bio->bi_vcnt=%d\n", rq->bio->bi_vcnt);
1498 op_size = rbd_get_segment(&rbd_dev->header,
1499 rbd_dev->header.block_name,
1500 ofs, size,
1501 NULL, NULL);
1502 kref_get(&coll->kref);
1503 bio = bio_chain_clone(&rq_bio, &next_bio, &bp,
1504 op_size, GFP_ATOMIC);
1505 if (!bio) {
1506 rbd_coll_end_req_index(rq, coll, cur_seg,
1507 -ENOMEM, op_size);
1508 goto next_seg;
1509 }
1510
1511
1512 /* init OSD command: write or read */
1513 if (do_write)
1514 rbd_req_write(rq, rbd_dev,
1515 rbd_dev->header.snapc,
1516 ofs,
1517 op_size, bio,
1518 coll, cur_seg);
1519 else
1520 rbd_req_read(rq, rbd_dev,
1521 cur_snap_id(rbd_dev),
1522 ofs,
1523 op_size, bio,
1524 coll, cur_seg);
1525
1526 next_seg:
1527 size -= op_size;
1528 ofs += op_size;
1529
1530 cur_seg++;
1531 rq_bio = next_bio;
1532 } while (size > 0);
1533 kref_put(&coll->kref, rbd_coll_release);
1534
1535 if (bp)
1536 bio_pair_release(bp);
1537 spin_lock_irq(q->queue_lock);
1538 next:
1539 rq = blk_fetch_request(q);
1540 }
1541 }
1542
1543 /*
1544 * a queue callback. Makes sure that we don't create a bio that spans across
1545 * multiple osd objects. One exception would be with a single page bios,
1546 * which we handle later at bio_chain_clone
1547 */
1548 static int rbd_merge_bvec(struct request_queue *q, struct bvec_merge_data *bmd,
1549 struct bio_vec *bvec)
1550 {
1551 struct rbd_device *rbd_dev = q->queuedata;
1552 unsigned int chunk_sectors = 1 << (rbd_dev->header.obj_order - 9);
1553 sector_t sector = bmd->bi_sector + get_start_sect(bmd->bi_bdev);
1554 unsigned int bio_sectors = bmd->bi_size >> 9;
1555 int max;
1556
1557 max = (chunk_sectors - ((sector & (chunk_sectors - 1))
1558 + bio_sectors)) << 9;
1559 if (max < 0)
1560 max = 0; /* bio_add cannot handle a negative return */
1561 if (max <= bvec->bv_len && bio_sectors == 0)
1562 return bvec->bv_len;
1563 return max;
1564 }
1565
1566 static void rbd_free_disk(struct rbd_device *rbd_dev)
1567 {
1568 struct gendisk *disk = rbd_dev->disk;
1569
1570 if (!disk)
1571 return;
1572
1573 rbd_header_free(&rbd_dev->header);
1574
1575 if (disk->flags & GENHD_FL_UP)
1576 del_gendisk(disk);
1577 if (disk->queue)
1578 blk_cleanup_queue(disk->queue);
1579 put_disk(disk);
1580 }
1581
1582 /*
1583 * reload the ondisk the header
1584 */
1585 static int rbd_read_header(struct rbd_device *rbd_dev,
1586 struct rbd_image_header *header)
1587 {
1588 ssize_t rc;
1589 struct rbd_image_header_ondisk *dh;
1590 int snap_count = 0;
1591 u64 snap_names_len = 0;
1592 u64 ver;
1593
1594 while (1) {
1595 int len = sizeof(*dh) +
1596 snap_count * sizeof(struct rbd_image_snap_ondisk) +
1597 snap_names_len;
1598
1599 rc = -ENOMEM;
1600 dh = kmalloc(len, GFP_KERNEL);
1601 if (!dh)
1602 return -ENOMEM;
1603
1604 rc = rbd_req_sync_read(rbd_dev,
1605 NULL, CEPH_NOSNAP,
1606 rbd_dev->obj_md_name,
1607 0, len,
1608 (char *)dh, &ver);
1609 if (rc < 0)
1610 goto out_dh;
1611
1612 rc = rbd_header_from_disk(header, dh, snap_count, GFP_KERNEL);
1613 if (rc < 0)
1614 goto out_dh;
1615
1616 if (snap_count != header->total_snaps) {
1617 snap_count = header->total_snaps;
1618 snap_names_len = header->snap_names_len;
1619 rbd_header_free(header);
1620 kfree(dh);
1621 continue;
1622 }
1623 break;
1624 }
1625 header->obj_version = ver;
1626
1627 out_dh:
1628 kfree(dh);
1629 return rc;
1630 }
1631
1632 /*
1633 * create a snapshot
1634 */
1635 static int rbd_header_add_snap(struct rbd_device *dev,
1636 const char *snap_name,
1637 gfp_t gfp_flags)
1638 {
1639 int name_len = strlen(snap_name);
1640 u64 new_snapid;
1641 int ret;
1642 void *data, *p, *e;
1643 u64 ver;
1644
1645 /* we should create a snapshot only if we're pointing at the head */
1646 if (dev->cur_snap)
1647 return -EINVAL;
1648
1649 ret = ceph_monc_create_snapid(&dev->client->monc, dev->poolid,
1650 &new_snapid);
1651 dout("created snapid=%lld\n", new_snapid);
1652 if (ret < 0)
1653 return ret;
1654
1655 data = kmalloc(name_len + 16, gfp_flags);
1656 if (!data)
1657 return -ENOMEM;
1658
1659 p = data;
1660 e = data + name_len + 16;
1661
1662 ceph_encode_string_safe(&p, e, snap_name, name_len, bad);
1663 ceph_encode_64_safe(&p, e, new_snapid, bad);
1664
1665 ret = rbd_req_sync_exec(dev, dev->obj_md_name, "rbd", "snap_add",
1666 data, p - data, &ver);
1667
1668 kfree(data);
1669
1670 if (ret < 0)
1671 return ret;
1672
1673 dev->header.snapc->seq = new_snapid;
1674
1675 return 0;
1676 bad:
1677 return -ERANGE;
1678 }
1679
1680 static void __rbd_remove_all_snaps(struct rbd_device *rbd_dev)
1681 {
1682 struct rbd_snap *snap;
1683
1684 while (!list_empty(&rbd_dev->snaps)) {
1685 snap = list_first_entry(&rbd_dev->snaps, struct rbd_snap, node);
1686 __rbd_remove_snap_dev(rbd_dev, snap);
1687 }
1688 }
1689
1690 /*
1691 * only read the first part of the ondisk header, without the snaps info
1692 */
1693 static int __rbd_update_snaps(struct rbd_device *rbd_dev)
1694 {
1695 int ret;
1696 struct rbd_image_header h;
1697 u64 snap_seq;
1698 int follow_seq = 0;
1699
1700 ret = rbd_read_header(rbd_dev, &h);
1701 if (ret < 0)
1702 return ret;
1703
1704 /* resized? */
1705 set_capacity(rbd_dev->disk, h.image_size / 512ULL);
1706
1707 down_write(&rbd_dev->header.snap_rwsem);
1708
1709 snap_seq = rbd_dev->header.snapc->seq;
1710 if (rbd_dev->header.total_snaps &&
1711 rbd_dev->header.snapc->snaps[0] == snap_seq)
1712 /* pointing at the head, will need to follow that
1713 if head moves */
1714 follow_seq = 1;
1715
1716 kfree(rbd_dev->header.snapc);
1717 kfree(rbd_dev->header.snap_names);
1718 kfree(rbd_dev->header.snap_sizes);
1719
1720 rbd_dev->header.total_snaps = h.total_snaps;
1721 rbd_dev->header.snapc = h.snapc;
1722 rbd_dev->header.snap_names = h.snap_names;
1723 rbd_dev->header.snap_names_len = h.snap_names_len;
1724 rbd_dev->header.snap_sizes = h.snap_sizes;
1725 if (follow_seq)
1726 rbd_dev->header.snapc->seq = rbd_dev->header.snapc->snaps[0];
1727 else
1728 rbd_dev->header.snapc->seq = snap_seq;
1729
1730 ret = __rbd_init_snaps_header(rbd_dev);
1731
1732 up_write(&rbd_dev->header.snap_rwsem);
1733
1734 return ret;
1735 }
1736
1737 static int rbd_init_disk(struct rbd_device *rbd_dev)
1738 {
1739 struct gendisk *disk;
1740 struct request_queue *q;
1741 int rc;
1742 u64 total_size = 0;
1743
1744 /* contact OSD, request size info about the object being mapped */
1745 rc = rbd_read_header(rbd_dev, &rbd_dev->header);
1746 if (rc)
1747 return rc;
1748
1749 /* no need to lock here, as rbd_dev is not registered yet */
1750 rc = __rbd_init_snaps_header(rbd_dev);
1751 if (rc)
1752 return rc;
1753
1754 rc = rbd_header_set_snap(rbd_dev, rbd_dev->snap_name, &total_size);
1755 if (rc)
1756 return rc;
1757
1758 /* create gendisk info */
1759 rc = -ENOMEM;
1760 disk = alloc_disk(RBD_MINORS_PER_MAJOR);
1761 if (!disk)
1762 goto out;
1763
1764 snprintf(disk->disk_name, sizeof(disk->disk_name), DRV_NAME "%d",
1765 rbd_dev->id);
1766 disk->major = rbd_dev->major;
1767 disk->first_minor = 0;
1768 disk->fops = &rbd_bd_ops;
1769 disk->private_data = rbd_dev;
1770
1771 /* init rq */
1772 rc = -ENOMEM;
1773 q = blk_init_queue(rbd_rq_fn, &rbd_dev->lock);
1774 if (!q)
1775 goto out_disk;
1776
1777 /* set io sizes to object size */
1778 blk_queue_max_hw_sectors(q, rbd_obj_bytes(&rbd_dev->header) / 512ULL);
1779 blk_queue_max_segment_size(q, rbd_obj_bytes(&rbd_dev->header));
1780 blk_queue_io_min(q, rbd_obj_bytes(&rbd_dev->header));
1781 blk_queue_io_opt(q, rbd_obj_bytes(&rbd_dev->header));
1782
1783 blk_queue_merge_bvec(q, rbd_merge_bvec);
1784 disk->queue = q;
1785
1786 q->queuedata = rbd_dev;
1787
1788 rbd_dev->disk = disk;
1789 rbd_dev->q = q;
1790
1791 /* finally, announce the disk to the world */
1792 set_capacity(disk, total_size / 512ULL);
1793 add_disk(disk);
1794
1795 pr_info("%s: added with size 0x%llx\n",
1796 disk->disk_name, (unsigned long long)total_size);
1797 return 0;
1798
1799 out_disk:
1800 put_disk(disk);
1801 out:
1802 return rc;
1803 }
1804
1805 /*
1806 sysfs
1807 */
1808
1809 static ssize_t rbd_size_show(struct device *dev,
1810 struct device_attribute *attr, char *buf)
1811 {
1812 struct rbd_device *rbd_dev = dev_to_rbd(dev);
1813
1814 return sprintf(buf, "%llu\n", (unsigned long long)rbd_dev->header.image_size);
1815 }
1816
1817 static ssize_t rbd_major_show(struct device *dev,
1818 struct device_attribute *attr, char *buf)
1819 {
1820 struct rbd_device *rbd_dev = dev_to_rbd(dev);
1821
1822 return sprintf(buf, "%d\n", rbd_dev->major);
1823 }
1824
1825 static ssize_t rbd_client_id_show(struct device *dev,
1826 struct device_attribute *attr, char *buf)
1827 {
1828 struct rbd_device *rbd_dev = dev_to_rbd(dev);
1829
1830 return sprintf(buf, "client%lld\n", ceph_client_id(rbd_dev->client));
1831 }
1832
1833 static ssize_t rbd_pool_show(struct device *dev,
1834 struct device_attribute *attr, char *buf)
1835 {
1836 struct rbd_device *rbd_dev = dev_to_rbd(dev);
1837
1838 return sprintf(buf, "%s\n", rbd_dev->pool_name);
1839 }
1840
1841 static ssize_t rbd_name_show(struct device *dev,
1842 struct device_attribute *attr, char *buf)
1843 {
1844 struct rbd_device *rbd_dev = dev_to_rbd(dev);
1845
1846 return sprintf(buf, "%s\n", rbd_dev->obj);
1847 }
1848
1849 static ssize_t rbd_snap_show(struct device *dev,
1850 struct device_attribute *attr,
1851 char *buf)
1852 {
1853 struct rbd_device *rbd_dev = dev_to_rbd(dev);
1854
1855 return sprintf(buf, "%s\n", rbd_dev->snap_name);
1856 }
1857
1858 static ssize_t rbd_image_refresh(struct device *dev,
1859 struct device_attribute *attr,
1860 const char *buf,
1861 size_t size)
1862 {
1863 struct rbd_device *rbd_dev = dev_to_rbd(dev);
1864 int rc;
1865 int ret = size;
1866
1867 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
1868
1869 rc = __rbd_update_snaps(rbd_dev);
1870 if (rc < 0)
1871 ret = rc;
1872
1873 mutex_unlock(&ctl_mutex);
1874 return ret;
1875 }
1876
1877 static DEVICE_ATTR(size, S_IRUGO, rbd_size_show, NULL);
1878 static DEVICE_ATTR(major, S_IRUGO, rbd_major_show, NULL);
1879 static DEVICE_ATTR(client_id, S_IRUGO, rbd_client_id_show, NULL);
1880 static DEVICE_ATTR(pool, S_IRUGO, rbd_pool_show, NULL);
1881 static DEVICE_ATTR(name, S_IRUGO, rbd_name_show, NULL);
1882 static DEVICE_ATTR(refresh, S_IWUSR, NULL, rbd_image_refresh);
1883 static DEVICE_ATTR(current_snap, S_IRUGO, rbd_snap_show, NULL);
1884 static DEVICE_ATTR(create_snap, S_IWUSR, NULL, rbd_snap_add);
1885 static DEVICE_ATTR(rollback_snap, S_IWUSR, NULL, rbd_snap_rollback);
1886
1887 static struct attribute *rbd_attrs[] = {
1888 &dev_attr_size.attr,
1889 &dev_attr_major.attr,
1890 &dev_attr_client_id.attr,
1891 &dev_attr_pool.attr,
1892 &dev_attr_name.attr,
1893 &dev_attr_current_snap.attr,
1894 &dev_attr_refresh.attr,
1895 &dev_attr_create_snap.attr,
1896 &dev_attr_rollback_snap.attr,
1897 NULL
1898 };
1899
1900 static struct attribute_group rbd_attr_group = {
1901 .attrs = rbd_attrs,
1902 };
1903
1904 static const struct attribute_group *rbd_attr_groups[] = {
1905 &rbd_attr_group,
1906 NULL
1907 };
1908
1909 static void rbd_sysfs_dev_release(struct device *dev)
1910 {
1911 }
1912
1913 static struct device_type rbd_device_type = {
1914 .name = "rbd",
1915 .groups = rbd_attr_groups,
1916 .release = rbd_sysfs_dev_release,
1917 };
1918
1919
1920 /*
1921 sysfs - snapshots
1922 */
1923
1924 static ssize_t rbd_snap_size_show(struct device *dev,
1925 struct device_attribute *attr,
1926 char *buf)
1927 {
1928 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
1929
1930 return sprintf(buf, "%lld\n", (long long)snap->size);
1931 }
1932
1933 static ssize_t rbd_snap_id_show(struct device *dev,
1934 struct device_attribute *attr,
1935 char *buf)
1936 {
1937 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
1938
1939 return sprintf(buf, "%lld\n", (long long)snap->id);
1940 }
1941
1942 static DEVICE_ATTR(snap_size, S_IRUGO, rbd_snap_size_show, NULL);
1943 static DEVICE_ATTR(snap_id, S_IRUGO, rbd_snap_id_show, NULL);
1944
1945 static struct attribute *rbd_snap_attrs[] = {
1946 &dev_attr_snap_size.attr,
1947 &dev_attr_snap_id.attr,
1948 NULL,
1949 };
1950
1951 static struct attribute_group rbd_snap_attr_group = {
1952 .attrs = rbd_snap_attrs,
1953 };
1954
1955 static void rbd_snap_dev_release(struct device *dev)
1956 {
1957 struct rbd_snap *snap = container_of(dev, struct rbd_snap, dev);
1958 kfree(snap->name);
1959 kfree(snap);
1960 }
1961
1962 static const struct attribute_group *rbd_snap_attr_groups[] = {
1963 &rbd_snap_attr_group,
1964 NULL
1965 };
1966
1967 static struct device_type rbd_snap_device_type = {
1968 .groups = rbd_snap_attr_groups,
1969 .release = rbd_snap_dev_release,
1970 };
1971
1972 static void __rbd_remove_snap_dev(struct rbd_device *rbd_dev,
1973 struct rbd_snap *snap)
1974 {
1975 list_del(&snap->node);
1976 device_unregister(&snap->dev);
1977 }
1978
1979 static int rbd_register_snap_dev(struct rbd_device *rbd_dev,
1980 struct rbd_snap *snap,
1981 struct device *parent)
1982 {
1983 struct device *dev = &snap->dev;
1984 int ret;
1985
1986 dev->type = &rbd_snap_device_type;
1987 dev->parent = parent;
1988 dev->release = rbd_snap_dev_release;
1989 dev_set_name(dev, "snap_%s", snap->name);
1990 ret = device_register(dev);
1991
1992 return ret;
1993 }
1994
1995 static int __rbd_add_snap_dev(struct rbd_device *rbd_dev,
1996 int i, const char *name,
1997 struct rbd_snap **snapp)
1998 {
1999 int ret;
2000 struct rbd_snap *snap = kzalloc(sizeof(*snap), GFP_KERNEL);
2001 if (!snap)
2002 return -ENOMEM;
2003 snap->name = kstrdup(name, GFP_KERNEL);
2004 snap->size = rbd_dev->header.snap_sizes[i];
2005 snap->id = rbd_dev->header.snapc->snaps[i];
2006 if (device_is_registered(&rbd_dev->dev)) {
2007 ret = rbd_register_snap_dev(rbd_dev, snap,
2008 &rbd_dev->dev);
2009 if (ret < 0)
2010 goto err;
2011 }
2012 *snapp = snap;
2013 return 0;
2014 err:
2015 kfree(snap->name);
2016 kfree(snap);
2017 return ret;
2018 }
2019
2020 /*
2021 * search for the previous snap in a null delimited string list
2022 */
2023 const char *rbd_prev_snap_name(const char *name, const char *start)
2024 {
2025 if (name < start + 2)
2026 return NULL;
2027
2028 name -= 2;
2029 while (*name) {
2030 if (name == start)
2031 return start;
2032 name--;
2033 }
2034 return name + 1;
2035 }
2036
2037 /*
2038 * compare the old list of snapshots that we have to what's in the header
2039 * and update it accordingly. Note that the header holds the snapshots
2040 * in a reverse order (from newest to oldest) and we need to go from
2041 * older to new so that we don't get a duplicate snap name when
2042 * doing the process (e.g., removed snapshot and recreated a new
2043 * one with the same name.
2044 */
2045 static int __rbd_init_snaps_header(struct rbd_device *rbd_dev)
2046 {
2047 const char *name, *first_name;
2048 int i = rbd_dev->header.total_snaps;
2049 struct rbd_snap *snap, *old_snap = NULL;
2050 int ret;
2051 struct list_head *p, *n;
2052
2053 first_name = rbd_dev->header.snap_names;
2054 name = first_name + rbd_dev->header.snap_names_len;
2055
2056 list_for_each_prev_safe(p, n, &rbd_dev->snaps) {
2057 u64 cur_id;
2058
2059 old_snap = list_entry(p, struct rbd_snap, node);
2060
2061 if (i)
2062 cur_id = rbd_dev->header.snapc->snaps[i - 1];
2063
2064 if (!i || old_snap->id < cur_id) {
2065 /* old_snap->id was skipped, thus was removed */
2066 __rbd_remove_snap_dev(rbd_dev, old_snap);
2067 continue;
2068 }
2069 if (old_snap->id == cur_id) {
2070 /* we have this snapshot already */
2071 i--;
2072 name = rbd_prev_snap_name(name, first_name);
2073 continue;
2074 }
2075 for (; i > 0;
2076 i--, name = rbd_prev_snap_name(name, first_name)) {
2077 if (!name) {
2078 WARN_ON(1);
2079 return -EINVAL;
2080 }
2081 cur_id = rbd_dev->header.snapc->snaps[i];
2082 /* snapshot removal? handle it above */
2083 if (cur_id >= old_snap->id)
2084 break;
2085 /* a new snapshot */
2086 ret = __rbd_add_snap_dev(rbd_dev, i - 1, name, &snap);
2087 if (ret < 0)
2088 return ret;
2089
2090 /* note that we add it backward so using n and not p */
2091 list_add(&snap->node, n);
2092 p = &snap->node;
2093 }
2094 }
2095 /* we're done going over the old snap list, just add what's left */
2096 for (; i > 0; i--) {
2097 name = rbd_prev_snap_name(name, first_name);
2098 if (!name) {
2099 WARN_ON(1);
2100 return -EINVAL;
2101 }
2102 ret = __rbd_add_snap_dev(rbd_dev, i - 1, name, &snap);
2103 if (ret < 0)
2104 return ret;
2105 list_add(&snap->node, &rbd_dev->snaps);
2106 }
2107
2108 return 0;
2109 }
2110
2111
2112 static void rbd_root_dev_release(struct device *dev)
2113 {
2114 }
2115
2116 static struct device rbd_root_dev = {
2117 .init_name = "rbd",
2118 .release = rbd_root_dev_release,
2119 };
2120
2121 static int rbd_bus_add_dev(struct rbd_device *rbd_dev)
2122 {
2123 int ret = -ENOMEM;
2124 struct device *dev;
2125 struct rbd_snap *snap;
2126
2127 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2128 dev = &rbd_dev->dev;
2129
2130 dev->bus = &rbd_bus_type;
2131 dev->type = &rbd_device_type;
2132 dev->parent = &rbd_root_dev;
2133 dev->release = rbd_dev_release;
2134 dev_set_name(dev, "%d", rbd_dev->id);
2135 ret = device_register(dev);
2136 if (ret < 0)
2137 goto done_free;
2138
2139 list_for_each_entry(snap, &rbd_dev->snaps, node) {
2140 ret = rbd_register_snap_dev(rbd_dev, snap,
2141 &rbd_dev->dev);
2142 if (ret < 0)
2143 break;
2144 }
2145
2146 mutex_unlock(&ctl_mutex);
2147 return 0;
2148 done_free:
2149 mutex_unlock(&ctl_mutex);
2150 return ret;
2151 }
2152
2153 static void rbd_bus_del_dev(struct rbd_device *rbd_dev)
2154 {
2155 device_unregister(&rbd_dev->dev);
2156 }
2157
2158 static int rbd_init_watch_dev(struct rbd_device *rbd_dev)
2159 {
2160 int ret, rc;
2161
2162 do {
2163 ret = rbd_req_sync_watch(rbd_dev, rbd_dev->obj_md_name,
2164 rbd_dev->header.obj_version);
2165 if (ret == -ERANGE) {
2166 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2167 rc = __rbd_update_snaps(rbd_dev);
2168 mutex_unlock(&ctl_mutex);
2169 if (rc < 0)
2170 return rc;
2171 }
2172 } while (ret == -ERANGE);
2173
2174 return ret;
2175 }
2176
2177 static ssize_t rbd_add(struct bus_type *bus,
2178 const char *buf,
2179 size_t count)
2180 {
2181 struct ceph_osd_client *osdc;
2182 struct rbd_device *rbd_dev;
2183 ssize_t rc = -ENOMEM;
2184 int irc, new_id = 0;
2185 struct list_head *tmp;
2186 char *mon_dev_name;
2187 char *options;
2188
2189 if (!try_module_get(THIS_MODULE))
2190 return -ENODEV;
2191
2192 mon_dev_name = kmalloc(RBD_MAX_OPT_LEN, GFP_KERNEL);
2193 if (!mon_dev_name)
2194 goto err_out_mod;
2195
2196 options = kmalloc(RBD_MAX_OPT_LEN, GFP_KERNEL);
2197 if (!options)
2198 goto err_mon_dev;
2199
2200 /* new rbd_device object */
2201 rbd_dev = kzalloc(sizeof(*rbd_dev), GFP_KERNEL);
2202 if (!rbd_dev)
2203 goto err_out_opt;
2204
2205 /* static rbd_device initialization */
2206 spin_lock_init(&rbd_dev->lock);
2207 INIT_LIST_HEAD(&rbd_dev->node);
2208 INIT_LIST_HEAD(&rbd_dev->snaps);
2209
2210 /* generate unique id: find highest unique id, add one */
2211 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2212
2213 list_for_each(tmp, &rbd_dev_list) {
2214 struct rbd_device *rbd_dev;
2215
2216 rbd_dev = list_entry(tmp, struct rbd_device, node);
2217 if (rbd_dev->id >= new_id)
2218 new_id = rbd_dev->id + 1;
2219 }
2220
2221 rbd_dev->id = new_id;
2222
2223 /* add to global list */
2224 list_add_tail(&rbd_dev->node, &rbd_dev_list);
2225
2226 /* parse add command */
2227 if (sscanf(buf, "%" __stringify(RBD_MAX_OPT_LEN) "s "
2228 "%" __stringify(RBD_MAX_OPT_LEN) "s "
2229 "%" __stringify(RBD_MAX_POOL_NAME_LEN) "s "
2230 "%" __stringify(RBD_MAX_OBJ_NAME_LEN) "s"
2231 "%" __stringify(RBD_MAX_SNAP_NAME_LEN) "s",
2232 mon_dev_name, options, rbd_dev->pool_name,
2233 rbd_dev->obj, rbd_dev->snap_name) < 4) {
2234 rc = -EINVAL;
2235 goto err_out_slot;
2236 }
2237
2238 if (rbd_dev->snap_name[0] == 0)
2239 rbd_dev->snap_name[0] = '-';
2240
2241 rbd_dev->obj_len = strlen(rbd_dev->obj);
2242 snprintf(rbd_dev->obj_md_name, sizeof(rbd_dev->obj_md_name), "%s%s",
2243 rbd_dev->obj, RBD_SUFFIX);
2244
2245 /* initialize rest of new object */
2246 snprintf(rbd_dev->name, DEV_NAME_LEN, DRV_NAME "%d", rbd_dev->id);
2247 rc = rbd_get_client(rbd_dev, mon_dev_name, options);
2248 if (rc < 0)
2249 goto err_out_slot;
2250
2251 mutex_unlock(&ctl_mutex);
2252
2253 /* pick the pool */
2254 osdc = &rbd_dev->client->osdc;
2255 rc = ceph_pg_poolid_by_name(osdc->osdmap, rbd_dev->pool_name);
2256 if (rc < 0)
2257 goto err_out_client;
2258 rbd_dev->poolid = rc;
2259
2260 /* register our block device */
2261 irc = register_blkdev(0, rbd_dev->name);
2262 if (irc < 0) {
2263 rc = irc;
2264 goto err_out_client;
2265 }
2266 rbd_dev->major = irc;
2267
2268 rc = rbd_bus_add_dev(rbd_dev);
2269 if (rc)
2270 goto err_out_blkdev;
2271
2272 /* set up and announce blkdev mapping */
2273 rc = rbd_init_disk(rbd_dev);
2274 if (rc)
2275 goto err_out_bus;
2276
2277 rc = rbd_init_watch_dev(rbd_dev);
2278 if (rc)
2279 goto err_out_bus;
2280
2281 return count;
2282
2283 err_out_bus:
2284 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2285 list_del_init(&rbd_dev->node);
2286 mutex_unlock(&ctl_mutex);
2287
2288 /* this will also clean up rest of rbd_dev stuff */
2289
2290 rbd_bus_del_dev(rbd_dev);
2291 kfree(options);
2292 kfree(mon_dev_name);
2293 return rc;
2294
2295 err_out_blkdev:
2296 unregister_blkdev(rbd_dev->major, rbd_dev->name);
2297 err_out_client:
2298 rbd_put_client(rbd_dev);
2299 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2300 err_out_slot:
2301 list_del_init(&rbd_dev->node);
2302 mutex_unlock(&ctl_mutex);
2303
2304 kfree(rbd_dev);
2305 err_out_opt:
2306 kfree(options);
2307 err_mon_dev:
2308 kfree(mon_dev_name);
2309 err_out_mod:
2310 dout("Error adding device %s\n", buf);
2311 module_put(THIS_MODULE);
2312 return rc;
2313 }
2314
2315 static struct rbd_device *__rbd_get_dev(unsigned long id)
2316 {
2317 struct list_head *tmp;
2318 struct rbd_device *rbd_dev;
2319
2320 list_for_each(tmp, &rbd_dev_list) {
2321 rbd_dev = list_entry(tmp, struct rbd_device, node);
2322 if (rbd_dev->id == id)
2323 return rbd_dev;
2324 }
2325 return NULL;
2326 }
2327
2328 static void rbd_dev_release(struct device *dev)
2329 {
2330 struct rbd_device *rbd_dev =
2331 container_of(dev, struct rbd_device, dev);
2332
2333 if (rbd_dev->watch_request)
2334 ceph_osdc_unregister_linger_request(&rbd_dev->client->osdc,
2335 rbd_dev->watch_request);
2336 if (rbd_dev->watch_event)
2337 rbd_req_sync_unwatch(rbd_dev, rbd_dev->obj_md_name);
2338
2339 rbd_put_client(rbd_dev);
2340
2341 /* clean up and free blkdev */
2342 rbd_free_disk(rbd_dev);
2343 unregister_blkdev(rbd_dev->major, rbd_dev->name);
2344 kfree(rbd_dev);
2345
2346 /* release module ref */
2347 module_put(THIS_MODULE);
2348 }
2349
2350 static ssize_t rbd_remove(struct bus_type *bus,
2351 const char *buf,
2352 size_t count)
2353 {
2354 struct rbd_device *rbd_dev = NULL;
2355 int target_id, rc;
2356 unsigned long ul;
2357 int ret = count;
2358
2359 rc = strict_strtoul(buf, 10, &ul);
2360 if (rc)
2361 return rc;
2362
2363 /* convert to int; abort if we lost anything in the conversion */
2364 target_id = (int) ul;
2365 if (target_id != ul)
2366 return -EINVAL;
2367
2368 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2369
2370 rbd_dev = __rbd_get_dev(target_id);
2371 if (!rbd_dev) {
2372 ret = -ENOENT;
2373 goto done;
2374 }
2375
2376 list_del_init(&rbd_dev->node);
2377
2378 __rbd_remove_all_snaps(rbd_dev);
2379 rbd_bus_del_dev(rbd_dev);
2380
2381 done:
2382 mutex_unlock(&ctl_mutex);
2383 return ret;
2384 }
2385
2386 static ssize_t rbd_snap_add(struct device *dev,
2387 struct device_attribute *attr,
2388 const char *buf,
2389 size_t count)
2390 {
2391 struct rbd_device *rbd_dev = dev_to_rbd(dev);
2392 int ret;
2393 char *name = kmalloc(count + 1, GFP_KERNEL);
2394 if (!name)
2395 return -ENOMEM;
2396
2397 snprintf(name, count, "%s", buf);
2398
2399 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2400
2401 ret = rbd_header_add_snap(rbd_dev,
2402 name, GFP_KERNEL);
2403 if (ret < 0)
2404 goto err_unlock;
2405
2406 ret = __rbd_update_snaps(rbd_dev);
2407 if (ret < 0)
2408 goto err_unlock;
2409
2410 /* shouldn't hold ctl_mutex when notifying.. notify might
2411 trigger a watch callback that would need to get that mutex */
2412 mutex_unlock(&ctl_mutex);
2413
2414 /* make a best effort, don't error if failed */
2415 rbd_req_sync_notify(rbd_dev, rbd_dev->obj_md_name);
2416
2417 ret = count;
2418 kfree(name);
2419 return ret;
2420
2421 err_unlock:
2422 mutex_unlock(&ctl_mutex);
2423 kfree(name);
2424 return ret;
2425 }
2426
2427 static ssize_t rbd_snap_rollback(struct device *dev,
2428 struct device_attribute *attr,
2429 const char *buf,
2430 size_t count)
2431 {
2432 struct rbd_device *rbd_dev = dev_to_rbd(dev);
2433 int ret;
2434 u64 snapid;
2435 u64 cur_ofs;
2436 char *seg_name = NULL;
2437 char *snap_name = kmalloc(count + 1, GFP_KERNEL);
2438 ret = -ENOMEM;
2439 if (!snap_name)
2440 return ret;
2441
2442 /* parse snaps add command */
2443 snprintf(snap_name, count, "%s", buf);
2444 seg_name = kmalloc(RBD_MAX_SEG_NAME_LEN + 1, GFP_NOIO);
2445 if (!seg_name)
2446 goto done;
2447
2448 mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2449
2450 ret = snap_by_name(&rbd_dev->header, snap_name, &snapid, NULL);
2451 if (ret < 0)
2452 goto done_unlock;
2453
2454 dout("snapid=%lld\n", snapid);
2455
2456 cur_ofs = 0;
2457 while (cur_ofs < rbd_dev->header.image_size) {
2458 cur_ofs += rbd_get_segment(&rbd_dev->header,
2459 rbd_dev->obj,
2460 cur_ofs, (u64)-1,
2461 seg_name, NULL);
2462 dout("seg_name=%s\n", seg_name);
2463
2464 ret = rbd_req_sync_rollback_obj(rbd_dev, snapid, seg_name);
2465 if (ret < 0)
2466 pr_warning("could not roll back obj %s err=%d\n",
2467 seg_name, ret);
2468 }
2469
2470 ret = __rbd_update_snaps(rbd_dev);
2471 if (ret < 0)
2472 goto done_unlock;
2473
2474 ret = count;
2475
2476 done_unlock:
2477 mutex_unlock(&ctl_mutex);
2478 done:
2479 kfree(seg_name);
2480 kfree(snap_name);
2481
2482 return ret;
2483 }
2484
2485 static struct bus_attribute rbd_bus_attrs[] = {
2486 __ATTR(add, S_IWUSR, NULL, rbd_add),
2487 __ATTR(remove, S_IWUSR, NULL, rbd_remove),
2488 __ATTR_NULL
2489 };
2490
2491 /*
2492 * create control files in sysfs
2493 * /sys/bus/rbd/...
2494 */
2495 static int rbd_sysfs_init(void)
2496 {
2497 int ret;
2498
2499 rbd_bus_type.bus_attrs = rbd_bus_attrs;
2500
2501 ret = bus_register(&rbd_bus_type);
2502 if (ret < 0)
2503 return ret;
2504
2505 ret = device_register(&rbd_root_dev);
2506
2507 return ret;
2508 }
2509
2510 static void rbd_sysfs_cleanup(void)
2511 {
2512 device_unregister(&rbd_root_dev);
2513 bus_unregister(&rbd_bus_type);
2514 }
2515
2516 int __init rbd_init(void)
2517 {
2518 int rc;
2519
2520 rc = rbd_sysfs_init();
2521 if (rc)
2522 return rc;
2523 spin_lock_init(&node_lock);
2524 pr_info("loaded " DRV_NAME_LONG "\n");
2525 return 0;
2526 }
2527
2528 void __exit rbd_exit(void)
2529 {
2530 rbd_sysfs_cleanup();
2531 }
2532
2533 module_init(rbd_init);
2534 module_exit(rbd_exit);
2535
2536 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
2537 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
2538 MODULE_DESCRIPTION("rados block device");
2539
2540 /* following authorship retained from original osdblk.c */
2541 MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
2542
2543 MODULE_LICENSE("GPL");
linux-next