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 */
  63struct 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
  81struct rbd_options {
  82	int	notify_timeout;
  83};
  84
  85/*
  86 * an instance of the client.  multiple devices may share a client.
  87 */
  88struct rbd_client {
  89	struct ceph_client	*client;
  90	struct rbd_options	*rbd_opts;
  91	struct kref		kref;
  92	struct list_head	node;
  93};
  94
  95struct rbd_req_coll;
  96
  97/*
  98 * a single io request
  99 */
 100struct 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
 109struct rbd_req_status {
 110	int done;
 111	int rc;
 112	u64 bytes;
 113};
 114
 115/*
 116 * a collection of requests
 117 */
 118struct rbd_req_coll {
 119	int			total;
 120	int			num_done;
 121	struct kref		kref;
 122	struct rbd_req_status	status[0];
 123};
 124
 
 
 
 
 
 
 
 
 
 
 
 
 125struct 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 */
 136struct 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
 174static struct bus_type rbd_bus_type = {
 175	.name		= "rbd",
 176};
 177
 178static spinlock_t node_lock;      /* protects client get/put */
 179
 180static DEFINE_MUTEX(ctl_mutex);	  /* Serialize open/close/setup/teardown */
 
 181static LIST_HEAD(rbd_dev_list);    /* devices */
 182static LIST_HEAD(rbd_client_list);      /* clients */
 
 
 
 183
 184static int __rbd_init_snaps_header(struct rbd_device *rbd_dev);
 185static void rbd_dev_release(struct device *dev);
 186static ssize_t rbd_snap_rollback(struct device *dev,
 187				 struct device_attribute *attr,
 188				 const char *buf,
 189				 size_t size);
 190static ssize_t rbd_snap_add(struct device *dev,
 191			    struct device_attribute *attr,
 192			    const char *buf,
 193			    size_t count);
 194static void __rbd_remove_snap_dev(struct rbd_device *rbd_dev,
 195				  struct rbd_snap *snap);;
 196
 
 
 
 
 197
 198static struct rbd_device *dev_to_rbd(struct device *dev)
 
 
 
 
 
 
 
 
 
 
 
 199{
 200	return container_of(dev, struct rbd_device, dev);
 201}
 202
 
 
 
 
 
 
 203static struct device *rbd_get_dev(struct rbd_device *rbd_dev)
 204{
 205	return get_device(&rbd_dev->dev);
 206}
 207
 208static void rbd_put_dev(struct rbd_device *rbd_dev)
 209{
 210	put_device(&rbd_dev->dev);
 211}
 212
 213static int __rbd_update_snaps(struct rbd_device *rbd_dev);
 214
 215static 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
 230static 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
 239static 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 */
 249static 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
 281out_err:
 282	ceph_destroy_client(rbdc->client);
 283out_rbdc:
 
 284	kfree(rbdc);
 285out_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 */
 294static 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 */
 310enum {
 311	Opt_notify_timeout,
 312	Opt_last_int,
 313	/* int args above */
 314	Opt_last_string,
 315	/* string args above */
 316};
 317
 318static 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
 325static 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 */
 364static 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;
 406done_err:
 407	kfree(rbd_opts);
 408	return ret;
 409}
 410
 411/*
 412 * Destroy ceph client
 
 
 413 */
 414static 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 */
 432static 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 */
 442static 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 */
 455static 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
 514err_names:
 515	kfree(header->snap_names);
 516err_snapc:
 517	kfree(header->snapc);
 518	return ret;
 519}
 520
 521static int snap_index(struct rbd_image_header *header, int snap_num)
 522{
 523	return header->total_snaps - snap_num;
 524}
 525
 526static 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
 536static 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
 557static 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;
 589done:
 590	up_write(&header->snap_rwsem);
 591	return ret;
 592}
 593
 594static 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 */
 604static 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
 624static 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 */
 635static u64 rbd_obj_bytes(struct rbd_image_header *header)
 636{
 637	return 1 << header->obj_order;
 638}
 639
 640/*
 641 * bio helpers
 642 */
 643
 644static 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 */
 658static 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 */
 686static 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
 753err_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 */
 762static 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
 780static void rbd_destroy_ops(struct ceph_osd_req_op *ops)
 781{
 782	kfree(ops);
 783}
 784
 785static 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
 823static 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 */
 832static 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
 938done_err:
 939	bio_chain_put(req_data->bio);
 940	ceph_osdc_put_request(req);
 941done_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 */
 950static 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
 986static 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 */
 994static 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
1046done_ops:
1047	if (!orig_ops)
1048		rbd_destroy_ops(ops);
1049done:
1050	ceph_release_page_vector(pages, num_pages);
1051	return ret;
1052}
1053
1054/*
1055 * Do an asynchronous ceph osd operation
1056 */
1057static 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);
1105done:
1106	kfree(seg_name);
1107	return ret;
1108}
1109
1110/*
1111 * Request async osd write
1112 */
1113static 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 */
1131static 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 */
1150static 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 */
1169static 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
1199static 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 */
1222static 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
1256fail_event:
1257	ceph_osdc_cancel_event(dev->watch_event);
1258	dev->watch_event = NULL;
1259fail:
1260	rbd_destroy_ops(ops);
1261	return ret;
1262}
1263
1264/*
1265 * Request sync osd unwatch
1266 */
1267static 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
1293struct rbd_notify_info {
1294	struct rbd_device *dev;
1295};
1296
1297static 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 */
1310static 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
1351fail_event:
1352	ceph_osdc_cancel_event(event);
1353fail:
1354	rbd_destroy_ops(ops);
1355	return ret;
1356}
1357
1358/*
1359 * Request sync osd rollback
1360 */
1361static 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 */
1387static 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
1424static 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 */
1441static 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
1526next_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);
1538next:
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 */
1548static 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
1566static 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 */
1585static 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
1627out_dh:
1628	kfree(dh);
1629	return rc;
1630}
1631
1632/*
1633 * create a snapshot
1634 */
1635static 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;
1676bad:
1677	return -ERANGE;
1678}
1679
1680static 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 */
1693static 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
1737static 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
1799out_disk:
1800	put_disk(disk);
1801out:
1802	return rc;
1803}
1804
1805/*
1806  sysfs
1807*/
1808
 
 
 
 
 
1809static 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
1817static 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
1825static 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
1833static 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
1841static 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
1849static 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
1858static 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
1877static DEVICE_ATTR(size, S_IRUGO, rbd_size_show, NULL);
1878static DEVICE_ATTR(major, S_IRUGO, rbd_major_show, NULL);
1879static DEVICE_ATTR(client_id, S_IRUGO, rbd_client_id_show, NULL);
1880static DEVICE_ATTR(pool, S_IRUGO, rbd_pool_show, NULL);
1881static DEVICE_ATTR(name, S_IRUGO, rbd_name_show, NULL);
1882static DEVICE_ATTR(refresh, S_IWUSR, NULL, rbd_image_refresh);
1883static DEVICE_ATTR(current_snap, S_IRUGO, rbd_snap_show, NULL);
1884static DEVICE_ATTR(create_snap, S_IWUSR, NULL, rbd_snap_add);
1885static DEVICE_ATTR(rollback_snap, S_IWUSR, NULL, rbd_snap_rollback);
1886
1887static 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
1900static struct attribute_group rbd_attr_group = {
1901	.attrs = rbd_attrs,
1902};
1903
1904static const struct attribute_group *rbd_attr_groups[] = {
1905	&rbd_attr_group,
1906	NULL
1907};
1908
1909static void rbd_sysfs_dev_release(struct device *dev)
1910{
1911}
1912
1913static 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
1924static 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
1933static 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
1942static DEVICE_ATTR(snap_size, S_IRUGO, rbd_snap_size_show, NULL);
1943static DEVICE_ATTR(snap_id, S_IRUGO, rbd_snap_id_show, NULL);
1944
1945static struct attribute *rbd_snap_attrs[] = {
1946	&dev_attr_snap_size.attr,
1947	&dev_attr_snap_id.attr,
1948	NULL,
1949};
1950
1951static struct attribute_group rbd_snap_attr_group = {
1952	.attrs = rbd_snap_attrs,
1953};
1954
1955static 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
1962static const struct attribute_group *rbd_snap_attr_groups[] = {
1963	&rbd_snap_attr_group,
1964	NULL
1965};
1966
1967static struct device_type rbd_snap_device_type = {
1968	.groups		= rbd_snap_attr_groups,
1969	.release	= rbd_snap_dev_release,
1970};
1971
1972static 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
1979static 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
1995static 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;
2014err:
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 */
2023const 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 */
2045static 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
2112static void rbd_root_dev_release(struct device *dev)
2113{
2114}
2115
2116static struct device rbd_root_dev = {
2117	.init_name =    "rbd",
2118	.release =      rbd_root_dev_release,
2119};
2120
2121static 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;
2148done_free:
2149	mutex_unlock(&ctl_mutex);
2150	return ret;
2151}
2152
2153static void rbd_bus_del_dev(struct rbd_device *rbd_dev)
2154{
2155	device_unregister(&rbd_dev->dev);
2156}
2157
2158static 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
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
2177static 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
2283err_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
2295err_out_blkdev:
2296	unregister_blkdev(rbd_dev->major, rbd_dev->name);
2297err_out_client:
2298	rbd_put_client(rbd_dev);
2299	mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
2300err_out_slot:
2301	list_del_init(&rbd_dev->node);
2302	mutex_unlock(&ctl_mutex);
2303
2304	kfree(rbd_dev);
2305err_out_opt:
2306	kfree(options);
2307err_mon_dev:
2308	kfree(mon_dev_name);
2309err_out_mod:
2310	dout("Error adding device %s\n", buf);
2311	module_put(THIS_MODULE);
2312	return rc;
 
2313}
2314
2315static 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
2328static 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
2350static 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
2381done:
2382	mutex_unlock(&ctl_mutex);
2383	return ret;
2384}
2385
2386static 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
2421err_unlock:
2422	mutex_unlock(&ctl_mutex);
2423	kfree(name);
2424	return ret;
2425}
2426
2427static 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
2476done_unlock:
2477	mutex_unlock(&ctl_mutex);
2478done:
2479	kfree(seg_name);
2480	kfree(snap_name);
2481
2482	return ret;
2483}
2484
2485static 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 */
2495static 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
2510static void rbd_sysfs_cleanup(void)
2511{
2512	device_unregister(&rbd_root_dev);
2513	bus_unregister(&rbd_bus_type);
 
2514}
2515
2516int __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
2528void __exit rbd_exit(void)
2529{
2530	rbd_sysfs_cleanup();
2531}
2532
2533module_init(rbd_init);
2534module_exit(rbd_exit);
2535
2536MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
2537MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
2538MODULE_DESCRIPTION("rados block device");
2539
2540/* following authorship retained from original osdblk.c */
2541MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
2542
2543MODULE_LICENSE("GPL");