rbd.c

   1
   2/*
   3   rbd.c -- Export ceph rados objects as a Linux block device
   4
   5
   6   based on drivers/block/osdblk.c:
   7
   8   Copyright 2009 Red Hat, Inc.
   9
  10   This program is free software; you can redistribute it and/or modify
  11   it under the terms of the GNU General Public License as published by
  12   the Free Software Foundation.
  13
  14   This program is distributed in the hope that it will be useful,
  15   but WITHOUT ANY WARRANTY; without even the implied warranty of
  16   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17   GNU General Public License for more details.
  18
  19   You should have received a copy of the GNU General Public License
  20   along with this program; see the file COPYING.  If not, write to
  21   the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
  22
  23
  24
  25   For usage instructions, please refer to:
  26
  27                 Documentation/ABI/testing/sysfs-bus-rbd
  28
  29 */
  30
  31#include <linux/ceph/libceph.h>
  32#include <linux/ceph/osd_client.h>
  33#include <linux/ceph/mon_client.h>
  34#include <linux/ceph/cls_lock_client.h>
  35#include <linux/ceph/striper.h>
  36#include <linux/ceph/decode.h>
  37#include <linux/fs_parser.h>
  38#include <linux/bsearch.h>
  39
  40#include <linux/kernel.h>
  41#include <linux/device.h>
  42#include <linux/module.h>
  43#include <linux/blk-mq.h>
  44#include <linux/fs.h>
  45#include <linux/blkdev.h>
  46#include <linux/slab.h>
  47#include <linux/idr.h>
  48#include <linux/workqueue.h>
  49
  50#include "rbd_types.h"
  51
  52#define RBD_DEBUG	/* Activate rbd_assert() calls */
  53
  54/*
  55 * Increment the given counter and return its updated value.
  56 * If the counter is already 0 it will not be incremented.
  57 * If the counter is already at its maximum value returns
  58 * -EINVAL without updating it.
  59 */
  60static int atomic_inc_return_safe(atomic_t *v)
  61{
  62	unsigned int counter;
  63
  64	counter = (unsigned int)atomic_fetch_add_unless(v, 1, 0);
  65	if (counter <= (unsigned int)INT_MAX)
  66		return (int)counter;
  67
  68	atomic_dec(v);
  69
  70	return -EINVAL;
  71}
  72
  73/* Decrement the counter.  Return the resulting value, or -EINVAL */
  74static int atomic_dec_return_safe(atomic_t *v)
  75{
  76	int counter;
  77
  78	counter = atomic_dec_return(v);
  79	if (counter >= 0)
  80		return counter;
  81
  82	atomic_inc(v);
  83
  84	return -EINVAL;
  85}
  86
  87#define RBD_DRV_NAME "rbd"
  88
  89#define RBD_MINORS_PER_MAJOR		256
  90#define RBD_SINGLE_MAJOR_PART_SHIFT	4
  91
  92#define RBD_MAX_PARENT_CHAIN_LEN	16
  93
  94#define RBD_SNAP_DEV_NAME_PREFIX	"snap_"
  95#define RBD_MAX_SNAP_NAME_LEN	\
  96			(NAME_MAX - (sizeof (RBD_SNAP_DEV_NAME_PREFIX) - 1))
  97
  98#define RBD_MAX_SNAP_COUNT	510	/* allows max snapc to fit in 4KB */
  99
 100#define RBD_SNAP_HEAD_NAME	"-"
 101
 102#define	BAD_SNAP_INDEX	U32_MAX		/* invalid index into snap array */
 103
 104/* This allows a single page to hold an image name sent by OSD */
 105#define RBD_IMAGE_NAME_LEN_MAX	(PAGE_SIZE - sizeof (__le32) - 1)
 106#define RBD_IMAGE_ID_LEN_MAX	64
 107
 108#define RBD_OBJ_PREFIX_LEN_MAX	64
 109
 110#define RBD_NOTIFY_TIMEOUT	5	/* seconds */
 111#define RBD_RETRY_DELAY		msecs_to_jiffies(1000)
 112
 113/* Feature bits */
 114
 115#define RBD_FEATURE_LAYERING		(1ULL<<0)
 116#define RBD_FEATURE_STRIPINGV2		(1ULL<<1)
 117#define RBD_FEATURE_EXCLUSIVE_LOCK	(1ULL<<2)
 118#define RBD_FEATURE_OBJECT_MAP		(1ULL<<3)
 119#define RBD_FEATURE_FAST_DIFF		(1ULL<<4)
 120#define RBD_FEATURE_DEEP_FLATTEN	(1ULL<<5)
 121#define RBD_FEATURE_DATA_POOL		(1ULL<<7)
 122#define RBD_FEATURE_OPERATIONS		(1ULL<<8)
 123
 124#define RBD_FEATURES_ALL	(RBD_FEATURE_LAYERING |		\
 125				 RBD_FEATURE_STRIPINGV2 |	\
 126				 RBD_FEATURE_EXCLUSIVE_LOCK |	\
 127				 RBD_FEATURE_OBJECT_MAP |	\
 128				 RBD_FEATURE_FAST_DIFF |	\
 129				 RBD_FEATURE_DEEP_FLATTEN |	\
 130				 RBD_FEATURE_DATA_POOL |	\
 131				 RBD_FEATURE_OPERATIONS)
 132
 133/* Features supported by this (client software) implementation. */
 134
 135#define RBD_FEATURES_SUPPORTED	(RBD_FEATURES_ALL)
 136
 137/*
 138 * An RBD device name will be "rbd#", where the "rbd" comes from
 139 * RBD_DRV_NAME above, and # is a unique integer identifier.
 140 */
 141#define DEV_NAME_LEN		32
 142
 143/*
 144 * block device image metadata (in-memory version)
 145 */
 146struct rbd_image_header {
 147	/* These six fields never change for a given rbd image */
 148	char *object_prefix;
 149	__u8 obj_order;
 150	u64 stripe_unit;
 151	u64 stripe_count;
 152	s64 data_pool_id;
 153	u64 features;		/* Might be changeable someday? */
 154
 155	/* The remaining fields need to be updated occasionally */
 156	u64 image_size;
 157	struct ceph_snap_context *snapc;
 158	char *snap_names;	/* format 1 only */
 159	u64 *snap_sizes;	/* format 1 only */
 160};
 161
 162/*
 163 * An rbd image specification.
 164 *
 165 * The tuple (pool_id, image_id, snap_id) is sufficient to uniquely
 166 * identify an image.  Each rbd_dev structure includes a pointer to
 167 * an rbd_spec structure that encapsulates this identity.
 168 *
 169 * Each of the id's in an rbd_spec has an associated name.  For a
 170 * user-mapped image, the names are supplied and the id's associated
 171 * with them are looked up.  For a layered image, a parent image is
 172 * defined by the tuple, and the names are looked up.
 173 *
 174 * An rbd_dev structure contains a parent_spec pointer which is
 175 * non-null if the image it represents is a child in a layered
 176 * image.  This pointer will refer to the rbd_spec structure used
 177 * by the parent rbd_dev for its own identity (i.e., the structure
 178 * is shared between the parent and child).
 179 *
 180 * Since these structures are populated once, during the discovery
 181 * phase of image construction, they are effectively immutable so
 182 * we make no effort to synchronize access to them.
 183 *
 184 * Note that code herein does not assume the image name is known (it
 185 * could be a null pointer).
 186 */
 187struct rbd_spec {
 188	u64		pool_id;
 189	const char	*pool_name;
 190	const char	*pool_ns;	/* NULL if default, never "" */
 191
 192	const char	*image_id;
 193	const char	*image_name;
 194
 195	u64		snap_id;
 196	const char	*snap_name;
 197
 198	struct kref	kref;
 199};
 200
 201/*
 202 * an instance of the client.  multiple devices may share an rbd client.
 203 */
 204struct rbd_client {
 205	struct ceph_client	*client;
 206	struct kref		kref;
 207	struct list_head	node;
 208};
 209
 210struct pending_result {
 211	int			result;		/* first nonzero result */
 212	int			num_pending;
 213};
 214
 215struct rbd_img_request;
 216
 217enum obj_request_type {
 218	OBJ_REQUEST_NODATA = 1,
 219	OBJ_REQUEST_BIO,	/* pointer into provided bio (list) */
 220	OBJ_REQUEST_BVECS,	/* pointer into provided bio_vec array */
 221	OBJ_REQUEST_OWN_BVECS,	/* private bio_vec array, doesn't own pages */
 222};
 223
 224enum obj_operation_type {
 225	OBJ_OP_READ = 1,
 226	OBJ_OP_WRITE,
 227	OBJ_OP_DISCARD,
 228	OBJ_OP_ZEROOUT,
 229};
 230
 231#define RBD_OBJ_FLAG_DELETION			(1U << 0)
 232#define RBD_OBJ_FLAG_COPYUP_ENABLED		(1U << 1)
 233#define RBD_OBJ_FLAG_COPYUP_ZEROS		(1U << 2)
 234#define RBD_OBJ_FLAG_MAY_EXIST			(1U << 3)
 235#define RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT	(1U << 4)
 236
 237enum rbd_obj_read_state {
 238	RBD_OBJ_READ_START = 1,
 239	RBD_OBJ_READ_OBJECT,
 240	RBD_OBJ_READ_PARENT,
 241};
 242
 243/*
 244 * Writes go through the following state machine to deal with
 245 * layering:
 246 *
 247 *            . . . . . RBD_OBJ_WRITE_GUARD. . . . . . . . . . . . . .
 248 *            .                 |                                    .
 249 *            .                 v                                    .
 250 *            .    RBD_OBJ_WRITE_READ_FROM_PARENT. . .               .
 251 *            .                 |                    .               .
 252 *            .                 v                    v (deep-copyup  .
 253 *    (image  .   RBD_OBJ_WRITE_COPYUP_EMPTY_SNAPC   .  not needed)  .
 254 * flattened) v                 |                    .               .
 255 *            .                 v                    .               .
 256 *            . . . .RBD_OBJ_WRITE_COPYUP_OPS. . . . .      (copyup  .
 257 *                              |                        not needed) v
 258 *                              v                                    .
 259 *                            done . . . . . . . . . . . . . . . . . .
 260 *                              ^
 261 *                              |
 262 *                     RBD_OBJ_WRITE_FLAT
 263 *
 264 * Writes start in RBD_OBJ_WRITE_GUARD or _FLAT, depending on whether
 265 * assert_exists guard is needed or not (in some cases it's not needed
 266 * even if there is a parent).
 267 */
 268enum rbd_obj_write_state {
 269	RBD_OBJ_WRITE_START = 1,
 270	RBD_OBJ_WRITE_PRE_OBJECT_MAP,
 271	RBD_OBJ_WRITE_OBJECT,
 272	__RBD_OBJ_WRITE_COPYUP,
 273	RBD_OBJ_WRITE_COPYUP,
 274	RBD_OBJ_WRITE_POST_OBJECT_MAP,
 275};
 276
 277enum rbd_obj_copyup_state {
 278	RBD_OBJ_COPYUP_START = 1,
 279	RBD_OBJ_COPYUP_READ_PARENT,
 280	__RBD_OBJ_COPYUP_OBJECT_MAPS,
 281	RBD_OBJ_COPYUP_OBJECT_MAPS,
 282	__RBD_OBJ_COPYUP_WRITE_OBJECT,
 283	RBD_OBJ_COPYUP_WRITE_OBJECT,
 284};
 285
 286struct rbd_obj_request {
 287	struct ceph_object_extent ex;
 288	unsigned int		flags;	/* RBD_OBJ_FLAG_* */
 289	union {
 290		enum rbd_obj_read_state	 read_state;	/* for reads */
 291		enum rbd_obj_write_state write_state;	/* for writes */
 292	};
 293
 294	struct rbd_img_request	*img_request;
 295	struct ceph_file_extent	*img_extents;
 296	u32			num_img_extents;
 297
 298	union {
 299		struct ceph_bio_iter	bio_pos;
 300		struct {
 301			struct ceph_bvec_iter	bvec_pos;
 302			u32			bvec_count;
 303			u32			bvec_idx;
 304		};
 305	};
 306
 307	enum rbd_obj_copyup_state copyup_state;
 308	struct bio_vec		*copyup_bvecs;
 309	u32			copyup_bvec_count;
 310
 311	struct list_head	osd_reqs;	/* w/ r_private_item */
 312
 313	struct mutex		state_mutex;
 314	struct pending_result	pending;
 315	struct kref		kref;
 316};
 317
 318enum img_req_flags {
 319	IMG_REQ_CHILD,		/* initiator: block = 0, child image = 1 */
 320	IMG_REQ_LAYERED,	/* ENOENT handling: normal = 0, layered = 1 */
 321};
 322
 323enum rbd_img_state {
 324	RBD_IMG_START = 1,
 325	RBD_IMG_EXCLUSIVE_LOCK,
 326	__RBD_IMG_OBJECT_REQUESTS,
 327	RBD_IMG_OBJECT_REQUESTS,
 328};
 329
 330struct rbd_img_request {
 331	struct rbd_device	*rbd_dev;
 332	enum obj_operation_type	op_type;
 333	enum obj_request_type	data_type;
 334	unsigned long		flags;
 335	enum rbd_img_state	state;
 336	union {
 337		u64			snap_id;	/* for reads */
 338		struct ceph_snap_context *snapc;	/* for writes */
 339	};
 340	struct rbd_obj_request	*obj_request;	/* obj req initiator */
 341
 342	struct list_head	lock_item;
 343	struct list_head	object_extents;	/* obj_req.ex structs */
 344
 345	struct mutex		state_mutex;
 346	struct pending_result	pending;
 347	struct work_struct	work;
 348	int			work_result;
 349};
 350
 351#define for_each_obj_request(ireq, oreq) \
 352	list_for_each_entry(oreq, &(ireq)->object_extents, ex.oe_item)
 353#define for_each_obj_request_safe(ireq, oreq, n) \
 354	list_for_each_entry_safe(oreq, n, &(ireq)->object_extents, ex.oe_item)
 355
 356enum rbd_watch_state {
 357	RBD_WATCH_STATE_UNREGISTERED,
 358	RBD_WATCH_STATE_REGISTERED,
 359	RBD_WATCH_STATE_ERROR,
 360};
 361
 362enum rbd_lock_state {
 363	RBD_LOCK_STATE_UNLOCKED,
 364	RBD_LOCK_STATE_LOCKED,
 365	RBD_LOCK_STATE_QUIESCING,
 366};
 367
 368/* WatchNotify::ClientId */
 369struct rbd_client_id {
 370	u64 gid;
 371	u64 handle;
 372};
 373
 374struct rbd_mapping {
 375	u64                     size;
 376};
 377
 378/*
 379 * a single device
 380 */
 381struct rbd_device {
 382	int			dev_id;		/* blkdev unique id */
 383
 384	int			major;		/* blkdev assigned major */
 385	int			minor;
 386	struct gendisk		*disk;		/* blkdev's gendisk and rq */
 387
 388	u32			image_format;	/* Either 1 or 2 */
 389	struct rbd_client	*rbd_client;
 390
 391	char			name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */
 392
 393	spinlock_t		lock;		/* queue, flags, open_count */
 394
 395	struct rbd_image_header	header;
 396	unsigned long		flags;		/* possibly lock protected */
 397	struct rbd_spec		*spec;
 398	struct rbd_options	*opts;
 399	char			*config_info;	/* add{,_single_major} string */
 400
 401	struct ceph_object_id	header_oid;
 402	struct ceph_object_locator header_oloc;
 403
 404	struct ceph_file_layout	layout;		/* used for all rbd requests */
 405
 406	struct mutex		watch_mutex;
 407	enum rbd_watch_state	watch_state;
 408	struct ceph_osd_linger_request *watch_handle;
 409	u64			watch_cookie;
 410	struct delayed_work	watch_dwork;
 411
 412	struct rw_semaphore	lock_rwsem;
 413	enum rbd_lock_state	lock_state;
 414	char			lock_cookie[32];
 415	struct rbd_client_id	owner_cid;
 416	struct work_struct	acquired_lock_work;
 417	struct work_struct	released_lock_work;
 418	struct delayed_work	lock_dwork;
 419	struct work_struct	unlock_work;
 420	spinlock_t		lock_lists_lock;
 421	struct list_head	acquiring_list;
 422	struct list_head	running_list;
 423	struct completion	acquire_wait;
 424	int			acquire_err;
 425	struct completion	quiescing_wait;
 426
 427	spinlock_t		object_map_lock;
 428	u8			*object_map;
 429	u64			object_map_size;	/* in objects */
 430	u64			object_map_flags;
 431
 432	struct workqueue_struct	*task_wq;
 433
 434	struct rbd_spec		*parent_spec;
 435	u64			parent_overlap;
 436	atomic_t		parent_ref;
 437	struct rbd_device	*parent;
 438
 439	/* Block layer tags. */
 440	struct blk_mq_tag_set	tag_set;
 441
 442	/* protects updating the header */
 443	struct rw_semaphore     header_rwsem;
 444
 445	struct rbd_mapping	mapping;
 446
 447	struct list_head	node;
 448
 449	/* sysfs related */
 450	struct device		dev;
 451	unsigned long		open_count;	/* protected by lock */
 452};
 453
 454/*
 455 * Flag bits for rbd_dev->flags:
 456 * - REMOVING (which is coupled with rbd_dev->open_count) is protected
 457 *   by rbd_dev->lock
 458 */
 459enum rbd_dev_flags {
 460	RBD_DEV_FLAG_EXISTS,	/* rbd_dev_device_setup() ran */
 461	RBD_DEV_FLAG_REMOVING,	/* this mapping is being removed */
 462	RBD_DEV_FLAG_READONLY,  /* -o ro or snapshot */
 463};
 464
 465static DEFINE_MUTEX(client_mutex);	/* Serialize client creation */
 466
 467static LIST_HEAD(rbd_dev_list);    /* devices */
 468static DEFINE_SPINLOCK(rbd_dev_list_lock);
 469
 470static LIST_HEAD(rbd_client_list);		/* clients */
 471static DEFINE_SPINLOCK(rbd_client_list_lock);
 472
 473/* Slab caches for frequently-allocated structures */
 474
 475static struct kmem_cache	*rbd_img_request_cache;
 476static struct kmem_cache	*rbd_obj_request_cache;
 477
 478static int rbd_major;
 479static DEFINE_IDA(rbd_dev_id_ida);
 480
 481static struct workqueue_struct *rbd_wq;
 482
 483static struct ceph_snap_context rbd_empty_snapc = {
 484	.nref = REFCOUNT_INIT(1),
 485};
 486
 487/*
 488 * single-major requires >= 0.75 version of userspace rbd utility.
 489 */
 490static bool single_major = true;
 491module_param(single_major, bool, 0444);
 492MODULE_PARM_DESC(single_major, "Use a single major number for all rbd devices (default: true)");
 493
 494static ssize_t add_store(const struct bus_type *bus, const char *buf, size_t count);
 495static ssize_t remove_store(const struct bus_type *bus, const char *buf,
 496			    size_t count);
 497static ssize_t add_single_major_store(const struct bus_type *bus, const char *buf,
 498				      size_t count);
 499static ssize_t remove_single_major_store(const struct bus_type *bus, const char *buf,
 500					 size_t count);
 501static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth);
 502
 503static int rbd_dev_id_to_minor(int dev_id)
 504{
 505	return dev_id << RBD_SINGLE_MAJOR_PART_SHIFT;
 506}
 507
 508static int minor_to_rbd_dev_id(int minor)
 509{
 510	return minor >> RBD_SINGLE_MAJOR_PART_SHIFT;
 511}
 512
 513static bool rbd_is_ro(struct rbd_device *rbd_dev)
 514{
 515	return test_bit(RBD_DEV_FLAG_READONLY, &rbd_dev->flags);
 516}
 517
 518static bool rbd_is_snap(struct rbd_device *rbd_dev)
 519{
 520	return rbd_dev->spec->snap_id != CEPH_NOSNAP;
 521}
 522
 523static bool __rbd_is_lock_owner(struct rbd_device *rbd_dev)
 524{
 525	lockdep_assert_held(&rbd_dev->lock_rwsem);
 526
 527	return rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED ||
 528	       rbd_dev->lock_state == RBD_LOCK_STATE_QUIESCING;
 529}
 530
 531static bool rbd_is_lock_owner(struct rbd_device *rbd_dev)
 532{
 533	bool is_lock_owner;
 534
 535	down_read(&rbd_dev->lock_rwsem);
 536	is_lock_owner = __rbd_is_lock_owner(rbd_dev);
 537	up_read(&rbd_dev->lock_rwsem);
 538	return is_lock_owner;
 539}
 540
 541static ssize_t supported_features_show(const struct bus_type *bus, char *buf)
 542{
 543	return sprintf(buf, "0x%llx\n", RBD_FEATURES_SUPPORTED);
 544}
 545
 546static BUS_ATTR_WO(add);
 547static BUS_ATTR_WO(remove);
 548static BUS_ATTR_WO(add_single_major);
 549static BUS_ATTR_WO(remove_single_major);
 550static BUS_ATTR_RO(supported_features);
 551
 552static struct attribute *rbd_bus_attrs[] = {
 553	&bus_attr_add.attr,
 554	&bus_attr_remove.attr,
 555	&bus_attr_add_single_major.attr,
 556	&bus_attr_remove_single_major.attr,
 557	&bus_attr_supported_features.attr,
 558	NULL,
 559};
 560
 561static umode_t rbd_bus_is_visible(struct kobject *kobj,
 562				  struct attribute *attr, int index)
 563{
 564	if (!single_major &&
 565	    (attr == &bus_attr_add_single_major.attr ||
 566	     attr == &bus_attr_remove_single_major.attr))
 567		return 0;
 568
 569	return attr->mode;
 570}
 571
 572static const struct attribute_group rbd_bus_group = {
 573	.attrs = rbd_bus_attrs,
 574	.is_visible = rbd_bus_is_visible,
 575};
 576__ATTRIBUTE_GROUPS(rbd_bus);
 577
 578static const struct bus_type rbd_bus_type = {
 579	.name		= "rbd",
 580	.bus_groups	= rbd_bus_groups,
 581};
 582
 583static void rbd_root_dev_release(struct device *dev)
 584{
 585}
 586
 587static struct device rbd_root_dev = {
 588	.init_name =    "rbd",
 589	.release =      rbd_root_dev_release,
 590};
 591
 592static __printf(2, 3)
 593void rbd_warn(struct rbd_device *rbd_dev, const char *fmt, ...)
 594{
 595	struct va_format vaf;
 596	va_list args;
 597
 598	va_start(args, fmt);
 599	vaf.fmt = fmt;
 600	vaf.va = &args;
 601
 602	if (!rbd_dev)
 603		printk(KERN_WARNING "%s: %pV\n", RBD_DRV_NAME, &vaf);
 604	else if (rbd_dev->disk)
 605		printk(KERN_WARNING "%s: %s: %pV\n",
 606			RBD_DRV_NAME, rbd_dev->disk->disk_name, &vaf);
 607	else if (rbd_dev->spec && rbd_dev->spec->image_name)
 608		printk(KERN_WARNING "%s: image %s: %pV\n",
 609			RBD_DRV_NAME, rbd_dev->spec->image_name, &vaf);
 610	else if (rbd_dev->spec && rbd_dev->spec->image_id)
 611		printk(KERN_WARNING "%s: id %s: %pV\n",
 612			RBD_DRV_NAME, rbd_dev->spec->image_id, &vaf);
 613	else	/* punt */
 614		printk(KERN_WARNING "%s: rbd_dev %p: %pV\n",
 615			RBD_DRV_NAME, rbd_dev, &vaf);
 616	va_end(args);
 617}
 618
 619#ifdef RBD_DEBUG
 620#define rbd_assert(expr)						\
 621		if (unlikely(!(expr))) {				\
 622			printk(KERN_ERR "\nAssertion failure in %s() "	\
 623						"at line %d:\n\n"	\
 624					"\trbd_assert(%s);\n\n",	\
 625					__func__, __LINE__, #expr);	\
 626			BUG();						\
 627		}
 628#else /* !RBD_DEBUG */
 629#  define rbd_assert(expr)	((void) 0)
 630#endif /* !RBD_DEBUG */
 631
 632static void rbd_dev_remove_parent(struct rbd_device *rbd_dev);
 633
 634static int rbd_dev_refresh(struct rbd_device *rbd_dev);
 635static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev,
 636				     struct rbd_image_header *header);
 637static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
 638					u64 snap_id);
 639static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
 640				u8 *order, u64 *snap_size);
 641static int rbd_dev_v2_get_flags(struct rbd_device *rbd_dev);
 642
 643static void rbd_obj_handle_request(struct rbd_obj_request *obj_req, int result);
 644static void rbd_img_handle_request(struct rbd_img_request *img_req, int result);
 645
 646/*
 647 * Return true if nothing else is pending.
 648 */
 649static bool pending_result_dec(struct pending_result *pending, int *result)
 650{
 651	rbd_assert(pending->num_pending > 0);
 652
 653	if (*result && !pending->result)
 654		pending->result = *result;
 655	if (--pending->num_pending)
 656		return false;
 657
 658	*result = pending->result;
 659	return true;
 660}
 661
 662static int rbd_open(struct gendisk *disk, blk_mode_t mode)
 663{
 664	struct rbd_device *rbd_dev = disk->private_data;
 665	bool removing = false;
 666
 667	spin_lock_irq(&rbd_dev->lock);
 668	if (test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
 669		removing = true;
 670	else
 671		rbd_dev->open_count++;
 672	spin_unlock_irq(&rbd_dev->lock);
 673	if (removing)
 674		return -ENOENT;
 675
 676	(void) get_device(&rbd_dev->dev);
 677
 678	return 0;
 679}
 680
 681static void rbd_release(struct gendisk *disk)
 682{
 683	struct rbd_device *rbd_dev = disk->private_data;
 684	unsigned long open_count_before;
 685
 686	spin_lock_irq(&rbd_dev->lock);
 687	open_count_before = rbd_dev->open_count--;
 688	spin_unlock_irq(&rbd_dev->lock);
 689	rbd_assert(open_count_before > 0);
 690
 691	put_device(&rbd_dev->dev);
 692}
 693
 694static const struct block_device_operations rbd_bd_ops = {
 695	.owner			= THIS_MODULE,
 696	.open			= rbd_open,
 697	.release		= rbd_release,
 698};
 699
 700/*
 701 * Initialize an rbd client instance.  Success or not, this function
 702 * consumes ceph_opts.  Caller holds client_mutex.
 703 */
 704static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
 705{
 706	struct rbd_client *rbdc;
 707	int ret = -ENOMEM;
 708
 709	dout("%s:\n", __func__);
 710	rbdc = kmalloc(sizeof(struct rbd_client), GFP_KERNEL);
 711	if (!rbdc)
 712		goto out_opt;
 713
 714	kref_init(&rbdc->kref);
 715	INIT_LIST_HEAD(&rbdc->node);
 716
 717	rbdc->client = ceph_create_client(ceph_opts, rbdc);
 718	if (IS_ERR(rbdc->client))
 719		goto out_rbdc;
 720	ceph_opts = NULL; /* Now rbdc->client is responsible for ceph_opts */
 721
 722	ret = ceph_open_session(rbdc->client);
 723	if (ret < 0)
 724		goto out_client;
 725
 726	spin_lock(&rbd_client_list_lock);
 727	list_add_tail(&rbdc->node, &rbd_client_list);
 728	spin_unlock(&rbd_client_list_lock);
 729
 730	dout("%s: rbdc %p\n", __func__, rbdc);
 731
 732	return rbdc;
 733out_client:
 734	ceph_destroy_client(rbdc->client);
 735out_rbdc:
 736	kfree(rbdc);
 737out_opt:
 738	if (ceph_opts)
 739		ceph_destroy_options(ceph_opts);
 740	dout("%s: error %d\n", __func__, ret);
 741
 742	return ERR_PTR(ret);
 743}
 744
 745static struct rbd_client *__rbd_get_client(struct rbd_client *rbdc)
 746{
 747	kref_get(&rbdc->kref);
 748
 749	return rbdc;
 750}
 751
 752/*
 753 * Find a ceph client with specific addr and configuration.  If
 754 * found, bump its reference count.
 755 */
 756static struct rbd_client *rbd_client_find(struct ceph_options *ceph_opts)
 757{
 758	struct rbd_client *rbdc = NULL, *iter;
 759
 760	if (ceph_opts->flags & CEPH_OPT_NOSHARE)
 761		return NULL;
 762
 763	spin_lock(&rbd_client_list_lock);
 764	list_for_each_entry(iter, &rbd_client_list, node) {
 765		if (!ceph_compare_options(ceph_opts, iter->client)) {
 766			__rbd_get_client(iter);
 767
 768			rbdc = iter;
 769			break;
 770		}
 771	}
 772	spin_unlock(&rbd_client_list_lock);
 773
 774	return rbdc;
 775}
 776
 777/*
 778 * (Per device) rbd map options
 779 */
 780enum {
 781	Opt_queue_depth,
 782	Opt_alloc_size,
 783	Opt_lock_timeout,
 784	/* int args above */
 785	Opt_pool_ns,
 786	Opt_compression_hint,
 787	/* string args above */
 788	Opt_read_only,
 789	Opt_read_write,
 790	Opt_lock_on_read,
 791	Opt_exclusive,
 792	Opt_notrim,
 793};
 794
 795enum {
 796	Opt_compression_hint_none,
 797	Opt_compression_hint_compressible,
 798	Opt_compression_hint_incompressible,
 799};
 800
 801static const struct constant_table rbd_param_compression_hint[] = {
 802	{"none",		Opt_compression_hint_none},
 803	{"compressible",	Opt_compression_hint_compressible},
 804	{"incompressible",	Opt_compression_hint_incompressible},
 805	{}
 806};
 807
 808static const struct fs_parameter_spec rbd_parameters[] = {
 809	fsparam_u32	("alloc_size",			Opt_alloc_size),
 810	fsparam_enum	("compression_hint",		Opt_compression_hint,
 811			 rbd_param_compression_hint),
 812	fsparam_flag	("exclusive",			Opt_exclusive),
 813	fsparam_flag	("lock_on_read",		Opt_lock_on_read),
 814	fsparam_u32	("lock_timeout",		Opt_lock_timeout),
 815	fsparam_flag	("notrim",			Opt_notrim),
 816	fsparam_string	("_pool_ns",			Opt_pool_ns),
 817	fsparam_u32	("queue_depth",			Opt_queue_depth),
 818	fsparam_flag	("read_only",			Opt_read_only),
 819	fsparam_flag	("read_write",			Opt_read_write),
 820	fsparam_flag	("ro",				Opt_read_only),
 821	fsparam_flag	("rw",				Opt_read_write),
 822	{}
 823};
 824
 825struct rbd_options {
 826	int	queue_depth;
 827	int	alloc_size;
 828	unsigned long	lock_timeout;
 829	bool	read_only;
 830	bool	lock_on_read;
 831	bool	exclusive;
 832	bool	trim;
 833
 834	u32 alloc_hint_flags;  /* CEPH_OSD_OP_ALLOC_HINT_FLAG_* */
 835};
 836
 837#define RBD_QUEUE_DEPTH_DEFAULT	BLKDEV_DEFAULT_RQ
 838#define RBD_ALLOC_SIZE_DEFAULT	(64 * 1024)
 839#define RBD_LOCK_TIMEOUT_DEFAULT 0  /* no timeout */
 840#define RBD_READ_ONLY_DEFAULT	false
 841#define RBD_LOCK_ON_READ_DEFAULT false
 842#define RBD_EXCLUSIVE_DEFAULT	false
 843#define RBD_TRIM_DEFAULT	true
 844
 845struct rbd_parse_opts_ctx {
 846	struct rbd_spec		*spec;
 847	struct ceph_options	*copts;
 848	struct rbd_options	*opts;
 849};
 850
 851static char* obj_op_name(enum obj_operation_type op_type)
 852{
 853	switch (op_type) {
 854	case OBJ_OP_READ:
 855		return "read";
 856	case OBJ_OP_WRITE:
 857		return "write";
 858	case OBJ_OP_DISCARD:
 859		return "discard";
 860	case OBJ_OP_ZEROOUT:
 861		return "zeroout";
 862	default:
 863		return "???";
 864	}
 865}
 866
 867/*
 868 * Destroy ceph client
 869 *
 870 * Caller must hold rbd_client_list_lock.
 871 */
 872static void rbd_client_release(struct kref *kref)
 873{
 874	struct rbd_client *rbdc = container_of(kref, struct rbd_client, kref);
 875
 876	dout("%s: rbdc %p\n", __func__, rbdc);
 877	spin_lock(&rbd_client_list_lock);
 878	list_del(&rbdc->node);
 879	spin_unlock(&rbd_client_list_lock);
 880
 881	ceph_destroy_client(rbdc->client);
 882	kfree(rbdc);
 883}
 884
 885/*
 886 * Drop reference to ceph client node. If it's not referenced anymore, release
 887 * it.
 888 */
 889static void rbd_put_client(struct rbd_client *rbdc)
 890{
 891	if (rbdc)
 892		kref_put(&rbdc->kref, rbd_client_release);
 893}
 894
 895/*
 896 * Get a ceph client with specific addr and configuration, if one does
 897 * not exist create it.  Either way, ceph_opts is consumed by this
 898 * function.
 899 */
 900static struct rbd_client *rbd_get_client(struct ceph_options *ceph_opts)
 901{
 902	struct rbd_client *rbdc;
 903	int ret;
 904
 905	mutex_lock(&client_mutex);
 906	rbdc = rbd_client_find(ceph_opts);
 907	if (rbdc) {
 908		ceph_destroy_options(ceph_opts);
 909
 910		/*
 911		 * Using an existing client.  Make sure ->pg_pools is up to
 912		 * date before we look up the pool id in do_rbd_add().
 913		 */
 914		ret = ceph_wait_for_latest_osdmap(rbdc->client,
 915					rbdc->client->options->mount_timeout);
 916		if (ret) {
 917			rbd_warn(NULL, "failed to get latest osdmap: %d", ret);
 918			rbd_put_client(rbdc);
 919			rbdc = ERR_PTR(ret);
 920		}
 921	} else {
 922		rbdc = rbd_client_create(ceph_opts);
 923	}
 924	mutex_unlock(&client_mutex);
 925
 926	return rbdc;
 927}
 928
 929static bool rbd_image_format_valid(u32 image_format)
 930{
 931	return image_format == 1 || image_format == 2;
 932}
 933
 934static bool rbd_dev_ondisk_valid(struct rbd_image_header_ondisk *ondisk)
 935{
 936	size_t size;
 937	u32 snap_count;
 938
 939	/* The header has to start with the magic rbd header text */
 940	if (memcmp(&ondisk->text, RBD_HEADER_TEXT, sizeof (RBD_HEADER_TEXT)))
 941		return false;
 942
 943	/* The bio layer requires at least sector-sized I/O */
 944
 945	if (ondisk->options.order < SECTOR_SHIFT)
 946		return false;
 947
 948	/* If we use u64 in a few spots we may be able to loosen this */
 949
 950	if (ondisk->options.order > 8 * sizeof (int) - 1)
 951		return false;
 952
 953	/*
 954	 * The size of a snapshot header has to fit in a size_t, and
 955	 * that limits the number of snapshots.
 956	 */
 957	snap_count = le32_to_cpu(ondisk->snap_count);
 958	size = SIZE_MAX - sizeof (struct ceph_snap_context);
 959	if (snap_count > size / sizeof (__le64))
 960		return false;
 961
 962	/*
 963	 * Not only that, but the size of the entire the snapshot
 964	 * header must also be representable in a size_t.
 965	 */
 966	size -= snap_count * sizeof (__le64);
 967	if ((u64) size < le64_to_cpu(ondisk->snap_names_len))
 968		return false;
 969
 970	return true;
 971}
 972
 973/*
 974 * returns the size of an object in the image
 975 */
 976static u32 rbd_obj_bytes(struct rbd_image_header *header)
 977{
 978	return 1U << header->obj_order;
 979}
 980
 981static void rbd_init_layout(struct rbd_device *rbd_dev)
 982{
 983	if (rbd_dev->header.stripe_unit == 0 ||
 984	    rbd_dev->header.stripe_count == 0) {
 985		rbd_dev->header.stripe_unit = rbd_obj_bytes(&rbd_dev->header);
 986		rbd_dev->header.stripe_count = 1;
 987	}
 988
 989	rbd_dev->layout.stripe_unit = rbd_dev->header.stripe_unit;
 990	rbd_dev->layout.stripe_count = rbd_dev->header.stripe_count;
 991	rbd_dev->layout.object_size = rbd_obj_bytes(&rbd_dev->header);
 992	rbd_dev->layout.pool_id = rbd_dev->header.data_pool_id == CEPH_NOPOOL ?
 993			  rbd_dev->spec->pool_id : rbd_dev->header.data_pool_id;
 994	RCU_INIT_POINTER(rbd_dev->layout.pool_ns, NULL);
 995}
 996
 997static void rbd_image_header_cleanup(struct rbd_image_header *header)
 998{
 999	kfree(header->object_prefix);
1000	ceph_put_snap_context(header->snapc);
1001	kfree(header->snap_sizes);
1002	kfree(header->snap_names);
1003
1004	memset(header, 0, sizeof(*header));
1005}
1006
1007/*
1008 * Fill an rbd image header with information from the given format 1
1009 * on-disk header.
1010 */
1011static int rbd_header_from_disk(struct rbd_image_header *header,
1012				struct rbd_image_header_ondisk *ondisk,
1013				bool first_time)
1014{
1015	struct ceph_snap_context *snapc;
1016	char *object_prefix = NULL;
1017	char *snap_names = NULL;
1018	u64 *snap_sizes = NULL;
1019	u32 snap_count;
1020	int ret = -ENOMEM;
1021	u32 i;
1022
1023	/* Allocate this now to avoid having to handle failure below */
1024
1025	if (first_time) {
1026		object_prefix = kstrndup(ondisk->object_prefix,
1027					 sizeof(ondisk->object_prefix),
1028					 GFP_KERNEL);
1029		if (!object_prefix)
1030			return -ENOMEM;
1031	}
1032
1033	/* Allocate the snapshot context and fill it in */
1034
1035	snap_count = le32_to_cpu(ondisk->snap_count);
1036	snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
1037	if (!snapc)
1038		goto out_err;
1039	snapc->seq = le64_to_cpu(ondisk->snap_seq);
1040	if (snap_count) {
1041		struct rbd_image_snap_ondisk *snaps;
1042		u64 snap_names_len = le64_to_cpu(ondisk->snap_names_len);
1043
1044		/* We'll keep a copy of the snapshot names... */
1045
1046		if (snap_names_len > (u64)SIZE_MAX)
1047			goto out_2big;
1048		snap_names = kmalloc(snap_names_len, GFP_KERNEL);
1049		if (!snap_names)
1050			goto out_err;
1051
1052		/* ...as well as the array of their sizes. */
1053		snap_sizes = kmalloc_array(snap_count,
1054					   sizeof(*header->snap_sizes),
1055					   GFP_KERNEL);
1056		if (!snap_sizes)
1057			goto out_err;
1058
1059		/*
1060		 * Copy the names, and fill in each snapshot's id
1061		 * and size.
1062		 *
1063		 * Note that rbd_dev_v1_header_info() guarantees the
1064		 * ondisk buffer we're working with has
1065		 * snap_names_len bytes beyond the end of the
1066		 * snapshot id array, this memcpy() is safe.
1067		 */
1068		memcpy(snap_names, &ondisk->snaps[snap_count], snap_names_len);
1069		snaps = ondisk->snaps;
1070		for (i = 0; i < snap_count; i++) {
1071			snapc->snaps[i] = le64_to_cpu(snaps[i].id);
1072			snap_sizes[i] = le64_to_cpu(snaps[i].image_size);
1073		}
1074	}
1075
1076	/* We won't fail any more, fill in the header */
1077
1078	if (first_time) {
1079		header->object_prefix = object_prefix;
1080		header->obj_order = ondisk->options.order;
1081	}
1082
1083	/* The remaining fields always get updated (when we refresh) */
1084
1085	header->image_size = le64_to_cpu(ondisk->image_size);
1086	header->snapc = snapc;
1087	header->snap_names = snap_names;
1088	header->snap_sizes = snap_sizes;
1089
1090	return 0;
1091out_2big:
1092	ret = -EIO;
1093out_err:
1094	kfree(snap_sizes);
1095	kfree(snap_names);
1096	ceph_put_snap_context(snapc);
1097	kfree(object_prefix);
1098
1099	return ret;
1100}
1101
1102static const char *_rbd_dev_v1_snap_name(struct rbd_device *rbd_dev, u32 which)
1103{
1104	const char *snap_name;
1105
1106	rbd_assert(which < rbd_dev->header.snapc->num_snaps);
1107
1108	/* Skip over names until we find the one we are looking for */
1109
1110	snap_name = rbd_dev->header.snap_names;
1111	while (which--)
1112		snap_name += strlen(snap_name) + 1;
1113
1114	return kstrdup(snap_name, GFP_KERNEL);
1115}
1116
1117/*
1118 * Snapshot id comparison function for use with qsort()/bsearch().
1119 * Note that result is for snapshots in *descending* order.
1120 */
1121static int snapid_compare_reverse(const void *s1, const void *s2)
1122{
1123	u64 snap_id1 = *(u64 *)s1;
1124	u64 snap_id2 = *(u64 *)s2;
1125
1126	if (snap_id1 < snap_id2)
1127		return 1;
1128	return snap_id1 == snap_id2 ? 0 : -1;
1129}
1130
1131/*
1132 * Search a snapshot context to see if the given snapshot id is
1133 * present.
1134 *
1135 * Returns the position of the snapshot id in the array if it's found,
1136 * or BAD_SNAP_INDEX otherwise.
1137 *
1138 * Note: The snapshot array is in kept sorted (by the osd) in
1139 * reverse order, highest snapshot id first.
1140 */
1141static u32 rbd_dev_snap_index(struct rbd_device *rbd_dev, u64 snap_id)
1142{
1143	struct ceph_snap_context *snapc = rbd_dev->header.snapc;
1144	u64 *found;
1145
1146	found = bsearch(&snap_id, &snapc->snaps, snapc->num_snaps,
1147				sizeof (snap_id), snapid_compare_reverse);
1148
1149	return found ? (u32)(found - &snapc->snaps[0]) : BAD_SNAP_INDEX;
1150}
1151
1152static const char *rbd_dev_v1_snap_name(struct rbd_device *rbd_dev,
1153					u64 snap_id)
1154{
1155	u32 which;
1156	const char *snap_name;
1157
1158	which = rbd_dev_snap_index(rbd_dev, snap_id);
1159	if (which == BAD_SNAP_INDEX)
1160		return ERR_PTR(-ENOENT);
1161
1162	snap_name = _rbd_dev_v1_snap_name(rbd_dev, which);
1163	return snap_name ? snap_name : ERR_PTR(-ENOMEM);
1164}
1165
1166static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id)
1167{
1168	if (snap_id == CEPH_NOSNAP)
1169		return RBD_SNAP_HEAD_NAME;
1170
1171	rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1172	if (rbd_dev->image_format == 1)
1173		return rbd_dev_v1_snap_name(rbd_dev, snap_id);
1174
1175	return rbd_dev_v2_snap_name(rbd_dev, snap_id);
1176}
1177
1178static int rbd_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
1179				u64 *snap_size)
1180{
1181	rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
1182	if (snap_id == CEPH_NOSNAP) {
1183		*snap_size = rbd_dev->header.image_size;
1184	} else if (rbd_dev->image_format == 1) {
1185		u32 which;
1186
1187		which = rbd_dev_snap_index(rbd_dev, snap_id);
1188		if (which == BAD_SNAP_INDEX)
1189			return -ENOENT;
1190
1191		*snap_size = rbd_dev->header.snap_sizes[which];
1192	} else {
1193		u64 size = 0;
1194		int ret;
1195
1196		ret = _rbd_dev_v2_snap_size(rbd_dev, snap_id, NULL, &size);
1197		if (ret)
1198			return ret;
1199
1200		*snap_size = size;
1201	}
1202	return 0;
1203}
1204
1205static int rbd_dev_mapping_set(struct rbd_device *rbd_dev)
1206{
1207	u64 snap_id = rbd_dev->spec->snap_id;
1208	u64 size = 0;
1209	int ret;
1210
1211	ret = rbd_snap_size(rbd_dev, snap_id, &size);
1212	if (ret)
1213		return ret;
1214
1215	rbd_dev->mapping.size = size;
1216	return 0;
1217}
1218
1219static void rbd_dev_mapping_clear(struct rbd_device *rbd_dev)
1220{
1221	rbd_dev->mapping.size = 0;
1222}
1223
1224static void zero_bios(struct ceph_bio_iter *bio_pos, u32 off, u32 bytes)
1225{
1226	struct ceph_bio_iter it = *bio_pos;
1227
1228	ceph_bio_iter_advance(&it, off);
1229	ceph_bio_iter_advance_step(&it, bytes, ({
1230		memzero_bvec(&bv);
1231	}));
1232}
1233
1234static void zero_bvecs(struct ceph_bvec_iter *bvec_pos, u32 off, u32 bytes)
1235{
1236	struct ceph_bvec_iter it = *bvec_pos;
1237
1238	ceph_bvec_iter_advance(&it, off);
1239	ceph_bvec_iter_advance_step(&it, bytes, ({
1240		memzero_bvec(&bv);
1241	}));
1242}
1243
1244/*
1245 * Zero a range in @obj_req data buffer defined by a bio (list) or
1246 * (private) bio_vec array.
1247 *
1248 * @off is relative to the start of the data buffer.
1249 */
1250static void rbd_obj_zero_range(struct rbd_obj_request *obj_req, u32 off,
1251			       u32 bytes)
1252{
1253	dout("%s %p data buf %u~%u\n", __func__, obj_req, off, bytes);
1254
1255	switch (obj_req->img_request->data_type) {
1256	case OBJ_REQUEST_BIO:
1257		zero_bios(&obj_req->bio_pos, off, bytes);
1258		break;
1259	case OBJ_REQUEST_BVECS:
1260	case OBJ_REQUEST_OWN_BVECS:
1261		zero_bvecs(&obj_req->bvec_pos, off, bytes);
1262		break;
1263	default:
1264		BUG();
1265	}
1266}
1267
1268static void rbd_obj_request_destroy(struct kref *kref);
1269static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
1270{
1271	rbd_assert(obj_request != NULL);
1272	dout("%s: obj %p (was %d)\n", __func__, obj_request,
1273		kref_read(&obj_request->kref));
1274	kref_put(&obj_request->kref, rbd_obj_request_destroy);
1275}
1276
1277static inline void rbd_img_obj_request_add(struct rbd_img_request *img_request,
1278					struct rbd_obj_request *obj_request)
1279{
1280	rbd_assert(obj_request->img_request == NULL);
1281
1282	/* Image request now owns object's original reference */
1283	obj_request->img_request = img_request;
1284	dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
1285}
1286
1287static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
1288					struct rbd_obj_request *obj_request)
1289{
1290	dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
1291	list_del(&obj_request->ex.oe_item);
1292	rbd_assert(obj_request->img_request == img_request);
1293	rbd_obj_request_put(obj_request);
1294}
1295
1296static void rbd_osd_submit(struct ceph_osd_request *osd_req)
1297{
1298	struct rbd_obj_request *obj_req = osd_req->r_priv;
1299
1300	dout("%s osd_req %p for obj_req %p objno %llu %llu~%llu\n",
1301	     __func__, osd_req, obj_req, obj_req->ex.oe_objno,
1302	     obj_req->ex.oe_off, obj_req->ex.oe_len);
1303	ceph_osdc_start_request(osd_req->r_osdc, osd_req);
1304}
1305
1306/*
1307 * The default/initial value for all image request flags is 0.  Each
1308 * is conditionally set to 1 at image request initialization time
1309 * and currently never change thereafter.
1310 */
1311static void img_request_layered_set(struct rbd_img_request *img_request)
1312{
1313	set_bit(IMG_REQ_LAYERED, &img_request->flags);
1314}
1315
1316static bool img_request_layered_test(struct rbd_img_request *img_request)
1317{
1318	return test_bit(IMG_REQ_LAYERED, &img_request->flags) != 0;
1319}
1320
1321static bool rbd_obj_is_entire(struct rbd_obj_request *obj_req)
1322{
1323	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1324
1325	return !obj_req->ex.oe_off &&
1326	       obj_req->ex.oe_len == rbd_dev->layout.object_size;
1327}
1328
1329static bool rbd_obj_is_tail(struct rbd_obj_request *obj_req)
1330{
1331	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1332
1333	return obj_req->ex.oe_off + obj_req->ex.oe_len ==
1334					rbd_dev->layout.object_size;
1335}
1336
1337/*
1338 * Must be called after rbd_obj_calc_img_extents().
1339 */
1340static void rbd_obj_set_copyup_enabled(struct rbd_obj_request *obj_req)
1341{
1342	rbd_assert(obj_req->img_request->snapc);
1343
1344	if (obj_req->img_request->op_type == OBJ_OP_DISCARD) {
1345		dout("%s %p objno %llu discard\n", __func__, obj_req,
1346		     obj_req->ex.oe_objno);
1347		return;
1348	}
1349
1350	if (!obj_req->num_img_extents) {
1351		dout("%s %p objno %llu not overlapping\n", __func__, obj_req,
1352		     obj_req->ex.oe_objno);
1353		return;
1354	}
1355
1356	if (rbd_obj_is_entire(obj_req) &&
1357	    !obj_req->img_request->snapc->num_snaps) {
1358		dout("%s %p objno %llu entire\n", __func__, obj_req,
1359		     obj_req->ex.oe_objno);
1360		return;
1361	}
1362
1363	obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ENABLED;
1364}
1365
1366static u64 rbd_obj_img_extents_bytes(struct rbd_obj_request *obj_req)
1367{
1368	return ceph_file_extents_bytes(obj_req->img_extents,
1369				       obj_req->num_img_extents);
1370}
1371
1372static bool rbd_img_is_write(struct rbd_img_request *img_req)
1373{
1374	switch (img_req->op_type) {
1375	case OBJ_OP_READ:
1376		return false;
1377	case OBJ_OP_WRITE:
1378	case OBJ_OP_DISCARD:
1379	case OBJ_OP_ZEROOUT:
1380		return true;
1381	default:
1382		BUG();
1383	}
1384}
1385
1386static void rbd_osd_req_callback(struct ceph_osd_request *osd_req)
1387{
1388	struct rbd_obj_request *obj_req = osd_req->r_priv;
1389	int result;
1390
1391	dout("%s osd_req %p result %d for obj_req %p\n", __func__, osd_req,
1392	     osd_req->r_result, obj_req);
1393
1394	/*
1395	 * Writes aren't allowed to return a data payload.  In some
1396	 * guarded write cases (e.g. stat + zero on an empty object)
1397	 * a stat response makes it through, but we don't care.
1398	 */
1399	if (osd_req->r_result > 0 && rbd_img_is_write(obj_req->img_request))
1400		result = 0;
1401	else
1402		result = osd_req->r_result;
1403
1404	rbd_obj_handle_request(obj_req, result);
1405}
1406
1407static void rbd_osd_format_read(struct ceph_osd_request *osd_req)
1408{
1409	struct rbd_obj_request *obj_request = osd_req->r_priv;
1410	struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev;
1411	struct ceph_options *opt = rbd_dev->rbd_client->client->options;
1412
1413	osd_req->r_flags = CEPH_OSD_FLAG_READ | opt->read_from_replica;
1414	osd_req->r_snapid = obj_request->img_request->snap_id;
1415}
1416
1417static void rbd_osd_format_write(struct ceph_osd_request *osd_req)
1418{
1419	struct rbd_obj_request *obj_request = osd_req->r_priv;
1420
1421	osd_req->r_flags = CEPH_OSD_FLAG_WRITE;
1422	ktime_get_real_ts64(&osd_req->r_mtime);
1423	osd_req->r_data_offset = obj_request->ex.oe_off;
1424}
1425
1426static struct ceph_osd_request *
1427__rbd_obj_add_osd_request(struct rbd_obj_request *obj_req,
1428			  struct ceph_snap_context *snapc, int num_ops)
1429{
1430	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1431	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1432	struct ceph_osd_request *req;
1433	const char *name_format = rbd_dev->image_format == 1 ?
1434				      RBD_V1_DATA_FORMAT : RBD_V2_DATA_FORMAT;
1435	int ret;
1436
1437	req = ceph_osdc_alloc_request(osdc, snapc, num_ops, false, GFP_NOIO);
1438	if (!req)
1439		return ERR_PTR(-ENOMEM);
1440
1441	list_add_tail(&req->r_private_item, &obj_req->osd_reqs);
1442	req->r_callback = rbd_osd_req_callback;
1443	req->r_priv = obj_req;
1444
1445	/*
1446	 * Data objects may be stored in a separate pool, but always in
1447	 * the same namespace in that pool as the header in its pool.
1448	 */
1449	ceph_oloc_copy(&req->r_base_oloc, &rbd_dev->header_oloc);
1450	req->r_base_oloc.pool = rbd_dev->layout.pool_id;
1451
1452	ret = ceph_oid_aprintf(&req->r_base_oid, GFP_NOIO, name_format,
1453			       rbd_dev->header.object_prefix,
1454			       obj_req->ex.oe_objno);
1455	if (ret)
1456		return ERR_PTR(ret);
1457
1458	return req;
1459}
1460
1461static struct ceph_osd_request *
1462rbd_obj_add_osd_request(struct rbd_obj_request *obj_req, int num_ops)
1463{
1464	rbd_assert(obj_req->img_request->snapc);
1465	return __rbd_obj_add_osd_request(obj_req, obj_req->img_request->snapc,
1466					 num_ops);
1467}
1468
1469static struct rbd_obj_request *rbd_obj_request_create(void)
1470{
1471	struct rbd_obj_request *obj_request;
1472
1473	obj_request = kmem_cache_zalloc(rbd_obj_request_cache, GFP_NOIO);
1474	if (!obj_request)
1475		return NULL;
1476
1477	ceph_object_extent_init(&obj_request->ex);
1478	INIT_LIST_HEAD(&obj_request->osd_reqs);
1479	mutex_init(&obj_request->state_mutex);
1480	kref_init(&obj_request->kref);
1481
1482	dout("%s %p\n", __func__, obj_request);
1483	return obj_request;
1484}
1485
1486static void rbd_obj_request_destroy(struct kref *kref)
1487{
1488	struct rbd_obj_request *obj_request;
1489	struct ceph_osd_request *osd_req;
1490	u32 i;
1491
1492	obj_request = container_of(kref, struct rbd_obj_request, kref);
1493
1494	dout("%s: obj %p\n", __func__, obj_request);
1495
1496	while (!list_empty(&obj_request->osd_reqs)) {
1497		osd_req = list_first_entry(&obj_request->osd_reqs,
1498				    struct ceph_osd_request, r_private_item);
1499		list_del_init(&osd_req->r_private_item);
1500		ceph_osdc_put_request(osd_req);
1501	}
1502
1503	switch (obj_request->img_request->data_type) {
1504	case OBJ_REQUEST_NODATA:
1505	case OBJ_REQUEST_BIO:
1506	case OBJ_REQUEST_BVECS:
1507		break;		/* Nothing to do */
1508	case OBJ_REQUEST_OWN_BVECS:
1509		kfree(obj_request->bvec_pos.bvecs);
1510		break;
1511	default:
1512		BUG();
1513	}
1514
1515	kfree(obj_request->img_extents);
1516	if (obj_request->copyup_bvecs) {
1517		for (i = 0; i < obj_request->copyup_bvec_count; i++) {
1518			if (obj_request->copyup_bvecs[i].bv_page)
1519				__free_page(obj_request->copyup_bvecs[i].bv_page);
1520		}
1521		kfree(obj_request->copyup_bvecs);
1522	}
1523
1524	kmem_cache_free(rbd_obj_request_cache, obj_request);
1525}
1526
1527/* It's OK to call this for a device with no parent */
1528
1529static void rbd_spec_put(struct rbd_spec *spec);
1530static void rbd_dev_unparent(struct rbd_device *rbd_dev)
1531{
1532	rbd_dev_remove_parent(rbd_dev);
1533	rbd_spec_put(rbd_dev->parent_spec);
1534	rbd_dev->parent_spec = NULL;
1535	rbd_dev->parent_overlap = 0;
1536}
1537
1538/*
1539 * Parent image reference counting is used to determine when an
1540 * image's parent fields can be safely torn down--after there are no
1541 * more in-flight requests to the parent image.  When the last
1542 * reference is dropped, cleaning them up is safe.
1543 */
1544static void rbd_dev_parent_put(struct rbd_device *rbd_dev)
1545{
1546	int counter;
1547
1548	if (!rbd_dev->parent_spec)
1549		return;
1550
1551	counter = atomic_dec_return_safe(&rbd_dev->parent_ref);
1552	if (counter > 0)
1553		return;
1554
1555	/* Last reference; clean up parent data structures */
1556
1557	if (!counter)
1558		rbd_dev_unparent(rbd_dev);
1559	else
1560		rbd_warn(rbd_dev, "parent reference underflow");
1561}
1562
1563/*
1564 * If an image has a non-zero parent overlap, get a reference to its
1565 * parent.
1566 *
1567 * Returns true if the rbd device has a parent with a non-zero
1568 * overlap and a reference for it was successfully taken, or
1569 * false otherwise.
1570 */
1571static bool rbd_dev_parent_get(struct rbd_device *rbd_dev)
1572{
1573	int counter = 0;
1574
1575	if (!rbd_dev->parent_spec)
1576		return false;
1577
1578	if (rbd_dev->parent_overlap)
1579		counter = atomic_inc_return_safe(&rbd_dev->parent_ref);
1580
1581	if (counter < 0)
1582		rbd_warn(rbd_dev, "parent reference overflow");
1583
1584	return counter > 0;
1585}
1586
1587static void rbd_img_request_init(struct rbd_img_request *img_request,
1588				 struct rbd_device *rbd_dev,
1589				 enum obj_operation_type op_type)
1590{
1591	memset(img_request, 0, sizeof(*img_request));
1592
1593	img_request->rbd_dev = rbd_dev;
1594	img_request->op_type = op_type;
1595
1596	INIT_LIST_HEAD(&img_request->lock_item);
1597	INIT_LIST_HEAD(&img_request->object_extents);
1598	mutex_init(&img_request->state_mutex);
1599}
1600
1601/*
1602 * Only snap_id is captured here, for reads.  For writes, snapshot
1603 * context is captured in rbd_img_object_requests() after exclusive
1604 * lock is ensured to be held.
1605 */
1606static void rbd_img_capture_header(struct rbd_img_request *img_req)
1607{
1608	struct rbd_device *rbd_dev = img_req->rbd_dev;
1609
1610	lockdep_assert_held(&rbd_dev->header_rwsem);
1611
1612	if (!rbd_img_is_write(img_req))
1613		img_req->snap_id = rbd_dev->spec->snap_id;
1614
1615	if (rbd_dev_parent_get(rbd_dev))
1616		img_request_layered_set(img_req);
1617}
1618
1619static void rbd_img_request_destroy(struct rbd_img_request *img_request)
1620{
1621	struct rbd_obj_request *obj_request;
1622	struct rbd_obj_request *next_obj_request;
1623
1624	dout("%s: img %p\n", __func__, img_request);
1625
1626	WARN_ON(!list_empty(&img_request->lock_item));
1627	for_each_obj_request_safe(img_request, obj_request, next_obj_request)
1628		rbd_img_obj_request_del(img_request, obj_request);
1629
1630	if (img_request_layered_test(img_request))
1631		rbd_dev_parent_put(img_request->rbd_dev);
1632
1633	if (rbd_img_is_write(img_request))
1634		ceph_put_snap_context(img_request->snapc);
1635
1636	if (test_bit(IMG_REQ_CHILD, &img_request->flags))
1637		kmem_cache_free(rbd_img_request_cache, img_request);
1638}
1639
1640#define BITS_PER_OBJ	2
1641#define OBJS_PER_BYTE	(BITS_PER_BYTE / BITS_PER_OBJ)
1642#define OBJ_MASK	((1 << BITS_PER_OBJ) - 1)
1643
1644static void __rbd_object_map_index(struct rbd_device *rbd_dev, u64 objno,
1645				   u64 *index, u8 *shift)
1646{
1647	u32 off;
1648
1649	rbd_assert(objno < rbd_dev->object_map_size);
1650	*index = div_u64_rem(objno, OBJS_PER_BYTE, &off);
1651	*shift = (OBJS_PER_BYTE - off - 1) * BITS_PER_OBJ;
1652}
1653
1654static u8 __rbd_object_map_get(struct rbd_device *rbd_dev, u64 objno)
1655{
1656	u64 index;
1657	u8 shift;
1658
1659	lockdep_assert_held(&rbd_dev->object_map_lock);
1660	__rbd_object_map_index(rbd_dev, objno, &index, &shift);
1661	return (rbd_dev->object_map[index] >> shift) & OBJ_MASK;
1662}
1663
1664static void __rbd_object_map_set(struct rbd_device *rbd_dev, u64 objno, u8 val)
1665{
1666	u64 index;
1667	u8 shift;
1668	u8 *p;
1669
1670	lockdep_assert_held(&rbd_dev->object_map_lock);
1671	rbd_assert(!(val & ~OBJ_MASK));
1672
1673	__rbd_object_map_index(rbd_dev, objno, &index, &shift);
1674	p = &rbd_dev->object_map[index];
1675	*p = (*p & ~(OBJ_MASK << shift)) | (val << shift);
1676}
1677
1678static u8 rbd_object_map_get(struct rbd_device *rbd_dev, u64 objno)
1679{
1680	u8 state;
1681
1682	spin_lock(&rbd_dev->object_map_lock);
1683	state = __rbd_object_map_get(rbd_dev, objno);
1684	spin_unlock(&rbd_dev->object_map_lock);
1685	return state;
1686}
1687
1688static bool use_object_map(struct rbd_device *rbd_dev)
1689{
1690	/*
1691	 * An image mapped read-only can't use the object map -- it isn't
1692	 * loaded because the header lock isn't acquired.  Someone else can
1693	 * write to the image and update the object map behind our back.
1694	 *
1695	 * A snapshot can't be written to, so using the object map is always
1696	 * safe.
1697	 */
1698	if (!rbd_is_snap(rbd_dev) && rbd_is_ro(rbd_dev))
1699		return false;
1700
1701	return ((rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP) &&
1702		!(rbd_dev->object_map_flags & RBD_FLAG_OBJECT_MAP_INVALID));
1703}
1704
1705static bool rbd_object_map_may_exist(struct rbd_device *rbd_dev, u64 objno)
1706{
1707	u8 state;
1708
1709	/* fall back to default logic if object map is disabled or invalid */
1710	if (!use_object_map(rbd_dev))
1711		return true;
1712
1713	state = rbd_object_map_get(rbd_dev, objno);
1714	return state != OBJECT_NONEXISTENT;
1715}
1716
1717static void rbd_object_map_name(struct rbd_device *rbd_dev, u64 snap_id,
1718				struct ceph_object_id *oid)
1719{
1720	if (snap_id == CEPH_NOSNAP)
1721		ceph_oid_printf(oid, "%s%s", RBD_OBJECT_MAP_PREFIX,
1722				rbd_dev->spec->image_id);
1723	else
1724		ceph_oid_printf(oid, "%s%s.%016llx", RBD_OBJECT_MAP_PREFIX,
1725				rbd_dev->spec->image_id, snap_id);
1726}
1727
1728static int rbd_object_map_lock(struct rbd_device *rbd_dev)
1729{
1730	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1731	CEPH_DEFINE_OID_ONSTACK(oid);
1732	u8 lock_type;
1733	char *lock_tag;
1734	struct ceph_locker *lockers;
1735	u32 num_lockers;
1736	bool broke_lock = false;
1737	int ret;
1738
1739	rbd_object_map_name(rbd_dev, CEPH_NOSNAP, &oid);
1740
1741again:
1742	ret = ceph_cls_lock(osdc, &oid, &rbd_dev->header_oloc, RBD_LOCK_NAME,
1743			    CEPH_CLS_LOCK_EXCLUSIVE, "", "", "", 0);
1744	if (ret != -EBUSY || broke_lock) {
1745		if (ret == -EEXIST)
1746			ret = 0; /* already locked by myself */
1747		if (ret)
1748			rbd_warn(rbd_dev, "failed to lock object map: %d", ret);
1749		return ret;
1750	}
1751
1752	ret = ceph_cls_lock_info(osdc, &oid, &rbd_dev->header_oloc,
1753				 RBD_LOCK_NAME, &lock_type, &lock_tag,
1754				 &lockers, &num_lockers);
1755	if (ret) {
1756		if (ret == -ENOENT)
1757			goto again;
1758
1759		rbd_warn(rbd_dev, "failed to get object map lockers: %d", ret);
1760		return ret;
1761	}
1762
1763	kfree(lock_tag);
1764	if (num_lockers == 0)
1765		goto again;
1766
1767	rbd_warn(rbd_dev, "breaking object map lock owned by %s%llu",
1768		 ENTITY_NAME(lockers[0].id.name));
1769
1770	ret = ceph_cls_break_lock(osdc, &oid, &rbd_dev->header_oloc,
1771				  RBD_LOCK_NAME, lockers[0].id.cookie,
1772				  &lockers[0].id.name);
1773	ceph_free_lockers(lockers, num_lockers);
1774	if (ret) {
1775		if (ret == -ENOENT)
1776			goto again;
1777
1778		rbd_warn(rbd_dev, "failed to break object map lock: %d", ret);
1779		return ret;
1780	}
1781
1782	broke_lock = true;
1783	goto again;
1784}
1785
1786static void rbd_object_map_unlock(struct rbd_device *rbd_dev)
1787{
1788	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1789	CEPH_DEFINE_OID_ONSTACK(oid);
1790	int ret;
1791
1792	rbd_object_map_name(rbd_dev, CEPH_NOSNAP, &oid);
1793
1794	ret = ceph_cls_unlock(osdc, &oid, &rbd_dev->header_oloc, RBD_LOCK_NAME,
1795			      "");
1796	if (ret && ret != -ENOENT)
1797		rbd_warn(rbd_dev, "failed to unlock object map: %d", ret);
1798}
1799
1800static int decode_object_map_header(void **p, void *end, u64 *object_map_size)
1801{
1802	u8 struct_v;
1803	u32 struct_len;
1804	u32 header_len;
1805	void *header_end;
1806	int ret;
1807
1808	ceph_decode_32_safe(p, end, header_len, e_inval);
1809	header_end = *p + header_len;
1810
1811	ret = ceph_start_decoding(p, end, 1, "BitVector header", &struct_v,
1812				  &struct_len);
1813	if (ret)
1814		return ret;
1815
1816	ceph_decode_64_safe(p, end, *object_map_size, e_inval);
1817
1818	*p = header_end;
1819	return 0;
1820
1821e_inval:
1822	return -EINVAL;
1823}
1824
1825static int __rbd_object_map_load(struct rbd_device *rbd_dev)
1826{
1827	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
1828	CEPH_DEFINE_OID_ONSTACK(oid);
1829	struct page **pages;
1830	void *p, *end;
1831	size_t reply_len;
1832	u64 num_objects;
1833	u64 object_map_bytes;
1834	u64 object_map_size;
1835	int num_pages;
1836	int ret;
1837
1838	rbd_assert(!rbd_dev->object_map && !rbd_dev->object_map_size);
1839
1840	num_objects = ceph_get_num_objects(&rbd_dev->layout,
1841					   rbd_dev->mapping.size);
1842	object_map_bytes = DIV_ROUND_UP_ULL(num_objects * BITS_PER_OBJ,
1843					    BITS_PER_BYTE);
1844	num_pages = calc_pages_for(0, object_map_bytes) + 1;
1845	pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
1846	if (IS_ERR(pages))
1847		return PTR_ERR(pages);
1848
1849	reply_len = num_pages * PAGE_SIZE;
1850	rbd_object_map_name(rbd_dev, rbd_dev->spec->snap_id, &oid);
1851	ret = ceph_osdc_call(osdc, &oid, &rbd_dev->header_oloc,
1852			     "rbd", "object_map_load", CEPH_OSD_FLAG_READ,
1853			     NULL, 0, pages, &reply_len);
1854	if (ret)
1855		goto out;
1856
1857	p = page_address(pages[0]);
1858	end = p + min(reply_len, (size_t)PAGE_SIZE);
1859	ret = decode_object_map_header(&p, end, &object_map_size);
1860	if (ret)
1861		goto out;
1862
1863	if (object_map_size != num_objects) {
1864		rbd_warn(rbd_dev, "object map size mismatch: %llu vs %llu",
1865			 object_map_size, num_objects);
1866		ret = -EINVAL;
1867		goto out;
1868	}
1869
1870	if (offset_in_page(p) + object_map_bytes > reply_len) {
1871		ret = -EINVAL;
1872		goto out;
1873	}
1874
1875	rbd_dev->object_map = kvmalloc(object_map_bytes, GFP_KERNEL);
1876	if (!rbd_dev->object_map) {
1877		ret = -ENOMEM;
1878		goto out;
1879	}
1880
1881	rbd_dev->object_map_size = object_map_size;
1882	ceph_copy_from_page_vector(pages, rbd_dev->object_map,
1883				   offset_in_page(p), object_map_bytes);
1884
1885out:
1886	ceph_release_page_vector(pages, num_pages);
1887	return ret;
1888}
1889
1890static void rbd_object_map_free(struct rbd_device *rbd_dev)
1891{
1892	kvfree(rbd_dev->object_map);
1893	rbd_dev->object_map = NULL;
1894	rbd_dev->object_map_size = 0;
1895}
1896
1897static int rbd_object_map_load(struct rbd_device *rbd_dev)
1898{
1899	int ret;
1900
1901	ret = __rbd_object_map_load(rbd_dev);
1902	if (ret)
1903		return ret;
1904
1905	ret = rbd_dev_v2_get_flags(rbd_dev);
1906	if (ret) {
1907		rbd_object_map_free(rbd_dev);
1908		return ret;
1909	}
1910
1911	if (rbd_dev->object_map_flags & RBD_FLAG_OBJECT_MAP_INVALID)
1912		rbd_warn(rbd_dev, "object map is invalid");
1913
1914	return 0;
1915}
1916
1917static int rbd_object_map_open(struct rbd_device *rbd_dev)
1918{
1919	int ret;
1920
1921	ret = rbd_object_map_lock(rbd_dev);
1922	if (ret)
1923		return ret;
1924
1925	ret = rbd_object_map_load(rbd_dev);
1926	if (ret) {
1927		rbd_object_map_unlock(rbd_dev);
1928		return ret;
1929	}
1930
1931	return 0;
1932}
1933
1934static void rbd_object_map_close(struct rbd_device *rbd_dev)
1935{
1936	rbd_object_map_free(rbd_dev);
1937	rbd_object_map_unlock(rbd_dev);
1938}
1939
1940/*
1941 * This function needs snap_id (or more precisely just something to
1942 * distinguish between HEAD and snapshot object maps), new_state and
1943 * current_state that were passed to rbd_object_map_update().
1944 *
1945 * To avoid allocating and stashing a context we piggyback on the OSD
1946 * request.  A HEAD update has two ops (assert_locked).  For new_state
1947 * and current_state we decode our own object_map_update op, encoded in
1948 * rbd_cls_object_map_update().
1949 */
1950static int rbd_object_map_update_finish(struct rbd_obj_request *obj_req,
1951					struct ceph_osd_request *osd_req)
1952{
1953	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
1954	struct ceph_osd_data *osd_data;
1955	u64 objno;
1956	u8 state, new_state, current_state;
1957	bool has_current_state;
1958	void *p;
1959
1960	if (osd_req->r_result)
1961		return osd_req->r_result;
1962
1963	/*
1964	 * Nothing to do for a snapshot object map.
1965	 */
1966	if (osd_req->r_num_ops == 1)
1967		return 0;
1968
1969	/*
1970	 * Update in-memory HEAD object map.
1971	 */
1972	rbd_assert(osd_req->r_num_ops == 2);
1973	osd_data = osd_req_op_data(osd_req, 1, cls, request_data);
1974	rbd_assert(osd_data->type == CEPH_OSD_DATA_TYPE_PAGES);
1975
1976	p = page_address(osd_data->pages[0]);
1977	objno = ceph_decode_64(&p);
1978	rbd_assert(objno == obj_req->ex.oe_objno);
1979	rbd_assert(ceph_decode_64(&p) == objno + 1);
1980	new_state = ceph_decode_8(&p);
1981	has_current_state = ceph_decode_8(&p);
1982	if (has_current_state)
1983		current_state = ceph_decode_8(&p);
1984
1985	spin_lock(&rbd_dev->object_map_lock);
1986	state = __rbd_object_map_get(rbd_dev, objno);
1987	if (!has_current_state || current_state == state ||
1988	    (current_state == OBJECT_EXISTS && state == OBJECT_EXISTS_CLEAN))
1989		__rbd_object_map_set(rbd_dev, objno, new_state);
1990	spin_unlock(&rbd_dev->object_map_lock);
1991
1992	return 0;
1993}
1994
1995static void rbd_object_map_callback(struct ceph_osd_request *osd_req)
1996{
1997	struct rbd_obj_request *obj_req = osd_req->r_priv;
1998	int result;
1999
2000	dout("%s osd_req %p result %d for obj_req %p\n", __func__, osd_req,
2001	     osd_req->r_result, obj_req);
2002
2003	result = rbd_object_map_update_finish(obj_req, osd_req);
2004	rbd_obj_handle_request(obj_req, result);
2005}
2006
2007static bool update_needed(struct rbd_device *rbd_dev, u64 objno, u8 new_state)
2008{
2009	u8 state = rbd_object_map_get(rbd_dev, objno);
2010
2011	if (state == new_state ||
2012	    (new_state == OBJECT_PENDING && state == OBJECT_NONEXISTENT) ||
2013	    (new_state == OBJECT_NONEXISTENT && state != OBJECT_PENDING))
2014		return false;
2015
2016	return true;
2017}
2018
2019static int rbd_cls_object_map_update(struct ceph_osd_request *req,
2020				     int which, u64 objno, u8 new_state,
2021				     const u8 *current_state)
2022{
2023	struct page **pages;
2024	void *p, *start;
2025	int ret;
2026
2027	ret = osd_req_op_cls_init(req, which, "rbd", "object_map_update");
2028	if (ret)
2029		return ret;
2030
2031	pages = ceph_alloc_page_vector(1, GFP_NOIO);
2032	if (IS_ERR(pages))
2033		return PTR_ERR(pages);
2034
2035	p = start = page_address(pages[0]);
2036	ceph_encode_64(&p, objno);
2037	ceph_encode_64(&p, objno + 1);
2038	ceph_encode_8(&p, new_state);
2039	if (current_state) {
2040		ceph_encode_8(&p, 1);
2041		ceph_encode_8(&p, *current_state);
2042	} else {
2043		ceph_encode_8(&p, 0);
2044	}
2045
2046	osd_req_op_cls_request_data_pages(req, which, pages, p - start, 0,
2047					  false, true);
2048	return 0;
2049}
2050
2051/*
2052 * Return:
2053 *   0 - object map update sent
2054 *   1 - object map update isn't needed
2055 *  <0 - error
2056 */
2057static int rbd_object_map_update(struct rbd_obj_request *obj_req, u64 snap_id,
2058				 u8 new_state, const u8 *current_state)
2059{
2060	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2061	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
2062	struct ceph_osd_request *req;
2063	int num_ops = 1;
2064	int which = 0;
2065	int ret;
2066
2067	if (snap_id == CEPH_NOSNAP) {
2068		if (!update_needed(rbd_dev, obj_req->ex.oe_objno, new_state))
2069			return 1;
2070
2071		num_ops++; /* assert_locked */
2072	}
2073
2074	req = ceph_osdc_alloc_request(osdc, NULL, num_ops, false, GFP_NOIO);
2075	if (!req)
2076		return -ENOMEM;
2077
2078	list_add_tail(&req->r_private_item, &obj_req->osd_reqs);
2079	req->r_callback = rbd_object_map_callback;
2080	req->r_priv = obj_req;
2081
2082	rbd_object_map_name(rbd_dev, snap_id, &req->r_base_oid);
2083	ceph_oloc_copy(&req->r_base_oloc, &rbd_dev->header_oloc);
2084	req->r_flags = CEPH_OSD_FLAG_WRITE;
2085	ktime_get_real_ts64(&req->r_mtime);
2086
2087	if (snap_id == CEPH_NOSNAP) {
2088		/*
2089		 * Protect against possible race conditions during lock
2090		 * ownership transitions.
2091		 */
2092		ret = ceph_cls_assert_locked(req, which++, RBD_LOCK_NAME,
2093					     CEPH_CLS_LOCK_EXCLUSIVE, "", "");
2094		if (ret)
2095			return ret;
2096	}
2097
2098	ret = rbd_cls_object_map_update(req, which, obj_req->ex.oe_objno,
2099					new_state, current_state);
2100	if (ret)
2101		return ret;
2102
2103	ret = ceph_osdc_alloc_messages(req, GFP_NOIO);
2104	if (ret)
2105		return ret;
2106
2107	ceph_osdc_start_request(osdc, req);
2108	return 0;
2109}
2110
2111static void prune_extents(struct ceph_file_extent *img_extents,
2112			  u32 *num_img_extents, u64 overlap)
2113{
2114	u32 cnt = *num_img_extents;
2115
2116	/* drop extents completely beyond the overlap */
2117	while (cnt && img_extents[cnt - 1].fe_off >= overlap)
2118		cnt--;
2119
2120	if (cnt) {
2121		struct ceph_file_extent *ex = &img_extents[cnt - 1];
2122
2123		/* trim final overlapping extent */
2124		if (ex->fe_off + ex->fe_len > overlap)
2125			ex->fe_len = overlap - ex->fe_off;
2126	}
2127
2128	*num_img_extents = cnt;
2129}
2130
2131/*
2132 * Determine the byte range(s) covered by either just the object extent
2133 * or the entire object in the parent image.
2134 */
2135static int rbd_obj_calc_img_extents(struct rbd_obj_request *obj_req,
2136				    bool entire)
2137{
2138	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2139	int ret;
2140
2141	if (!rbd_dev->parent_overlap)
2142		return 0;
2143
2144	ret = ceph_extent_to_file(&rbd_dev->layout, obj_req->ex.oe_objno,
2145				  entire ? 0 : obj_req->ex.oe_off,
2146				  entire ? rbd_dev->layout.object_size :
2147							obj_req->ex.oe_len,
2148				  &obj_req->img_extents,
2149				  &obj_req->num_img_extents);
2150	if (ret)
2151		return ret;
2152
2153	prune_extents(obj_req->img_extents, &obj_req->num_img_extents,
2154		      rbd_dev->parent_overlap);
2155	return 0;
2156}
2157
2158static void rbd_osd_setup_data(struct ceph_osd_request *osd_req, int which)
2159{
2160	struct rbd_obj_request *obj_req = osd_req->r_priv;
2161
2162	switch (obj_req->img_request->data_type) {
2163	case OBJ_REQUEST_BIO:
2164		osd_req_op_extent_osd_data_bio(osd_req, which,
2165					       &obj_req->bio_pos,
2166					       obj_req->ex.oe_len);
2167		break;
2168	case OBJ_REQUEST_BVECS:
2169	case OBJ_REQUEST_OWN_BVECS:
2170		rbd_assert(obj_req->bvec_pos.iter.bi_size ==
2171							obj_req->ex.oe_len);
2172		rbd_assert(obj_req->bvec_idx == obj_req->bvec_count);
2173		osd_req_op_extent_osd_data_bvec_pos(osd_req, which,
2174						    &obj_req->bvec_pos);
2175		break;
2176	default:
2177		BUG();
2178	}
2179}
2180
2181static int rbd_osd_setup_stat(struct ceph_osd_request *osd_req, int which)
2182{
2183	struct page **pages;
2184
2185	/*
2186	 * The response data for a STAT call consists of:
2187	 *     le64 length;
2188	 *     struct {
2189	 *         le32 tv_sec;
2190	 *         le32 tv_nsec;
2191	 *     } mtime;
2192	 */
2193	pages = ceph_alloc_page_vector(1, GFP_NOIO);
2194	if (IS_ERR(pages))
2195		return PTR_ERR(pages);
2196
2197	osd_req_op_init(osd_req, which, CEPH_OSD_OP_STAT, 0);
2198	osd_req_op_raw_data_in_pages(osd_req, which, pages,
2199				     8 + sizeof(struct ceph_timespec),
2200				     0, false, true);
2201	return 0;
2202}
2203
2204static int rbd_osd_setup_copyup(struct ceph_osd_request *osd_req, int which,
2205				u32 bytes)
2206{
2207	struct rbd_obj_request *obj_req = osd_req->r_priv;
2208	int ret;
2209
2210	ret = osd_req_op_cls_init(osd_req, which, "rbd", "copyup");
2211	if (ret)
2212		return ret;
2213
2214	osd_req_op_cls_request_data_bvecs(osd_req, which, obj_req->copyup_bvecs,
2215					  obj_req->copyup_bvec_count, bytes);
2216	return 0;
2217}
2218
2219static int rbd_obj_init_read(struct rbd_obj_request *obj_req)
2220{
2221	obj_req->read_state = RBD_OBJ_READ_START;
2222	return 0;
2223}
2224
2225static void __rbd_osd_setup_write_ops(struct ceph_osd_request *osd_req,
2226				      int which)
2227{
2228	struct rbd_obj_request *obj_req = osd_req->r_priv;
2229	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2230	u16 opcode;
2231
2232	if (!use_object_map(rbd_dev) ||
2233	    !(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST)) {
2234		osd_req_op_alloc_hint_init(osd_req, which++,
2235					   rbd_dev->layout.object_size,
2236					   rbd_dev->layout.object_size,
2237					   rbd_dev->opts->alloc_hint_flags);
2238	}
2239
2240	if (rbd_obj_is_entire(obj_req))
2241		opcode = CEPH_OSD_OP_WRITEFULL;
2242	else
2243		opcode = CEPH_OSD_OP_WRITE;
2244
2245	osd_req_op_extent_init(osd_req, which, opcode,
2246			       obj_req->ex.oe_off, obj_req->ex.oe_len, 0, 0);
2247	rbd_osd_setup_data(osd_req, which);
2248}
2249
2250static int rbd_obj_init_write(struct rbd_obj_request *obj_req)
2251{
2252	int ret;
2253
2254	/* reverse map the entire object onto the parent */
2255	ret = rbd_obj_calc_img_extents(obj_req, true);
2256	if (ret)
2257		return ret;
2258
2259	obj_req->write_state = RBD_OBJ_WRITE_START;
2260	return 0;
2261}
2262
2263static u16 truncate_or_zero_opcode(struct rbd_obj_request *obj_req)
2264{
2265	return rbd_obj_is_tail(obj_req) ? CEPH_OSD_OP_TRUNCATE :
2266					  CEPH_OSD_OP_ZERO;
2267}
2268
2269static void __rbd_osd_setup_discard_ops(struct ceph_osd_request *osd_req,
2270					int which)
2271{
2272	struct rbd_obj_request *obj_req = osd_req->r_priv;
2273
2274	if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents) {
2275		rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
2276		osd_req_op_init(osd_req, which, CEPH_OSD_OP_DELETE, 0);
2277	} else {
2278		osd_req_op_extent_init(osd_req, which,
2279				       truncate_or_zero_opcode(obj_req),
2280				       obj_req->ex.oe_off, obj_req->ex.oe_len,
2281				       0, 0);
2282	}
2283}
2284
2285static int rbd_obj_init_discard(struct rbd_obj_request *obj_req)
2286{
2287	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2288	u64 off, next_off;
2289	int ret;
2290
2291	/*
2292	 * Align the range to alloc_size boundary and punt on discards
2293	 * that are too small to free up any space.
2294	 *
2295	 * alloc_size == object_size && is_tail() is a special case for
2296	 * filestore with filestore_punch_hole = false, needed to allow
2297	 * truncate (in addition to delete).
2298	 */
2299	if (rbd_dev->opts->alloc_size != rbd_dev->layout.object_size ||
2300	    !rbd_obj_is_tail(obj_req)) {
2301		off = round_up(obj_req->ex.oe_off, rbd_dev->opts->alloc_size);
2302		next_off = round_down(obj_req->ex.oe_off + obj_req->ex.oe_len,
2303				      rbd_dev->opts->alloc_size);
2304		if (off >= next_off)
2305			return 1;
2306
2307		dout("%s %p %llu~%llu -> %llu~%llu\n", __func__,
2308		     obj_req, obj_req->ex.oe_off, obj_req->ex.oe_len,
2309		     off, next_off - off);
2310		obj_req->ex.oe_off = off;
2311		obj_req->ex.oe_len = next_off - off;
2312	}
2313
2314	/* reverse map the entire object onto the parent */
2315	ret = rbd_obj_calc_img_extents(obj_req, true);
2316	if (ret)
2317		return ret;
2318
2319	obj_req->flags |= RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT;
2320	if (rbd_obj_is_entire(obj_req) && !obj_req->num_img_extents)
2321		obj_req->flags |= RBD_OBJ_FLAG_DELETION;
2322
2323	obj_req->write_state = RBD_OBJ_WRITE_START;
2324	return 0;
2325}
2326
2327static void __rbd_osd_setup_zeroout_ops(struct ceph_osd_request *osd_req,
2328					int which)
2329{
2330	struct rbd_obj_request *obj_req = osd_req->r_priv;
2331	u16 opcode;
2332
2333	if (rbd_obj_is_entire(obj_req)) {
2334		if (obj_req->num_img_extents) {
2335			if (!(obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED))
2336				osd_req_op_init(osd_req, which++,
2337						CEPH_OSD_OP_CREATE, 0);
2338			opcode = CEPH_OSD_OP_TRUNCATE;
2339		} else {
2340			rbd_assert(obj_req->flags & RBD_OBJ_FLAG_DELETION);
2341			osd_req_op_init(osd_req, which++,
2342					CEPH_OSD_OP_DELETE, 0);
2343			opcode = 0;
2344		}
2345	} else {
2346		opcode = truncate_or_zero_opcode(obj_req);
2347	}
2348
2349	if (opcode)
2350		osd_req_op_extent_init(osd_req, which, opcode,
2351				       obj_req->ex.oe_off, obj_req->ex.oe_len,
2352				       0, 0);
2353}
2354
2355static int rbd_obj_init_zeroout(struct rbd_obj_request *obj_req)
2356{
2357	int ret;
2358
2359	/* reverse map the entire object onto the parent */
2360	ret = rbd_obj_calc_img_extents(obj_req, true);
2361	if (ret)
2362		return ret;
2363
2364	if (!obj_req->num_img_extents) {
2365		obj_req->flags |= RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT;
2366		if (rbd_obj_is_entire(obj_req))
2367			obj_req->flags |= RBD_OBJ_FLAG_DELETION;
2368	}
2369
2370	obj_req->write_state = RBD_OBJ_WRITE_START;
2371	return 0;
2372}
2373
2374static int count_write_ops(struct rbd_obj_request *obj_req)
2375{
2376	struct rbd_img_request *img_req = obj_req->img_request;
2377
2378	switch (img_req->op_type) {
2379	case OBJ_OP_WRITE:
2380		if (!use_object_map(img_req->rbd_dev) ||
2381		    !(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST))
2382			return 2; /* setallochint + write/writefull */
2383
2384		return 1; /* write/writefull */
2385	case OBJ_OP_DISCARD:
2386		return 1; /* delete/truncate/zero */
2387	case OBJ_OP_ZEROOUT:
2388		if (rbd_obj_is_entire(obj_req) && obj_req->num_img_extents &&
2389		    !(obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED))
2390			return 2; /* create + truncate */
2391
2392		return 1; /* delete/truncate/zero */
2393	default:
2394		BUG();
2395	}
2396}
2397
2398static void rbd_osd_setup_write_ops(struct ceph_osd_request *osd_req,
2399				    int which)
2400{
2401	struct rbd_obj_request *obj_req = osd_req->r_priv;
2402
2403	switch (obj_req->img_request->op_type) {
2404	case OBJ_OP_WRITE:
2405		__rbd_osd_setup_write_ops(osd_req, which);
2406		break;
2407	case OBJ_OP_DISCARD:
2408		__rbd_osd_setup_discard_ops(osd_req, which);
2409		break;
2410	case OBJ_OP_ZEROOUT:
2411		__rbd_osd_setup_zeroout_ops(osd_req, which);
2412		break;
2413	default:
2414		BUG();
2415	}
2416}
2417
2418/*
2419 * Prune the list of object requests (adjust offset and/or length, drop
2420 * redundant requests).  Prepare object request state machines and image
2421 * request state machine for execution.
2422 */
2423static int __rbd_img_fill_request(struct rbd_img_request *img_req)
2424{
2425	struct rbd_obj_request *obj_req, *next_obj_req;
2426	int ret;
2427
2428	for_each_obj_request_safe(img_req, obj_req, next_obj_req) {
2429		switch (img_req->op_type) {
2430		case OBJ_OP_READ:
2431			ret = rbd_obj_init_read(obj_req);
2432			break;
2433		case OBJ_OP_WRITE:
2434			ret = rbd_obj_init_write(obj_req);
2435			break;
2436		case OBJ_OP_DISCARD:
2437			ret = rbd_obj_init_discard(obj_req);
2438			break;
2439		case OBJ_OP_ZEROOUT:
2440			ret = rbd_obj_init_zeroout(obj_req);
2441			break;
2442		default:
2443			BUG();
2444		}
2445		if (ret < 0)
2446			return ret;
2447		if (ret > 0) {
2448			rbd_img_obj_request_del(img_req, obj_req);
2449			continue;
2450		}
2451	}
2452
2453	img_req->state = RBD_IMG_START;
2454	return 0;
2455}
2456
2457union rbd_img_fill_iter {
2458	struct ceph_bio_iter	bio_iter;
2459	struct ceph_bvec_iter	bvec_iter;
2460};
2461
2462struct rbd_img_fill_ctx {
2463	enum obj_request_type	pos_type;
2464	union rbd_img_fill_iter	*pos;
2465	union rbd_img_fill_iter	iter;
2466	ceph_object_extent_fn_t	set_pos_fn;
2467	ceph_object_extent_fn_t	count_fn;
2468	ceph_object_extent_fn_t	copy_fn;
2469};
2470
2471static struct ceph_object_extent *alloc_object_extent(void *arg)
2472{
2473	struct rbd_img_request *img_req = arg;
2474	struct rbd_obj_request *obj_req;
2475
2476	obj_req = rbd_obj_request_create();
2477	if (!obj_req)
2478		return NULL;
2479
2480	rbd_img_obj_request_add(img_req, obj_req);
2481	return &obj_req->ex;
2482}
2483
2484/*
2485 * While su != os && sc == 1 is technically not fancy (it's the same
2486 * layout as su == os && sc == 1), we can't use the nocopy path for it
2487 * because ->set_pos_fn() should be called only once per object.
2488 * ceph_file_to_extents() invokes action_fn once per stripe unit, so
2489 * treat su != os && sc == 1 as fancy.
2490 */
2491static bool rbd_layout_is_fancy(struct ceph_file_layout *l)
2492{
2493	return l->stripe_unit != l->object_size;
2494}
2495
2496static int rbd_img_fill_request_nocopy(struct rbd_img_request *img_req,
2497				       struct ceph_file_extent *img_extents,
2498				       u32 num_img_extents,
2499				       struct rbd_img_fill_ctx *fctx)
2500{
2501	u32 i;
2502	int ret;
2503
2504	img_req->data_type = fctx->pos_type;
2505
2506	/*
2507	 * Create object requests and set each object request's starting
2508	 * position in the provided bio (list) or bio_vec array.
2509	 */
2510	fctx->iter = *fctx->pos;
2511	for (i = 0; i < num_img_extents; i++) {
2512		ret = ceph_file_to_extents(&img_req->rbd_dev->layout,
2513					   img_extents[i].fe_off,
2514					   img_extents[i].fe_len,
2515					   &img_req->object_extents,
2516					   alloc_object_extent, img_req,
2517					   fctx->set_pos_fn, &fctx->iter);
2518		if (ret)
2519			return ret;
2520	}
2521
2522	return __rbd_img_fill_request(img_req);
2523}
2524
2525/*
2526 * Map a list of image extents to a list of object extents, create the
2527 * corresponding object requests (normally each to a different object,
2528 * but not always) and add them to @img_req.  For each object request,
2529 * set up its data descriptor to point to the corresponding chunk(s) of
2530 * @fctx->pos data buffer.
2531 *
2532 * Because ceph_file_to_extents() will merge adjacent object extents
2533 * together, each object request's data descriptor may point to multiple
2534 * different chunks of @fctx->pos data buffer.
2535 *
2536 * @fctx->pos data buffer is assumed to be large enough.
2537 */
2538static int rbd_img_fill_request(struct rbd_img_request *img_req,
2539				struct ceph_file_extent *img_extents,
2540				u32 num_img_extents,
2541				struct rbd_img_fill_ctx *fctx)
2542{
2543	struct rbd_device *rbd_dev = img_req->rbd_dev;
2544	struct rbd_obj_request *obj_req;
2545	u32 i;
2546	int ret;
2547
2548	if (fctx->pos_type == OBJ_REQUEST_NODATA ||
2549	    !rbd_layout_is_fancy(&rbd_dev->layout))
2550		return rbd_img_fill_request_nocopy(img_req, img_extents,
2551						   num_img_extents, fctx);
2552
2553	img_req->data_type = OBJ_REQUEST_OWN_BVECS;
2554
2555	/*
2556	 * Create object requests and determine ->bvec_count for each object
2557	 * request.  Note that ->bvec_count sum over all object requests may
2558	 * be greater than the number of bio_vecs in the provided bio (list)
2559	 * or bio_vec array because when mapped, those bio_vecs can straddle
2560	 * stripe unit boundaries.
2561	 */
2562	fctx->iter = *fctx->pos;
2563	for (i = 0; i < num_img_extents; i++) {
2564		ret = ceph_file_to_extents(&rbd_dev->layout,
2565					   img_extents[i].fe_off,
2566					   img_extents[i].fe_len,
2567					   &img_req->object_extents,
2568					   alloc_object_extent, img_req,
2569					   fctx->count_fn, &fctx->iter);
2570		if (ret)
2571			return ret;
2572	}
2573
2574	for_each_obj_request(img_req, obj_req) {
2575		obj_req->bvec_pos.bvecs = kmalloc_array(obj_req->bvec_count,
2576					      sizeof(*obj_req->bvec_pos.bvecs),
2577					      GFP_NOIO);
2578		if (!obj_req->bvec_pos.bvecs)
2579			return -ENOMEM;
2580	}
2581
2582	/*
2583	 * Fill in each object request's private bio_vec array, splitting and
2584	 * rearranging the provided bio_vecs in stripe unit chunks as needed.
2585	 */
2586	fctx->iter = *fctx->pos;
2587	for (i = 0; i < num_img_extents; i++) {
2588		ret = ceph_iterate_extents(&rbd_dev->layout,
2589					   img_extents[i].fe_off,
2590					   img_extents[i].fe_len,
2591					   &img_req->object_extents,
2592					   fctx->copy_fn, &fctx->iter);
2593		if (ret)
2594			return ret;
2595	}
2596
2597	return __rbd_img_fill_request(img_req);
2598}
2599
2600static int rbd_img_fill_nodata(struct rbd_img_request *img_req,
2601			       u64 off, u64 len)
2602{
2603	struct ceph_file_extent ex = { off, len };
2604	union rbd_img_fill_iter dummy = {};
2605	struct rbd_img_fill_ctx fctx = {
2606		.pos_type = OBJ_REQUEST_NODATA,
2607		.pos = &dummy,
2608	};
2609
2610	return rbd_img_fill_request(img_req, &ex, 1, &fctx);
2611}
2612
2613static void set_bio_pos(struct ceph_object_extent *ex, u32 bytes, void *arg)
2614{
2615	struct rbd_obj_request *obj_req =
2616	    container_of(ex, struct rbd_obj_request, ex);
2617	struct ceph_bio_iter *it = arg;
2618
2619	dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2620	obj_req->bio_pos = *it;
2621	ceph_bio_iter_advance(it, bytes);
2622}
2623
2624static void count_bio_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2625{
2626	struct rbd_obj_request *obj_req =
2627	    container_of(ex, struct rbd_obj_request, ex);
2628	struct ceph_bio_iter *it = arg;
2629
2630	dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2631	ceph_bio_iter_advance_step(it, bytes, ({
2632		obj_req->bvec_count++;
2633	}));
2634
2635}
2636
2637static void copy_bio_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2638{
2639	struct rbd_obj_request *obj_req =
2640	    container_of(ex, struct rbd_obj_request, ex);
2641	struct ceph_bio_iter *it = arg;
2642
2643	dout("%s objno %llu bytes %u\n", __func__, ex->oe_objno, bytes);
2644	ceph_bio_iter_advance_step(it, bytes, ({
2645		obj_req->bvec_pos.bvecs[obj_req->bvec_idx++] = bv;
2646		obj_req->bvec_pos.iter.bi_size += bv.bv_len;
2647	}));
2648}
2649
2650static int __rbd_img_fill_from_bio(struct rbd_img_request *img_req,
2651				   struct ceph_file_extent *img_extents,
2652				   u32 num_img_extents,
2653				   struct ceph_bio_iter *bio_pos)
2654{
2655	struct rbd_img_fill_ctx fctx = {
2656		.pos_type = OBJ_REQUEST_BIO,
2657		.pos = (union rbd_img_fill_iter *)bio_pos,
2658		.set_pos_fn = set_bio_pos,
2659		.count_fn = count_bio_bvecs,
2660		.copy_fn = copy_bio_bvecs,
2661	};
2662
2663	return rbd_img_fill_request(img_req, img_extents, num_img_extents,
2664				    &fctx);
2665}
2666
2667static int rbd_img_fill_from_bio(struct rbd_img_request *img_req,
2668				 u64 off, u64 len, struct bio *bio)
2669{
2670	struct ceph_file_extent ex = { off, len };
2671	struct ceph_bio_iter it = { .bio = bio, .iter = bio->bi_iter };
2672
2673	return __rbd_img_fill_from_bio(img_req, &ex, 1, &it);
2674}
2675
2676static void set_bvec_pos(struct ceph_object_extent *ex, u32 bytes, void *arg)
2677{
2678	struct rbd_obj_request *obj_req =
2679	    container_of(ex, struct rbd_obj_request, ex);
2680	struct ceph_bvec_iter *it = arg;
2681
2682	obj_req->bvec_pos = *it;
2683	ceph_bvec_iter_shorten(&obj_req->bvec_pos, bytes);
2684	ceph_bvec_iter_advance(it, bytes);
2685}
2686
2687static void count_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2688{
2689	struct rbd_obj_request *obj_req =
2690	    container_of(ex, struct rbd_obj_request, ex);
2691	struct ceph_bvec_iter *it = arg;
2692
2693	ceph_bvec_iter_advance_step(it, bytes, ({
2694		obj_req->bvec_count++;
2695	}));
2696}
2697
2698static void copy_bvecs(struct ceph_object_extent *ex, u32 bytes, void *arg)
2699{
2700	struct rbd_obj_request *obj_req =
2701	    container_of(ex, struct rbd_obj_request, ex);
2702	struct ceph_bvec_iter *it = arg;
2703
2704	ceph_bvec_iter_advance_step(it, bytes, ({
2705		obj_req->bvec_pos.bvecs[obj_req->bvec_idx++] = bv;
2706		obj_req->bvec_pos.iter.bi_size += bv.bv_len;
2707	}));
2708}
2709
2710static int __rbd_img_fill_from_bvecs(struct rbd_img_request *img_req,
2711				     struct ceph_file_extent *img_extents,
2712				     u32 num_img_extents,
2713				     struct ceph_bvec_iter *bvec_pos)
2714{
2715	struct rbd_img_fill_ctx fctx = {
2716		.pos_type = OBJ_REQUEST_BVECS,
2717		.pos = (union rbd_img_fill_iter *)bvec_pos,
2718		.set_pos_fn = set_bvec_pos,
2719		.count_fn = count_bvecs,
2720		.copy_fn = copy_bvecs,
2721	};
2722
2723	return rbd_img_fill_request(img_req, img_extents, num_img_extents,
2724				    &fctx);
2725}
2726
2727static int rbd_img_fill_from_bvecs(struct rbd_img_request *img_req,
2728				   struct ceph_file_extent *img_extents,
2729				   u32 num_img_extents,
2730				   struct bio_vec *bvecs)
2731{
2732	struct ceph_bvec_iter it = {
2733		.bvecs = bvecs,
2734		.iter = { .bi_size = ceph_file_extents_bytes(img_extents,
2735							     num_img_extents) },
2736	};
2737
2738	return __rbd_img_fill_from_bvecs(img_req, img_extents, num_img_extents,
2739					 &it);
2740}
2741
2742static void rbd_img_handle_request_work(struct work_struct *work)
2743{
2744	struct rbd_img_request *img_req =
2745	    container_of(work, struct rbd_img_request, work);
2746
2747	rbd_img_handle_request(img_req, img_req->work_result);
2748}
2749
2750static void rbd_img_schedule(struct rbd_img_request *img_req, int result)
2751{
2752	INIT_WORK(&img_req->work, rbd_img_handle_request_work);
2753	img_req->work_result = result;
2754	queue_work(rbd_wq, &img_req->work);
2755}
2756
2757static bool rbd_obj_may_exist(struct rbd_obj_request *obj_req)
2758{
2759	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2760
2761	if (rbd_object_map_may_exist(rbd_dev, obj_req->ex.oe_objno)) {
2762		obj_req->flags |= RBD_OBJ_FLAG_MAY_EXIST;
2763		return true;
2764	}
2765
2766	dout("%s %p objno %llu assuming dne\n", __func__, obj_req,
2767	     obj_req->ex.oe_objno);
2768	return false;
2769}
2770
2771static int rbd_obj_read_object(struct rbd_obj_request *obj_req)
2772{
2773	struct ceph_osd_request *osd_req;
2774	int ret;
2775
2776	osd_req = __rbd_obj_add_osd_request(obj_req, NULL, 1);
2777	if (IS_ERR(osd_req))
2778		return PTR_ERR(osd_req);
2779
2780	osd_req_op_extent_init(osd_req, 0, CEPH_OSD_OP_READ,
2781			       obj_req->ex.oe_off, obj_req->ex.oe_len, 0, 0);
2782	rbd_osd_setup_data(osd_req, 0);
2783	rbd_osd_format_read(osd_req);
2784
2785	ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
2786	if (ret)
2787		return ret;
2788
2789	rbd_osd_submit(osd_req);
2790	return 0;
2791}
2792
2793static int rbd_obj_read_from_parent(struct rbd_obj_request *obj_req)
2794{
2795	struct rbd_img_request *img_req = obj_req->img_request;
2796	struct rbd_device *parent = img_req->rbd_dev->parent;
2797	struct rbd_img_request *child_img_req;
2798	int ret;
2799
2800	child_img_req = kmem_cache_alloc(rbd_img_request_cache, GFP_NOIO);
2801	if (!child_img_req)
2802		return -ENOMEM;
2803
2804	rbd_img_request_init(child_img_req, parent, OBJ_OP_READ);
2805	__set_bit(IMG_REQ_CHILD, &child_img_req->flags);
2806	child_img_req->obj_request = obj_req;
2807
2808	down_read(&parent->header_rwsem);
2809	rbd_img_capture_header(child_img_req);
2810	up_read(&parent->header_rwsem);
2811
2812	dout("%s child_img_req %p for obj_req %p\n", __func__, child_img_req,
2813	     obj_req);
2814
2815	if (!rbd_img_is_write(img_req)) {
2816		switch (img_req->data_type) {
2817		case OBJ_REQUEST_BIO:
2818			ret = __rbd_img_fill_from_bio(child_img_req,
2819						      obj_req->img_extents,
2820						      obj_req->num_img_extents,
2821						      &obj_req->bio_pos);
2822			break;
2823		case OBJ_REQUEST_BVECS:
2824		case OBJ_REQUEST_OWN_BVECS:
2825			ret = __rbd_img_fill_from_bvecs(child_img_req,
2826						      obj_req->img_extents,
2827						      obj_req->num_img_extents,
2828						      &obj_req->bvec_pos);
2829			break;
2830		default:
2831			BUG();
2832		}
2833	} else {
2834		ret = rbd_img_fill_from_bvecs(child_img_req,
2835					      obj_req->img_extents,
2836					      obj_req->num_img_extents,
2837					      obj_req->copyup_bvecs);
2838	}
2839	if (ret) {
2840		rbd_img_request_destroy(child_img_req);
2841		return ret;
2842	}
2843
2844	/* avoid parent chain recursion */
2845	rbd_img_schedule(child_img_req, 0);
2846	return 0;
2847}
2848
2849static bool rbd_obj_advance_read(struct rbd_obj_request *obj_req, int *result)
2850{
2851	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2852	int ret;
2853
2854again:
2855	switch (obj_req->read_state) {
2856	case RBD_OBJ_READ_START:
2857		rbd_assert(!*result);
2858
2859		if (!rbd_obj_may_exist(obj_req)) {
2860			*result = -ENOENT;
2861			obj_req->read_state = RBD_OBJ_READ_OBJECT;
2862			goto again;
2863		}
2864
2865		ret = rbd_obj_read_object(obj_req);
2866		if (ret) {
2867			*result = ret;
2868			return true;
2869		}
2870		obj_req->read_state = RBD_OBJ_READ_OBJECT;
2871		return false;
2872	case RBD_OBJ_READ_OBJECT:
2873		if (*result == -ENOENT && rbd_dev->parent_overlap) {
2874			/* reverse map this object extent onto the parent */
2875			ret = rbd_obj_calc_img_extents(obj_req, false);
2876			if (ret) {
2877				*result = ret;
2878				return true;
2879			}
2880			if (obj_req->num_img_extents) {
2881				ret = rbd_obj_read_from_parent(obj_req);
2882				if (ret) {
2883					*result = ret;
2884					return true;
2885				}
2886				obj_req->read_state = RBD_OBJ_READ_PARENT;
2887				return false;
2888			}
2889		}
2890
2891		/*
2892		 * -ENOENT means a hole in the image -- zero-fill the entire
2893		 * length of the request.  A short read also implies zero-fill
2894		 * to the end of the request.
2895		 */
2896		if (*result == -ENOENT) {
2897			rbd_obj_zero_range(obj_req, 0, obj_req->ex.oe_len);
2898			*result = 0;
2899		} else if (*result >= 0) {
2900			if (*result < obj_req->ex.oe_len)
2901				rbd_obj_zero_range(obj_req, *result,
2902						obj_req->ex.oe_len - *result);
2903			else
2904				rbd_assert(*result == obj_req->ex.oe_len);
2905			*result = 0;
2906		}
2907		return true;
2908	case RBD_OBJ_READ_PARENT:
2909		/*
2910		 * The parent image is read only up to the overlap -- zero-fill
2911		 * from the overlap to the end of the request.
2912		 */
2913		if (!*result) {
2914			u32 obj_overlap = rbd_obj_img_extents_bytes(obj_req);
2915
2916			if (obj_overlap < obj_req->ex.oe_len)
2917				rbd_obj_zero_range(obj_req, obj_overlap,
2918					    obj_req->ex.oe_len - obj_overlap);
2919		}
2920		return true;
2921	default:
2922		BUG();
2923	}
2924}
2925
2926static bool rbd_obj_write_is_noop(struct rbd_obj_request *obj_req)
2927{
2928	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2929
2930	if (rbd_object_map_may_exist(rbd_dev, obj_req->ex.oe_objno))
2931		obj_req->flags |= RBD_OBJ_FLAG_MAY_EXIST;
2932
2933	if (!(obj_req->flags & RBD_OBJ_FLAG_MAY_EXIST) &&
2934	    (obj_req->flags & RBD_OBJ_FLAG_NOOP_FOR_NONEXISTENT)) {
2935		dout("%s %p noop for nonexistent\n", __func__, obj_req);
2936		return true;
2937	}
2938
2939	return false;
2940}
2941
2942/*
2943 * Return:
2944 *   0 - object map update sent
2945 *   1 - object map update isn't needed
2946 *  <0 - error
2947 */
2948static int rbd_obj_write_pre_object_map(struct rbd_obj_request *obj_req)
2949{
2950	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
2951	u8 new_state;
2952
2953	if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
2954		return 1;
2955
2956	if (obj_req->flags & RBD_OBJ_FLAG_DELETION)
2957		new_state = OBJECT_PENDING;
2958	else
2959		new_state = OBJECT_EXISTS;
2960
2961	return rbd_object_map_update(obj_req, CEPH_NOSNAP, new_state, NULL);
2962}
2963
2964static int rbd_obj_write_object(struct rbd_obj_request *obj_req)
2965{
2966	struct ceph_osd_request *osd_req;
2967	int num_ops = count_write_ops(obj_req);
2968	int which = 0;
2969	int ret;
2970
2971	if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED)
2972		num_ops++; /* stat */
2973
2974	osd_req = rbd_obj_add_osd_request(obj_req, num_ops);
2975	if (IS_ERR(osd_req))
2976		return PTR_ERR(osd_req);
2977
2978	if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
2979		ret = rbd_osd_setup_stat(osd_req, which++);
2980		if (ret)
2981			return ret;
2982	}
2983
2984	rbd_osd_setup_write_ops(osd_req, which);
2985	rbd_osd_format_write(osd_req);
2986
2987	ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
2988	if (ret)
2989		return ret;
2990
2991	rbd_osd_submit(osd_req);
2992	return 0;
2993}
2994
2995/*
2996 * copyup_bvecs pages are never highmem pages
2997 */
2998static bool is_zero_bvecs(struct bio_vec *bvecs, u32 bytes)
2999{
3000	struct ceph_bvec_iter it = {
3001		.bvecs = bvecs,
3002		.iter = { .bi_size = bytes },
3003	};
3004
3005	ceph_bvec_iter_advance_step(&it, bytes, ({
3006		if (memchr_inv(bvec_virt(&bv), 0, bv.bv_len))
3007			return false;
3008	}));
3009	return true;
3010}
3011
3012#define MODS_ONLY	U32_MAX
3013
3014static int rbd_obj_copyup_empty_snapc(struct rbd_obj_request *obj_req,
3015				      u32 bytes)
3016{
3017	struct ceph_osd_request *osd_req;
3018	int ret;
3019
3020	dout("%s obj_req %p bytes %u\n", __func__, obj_req, bytes);
3021	rbd_assert(bytes > 0 && bytes != MODS_ONLY);
3022
3023	osd_req = __rbd_obj_add_osd_request(obj_req, &rbd_empty_snapc, 1);
3024	if (IS_ERR(osd_req))
3025		return PTR_ERR(osd_req);
3026
3027	ret = rbd_osd_setup_copyup(osd_req, 0, bytes);
3028	if (ret)
3029		return ret;
3030
3031	rbd_osd_format_write(osd_req);
3032
3033	ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
3034	if (ret)
3035		return ret;
3036
3037	rbd_osd_submit(osd_req);
3038	return 0;
3039}
3040
3041static int rbd_obj_copyup_current_snapc(struct rbd_obj_request *obj_req,
3042					u32 bytes)
3043{
3044	struct ceph_osd_request *osd_req;
3045	int num_ops = count_write_ops(obj_req);
3046	int which = 0;
3047	int ret;
3048
3049	dout("%s obj_req %p bytes %u\n", __func__, obj_req, bytes);
3050
3051	if (bytes != MODS_ONLY)
3052		num_ops++; /* copyup */
3053
3054	osd_req = rbd_obj_add_osd_request(obj_req, num_ops);
3055	if (IS_ERR(osd_req))
3056		return PTR_ERR(osd_req);
3057
3058	if (bytes != MODS_ONLY) {
3059		ret = rbd_osd_setup_copyup(osd_req, which++, bytes);
3060		if (ret)
3061			return ret;
3062	}
3063
3064	rbd_osd_setup_write_ops(osd_req, which);
3065	rbd_osd_format_write(osd_req);
3066
3067	ret = ceph_osdc_alloc_messages(osd_req, GFP_NOIO);
3068	if (ret)
3069		return ret;
3070
3071	rbd_osd_submit(osd_req);
3072	return 0;
3073}
3074
3075static int setup_copyup_bvecs(struct rbd_obj_request *obj_req, u64 obj_overlap)
3076{
3077	u32 i;
3078
3079	rbd_assert(!obj_req->copyup_bvecs);
3080	obj_req->copyup_bvec_count = calc_pages_for(0, obj_overlap);
3081	obj_req->copyup_bvecs = kcalloc(obj_req->copyup_bvec_count,
3082					sizeof(*obj_req->copyup_bvecs),
3083					GFP_NOIO);
3084	if (!obj_req->copyup_bvecs)
3085		return -ENOMEM;
3086
3087	for (i = 0; i < obj_req->copyup_bvec_count; i++) {
3088		unsigned int len = min(obj_overlap, (u64)PAGE_SIZE);
3089		struct page *page = alloc_page(GFP_NOIO);
3090
3091		if (!page)
3092			return -ENOMEM;
3093
3094		bvec_set_page(&obj_req->copyup_bvecs[i], page, len, 0);
3095		obj_overlap -= len;
3096	}
3097
3098	rbd_assert(!obj_overlap);
3099	return 0;
3100}
3101
3102/*
3103 * The target object doesn't exist.  Read the data for the entire
3104 * target object up to the overlap point (if any) from the parent,
3105 * so we can use it for a copyup.
3106 */
3107static int rbd_obj_copyup_read_parent(struct rbd_obj_request *obj_req)
3108{
3109	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3110	int ret;
3111
3112	rbd_assert(obj_req->num_img_extents);
3113	prune_extents(obj_req->img_extents, &obj_req->num_img_extents,
3114		      rbd_dev->parent_overlap);
3115	if (!obj_req->num_img_extents) {
3116		/*
3117		 * The overlap has become 0 (most likely because the
3118		 * image has been flattened).  Re-submit the original write
3119		 * request -- pass MODS_ONLY since the copyup isn't needed
3120		 * anymore.
3121		 */
3122		return rbd_obj_copyup_current_snapc(obj_req, MODS_ONLY);
3123	}
3124
3125	ret = setup_copyup_bvecs(obj_req, rbd_obj_img_extents_bytes(obj_req));
3126	if (ret)
3127		return ret;
3128
3129	return rbd_obj_read_from_parent(obj_req);
3130}
3131
3132static void rbd_obj_copyup_object_maps(struct rbd_obj_request *obj_req)
3133{
3134	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3135	struct ceph_snap_context *snapc = obj_req->img_request->snapc;
3136	u8 new_state;
3137	u32 i;
3138	int ret;
3139
3140	rbd_assert(!obj_req->pending.result && !obj_req->pending.num_pending);
3141
3142	if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3143		return;
3144
3145	if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ZEROS)
3146		return;
3147
3148	for (i = 0; i < snapc->num_snaps; i++) {
3149		if ((rbd_dev->header.features & RBD_FEATURE_FAST_DIFF) &&
3150		    i + 1 < snapc->num_snaps)
3151			new_state = OBJECT_EXISTS_CLEAN;
3152		else
3153			new_state = OBJECT_EXISTS;
3154
3155		ret = rbd_object_map_update(obj_req, snapc->snaps[i],
3156					    new_state, NULL);
3157		if (ret < 0) {
3158			obj_req->pending.result = ret;
3159			return;
3160		}
3161
3162		rbd_assert(!ret);
3163		obj_req->pending.num_pending++;
3164	}
3165}
3166
3167static void rbd_obj_copyup_write_object(struct rbd_obj_request *obj_req)
3168{
3169	u32 bytes = rbd_obj_img_extents_bytes(obj_req);
3170	int ret;
3171
3172	rbd_assert(!obj_req->pending.result && !obj_req->pending.num_pending);
3173
3174	/*
3175	 * Only send non-zero copyup data to save some I/O and network
3176	 * bandwidth -- zero copyup data is equivalent to the object not
3177	 * existing.
3178	 */
3179	if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ZEROS)
3180		bytes = 0;
3181
3182	if (obj_req->img_request->snapc->num_snaps && bytes > 0) {
3183		/*
3184		 * Send a copyup request with an empty snapshot context to
3185		 * deep-copyup the object through all existing snapshots.
3186		 * A second request with the current snapshot context will be
3187		 * sent for the actual modification.
3188		 */
3189		ret = rbd_obj_copyup_empty_snapc(obj_req, bytes);
3190		if (ret) {
3191			obj_req->pending.result = ret;
3192			return;
3193		}
3194
3195		obj_req->pending.num_pending++;
3196		bytes = MODS_ONLY;
3197	}
3198
3199	ret = rbd_obj_copyup_current_snapc(obj_req, bytes);
3200	if (ret) {
3201		obj_req->pending.result = ret;
3202		return;
3203	}
3204
3205	obj_req->pending.num_pending++;
3206}
3207
3208static bool rbd_obj_advance_copyup(struct rbd_obj_request *obj_req, int *result)
3209{
3210	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3211	int ret;
3212
3213again:
3214	switch (obj_req->copyup_state) {
3215	case RBD_OBJ_COPYUP_START:
3216		rbd_assert(!*result);
3217
3218		ret = rbd_obj_copyup_read_parent(obj_req);
3219		if (ret) {
3220			*result = ret;
3221			return true;
3222		}
3223		if (obj_req->num_img_extents)
3224			obj_req->copyup_state = RBD_OBJ_COPYUP_READ_PARENT;
3225		else
3226			obj_req->copyup_state = RBD_OBJ_COPYUP_WRITE_OBJECT;
3227		return false;
3228	case RBD_OBJ_COPYUP_READ_PARENT:
3229		if (*result)
3230			return true;
3231
3232		if (is_zero_bvecs(obj_req->copyup_bvecs,
3233				  rbd_obj_img_extents_bytes(obj_req))) {
3234			dout("%s %p detected zeros\n", __func__, obj_req);
3235			obj_req->flags |= RBD_OBJ_FLAG_COPYUP_ZEROS;
3236		}
3237
3238		rbd_obj_copyup_object_maps(obj_req);
3239		if (!obj_req->pending.num_pending) {
3240			*result = obj_req->pending.result;
3241			obj_req->copyup_state = RBD_OBJ_COPYUP_OBJECT_MAPS;
3242			goto again;
3243		}
3244		obj_req->copyup_state = __RBD_OBJ_COPYUP_OBJECT_MAPS;
3245		return false;
3246	case __RBD_OBJ_COPYUP_OBJECT_MAPS:
3247		if (!pending_result_dec(&obj_req->pending, result))
3248			return false;
3249		fallthrough;
3250	case RBD_OBJ_COPYUP_OBJECT_MAPS:
3251		if (*result) {
3252			rbd_warn(rbd_dev, "snap object map update failed: %d",
3253				 *result);
3254			return true;
3255		}
3256
3257		rbd_obj_copyup_write_object(obj_req);
3258		if (!obj_req->pending.num_pending) {
3259			*result = obj_req->pending.result;
3260			obj_req->copyup_state = RBD_OBJ_COPYUP_WRITE_OBJECT;
3261			goto again;
3262		}
3263		obj_req->copyup_state = __RBD_OBJ_COPYUP_WRITE_OBJECT;
3264		return false;
3265	case __RBD_OBJ_COPYUP_WRITE_OBJECT:
3266		if (!pending_result_dec(&obj_req->pending, result))
3267			return false;
3268		fallthrough;
3269	case RBD_OBJ_COPYUP_WRITE_OBJECT:
3270		return true;
3271	default:
3272		BUG();
3273	}
3274}
3275
3276/*
3277 * Return:
3278 *   0 - object map update sent
3279 *   1 - object map update isn't needed
3280 *  <0 - error
3281 */
3282static int rbd_obj_write_post_object_map(struct rbd_obj_request *obj_req)
3283{
3284	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3285	u8 current_state = OBJECT_PENDING;
3286
3287	if (!(rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3288		return 1;
3289
3290	if (!(obj_req->flags & RBD_OBJ_FLAG_DELETION))
3291		return 1;
3292
3293	return rbd_object_map_update(obj_req, CEPH_NOSNAP, OBJECT_NONEXISTENT,
3294				     &current_state);
3295}
3296
3297static bool rbd_obj_advance_write(struct rbd_obj_request *obj_req, int *result)
3298{
3299	struct rbd_device *rbd_dev = obj_req->img_request->rbd_dev;
3300	int ret;
3301
3302again:
3303	switch (obj_req->write_state) {
3304	case RBD_OBJ_WRITE_START:
3305		rbd_assert(!*result);
3306
3307		rbd_obj_set_copyup_enabled(obj_req);
3308		if (rbd_obj_write_is_noop(obj_req))
3309			return true;
3310
3311		ret = rbd_obj_write_pre_object_map(obj_req);
3312		if (ret < 0) {
3313			*result = ret;
3314			return true;
3315		}
3316		obj_req->write_state = RBD_OBJ_WRITE_PRE_OBJECT_MAP;
3317		if (ret > 0)
3318			goto again;
3319		return false;
3320	case RBD_OBJ_WRITE_PRE_OBJECT_MAP:
3321		if (*result) {
3322			rbd_warn(rbd_dev, "pre object map update failed: %d",
3323				 *result);
3324			return true;
3325		}
3326		ret = rbd_obj_write_object(obj_req);
3327		if (ret) {
3328			*result = ret;
3329			return true;
3330		}
3331		obj_req->write_state = RBD_OBJ_WRITE_OBJECT;
3332		return false;
3333	case RBD_OBJ_WRITE_OBJECT:
3334		if (*result == -ENOENT) {
3335			if (obj_req->flags & RBD_OBJ_FLAG_COPYUP_ENABLED) {
3336				*result = 0;
3337				obj_req->copyup_state = RBD_OBJ_COPYUP_START;
3338				obj_req->write_state = __RBD_OBJ_WRITE_COPYUP;
3339				goto again;
3340			}
3341			/*
3342			 * On a non-existent object:
3343			 *   delete - -ENOENT, truncate/zero - 0
3344			 */
3345			if (obj_req->flags & RBD_OBJ_FLAG_DELETION)
3346				*result = 0;
3347		}
3348		if (*result)
3349			return true;
3350
3351		obj_req->write_state = RBD_OBJ_WRITE_COPYUP;
3352		goto again;
3353	case __RBD_OBJ_WRITE_COPYUP:
3354		if (!rbd_obj_advance_copyup(obj_req, result))
3355			return false;
3356		fallthrough;
3357	case RBD_OBJ_WRITE_COPYUP:
3358		if (*result) {
3359			rbd_warn(rbd_dev, "copyup failed: %d", *result);
3360			return true;
3361		}
3362		ret = rbd_obj_write_post_object_map(obj_req);
3363		if (ret < 0) {
3364			*result = ret;
3365			return true;
3366		}
3367		obj_req->write_state = RBD_OBJ_WRITE_POST_OBJECT_MAP;
3368		if (ret > 0)
3369			goto again;
3370		return false;
3371	case RBD_OBJ_WRITE_POST_OBJECT_MAP:
3372		if (*result)
3373			rbd_warn(rbd_dev, "post object map update failed: %d",
3374				 *result);
3375		return true;
3376	default:
3377		BUG();
3378	}
3379}
3380
3381/*
3382 * Return true if @obj_req is completed.
3383 */
3384static bool __rbd_obj_handle_request(struct rbd_obj_request *obj_req,
3385				     int *result)
3386{
3387	struct rbd_img_request *img_req = obj_req->img_request;
3388	struct rbd_device *rbd_dev = img_req->rbd_dev;
3389	bool done;
3390
3391	mutex_lock(&obj_req->state_mutex);
3392	if (!rbd_img_is_write(img_req))
3393		done = rbd_obj_advance_read(obj_req, result);
3394	else
3395		done = rbd_obj_advance_write(obj_req, result);
3396	mutex_unlock(&obj_req->state_mutex);
3397
3398	if (done && *result) {
3399		rbd_assert(*result < 0);
3400		rbd_warn(rbd_dev, "%s at objno %llu %llu~%llu result %d",
3401			 obj_op_name(img_req->op_type), obj_req->ex.oe_objno,
3402			 obj_req->ex.oe_off, obj_req->ex.oe_len, *result);
3403	}
3404	return done;
3405}
3406
3407/*
3408 * This is open-coded in rbd_img_handle_request() to avoid parent chain
3409 * recursion.
3410 */
3411static void rbd_obj_handle_request(struct rbd_obj_request *obj_req, int result)
3412{
3413	if (__rbd_obj_handle_request(obj_req, &result))
3414		rbd_img_handle_request(obj_req->img_request, result);
3415}
3416
3417static bool need_exclusive_lock(struct rbd_img_request *img_req)
3418{
3419	struct rbd_device *rbd_dev = img_req->rbd_dev;
3420
3421	if (!(rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK))
3422		return false;
3423
3424	if (rbd_is_ro(rbd_dev))
3425		return false;
3426
3427	rbd_assert(!test_bit(IMG_REQ_CHILD, &img_req->flags));
3428	if (rbd_dev->opts->lock_on_read ||
3429	    (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP))
3430		return true;
3431
3432	return rbd_img_is_write(img_req);
3433}
3434
3435static bool rbd_lock_add_request(struct rbd_img_request *img_req)
3436{
3437	struct rbd_device *rbd_dev = img_req->rbd_dev;
3438	bool locked;
3439
3440	lockdep_assert_held(&rbd_dev->lock_rwsem);
3441	locked = rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED;
3442	spin_lock(&rbd_dev->lock_lists_lock);
3443	rbd_assert(list_empty(&img_req->lock_item));
3444	if (!locked)
3445		list_add_tail(&img_req->lock_item, &rbd_dev->acquiring_list);
3446	else
3447		list_add_tail(&img_req->lock_item, &rbd_dev->running_list);
3448	spin_unlock(&rbd_dev->lock_lists_lock);
3449	return locked;
3450}
3451
3452static void rbd_lock_del_request(struct rbd_img_request *img_req)
3453{
3454	struct rbd_device *rbd_dev = img_req->rbd_dev;
3455	bool need_wakeup = false;
3456
3457	lockdep_assert_held(&rbd_dev->lock_rwsem);
3458	spin_lock(&rbd_dev->lock_lists_lock);
3459	if (!list_empty(&img_req->lock_item)) {
3460		rbd_assert(!list_empty(&rbd_dev->running_list));
3461		list_del_init(&img_req->lock_item);
3462		need_wakeup = (rbd_dev->lock_state == RBD_LOCK_STATE_QUIESCING &&
3463			       list_empty(&rbd_dev->running_list));
3464	}
3465	spin_unlock(&rbd_dev->lock_lists_lock);
3466	if (need_wakeup)
3467		complete(&rbd_dev->quiescing_wait);
3468}
3469
3470static int rbd_img_exclusive_lock(struct rbd_img_request *img_req)
3471{
3472	struct rbd_device *rbd_dev = img_req->rbd_dev;
3473
3474	if (!need_exclusive_lock(img_req))
3475		return 1;
3476
3477	if (rbd_lock_add_request(img_req))
3478		return 1;
3479
3480	/*
3481	 * Note the use of mod_delayed_work() in rbd_acquire_lock()
3482	 * and cancel_delayed_work() in wake_lock_waiters().
3483	 */
3484	dout("%s rbd_dev %p queueing lock_dwork\n", __func__, rbd_dev);
3485	queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
3486	return 0;
3487}
3488
3489static void rbd_img_object_requests(struct rbd_img_request *img_req)
3490{
3491	struct rbd_device *rbd_dev = img_req->rbd_dev;
3492	struct rbd_obj_request *obj_req;
3493
3494	rbd_assert(!img_req->pending.result && !img_req->pending.num_pending);
3495	rbd_assert(!need_exclusive_lock(img_req) ||
3496		   __rbd_is_lock_owner(rbd_dev));
3497
3498	if (test_bit(IMG_REQ_CHILD, &img_req->flags)) {
3499		rbd_assert(!rbd_img_is_write(img_req));
3500	} else {
3501		struct request *rq = blk_mq_rq_from_pdu(img_req);
3502		u64 off = (u64)blk_rq_pos(rq) << SECTOR_SHIFT;
3503		u64 len = blk_rq_bytes(rq);
3504		u64 mapping_size;
3505
3506		down_read(&rbd_dev->header_rwsem);
3507		mapping_size = rbd_dev->mapping.size;
3508		if (rbd_img_is_write(img_req)) {
3509			rbd_assert(!img_req->snapc);
3510			img_req->snapc =
3511			    ceph_get_snap_context(rbd_dev->header.snapc);
3512		}
3513		up_read(&rbd_dev->header_rwsem);
3514
3515		if (unlikely(off + len > mapping_size)) {
3516			rbd_warn(rbd_dev, "beyond EOD (%llu~%llu > %llu)",
3517				 off, len, mapping_size);
3518			img_req->pending.result = -EIO;
3519			return;
3520		}
3521	}
3522
3523	for_each_obj_request(img_req, obj_req) {
3524		int result = 0;
3525
3526		if (__rbd_obj_handle_request(obj_req, &result)) {
3527			if (result) {
3528				img_req->pending.result = result;
3529				return;
3530			}
3531		} else {
3532			img_req->pending.num_pending++;
3533		}
3534	}
3535}
3536
3537static bool rbd_img_advance(struct rbd_img_request *img_req, int *result)
3538{
3539	int ret;
3540
3541again:
3542	switch (img_req->state) {
3543	case RBD_IMG_START:
3544		rbd_assert(!*result);
3545
3546		ret = rbd_img_exclusive_lock(img_req);
3547		if (ret < 0) {
3548			*result = ret;
3549			return true;
3550		}
3551		img_req->state = RBD_IMG_EXCLUSIVE_LOCK;
3552		if (ret > 0)
3553			goto again;
3554		return false;
3555	case RBD_IMG_EXCLUSIVE_LOCK:
3556		if (*result)
3557			return true;
3558
3559		rbd_img_object_requests(img_req);
3560		if (!img_req->pending.num_pending) {
3561			*result = img_req->pending.result;
3562			img_req->state = RBD_IMG_OBJECT_REQUESTS;
3563			goto again;
3564		}
3565		img_req->state = __RBD_IMG_OBJECT_REQUESTS;
3566		return false;
3567	case __RBD_IMG_OBJECT_REQUESTS:
3568		if (!pending_result_dec(&img_req->pending, result))
3569			return false;
3570		fallthrough;
3571	case RBD_IMG_OBJECT_REQUESTS:
3572		return true;
3573	default:
3574		BUG();
3575	}
3576}
3577
3578/*
3579 * Return true if @img_req is completed.
3580 */
3581static bool __rbd_img_handle_request(struct rbd_img_request *img_req,
3582				     int *result)
3583{
3584	struct rbd_device *rbd_dev = img_req->rbd_dev;
3585	bool done;
3586
3587	if (need_exclusive_lock(img_req)) {
3588		down_read(&rbd_dev->lock_rwsem);
3589		mutex_lock(&img_req->state_mutex);
3590		done = rbd_img_advance(img_req, result);
3591		if (done)
3592			rbd_lock_del_request(img_req);
3593		mutex_unlock(&img_req->state_mutex);
3594		up_read(&rbd_dev->lock_rwsem);
3595	} else {
3596		mutex_lock(&img_req->state_mutex);
3597		done = rbd_img_advance(img_req, result);
3598		mutex_unlock(&img_req->state_mutex);
3599	}
3600
3601	if (done && *result) {
3602		rbd_assert(*result < 0);
3603		rbd_warn(rbd_dev, "%s%s result %d",
3604		      test_bit(IMG_REQ_CHILD, &img_req->flags) ? "child " : "",
3605		      obj_op_name(img_req->op_type), *result);
3606	}
3607	return done;
3608}
3609
3610static void rbd_img_handle_request(struct rbd_img_request *img_req, int result)
3611{
3612again:
3613	if (!__rbd_img_handle_request(img_req, &result))
3614		return;
3615
3616	if (test_bit(IMG_REQ_CHILD, &img_req->flags)) {
3617		struct rbd_obj_request *obj_req = img_req->obj_request;
3618
3619		rbd_img_request_destroy(img_req);
3620		if (__rbd_obj_handle_request(obj_req, &result)) {
3621			img_req = obj_req->img_request;
3622			goto again;
3623		}
3624	} else {
3625		struct request *rq = blk_mq_rq_from_pdu(img_req);
3626
3627		rbd_img_request_destroy(img_req);
3628		blk_mq_end_request(rq, errno_to_blk_status(result));
3629	}
3630}
3631
3632static const struct rbd_client_id rbd_empty_cid;
3633
3634static bool rbd_cid_equal(const struct rbd_client_id *lhs,
3635			  const struct rbd_client_id *rhs)
3636{
3637	return lhs->gid == rhs->gid && lhs->handle == rhs->handle;
3638}
3639
3640static struct rbd_client_id rbd_get_cid(struct rbd_device *rbd_dev)
3641{
3642	struct rbd_client_id cid;
3643
3644	mutex_lock(&rbd_dev->watch_mutex);
3645	cid.gid = ceph_client_gid(rbd_dev->rbd_client->client);
3646	cid.handle = rbd_dev->watch_cookie;
3647	mutex_unlock(&rbd_dev->watch_mutex);
3648	return cid;
3649}
3650
3651/*
3652 * lock_rwsem must be held for write
3653 */
3654static void rbd_set_owner_cid(struct rbd_device *rbd_dev,
3655			      const struct rbd_client_id *cid)
3656{
3657	dout("%s rbd_dev %p %llu-%llu -> %llu-%llu\n", __func__, rbd_dev,
3658	     rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle,
3659	     cid->gid, cid->handle);
3660	rbd_dev->owner_cid = *cid; /* struct */
3661}
3662
3663static void format_lock_cookie(struct rbd_device *rbd_dev, char *buf)
3664{
3665	mutex_lock(&rbd_dev->watch_mutex);
3666	sprintf(buf, "%s %llu", RBD_LOCK_COOKIE_PREFIX, rbd_dev->watch_cookie);
3667	mutex_unlock(&rbd_dev->watch_mutex);
3668}
3669
3670static void __rbd_lock(struct rbd_device *rbd_dev, const char *cookie)
3671{
3672	struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3673
3674	rbd_dev->lock_state = RBD_LOCK_STATE_LOCKED;
3675	strcpy(rbd_dev->lock_cookie, cookie);
3676	rbd_set_owner_cid(rbd_dev, &cid);
3677	queue_work(rbd_dev->task_wq, &rbd_dev->acquired_lock_work);
3678}
3679
3680/*
3681 * lock_rwsem must be held for write
3682 */
3683static int rbd_lock(struct rbd_device *rbd_dev)
3684{
3685	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3686	char cookie[32];
3687	int ret;
3688
3689	WARN_ON(__rbd_is_lock_owner(rbd_dev) ||
3690		rbd_dev->lock_cookie[0] != '\0');
3691
3692	format_lock_cookie(rbd_dev, cookie);
3693	ret = ceph_cls_lock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
3694			    RBD_LOCK_NAME, CEPH_CLS_LOCK_EXCLUSIVE, cookie,
3695			    RBD_LOCK_TAG, "", 0);
3696	if (ret && ret != -EEXIST)
3697		return ret;
3698
3699	__rbd_lock(rbd_dev, cookie);
3700	return 0;
3701}
3702
3703/*
3704 * lock_rwsem must be held for write
3705 */
3706static void rbd_unlock(struct rbd_device *rbd_dev)
3707{
3708	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3709	int ret;
3710
3711	WARN_ON(!__rbd_is_lock_owner(rbd_dev) ||
3712		rbd_dev->lock_cookie[0] == '\0');
3713
3714	ret = ceph_cls_unlock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
3715			      RBD_LOCK_NAME, rbd_dev->lock_cookie);
3716	if (ret && ret != -ENOENT)
3717		rbd_warn(rbd_dev, "failed to unlock header: %d", ret);
3718
3719	/* treat errors as the image is unlocked */
3720	rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
3721	rbd_dev->lock_cookie[0] = '\0';
3722	rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
3723	queue_work(rbd_dev->task_wq, &rbd_dev->released_lock_work);
3724}
3725
3726static int __rbd_notify_op_lock(struct rbd_device *rbd_dev,
3727				enum rbd_notify_op notify_op,
3728				struct page ***preply_pages,
3729				size_t *preply_len)
3730{
3731	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3732	struct rbd_client_id cid = rbd_get_cid(rbd_dev);
3733	char buf[4 + 8 + 8 + CEPH_ENCODING_START_BLK_LEN];
3734	int buf_size = sizeof(buf);
3735	void *p = buf;
3736
3737	dout("%s rbd_dev %p notify_op %d\n", __func__, rbd_dev, notify_op);
3738
3739	/* encode *LockPayload NotifyMessage (op + ClientId) */
3740	ceph_start_encoding(&p, 2, 1, buf_size - CEPH_ENCODING_START_BLK_LEN);
3741	ceph_encode_32(&p, notify_op);
3742	ceph_encode_64(&p, cid.gid);
3743	ceph_encode_64(&p, cid.handle);
3744
3745	return ceph_osdc_notify(osdc, &rbd_dev->header_oid,
3746				&rbd_dev->header_oloc, buf, buf_size,
3747				RBD_NOTIFY_TIMEOUT, preply_pages, preply_len);
3748}
3749
3750static void rbd_notify_op_lock(struct rbd_device *rbd_dev,
3751			       enum rbd_notify_op notify_op)
3752{
3753	__rbd_notify_op_lock(rbd_dev, notify_op, NULL, NULL);
3754}
3755
3756static void rbd_notify_acquired_lock(struct work_struct *work)
3757{
3758	struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3759						  acquired_lock_work);
3760
3761	rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_ACQUIRED_LOCK);
3762}
3763
3764static void rbd_notify_released_lock(struct work_struct *work)
3765{
3766	struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
3767						  released_lock_work);
3768
3769	rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_RELEASED_LOCK);
3770}
3771
3772static int rbd_request_lock(struct rbd_device *rbd_dev)
3773{
3774	struct page **reply_pages;
3775	size_t reply_len;
3776	bool lock_owner_responded = false;
3777	int ret;
3778
3779	dout("%s rbd_dev %p\n", __func__, rbd_dev);
3780
3781	ret = __rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_REQUEST_LOCK,
3782				   &reply_pages, &reply_len);
3783	if (ret && ret != -ETIMEDOUT) {
3784		rbd_warn(rbd_dev, "failed to request lock: %d", ret);
3785		goto out;
3786	}
3787
3788	if (reply_len > 0 && reply_len <= PAGE_SIZE) {
3789		void *p = page_address(reply_pages[0]);
3790		void *const end = p + reply_len;
3791		u32 n;
3792
3793		ceph_decode_32_safe(&p, end, n, e_inval); /* num_acks */
3794		while (n--) {
3795			u8 struct_v;
3796			u32 len;
3797
3798			ceph_decode_need(&p, end, 8 + 8, e_inval);
3799			p += 8 + 8; /* skip gid and cookie */
3800
3801			ceph_decode_32_safe(&p, end, len, e_inval);
3802			if (!len)
3803				continue;
3804
3805			if (lock_owner_responded) {
3806				rbd_warn(rbd_dev,
3807					 "duplicate lock owners detected");
3808				ret = -EIO;
3809				goto out;
3810			}
3811
3812			lock_owner_responded = true;
3813			ret = ceph_start_decoding(&p, end, 1, "ResponseMessage",
3814						  &struct_v, &len);
3815			if (ret) {
3816				rbd_warn(rbd_dev,
3817					 "failed to decode ResponseMessage: %d",
3818					 ret);
3819				goto e_inval;
3820			}
3821
3822			ret = ceph_decode_32(&p);
3823		}
3824	}
3825
3826	if (!lock_owner_responded) {
3827		rbd_warn(rbd_dev, "no lock owners detected");
3828		ret = -ETIMEDOUT;
3829	}
3830
3831out:
3832	ceph_release_page_vector(reply_pages, calc_pages_for(0, reply_len));
3833	return ret;
3834
3835e_inval:
3836	ret = -EINVAL;
3837	goto out;
3838}
3839
3840/*
3841 * Either image request state machine(s) or rbd_add_acquire_lock()
3842 * (i.e. "rbd map").
3843 */
3844static void wake_lock_waiters(struct rbd_device *rbd_dev, int result)
3845{
3846	struct rbd_img_request *img_req;
3847
3848	dout("%s rbd_dev %p result %d\n", __func__, rbd_dev, result);
3849	lockdep_assert_held_write(&rbd_dev->lock_rwsem);
3850
3851	cancel_delayed_work(&rbd_dev->lock_dwork);
3852	if (!completion_done(&rbd_dev->acquire_wait)) {
3853		rbd_assert(list_empty(&rbd_dev->acquiring_list) &&
3854			   list_empty(&rbd_dev->running_list));
3855		rbd_dev->acquire_err = result;
3856		complete_all(&rbd_dev->acquire_wait);
3857		return;
3858	}
3859
3860	while (!list_empty(&rbd_dev->acquiring_list)) {
3861		img_req = list_first_entry(&rbd_dev->acquiring_list,
3862					   struct rbd_img_request, lock_item);
3863		mutex_lock(&img_req->state_mutex);
3864		rbd_assert(img_req->state == RBD_IMG_EXCLUSIVE_LOCK);
3865		if (!result)
3866			list_move_tail(&img_req->lock_item,
3867				       &rbd_dev->running_list);
3868		else
3869			list_del_init(&img_req->lock_item);
3870		rbd_img_schedule(img_req, result);
3871		mutex_unlock(&img_req->state_mutex);
3872	}
3873}
3874
3875static bool locker_equal(const struct ceph_locker *lhs,
3876			 const struct ceph_locker *rhs)
3877{
3878	return lhs->id.name.type == rhs->id.name.type &&
3879	       lhs->id.name.num == rhs->id.name.num &&
3880	       !strcmp(lhs->id.cookie, rhs->id.cookie) &&
3881	       ceph_addr_equal_no_type(&lhs->info.addr, &rhs->info.addr);
3882}
3883
3884static void free_locker(struct ceph_locker *locker)
3885{
3886	if (locker)
3887		ceph_free_lockers(locker, 1);
3888}
3889
3890static struct ceph_locker *get_lock_owner_info(struct rbd_device *rbd_dev)
3891{
3892	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3893	struct ceph_locker *lockers;
3894	u32 num_lockers;
3895	u8 lock_type;
3896	char *lock_tag;
3897	u64 handle;
3898	int ret;
3899
3900	ret = ceph_cls_lock_info(osdc, &rbd_dev->header_oid,
3901				 &rbd_dev->header_oloc, RBD_LOCK_NAME,
3902				 &lock_type, &lock_tag, &lockers, &num_lockers);
3903	if (ret) {
3904		rbd_warn(rbd_dev, "failed to get header lockers: %d", ret);
3905		return ERR_PTR(ret);
3906	}
3907
3908	if (num_lockers == 0) {
3909		dout("%s rbd_dev %p no lockers detected\n", __func__, rbd_dev);
3910		lockers = NULL;
3911		goto out;
3912	}
3913
3914	if (strcmp(lock_tag, RBD_LOCK_TAG)) {
3915		rbd_warn(rbd_dev, "locked by external mechanism, tag %s",
3916			 lock_tag);
3917		goto err_busy;
3918	}
3919
3920	if (lock_type != CEPH_CLS_LOCK_EXCLUSIVE) {
3921		rbd_warn(rbd_dev, "incompatible lock type detected");
3922		goto err_busy;
3923	}
3924
3925	WARN_ON(num_lockers != 1);
3926	ret = sscanf(lockers[0].id.cookie, RBD_LOCK_COOKIE_PREFIX " %llu",
3927		     &handle);
3928	if (ret != 1) {
3929		rbd_warn(rbd_dev, "locked by external mechanism, cookie %s",
3930			 lockers[0].id.cookie);
3931		goto err_busy;
3932	}
3933	if (ceph_addr_is_blank(&lockers[0].info.addr)) {
3934		rbd_warn(rbd_dev, "locker has a blank address");
3935		goto err_busy;
3936	}
3937
3938	dout("%s rbd_dev %p got locker %s%llu@%pISpc/%u handle %llu\n",
3939	     __func__, rbd_dev, ENTITY_NAME(lockers[0].id.name),
3940	     &lockers[0].info.addr.in_addr,
3941	     le32_to_cpu(lockers[0].info.addr.nonce), handle);
3942
3943out:
3944	kfree(lock_tag);
3945	return lockers;
3946
3947err_busy:
3948	kfree(lock_tag);
3949	ceph_free_lockers(lockers, num_lockers);
3950	return ERR_PTR(-EBUSY);
3951}
3952
3953static int find_watcher(struct rbd_device *rbd_dev,
3954			const struct ceph_locker *locker)
3955{
3956	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
3957	struct ceph_watch_item *watchers;
3958	u32 num_watchers;
3959	u64 cookie;
3960	int i;
3961	int ret;
3962
3963	ret = ceph_osdc_list_watchers(osdc, &rbd_dev->header_oid,
3964				      &rbd_dev->header_oloc, &watchers,
3965				      &num_watchers);
3966	if (ret) {
3967		rbd_warn(rbd_dev, "failed to get watchers: %d", ret);
3968		return ret;
3969	}
3970
3971	sscanf(locker->id.cookie, RBD_LOCK_COOKIE_PREFIX " %llu", &cookie);
3972	for (i = 0; i < num_watchers; i++) {
3973		/*
3974		 * Ignore addr->type while comparing.  This mimics
3975		 * entity_addr_t::get_legacy_str() + strcmp().
3976		 */
3977		if (ceph_addr_equal_no_type(&watchers[i].addr,
3978					    &locker->info.addr) &&
3979		    watchers[i].cookie == cookie) {
3980			struct rbd_client_id cid = {
3981				.gid = le64_to_cpu(watchers[i].name.num),
3982				.handle = cookie,
3983			};
3984
3985			dout("%s rbd_dev %p found cid %llu-%llu\n", __func__,
3986			     rbd_dev, cid.gid, cid.handle);
3987			rbd_set_owner_cid(rbd_dev, &cid);
3988			ret = 1;
3989			goto out;
3990		}
3991	}
3992
3993	dout("%s rbd_dev %p no watchers\n", __func__, rbd_dev);
3994	ret = 0;
3995out:
3996	kfree(watchers);
3997	return ret;
3998}
3999
4000/*
4001 * lock_rwsem must be held for write
4002 */
4003static int rbd_try_lock(struct rbd_device *rbd_dev)
4004{
4005	struct ceph_client *client = rbd_dev->rbd_client->client;
4006	struct ceph_locker *locker, *refreshed_locker;
4007	int ret;
4008
4009	for (;;) {
4010		locker = refreshed_locker = NULL;
4011
4012		ret = rbd_lock(rbd_dev);
4013		if (!ret)
4014			goto out;
4015		if (ret != -EBUSY) {
4016			rbd_warn(rbd_dev, "failed to lock header: %d", ret);
4017			goto out;
4018		}
4019
4020		/* determine if the current lock holder is still alive */
4021		locker = get_lock_owner_info(rbd_dev);
4022		if (IS_ERR(locker)) {
4023			ret = PTR_ERR(locker);
4024			locker = NULL;
4025			goto out;
4026		}
4027		if (!locker)
4028			goto again;
4029
4030		ret = find_watcher(rbd_dev, locker);
4031		if (ret)
4032			goto out; /* request lock or error */
4033
4034		refreshed_locker = get_lock_owner_info(rbd_dev);
4035		if (IS_ERR(refreshed_locker)) {
4036			ret = PTR_ERR(refreshed_locker);
4037			refreshed_locker = NULL;
4038			goto out;
4039		}
4040		if (!refreshed_locker ||
4041		    !locker_equal(locker, refreshed_locker))
4042			goto again;
4043
4044		rbd_warn(rbd_dev, "breaking header lock owned by %s%llu",
4045			 ENTITY_NAME(locker->id.name));
4046
4047		ret = ceph_monc_blocklist_add(&client->monc,
4048					      &locker->info.addr);
4049		if (ret) {
4050			rbd_warn(rbd_dev, "failed to blocklist %s%llu: %d",
4051				 ENTITY_NAME(locker->id.name), ret);
4052			goto out;
4053		}
4054
4055		ret = ceph_cls_break_lock(&client->osdc, &rbd_dev->header_oid,
4056					  &rbd_dev->header_oloc, RBD_LOCK_NAME,
4057					  locker->id.cookie, &locker->id.name);
4058		if (ret && ret != -ENOENT) {
4059			rbd_warn(rbd_dev, "failed to break header lock: %d",
4060				 ret);
4061			goto out;
4062		}
4063
4064again:
4065		free_locker(refreshed_locker);
4066		free_locker(locker);
4067	}
4068
4069out:
4070	free_locker(refreshed_locker);
4071	free_locker(locker);
4072	return ret;
4073}
4074
4075static int rbd_post_acquire_action(struct rbd_device *rbd_dev)
4076{
4077	int ret;
4078
4079	ret = rbd_dev_refresh(rbd_dev);
4080	if (ret)
4081		return ret;
4082
4083	if (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP) {
4084		ret = rbd_object_map_open(rbd_dev);
4085		if (ret)
4086			return ret;
4087	}
4088
4089	return 0;
4090}
4091
4092/*
4093 * Return:
4094 *   0 - lock acquired
4095 *   1 - caller should call rbd_request_lock()
4096 *  <0 - error
4097 */
4098static int rbd_try_acquire_lock(struct rbd_device *rbd_dev)
4099{
4100	int ret;
4101
4102	down_read(&rbd_dev->lock_rwsem);
4103	dout("%s rbd_dev %p read lock_state %d\n", __func__, rbd_dev,
4104	     rbd_dev->lock_state);
4105	if (__rbd_is_lock_owner(rbd_dev)) {
4106		up_read(&rbd_dev->lock_rwsem);
4107		return 0;
4108	}
4109
4110	up_read(&rbd_dev->lock_rwsem);
4111	down_write(&rbd_dev->lock_rwsem);
4112	dout("%s rbd_dev %p write lock_state %d\n", __func__, rbd_dev,
4113	     rbd_dev->lock_state);
4114	if (__rbd_is_lock_owner(rbd_dev)) {
4115		up_write(&rbd_dev->lock_rwsem);
4116		return 0;
4117	}
4118
4119	ret = rbd_try_lock(rbd_dev);
4120	if (ret < 0) {
4121		rbd_warn(rbd_dev, "failed to acquire lock: %d", ret);
4122		goto out;
4123	}
4124	if (ret > 0) {
4125		up_write(&rbd_dev->lock_rwsem);
4126		return ret;
4127	}
4128
4129	rbd_assert(rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED);
4130	rbd_assert(list_empty(&rbd_dev->running_list));
4131
4132	ret = rbd_post_acquire_action(rbd_dev);
4133	if (ret) {
4134		rbd_warn(rbd_dev, "post-acquire action failed: %d", ret);
4135		/*
4136		 * Can't stay in RBD_LOCK_STATE_LOCKED because
4137		 * rbd_lock_add_request() would let the request through,
4138		 * assuming that e.g. object map is locked and loaded.
4139		 */
4140		rbd_unlock(rbd_dev);
4141	}
4142
4143out:
4144	wake_lock_waiters(rbd_dev, ret);
4145	up_write(&rbd_dev->lock_rwsem);
4146	return ret;
4147}
4148
4149static void rbd_acquire_lock(struct work_struct *work)
4150{
4151	struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
4152					    struct rbd_device, lock_dwork);
4153	int ret;
4154
4155	dout("%s rbd_dev %p\n", __func__, rbd_dev);
4156again:
4157	ret = rbd_try_acquire_lock(rbd_dev);
4158	if (ret <= 0) {
4159		dout("%s rbd_dev %p ret %d - done\n", __func__, rbd_dev, ret);
4160		return;
4161	}
4162
4163	ret = rbd_request_lock(rbd_dev);
4164	if (ret == -ETIMEDOUT) {
4165		goto again; /* treat this as a dead client */
4166	} else if (ret == -EROFS) {
4167		rbd_warn(rbd_dev, "peer will not release lock");
4168		down_write(&rbd_dev->lock_rwsem);
4169		wake_lock_waiters(rbd_dev, ret);
4170		up_write(&rbd_dev->lock_rwsem);
4171	} else if (ret < 0) {
4172		rbd_warn(rbd_dev, "error requesting lock: %d", ret);
4173		mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork,
4174				 RBD_RETRY_DELAY);
4175	} else {
4176		/*
4177		 * lock owner acked, but resend if we don't see them
4178		 * release the lock
4179		 */
4180		dout("%s rbd_dev %p requeuing lock_dwork\n", __func__,
4181		     rbd_dev);
4182		mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork,
4183		    msecs_to_jiffies(2 * RBD_NOTIFY_TIMEOUT * MSEC_PER_SEC));
4184	}
4185}
4186
4187static bool rbd_quiesce_lock(struct rbd_device *rbd_dev)
4188{
4189	dout("%s rbd_dev %p\n", __func__, rbd_dev);
4190	lockdep_assert_held_write(&rbd_dev->lock_rwsem);
4191
4192	if (rbd_dev->lock_state != RBD_LOCK_STATE_LOCKED)
4193		return false;
4194
4195	/*
4196	 * Ensure that all in-flight IO is flushed.
4197	 */
4198	rbd_dev->lock_state = RBD_LOCK_STATE_QUIESCING;
4199	rbd_assert(!completion_done(&rbd_dev->quiescing_wait));
4200	if (list_empty(&rbd_dev->running_list))
4201		return true;
4202
4203	up_write(&rbd_dev->lock_rwsem);
4204	wait_for_completion(&rbd_dev->quiescing_wait);
4205
4206	down_write(&rbd_dev->lock_rwsem);
4207	if (rbd_dev->lock_state != RBD_LOCK_STATE_QUIESCING)
4208		return false;
4209
4210	rbd_assert(list_empty(&rbd_dev->running_list));
4211	return true;
4212}
4213
4214static void rbd_pre_release_action(struct rbd_device *rbd_dev)
4215{
4216	if (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP)
4217		rbd_object_map_close(rbd_dev);
4218}
4219
4220static void __rbd_release_lock(struct rbd_device *rbd_dev)
4221{
4222	rbd_assert(list_empty(&rbd_dev->running_list));
4223
4224	rbd_pre_release_action(rbd_dev);
4225	rbd_unlock(rbd_dev);
4226}
4227
4228/*
4229 * lock_rwsem must be held for write
4230 */
4231static void rbd_release_lock(struct rbd_device *rbd_dev)
4232{
4233	if (!rbd_quiesce_lock(rbd_dev))
4234		return;
4235
4236	__rbd_release_lock(rbd_dev);
4237
4238	/*
4239	 * Give others a chance to grab the lock - we would re-acquire
4240	 * almost immediately if we got new IO while draining the running
4241	 * list otherwise.  We need to ack our own notifications, so this
4242	 * lock_dwork will be requeued from rbd_handle_released_lock() by
4243	 * way of maybe_kick_acquire().
4244	 */
4245	cancel_delayed_work(&rbd_dev->lock_dwork);
4246}
4247
4248static void rbd_release_lock_work(struct work_struct *work)
4249{
4250	struct rbd_device *rbd_dev = container_of(work, struct rbd_device,
4251						  unlock_work);
4252
4253	down_write(&rbd_dev->lock_rwsem);
4254	rbd_release_lock(rbd_dev);
4255	up_write(&rbd_dev->lock_rwsem);
4256}
4257
4258static void maybe_kick_acquire(struct rbd_device *rbd_dev)
4259{
4260	bool have_requests;
4261
4262	dout("%s rbd_dev %p\n", __func__, rbd_dev);
4263	if (__rbd_is_lock_owner(rbd_dev))
4264		return;
4265
4266	spin_lock(&rbd_dev->lock_lists_lock);
4267	have_requests = !list_empty(&rbd_dev->acquiring_list);
4268	spin_unlock(&rbd_dev->lock_lists_lock);
4269	if (have_requests || delayed_work_pending(&rbd_dev->lock_dwork)) {
4270		dout("%s rbd_dev %p kicking lock_dwork\n", __func__, rbd_dev);
4271		mod_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
4272	}
4273}
4274
4275static void rbd_handle_acquired_lock(struct rbd_device *rbd_dev, u8 struct_v,
4276				     void **p)
4277{
4278	struct rbd_client_id cid = { 0 };
4279
4280	if (struct_v >= 2) {
4281		cid.gid = ceph_decode_64(p);
4282		cid.handle = ceph_decode_64(p);
4283	}
4284
4285	dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4286	     cid.handle);
4287	if (!rbd_cid_equal(&cid, &rbd_empty_cid)) {
4288		down_write(&rbd_dev->lock_rwsem);
4289		if (rbd_cid_equal(&cid, &rbd_dev->owner_cid)) {
4290			dout("%s rbd_dev %p cid %llu-%llu == owner_cid\n",
4291			     __func__, rbd_dev, cid.gid, cid.handle);
4292		} else {
4293			rbd_set_owner_cid(rbd_dev, &cid);
4294		}
4295		downgrade_write(&rbd_dev->lock_rwsem);
4296	} else {
4297		down_read(&rbd_dev->lock_rwsem);
4298	}
4299
4300	maybe_kick_acquire(rbd_dev);
4301	up_read(&rbd_dev->lock_rwsem);
4302}
4303
4304static void rbd_handle_released_lock(struct rbd_device *rbd_dev, u8 struct_v,
4305				     void **p)
4306{
4307	struct rbd_client_id cid = { 0 };
4308
4309	if (struct_v >= 2) {
4310		cid.gid = ceph_decode_64(p);
4311		cid.handle = ceph_decode_64(p);
4312	}
4313
4314	dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4315	     cid.handle);
4316	if (!rbd_cid_equal(&cid, &rbd_empty_cid)) {
4317		down_write(&rbd_dev->lock_rwsem);
4318		if (!rbd_cid_equal(&cid, &rbd_dev->owner_cid)) {
4319			dout("%s rbd_dev %p cid %llu-%llu != owner_cid %llu-%llu\n",
4320			     __func__, rbd_dev, cid.gid, cid.handle,
4321			     rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle);
4322		} else {
4323			rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
4324		}
4325		downgrade_write(&rbd_dev->lock_rwsem);
4326	} else {
4327		down_read(&rbd_dev->lock_rwsem);
4328	}
4329
4330	maybe_kick_acquire(rbd_dev);
4331	up_read(&rbd_dev->lock_rwsem);
4332}
4333
4334/*
4335 * Returns result for ResponseMessage to be encoded (<= 0), or 1 if no
4336 * ResponseMessage is needed.
4337 */
4338static int rbd_handle_request_lock(struct rbd_device *rbd_dev, u8 struct_v,
4339				   void **p)
4340{
4341	struct rbd_client_id my_cid = rbd_get_cid(rbd_dev);
4342	struct rbd_client_id cid = { 0 };
4343	int result = 1;
4344
4345	if (struct_v >= 2) {
4346		cid.gid = ceph_decode_64(p);
4347		cid.handle = ceph_decode_64(p);
4348	}
4349
4350	dout("%s rbd_dev %p cid %llu-%llu\n", __func__, rbd_dev, cid.gid,
4351	     cid.handle);
4352	if (rbd_cid_equal(&cid, &my_cid))
4353		return result;
4354
4355	down_read(&rbd_dev->lock_rwsem);
4356	if (__rbd_is_lock_owner(rbd_dev)) {
4357		if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED &&
4358		    rbd_cid_equal(&rbd_dev->owner_cid, &rbd_empty_cid))
4359			goto out_unlock;
4360
4361		/*
4362		 * encode ResponseMessage(0) so the peer can detect
4363		 * a missing owner
4364		 */
4365		result = 0;
4366
4367		if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED) {
4368			if (!rbd_dev->opts->exclusive) {
4369				dout("%s rbd_dev %p queueing unlock_work\n",
4370				     __func__, rbd_dev);
4371				queue_work(rbd_dev->task_wq,
4372					   &rbd_dev->unlock_work);
4373			} else {
4374				/* refuse to release the lock */
4375				result = -EROFS;
4376			}
4377		}
4378	}
4379
4380out_unlock:
4381	up_read(&rbd_dev->lock_rwsem);
4382	return result;
4383}
4384
4385static void __rbd_acknowledge_notify(struct rbd_device *rbd_dev,
4386				     u64 notify_id, u64 cookie, s32 *result)
4387{
4388	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4389	char buf[4 + CEPH_ENCODING_START_BLK_LEN];
4390	int buf_size = sizeof(buf);
4391	int ret;
4392
4393	if (result) {
4394		void *p = buf;
4395
4396		/* encode ResponseMessage */
4397		ceph_start_encoding(&p, 1, 1,
4398				    buf_size - CEPH_ENCODING_START_BLK_LEN);
4399		ceph_encode_32(&p, *result);
4400	} else {
4401		buf_size = 0;
4402	}
4403
4404	ret = ceph_osdc_notify_ack(osdc, &rbd_dev->header_oid,
4405				   &rbd_dev->header_oloc, notify_id, cookie,
4406				   buf, buf_size);
4407	if (ret)
4408		rbd_warn(rbd_dev, "acknowledge_notify failed: %d", ret);
4409}
4410
4411static void rbd_acknowledge_notify(struct rbd_device *rbd_dev, u64 notify_id,
4412				   u64 cookie)
4413{
4414	dout("%s rbd_dev %p\n", __func__, rbd_dev);
4415	__rbd_acknowledge_notify(rbd_dev, notify_id, cookie, NULL);
4416}
4417
4418static void rbd_acknowledge_notify_result(struct rbd_device *rbd_dev,
4419					  u64 notify_id, u64 cookie, s32 result)
4420{
4421	dout("%s rbd_dev %p result %d\n", __func__, rbd_dev, result);
4422	__rbd_acknowledge_notify(rbd_dev, notify_id, cookie, &result);
4423}
4424
4425static void rbd_watch_cb(void *arg, u64 notify_id, u64 cookie,
4426			 u64 notifier_id, void *data, size_t data_len)
4427{
4428	struct rbd_device *rbd_dev = arg;
4429	void *p = data;
4430	void *const end = p + data_len;
4431	u8 struct_v = 0;
4432	u32 len;
4433	u32 notify_op;
4434	int ret;
4435
4436	dout("%s rbd_dev %p cookie %llu notify_id %llu data_len %zu\n",
4437	     __func__, rbd_dev, cookie, notify_id, data_len);
4438	if (data_len) {
4439		ret = ceph_start_decoding(&p, end, 1, "NotifyMessage",
4440					  &struct_v, &len);
4441		if (ret) {
4442			rbd_warn(rbd_dev, "failed to decode NotifyMessage: %d",
4443				 ret);
4444			return;
4445		}
4446
4447		notify_op = ceph_decode_32(&p);
4448	} else {
4449		/* legacy notification for header updates */
4450		notify_op = RBD_NOTIFY_OP_HEADER_UPDATE;
4451		len = 0;
4452	}
4453
4454	dout("%s rbd_dev %p notify_op %u\n", __func__, rbd_dev, notify_op);
4455	switch (notify_op) {
4456	case RBD_NOTIFY_OP_ACQUIRED_LOCK:
4457		rbd_handle_acquired_lock(rbd_dev, struct_v, &p);
4458		rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4459		break;
4460	case RBD_NOTIFY_OP_RELEASED_LOCK:
4461		rbd_handle_released_lock(rbd_dev, struct_v, &p);
4462		rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4463		break;
4464	case RBD_NOTIFY_OP_REQUEST_LOCK:
4465		ret = rbd_handle_request_lock(rbd_dev, struct_v, &p);
4466		if (ret <= 0)
4467			rbd_acknowledge_notify_result(rbd_dev, notify_id,
4468						      cookie, ret);
4469		else
4470			rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4471		break;
4472	case RBD_NOTIFY_OP_HEADER_UPDATE:
4473		ret = rbd_dev_refresh(rbd_dev);
4474		if (ret)
4475			rbd_warn(rbd_dev, "refresh failed: %d", ret);
4476
4477		rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4478		break;
4479	default:
4480		if (rbd_is_lock_owner(rbd_dev))
4481			rbd_acknowledge_notify_result(rbd_dev, notify_id,
4482						      cookie, -EOPNOTSUPP);
4483		else
4484			rbd_acknowledge_notify(rbd_dev, notify_id, cookie);
4485		break;
4486	}
4487}
4488
4489static void __rbd_unregister_watch(struct rbd_device *rbd_dev);
4490
4491static void rbd_watch_errcb(void *arg, u64 cookie, int err)
4492{
4493	struct rbd_device *rbd_dev = arg;
4494
4495	rbd_warn(rbd_dev, "encountered watch error: %d", err);
4496
4497	down_write(&rbd_dev->lock_rwsem);
4498	rbd_set_owner_cid(rbd_dev, &rbd_empty_cid);
4499	up_write(&rbd_dev->lock_rwsem);
4500
4501	mutex_lock(&rbd_dev->watch_mutex);
4502	if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED) {
4503		__rbd_unregister_watch(rbd_dev);
4504		rbd_dev->watch_state = RBD_WATCH_STATE_ERROR;
4505
4506		queue_delayed_work(rbd_dev->task_wq, &rbd_dev->watch_dwork, 0);
4507	}
4508	mutex_unlock(&rbd_dev->watch_mutex);
4509}
4510
4511/*
4512 * watch_mutex must be locked
4513 */
4514static int __rbd_register_watch(struct rbd_device *rbd_dev)
4515{
4516	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4517	struct ceph_osd_linger_request *handle;
4518
4519	rbd_assert(!rbd_dev->watch_handle);
4520	dout("%s rbd_dev %p\n", __func__, rbd_dev);
4521
4522	handle = ceph_osdc_watch(osdc, &rbd_dev->header_oid,
4523				 &rbd_dev->header_oloc, rbd_watch_cb,
4524				 rbd_watch_errcb, rbd_dev);
4525	if (IS_ERR(handle))
4526		return PTR_ERR(handle);
4527
4528	rbd_dev->watch_handle = handle;
4529	return 0;
4530}
4531
4532/*
4533 * watch_mutex must be locked
4534 */
4535static void __rbd_unregister_watch(struct rbd_device *rbd_dev)
4536{
4537	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4538	int ret;
4539
4540	rbd_assert(rbd_dev->watch_handle);
4541	dout("%s rbd_dev %p\n", __func__, rbd_dev);
4542
4543	ret = ceph_osdc_unwatch(osdc, rbd_dev->watch_handle);
4544	if (ret)
4545		rbd_warn(rbd_dev, "failed to unwatch: %d", ret);
4546
4547	rbd_dev->watch_handle = NULL;
4548}
4549
4550static int rbd_register_watch(struct rbd_device *rbd_dev)
4551{
4552	int ret;
4553
4554	mutex_lock(&rbd_dev->watch_mutex);
4555	rbd_assert(rbd_dev->watch_state == RBD_WATCH_STATE_UNREGISTERED);
4556	ret = __rbd_register_watch(rbd_dev);
4557	if (ret)
4558		goto out;
4559
4560	rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
4561	rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
4562
4563out:
4564	mutex_unlock(&rbd_dev->watch_mutex);
4565	return ret;
4566}
4567
4568static void cancel_tasks_sync(struct rbd_device *rbd_dev)
4569{
4570	dout("%s rbd_dev %p\n", __func__, rbd_dev);
4571
4572	cancel_work_sync(&rbd_dev->acquired_lock_work);
4573	cancel_work_sync(&rbd_dev->released_lock_work);
4574	cancel_delayed_work_sync(&rbd_dev->lock_dwork);
4575	cancel_work_sync(&rbd_dev->unlock_work);
4576}
4577
4578/*
4579 * header_rwsem must not be held to avoid a deadlock with
4580 * rbd_dev_refresh() when flushing notifies.
4581 */
4582static void rbd_unregister_watch(struct rbd_device *rbd_dev)
4583{
4584	cancel_tasks_sync(rbd_dev);
4585
4586	mutex_lock(&rbd_dev->watch_mutex);
4587	if (rbd_dev->watch_state == RBD_WATCH_STATE_REGISTERED)
4588		__rbd_unregister_watch(rbd_dev);
4589	rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
4590	mutex_unlock(&rbd_dev->watch_mutex);
4591
4592	cancel_delayed_work_sync(&rbd_dev->watch_dwork);
4593	ceph_osdc_flush_notifies(&rbd_dev->rbd_client->client->osdc);
4594}
4595
4596/*
4597 * lock_rwsem must be held for write
4598 */
4599static void rbd_reacquire_lock(struct rbd_device *rbd_dev)
4600{
4601	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4602	char cookie[32];
4603	int ret;
4604
4605	if (!rbd_quiesce_lock(rbd_dev))
4606		return;
4607
4608	format_lock_cookie(rbd_dev, cookie);
4609	ret = ceph_cls_set_cookie(osdc, &rbd_dev->header_oid,
4610				  &rbd_dev->header_oloc, RBD_LOCK_NAME,
4611				  CEPH_CLS_LOCK_EXCLUSIVE, rbd_dev->lock_cookie,
4612				  RBD_LOCK_TAG, cookie);
4613	if (ret) {
4614		if (ret != -EOPNOTSUPP)
4615			rbd_warn(rbd_dev, "failed to update lock cookie: %d",
4616				 ret);
4617
4618		if (rbd_dev->opts->exclusive)
4619			rbd_warn(rbd_dev,
4620			     "temporarily releasing lock on exclusive mapping");
4621
4622		/*
4623		 * Lock cookie cannot be updated on older OSDs, so do
4624		 * a manual release and queue an acquire.
4625		 */
4626		__rbd_release_lock(rbd_dev);
4627		queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
4628	} else {
4629		__rbd_lock(rbd_dev, cookie);
4630		wake_lock_waiters(rbd_dev, 0);
4631	}
4632}
4633
4634static void rbd_reregister_watch(struct work_struct *work)
4635{
4636	struct rbd_device *rbd_dev = container_of(to_delayed_work(work),
4637					    struct rbd_device, watch_dwork);
4638	int ret;
4639
4640	dout("%s rbd_dev %p\n", __func__, rbd_dev);
4641
4642	mutex_lock(&rbd_dev->watch_mutex);
4643	if (rbd_dev->watch_state != RBD_WATCH_STATE_ERROR) {
4644		mutex_unlock(&rbd_dev->watch_mutex);
4645		return;
4646	}
4647
4648	ret = __rbd_register_watch(rbd_dev);
4649	if (ret) {
4650		rbd_warn(rbd_dev, "failed to reregister watch: %d", ret);
4651		if (ret != -EBLOCKLISTED && ret != -ENOENT) {
4652			queue_delayed_work(rbd_dev->task_wq,
4653					   &rbd_dev->watch_dwork,
4654					   RBD_RETRY_DELAY);
4655			mutex_unlock(&rbd_dev->watch_mutex);
4656			return;
4657		}
4658
4659		mutex_unlock(&rbd_dev->watch_mutex);
4660		down_write(&rbd_dev->lock_rwsem);
4661		wake_lock_waiters(rbd_dev, ret);
4662		up_write(&rbd_dev->lock_rwsem);
4663		return;
4664	}
4665
4666	rbd_dev->watch_state = RBD_WATCH_STATE_REGISTERED;
4667	rbd_dev->watch_cookie = rbd_dev->watch_handle->linger_id;
4668	mutex_unlock(&rbd_dev->watch_mutex);
4669
4670	down_write(&rbd_dev->lock_rwsem);
4671	if (rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED)
4672		rbd_reacquire_lock(rbd_dev);
4673	up_write(&rbd_dev->lock_rwsem);
4674
4675	ret = rbd_dev_refresh(rbd_dev);
4676	if (ret)
4677		rbd_warn(rbd_dev, "reregistration refresh failed: %d", ret);
4678}
4679
4680/*
4681 * Synchronous osd object method call.  Returns the number of bytes
4682 * returned in the outbound buffer, or a negative error code.
4683 */
4684static int rbd_obj_method_sync(struct rbd_device *rbd_dev,
4685			     struct ceph_object_id *oid,
4686			     struct ceph_object_locator *oloc,
4687			     const char *method_name,
4688			     const void *outbound,
4689			     size_t outbound_size,
4690			     void *inbound,
4691			     size_t inbound_size)
4692{
4693	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4694	struct page *req_page = NULL;
4695	struct page *reply_page;
4696	int ret;
4697
4698	/*
4699	 * Method calls are ultimately read operations.  The result
4700	 * should placed into the inbound buffer provided.  They
4701	 * also supply outbound data--parameters for the object
4702	 * method.  Currently if this is present it will be a
4703	 * snapshot id.
4704	 */
4705	if (outbound) {
4706		if (outbound_size > PAGE_SIZE)
4707			return -E2BIG;
4708
4709		req_page = alloc_page(GFP_KERNEL);
4710		if (!req_page)
4711			return -ENOMEM;
4712
4713		memcpy(page_address(req_page), outbound, outbound_size);
4714	}
4715
4716	reply_page = alloc_page(GFP_KERNEL);
4717	if (!reply_page) {
4718		if (req_page)
4719			__free_page(req_page);
4720		return -ENOMEM;
4721	}
4722
4723	ret = ceph_osdc_call(osdc, oid, oloc, RBD_DRV_NAME, method_name,
4724			     CEPH_OSD_FLAG_READ, req_page, outbound_size,
4725			     &reply_page, &inbound_size);
4726	if (!ret) {
4727		memcpy(inbound, page_address(reply_page), inbound_size);
4728		ret = inbound_size;
4729	}
4730
4731	if (req_page)
4732		__free_page(req_page);
4733	__free_page(reply_page);
4734	return ret;
4735}
4736
4737static void rbd_queue_workfn(struct work_struct *work)
4738{
4739	struct rbd_img_request *img_request =
4740	    container_of(work, struct rbd_img_request, work);
4741	struct rbd_device *rbd_dev = img_request->rbd_dev;
4742	enum obj_operation_type op_type = img_request->op_type;
4743	struct request *rq = blk_mq_rq_from_pdu(img_request);
4744	u64 offset = (u64)blk_rq_pos(rq) << SECTOR_SHIFT;
4745	u64 length = blk_rq_bytes(rq);
4746	int result;
4747
4748	/* Ignore/skip any zero-length requests */
4749	if (!length) {
4750		dout("%s: zero-length request\n", __func__);
4751		result = 0;
4752		goto err_img_request;
4753	}
4754
4755	blk_mq_start_request(rq);
4756
4757	down_read(&rbd_dev->header_rwsem);
4758	rbd_img_capture_header(img_request);
4759	up_read(&rbd_dev->header_rwsem);
4760
4761	dout("%s rbd_dev %p img_req %p %s %llu~%llu\n", __func__, rbd_dev,
4762	     img_request, obj_op_name(op_type), offset, length);
4763
4764	if (op_type == OBJ_OP_DISCARD || op_type == OBJ_OP_ZEROOUT)
4765		result = rbd_img_fill_nodata(img_request, offset, length);
4766	else
4767		result = rbd_img_fill_from_bio(img_request, offset, length,
4768					       rq->bio);
4769	if (result)
4770		goto err_img_request;
4771
4772	rbd_img_handle_request(img_request, 0);
4773	return;
4774
4775err_img_request:
4776	rbd_img_request_destroy(img_request);
4777	if (result)
4778		rbd_warn(rbd_dev, "%s %llx at %llx result %d",
4779			 obj_op_name(op_type), length, offset, result);
4780	blk_mq_end_request(rq, errno_to_blk_status(result));
4781}
4782
4783static blk_status_t rbd_queue_rq(struct blk_mq_hw_ctx *hctx,
4784		const struct blk_mq_queue_data *bd)
4785{
4786	struct rbd_device *rbd_dev = hctx->queue->queuedata;
4787	struct rbd_img_request *img_req = blk_mq_rq_to_pdu(bd->rq);
4788	enum obj_operation_type op_type;
4789
4790	switch (req_op(bd->rq)) {
4791	case REQ_OP_DISCARD:
4792		op_type = OBJ_OP_DISCARD;
4793		break;
4794	case REQ_OP_WRITE_ZEROES:
4795		op_type = OBJ_OP_ZEROOUT;
4796		break;
4797	case REQ_OP_WRITE:
4798		op_type = OBJ_OP_WRITE;
4799		break;
4800	case REQ_OP_READ:
4801		op_type = OBJ_OP_READ;
4802		break;
4803	default:
4804		rbd_warn(rbd_dev, "unknown req_op %d", req_op(bd->rq));
4805		return BLK_STS_IOERR;
4806	}
4807
4808	rbd_img_request_init(img_req, rbd_dev, op_type);
4809
4810	if (rbd_img_is_write(img_req)) {
4811		if (rbd_is_ro(rbd_dev)) {
4812			rbd_warn(rbd_dev, "%s on read-only mapping",
4813				 obj_op_name(img_req->op_type));
4814			return BLK_STS_IOERR;
4815		}
4816		rbd_assert(!rbd_is_snap(rbd_dev));
4817	}
4818
4819	INIT_WORK(&img_req->work, rbd_queue_workfn);
4820	queue_work(rbd_wq, &img_req->work);
4821	return BLK_STS_OK;
4822}
4823
4824static void rbd_free_disk(struct rbd_device *rbd_dev)
4825{
4826	put_disk(rbd_dev->disk);
4827	blk_mq_free_tag_set(&rbd_dev->tag_set);
4828	rbd_dev->disk = NULL;
4829}
4830
4831static int rbd_obj_read_sync(struct rbd_device *rbd_dev,
4832			     struct ceph_object_id *oid,
4833			     struct ceph_object_locator *oloc,
4834			     void *buf, int buf_len)
4835
4836{
4837	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
4838	struct ceph_osd_request *req;
4839	struct page **pages;
4840	int num_pages = calc_pages_for(0, buf_len);
4841	int ret;
4842
4843	req = ceph_osdc_alloc_request(osdc, NULL, 1, false, GFP_KERNEL);
4844	if (!req)
4845		return -ENOMEM;
4846
4847	ceph_oid_copy(&req->r_base_oid, oid);
4848	ceph_oloc_copy(&req->r_base_oloc, oloc);
4849	req->r_flags = CEPH_OSD_FLAG_READ;
4850
4851	pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
4852	if (IS_ERR(pages)) {
4853		ret = PTR_ERR(pages);
4854		goto out_req;
4855	}
4856
4857	osd_req_op_extent_init(req, 0, CEPH_OSD_OP_READ, 0, buf_len, 0, 0);
4858	osd_req_op_extent_osd_data_pages(req, 0, pages, buf_len, 0, false,
4859					 true);
4860
4861	ret = ceph_osdc_alloc_messages(req, GFP_KERNEL);
4862	if (ret)
4863		goto out_req;
4864
4865	ceph_osdc_start_request(osdc, req);
4866	ret = ceph_osdc_wait_request(osdc, req);
4867	if (ret >= 0)
4868		ceph_copy_from_page_vector(pages, buf, 0, ret);
4869
4870out_req:
4871	ceph_osdc_put_request(req);
4872	return ret;
4873}
4874
4875/*
4876 * Read the complete header for the given rbd device.  On successful
4877 * return, the rbd_dev->header field will contain up-to-date
4878 * information about the image.
4879 */
4880static int rbd_dev_v1_header_info(struct rbd_device *rbd_dev,
4881				  struct rbd_image_header *header,
4882				  bool first_time)
4883{
4884	struct rbd_image_header_ondisk *ondisk = NULL;
4885	u32 snap_count = 0;
4886	u64 names_size = 0;
4887	u32 want_count;
4888	int ret;
4889
4890	/*
4891	 * The complete header will include an array of its 64-bit
4892	 * snapshot ids, followed by the names of those snapshots as
4893	 * a contiguous block of NUL-terminated strings.  Note that
4894	 * the number of snapshots could change by the time we read
4895	 * it in, in which case we re-read it.
4896	 */
4897	do {
4898		size_t size;
4899
4900		kfree(ondisk);
4901
4902		size = sizeof (*ondisk);
4903		size += snap_count * sizeof (struct rbd_image_snap_ondisk);
4904		size += names_size;
4905		ondisk = kmalloc(size, GFP_KERNEL);
4906		if (!ondisk)
4907			return -ENOMEM;
4908
4909		ret = rbd_obj_read_sync(rbd_dev, &rbd_dev->header_oid,
4910					&rbd_dev->header_oloc, ondisk, size);
4911		if (ret < 0)
4912			goto out;
4913		if ((size_t)ret < size) {
4914			ret = -ENXIO;
4915			rbd_warn(rbd_dev, "short header read (want %zd got %d)",
4916				size, ret);
4917			goto out;
4918		}
4919		if (!rbd_dev_ondisk_valid(ondisk)) {
4920			ret = -ENXIO;
4921			rbd_warn(rbd_dev, "invalid header");
4922			goto out;
4923		}
4924
4925		names_size = le64_to_cpu(ondisk->snap_names_len);
4926		want_count = snap_count;
4927		snap_count = le32_to_cpu(ondisk->snap_count);
4928	} while (snap_count != want_count);
4929
4930	ret = rbd_header_from_disk(header, ondisk, first_time);
4931out:
4932	kfree(ondisk);
4933
4934	return ret;
4935}
4936
4937static void rbd_dev_update_size(struct rbd_device *rbd_dev)
4938{
4939	sector_t size;
4940
4941	/*
4942	 * If EXISTS is not set, rbd_dev->disk may be NULL, so don't
4943	 * try to update its size.  If REMOVING is set, updating size
4944	 * is just useless work since the device can't be opened.
4945	 */
4946	if (test_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags) &&
4947	    !test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags)) {
4948		size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
4949		dout("setting size to %llu sectors", (unsigned long long)size);
4950		set_capacity_and_notify(rbd_dev->disk, size);
4951	}
4952}
4953
4954static const struct blk_mq_ops rbd_mq_ops = {
4955	.queue_rq	= rbd_queue_rq,
4956};
4957
4958static int rbd_init_disk(struct rbd_device *rbd_dev)
4959{
4960	struct gendisk *disk;
4961	unsigned int objset_bytes =
4962	    rbd_dev->layout.object_size * rbd_dev->layout.stripe_count;
4963	struct queue_limits lim = {
4964		.max_hw_sectors		= objset_bytes >> SECTOR_SHIFT,
4965		.io_opt			= objset_bytes,
4966		.io_min			= rbd_dev->opts->alloc_size,
4967		.max_segments		= USHRT_MAX,
4968		.max_segment_size	= UINT_MAX,
4969	};
4970	int err;
4971
4972	memset(&rbd_dev->tag_set, 0, sizeof(rbd_dev->tag_set));
4973	rbd_dev->tag_set.ops = &rbd_mq_ops;
4974	rbd_dev->tag_set.queue_depth = rbd_dev->opts->queue_depth;
4975	rbd_dev->tag_set.numa_node = NUMA_NO_NODE;
4976	rbd_dev->tag_set.nr_hw_queues = num_present_cpus();
4977	rbd_dev->tag_set.cmd_size = sizeof(struct rbd_img_request);
4978
4979	err = blk_mq_alloc_tag_set(&rbd_dev->tag_set);
4980	if (err)
4981		return err;
4982
4983	if (rbd_dev->opts->trim) {
4984		lim.discard_granularity = rbd_dev->opts->alloc_size;
4985		lim.max_hw_discard_sectors = objset_bytes >> SECTOR_SHIFT;
4986		lim.max_write_zeroes_sectors = objset_bytes >> SECTOR_SHIFT;
4987	}
4988
4989	if (!ceph_test_opt(rbd_dev->rbd_client->client, NOCRC))
4990		lim.features |= BLK_FEAT_STABLE_WRITES;
4991
4992	disk = blk_mq_alloc_disk(&rbd_dev->tag_set, &lim, rbd_dev);
4993	if (IS_ERR(disk)) {
4994		err = PTR_ERR(disk);
4995		goto out_tag_set;
4996	}
4997
4998	snprintf(disk->disk_name, sizeof(disk->disk_name), RBD_DRV_NAME "%d",
4999		 rbd_dev->dev_id);
5000	disk->major = rbd_dev->major;
5001	disk->first_minor = rbd_dev->minor;
5002	if (single_major)
5003		disk->minors = (1 << RBD_SINGLE_MAJOR_PART_SHIFT);
5004	else
5005		disk->minors = RBD_MINORS_PER_MAJOR;
5006	disk->fops = &rbd_bd_ops;
5007	disk->private_data = rbd_dev;
5008	rbd_dev->disk = disk;
5009
5010	return 0;
5011out_tag_set:
5012	blk_mq_free_tag_set(&rbd_dev->tag_set);
5013	return err;
5014}
5015
5016/*
5017  sysfs
5018*/
5019
5020static struct rbd_device *dev_to_rbd_dev(struct device *dev)
5021{
5022	return container_of(dev, struct rbd_device, dev);
5023}
5024
5025static ssize_t rbd_size_show(struct device *dev,
5026			     struct device_attribute *attr, char *buf)
5027{
5028	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5029
5030	return sprintf(buf, "%llu\n",
5031		(unsigned long long)rbd_dev->mapping.size);
5032}
5033
5034static ssize_t rbd_features_show(struct device *dev,
5035			     struct device_attribute *attr, char *buf)
5036{
5037	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5038
5039	return sprintf(buf, "0x%016llx\n", rbd_dev->header.features);
5040}
5041
5042static ssize_t rbd_major_show(struct device *dev,
5043			      struct device_attribute *attr, char *buf)
5044{
5045	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5046
5047	if (rbd_dev->major)
5048		return sprintf(buf, "%d\n", rbd_dev->major);
5049
5050	return sprintf(buf, "(none)\n");
5051}
5052
5053static ssize_t rbd_minor_show(struct device *dev,
5054			      struct device_attribute *attr, char *buf)
5055{
5056	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5057
5058	return sprintf(buf, "%d\n", rbd_dev->minor);
5059}
5060
5061static ssize_t rbd_client_addr_show(struct device *dev,
5062				    struct device_attribute *attr, char *buf)
5063{
5064	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5065	struct ceph_entity_addr *client_addr =
5066	    ceph_client_addr(rbd_dev->rbd_client->client);
5067
5068	return sprintf(buf, "%pISpc/%u\n", &client_addr->in_addr,
5069		       le32_to_cpu(client_addr->nonce));
5070}
5071
5072static ssize_t rbd_client_id_show(struct device *dev,
5073				  struct device_attribute *attr, char *buf)
5074{
5075	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5076
5077	return sprintf(buf, "client%lld\n",
5078		       ceph_client_gid(rbd_dev->rbd_client->client));
5079}
5080
5081static ssize_t rbd_cluster_fsid_show(struct device *dev,
5082				     struct device_attribute *attr, char *buf)
5083{
5084	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5085
5086	return sprintf(buf, "%pU\n", &rbd_dev->rbd_client->client->fsid);
5087}
5088
5089static ssize_t rbd_config_info_show(struct device *dev,
5090				    struct device_attribute *attr, char *buf)
5091{
5092	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5093
5094	if (!capable(CAP_SYS_ADMIN))
5095		return -EPERM;
5096
5097	return sprintf(buf, "%s\n", rbd_dev->config_info);
5098}
5099
5100static ssize_t rbd_pool_show(struct device *dev,
5101			     struct device_attribute *attr, char *buf)
5102{
5103	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5104
5105	return sprintf(buf, "%s\n", rbd_dev->spec->pool_name);
5106}
5107
5108static ssize_t rbd_pool_id_show(struct device *dev,
5109			     struct device_attribute *attr, char *buf)
5110{
5111	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5112
5113	return sprintf(buf, "%llu\n",
5114			(unsigned long long) rbd_dev->spec->pool_id);
5115}
5116
5117static ssize_t rbd_pool_ns_show(struct device *dev,
5118				struct device_attribute *attr, char *buf)
5119{
5120	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5121
5122	return sprintf(buf, "%s\n", rbd_dev->spec->pool_ns ?: "");
5123}
5124
5125static ssize_t rbd_name_show(struct device *dev,
5126			     struct device_attribute *attr, char *buf)
5127{
5128	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5129
5130	if (rbd_dev->spec->image_name)
5131		return sprintf(buf, "%s\n", rbd_dev->spec->image_name);
5132
5133	return sprintf(buf, "(unknown)\n");
5134}
5135
5136static ssize_t rbd_image_id_show(struct device *dev,
5137			     struct device_attribute *attr, char *buf)
5138{
5139	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5140
5141	return sprintf(buf, "%s\n", rbd_dev->spec->image_id);
5142}
5143
5144/*
5145 * Shows the name of the currently-mapped snapshot (or
5146 * RBD_SNAP_HEAD_NAME for the base image).
5147 */
5148static ssize_t rbd_snap_show(struct device *dev,
5149			     struct device_attribute *attr,
5150			     char *buf)
5151{
5152	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5153
5154	return sprintf(buf, "%s\n", rbd_dev->spec->snap_name);
5155}
5156
5157static ssize_t rbd_snap_id_show(struct device *dev,
5158				struct device_attribute *attr, char *buf)
5159{
5160	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5161
5162	return sprintf(buf, "%llu\n", rbd_dev->spec->snap_id);
5163}
5164
5165/*
5166 * For a v2 image, shows the chain of parent images, separated by empty
5167 * lines.  For v1 images or if there is no parent, shows "(no parent
5168 * image)".
5169 */
5170static ssize_t rbd_parent_show(struct device *dev,
5171			       struct device_attribute *attr,
5172			       char *buf)
5173{
5174	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5175	ssize_t count = 0;
5176
5177	if (!rbd_dev->parent)
5178		return sprintf(buf, "(no parent image)\n");
5179
5180	for ( ; rbd_dev->parent; rbd_dev = rbd_dev->parent) {
5181		struct rbd_spec *spec = rbd_dev->parent_spec;
5182
5183		count += sprintf(&buf[count], "%s"
5184			    "pool_id %llu\npool_name %s\n"
5185			    "pool_ns %s\n"
5186			    "image_id %s\nimage_name %s\n"
5187			    "snap_id %llu\nsnap_name %s\n"
5188			    "overlap %llu\n",
5189			    !count ? "" : "\n", /* first? */
5190			    spec->pool_id, spec->pool_name,
5191			    spec->pool_ns ?: "",
5192			    spec->image_id, spec->image_name ?: "(unknown)",
5193			    spec->snap_id, spec->snap_name,
5194			    rbd_dev->parent_overlap);
5195	}
5196
5197	return count;
5198}
5199
5200static ssize_t rbd_image_refresh(struct device *dev,
5201				 struct device_attribute *attr,
5202				 const char *buf,
5203				 size_t size)
5204{
5205	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5206	int ret;
5207
5208	if (!capable(CAP_SYS_ADMIN))
5209		return -EPERM;
5210
5211	ret = rbd_dev_refresh(rbd_dev);
5212	if (ret)
5213		return ret;
5214
5215	return size;
5216}
5217
5218static DEVICE_ATTR(size, 0444, rbd_size_show, NULL);
5219static DEVICE_ATTR(features, 0444, rbd_features_show, NULL);
5220static DEVICE_ATTR(major, 0444, rbd_major_show, NULL);
5221static DEVICE_ATTR(minor, 0444, rbd_minor_show, NULL);
5222static DEVICE_ATTR(client_addr, 0444, rbd_client_addr_show, NULL);
5223static DEVICE_ATTR(client_id, 0444, rbd_client_id_show, NULL);
5224static DEVICE_ATTR(cluster_fsid, 0444, rbd_cluster_fsid_show, NULL);
5225static DEVICE_ATTR(config_info, 0400, rbd_config_info_show, NULL);
5226static DEVICE_ATTR(pool, 0444, rbd_pool_show, NULL);
5227static DEVICE_ATTR(pool_id, 0444, rbd_pool_id_show, NULL);
5228static DEVICE_ATTR(pool_ns, 0444, rbd_pool_ns_show, NULL);
5229static DEVICE_ATTR(name, 0444, rbd_name_show, NULL);
5230static DEVICE_ATTR(image_id, 0444, rbd_image_id_show, NULL);
5231static DEVICE_ATTR(refresh, 0200, NULL, rbd_image_refresh);
5232static DEVICE_ATTR(current_snap, 0444, rbd_snap_show, NULL);
5233static DEVICE_ATTR(snap_id, 0444, rbd_snap_id_show, NULL);
5234static DEVICE_ATTR(parent, 0444, rbd_parent_show, NULL);
5235
5236static struct attribute *rbd_attrs[] = {
5237	&dev_attr_size.attr,
5238	&dev_attr_features.attr,
5239	&dev_attr_major.attr,
5240	&dev_attr_minor.attr,
5241	&dev_attr_client_addr.attr,
5242	&dev_attr_client_id.attr,
5243	&dev_attr_cluster_fsid.attr,
5244	&dev_attr_config_info.attr,
5245	&dev_attr_pool.attr,
5246	&dev_attr_pool_id.attr,
5247	&dev_attr_pool_ns.attr,
5248	&dev_attr_name.attr,
5249	&dev_attr_image_id.attr,
5250	&dev_attr_current_snap.attr,
5251	&dev_attr_snap_id.attr,
5252	&dev_attr_parent.attr,
5253	&dev_attr_refresh.attr,
5254	NULL
5255};
5256
5257static struct attribute_group rbd_attr_group = {
5258	.attrs = rbd_attrs,
5259};
5260
5261static const struct attribute_group *rbd_attr_groups[] = {
5262	&rbd_attr_group,
5263	NULL
5264};
5265
5266static void rbd_dev_release(struct device *dev);
5267
5268static const struct device_type rbd_device_type = {
5269	.name		= "rbd",
5270	.groups		= rbd_attr_groups,
5271	.release	= rbd_dev_release,
5272};
5273
5274static struct rbd_spec *rbd_spec_get(struct rbd_spec *spec)
5275{
5276	kref_get(&spec->kref);
5277
5278	return spec;
5279}
5280
5281static void rbd_spec_free(struct kref *kref);
5282static void rbd_spec_put(struct rbd_spec *spec)
5283{
5284	if (spec)
5285		kref_put(&spec->kref, rbd_spec_free);
5286}
5287
5288static struct rbd_spec *rbd_spec_alloc(void)
5289{
5290	struct rbd_spec *spec;
5291
5292	spec = kzalloc(sizeof (*spec), GFP_KERNEL);
5293	if (!spec)
5294		return NULL;
5295
5296	spec->pool_id = CEPH_NOPOOL;
5297	spec->snap_id = CEPH_NOSNAP;
5298	kref_init(&spec->kref);
5299
5300	return spec;
5301}
5302
5303static void rbd_spec_free(struct kref *kref)
5304{
5305	struct rbd_spec *spec = container_of(kref, struct rbd_spec, kref);
5306
5307	kfree(spec->pool_name);
5308	kfree(spec->pool_ns);
5309	kfree(spec->image_id);
5310	kfree(spec->image_name);
5311	kfree(spec->snap_name);
5312	kfree(spec);
5313}
5314
5315static void rbd_dev_free(struct rbd_device *rbd_dev)
5316{
5317	WARN_ON(rbd_dev->watch_state != RBD_WATCH_STATE_UNREGISTERED);
5318	WARN_ON(rbd_dev->lock_state != RBD_LOCK_STATE_UNLOCKED);
5319
5320	ceph_oid_destroy(&rbd_dev->header_oid);
5321	ceph_oloc_destroy(&rbd_dev->header_oloc);
5322	kfree(rbd_dev->config_info);
5323
5324	rbd_put_client(rbd_dev->rbd_client);
5325	rbd_spec_put(rbd_dev->spec);
5326	kfree(rbd_dev->opts);
5327	kfree(rbd_dev);
5328}
5329
5330static void rbd_dev_release(struct device *dev)
5331{
5332	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
5333	bool need_put = !!rbd_dev->opts;
5334
5335	if (need_put) {
5336		destroy_workqueue(rbd_dev->task_wq);
5337		ida_free(&rbd_dev_id_ida, rbd_dev->dev_id);
5338	}
5339
5340	rbd_dev_free(rbd_dev);
5341
5342	/*
5343	 * This is racy, but way better than putting module outside of
5344	 * the release callback.  The race window is pretty small, so
5345	 * doing something similar to dm (dm-builtin.c) is overkill.
5346	 */
5347	if (need_put)
5348		module_put(THIS_MODULE);
5349}
5350
5351static struct rbd_device *__rbd_dev_create(struct rbd_spec *spec)
5352{
5353	struct rbd_device *rbd_dev;
5354
5355	rbd_dev = kzalloc(sizeof(*rbd_dev), GFP_KERNEL);
5356	if (!rbd_dev)
5357		return NULL;
5358
5359	spin_lock_init(&rbd_dev->lock);
5360	INIT_LIST_HEAD(&rbd_dev->node);
5361	init_rwsem(&rbd_dev->header_rwsem);
5362
5363	rbd_dev->header.data_pool_id = CEPH_NOPOOL;
5364	ceph_oid_init(&rbd_dev->header_oid);
5365	rbd_dev->header_oloc.pool = spec->pool_id;
5366	if (spec->pool_ns) {
5367		WARN_ON(!*spec->pool_ns);
5368		rbd_dev->header_oloc.pool_ns =
5369		    ceph_find_or_create_string(spec->pool_ns,
5370					       strlen(spec->pool_ns));
5371	}
5372
5373	mutex_init(&rbd_dev->watch_mutex);
5374	rbd_dev->watch_state = RBD_WATCH_STATE_UNREGISTERED;
5375	INIT_DELAYED_WORK(&rbd_dev->watch_dwork, rbd_reregister_watch);
5376
5377	init_rwsem(&rbd_dev->lock_rwsem);
5378	rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED;
5379	INIT_WORK(&rbd_dev->acquired_lock_work, rbd_notify_acquired_lock);
5380	INIT_WORK(&rbd_dev->released_lock_work, rbd_notify_released_lock);
5381	INIT_DELAYED_WORK(&rbd_dev->lock_dwork, rbd_acquire_lock);
5382	INIT_WORK(&rbd_dev->unlock_work, rbd_release_lock_work);
5383	spin_lock_init(&rbd_dev->lock_lists_lock);
5384	INIT_LIST_HEAD(&rbd_dev->acquiring_list);
5385	INIT_LIST_HEAD(&rbd_dev->running_list);
5386	init_completion(&rbd_dev->acquire_wait);
5387	init_completion(&rbd_dev->quiescing_wait);
5388
5389	spin_lock_init(&rbd_dev->object_map_lock);
5390
5391	rbd_dev->dev.bus = &rbd_bus_type;
5392	rbd_dev->dev.type = &rbd_device_type;
5393	rbd_dev->dev.parent = &rbd_root_dev;
5394	device_initialize(&rbd_dev->dev);
5395
5396	return rbd_dev;
5397}
5398
5399/*
5400 * Create a mapping rbd_dev.
5401 */
5402static struct rbd_device *rbd_dev_create(struct rbd_client *rbdc,
5403					 struct rbd_spec *spec,
5404					 struct rbd_options *opts)
5405{
5406	struct rbd_device *rbd_dev;
5407
5408	rbd_dev = __rbd_dev_create(spec);
5409	if (!rbd_dev)
5410		return NULL;
5411
5412	/* get an id and fill in device name */
5413	rbd_dev->dev_id = ida_alloc_max(&rbd_dev_id_ida,
5414					minor_to_rbd_dev_id(1 << MINORBITS) - 1,
5415					GFP_KERNEL);
5416	if (rbd_dev->dev_id < 0)
5417		goto fail_rbd_dev;
5418
5419	sprintf(rbd_dev->name, RBD_DRV_NAME "%d", rbd_dev->dev_id);
5420	rbd_dev->task_wq = alloc_ordered_workqueue("%s-tasks", WQ_MEM_RECLAIM,
5421						   rbd_dev->name);
5422	if (!rbd_dev->task_wq)
5423		goto fail_dev_id;
5424
5425	/* we have a ref from do_rbd_add() */
5426	__module_get(THIS_MODULE);
5427
5428	rbd_dev->rbd_client = rbdc;
5429	rbd_dev->spec = spec;
5430	rbd_dev->opts = opts;
5431
5432	dout("%s rbd_dev %p dev_id %d\n", __func__, rbd_dev, rbd_dev->dev_id);
5433	return rbd_dev;
5434
5435fail_dev_id:
5436	ida_free(&rbd_dev_id_ida, rbd_dev->dev_id);
5437fail_rbd_dev:
5438	rbd_dev_free(rbd_dev);
5439	return NULL;
5440}
5441
5442static void rbd_dev_destroy(struct rbd_device *rbd_dev)
5443{
5444	if (rbd_dev)
5445		put_device(&rbd_dev->dev);
5446}
5447
5448/*
5449 * Get the size and object order for an image snapshot, or if
5450 * snap_id is CEPH_NOSNAP, gets this information for the base
5451 * image.
5452 */
5453static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
5454				u8 *order, u64 *snap_size)
5455{
5456	__le64 snapid = cpu_to_le64(snap_id);
5457	int ret;
5458	struct {
5459		u8 order;
5460		__le64 size;
5461	} __attribute__ ((packed)) size_buf = { 0 };
5462
5463	ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5464				  &rbd_dev->header_oloc, "get_size",
5465				  &snapid, sizeof(snapid),
5466				  &size_buf, sizeof(size_buf));
5467	dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5468	if (ret < 0)
5469		return ret;
5470	if (ret < sizeof (size_buf))
5471		return -ERANGE;
5472
5473	if (order) {
5474		*order = size_buf.order;
5475		dout("  order %u", (unsigned int)*order);
5476	}
5477	*snap_size = le64_to_cpu(size_buf.size);
5478
5479	dout("  snap_id 0x%016llx snap_size = %llu\n",
5480		(unsigned long long)snap_id,
5481		(unsigned long long)*snap_size);
5482
5483	return 0;
5484}
5485
5486static int rbd_dev_v2_object_prefix(struct rbd_device *rbd_dev,
5487				    char **pobject_prefix)
5488{
5489	size_t size;
5490	void *reply_buf;
5491	char *object_prefix;
5492	int ret;
5493	void *p;
5494
5495	/* Response will be an encoded string, which includes a length */
5496	size = sizeof(__le32) + RBD_OBJ_PREFIX_LEN_MAX;
5497	reply_buf = kzalloc(size, GFP_KERNEL);
5498	if (!reply_buf)
5499		return -ENOMEM;
5500
5501	ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5502				  &rbd_dev->header_oloc, "get_object_prefix",
5503				  NULL, 0, reply_buf, size);
5504	dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5505	if (ret < 0)
5506		goto out;
5507
5508	p = reply_buf;
5509	object_prefix = ceph_extract_encoded_string(&p, p + ret, NULL,
5510						    GFP_NOIO);
5511	if (IS_ERR(object_prefix)) {
5512		ret = PTR_ERR(object_prefix);
5513		goto out;
5514	}
5515	ret = 0;
5516
5517	*pobject_prefix = object_prefix;
5518	dout("  object_prefix = %s\n", object_prefix);
5519out:
5520	kfree(reply_buf);
5521
5522	return ret;
5523}
5524
5525static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
5526				     bool read_only, u64 *snap_features)
5527{
5528	struct {
5529		__le64 snap_id;
5530		u8 read_only;
5531	} features_in;
5532	struct {
5533		__le64 features;
5534		__le64 incompat;
5535	} __attribute__ ((packed)) features_buf = { 0 };
5536	u64 unsup;
5537	int ret;
5538
5539	features_in.snap_id = cpu_to_le64(snap_id);
5540	features_in.read_only = read_only;
5541
5542	ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5543				  &rbd_dev->header_oloc, "get_features",
5544				  &features_in, sizeof(features_in),
5545				  &features_buf, sizeof(features_buf));
5546	dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5547	if (ret < 0)
5548		return ret;
5549	if (ret < sizeof (features_buf))
5550		return -ERANGE;
5551
5552	unsup = le64_to_cpu(features_buf.incompat) & ~RBD_FEATURES_SUPPORTED;
5553	if (unsup) {
5554		rbd_warn(rbd_dev, "image uses unsupported features: 0x%llx",
5555			 unsup);
5556		return -ENXIO;
5557	}
5558
5559	*snap_features = le64_to_cpu(features_buf.features);
5560
5561	dout("  snap_id 0x%016llx features = 0x%016llx incompat = 0x%016llx\n",
5562		(unsigned long long)snap_id,
5563		(unsigned long long)*snap_features,
5564		(unsigned long long)le64_to_cpu(features_buf.incompat));
5565
5566	return 0;
5567}
5568
5569/*
5570 * These are generic image flags, but since they are used only for
5571 * object map, store them in rbd_dev->object_map_flags.
5572 *
5573 * For the same reason, this function is called only on object map
5574 * (re)load and not on header refresh.
5575 */
5576static int rbd_dev_v2_get_flags(struct rbd_device *rbd_dev)
5577{
5578	__le64 snapid = cpu_to_le64(rbd_dev->spec->snap_id);
5579	__le64 flags;
5580	int ret;
5581
5582	ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5583				  &rbd_dev->header_oloc, "get_flags",
5584				  &snapid, sizeof(snapid),
5585				  &flags, sizeof(flags));
5586	if (ret < 0)
5587		return ret;
5588	if (ret < sizeof(flags))
5589		return -EBADMSG;
5590
5591	rbd_dev->object_map_flags = le64_to_cpu(flags);
5592	return 0;
5593}
5594
5595struct parent_image_info {
5596	u64		pool_id;
5597	const char	*pool_ns;
5598	const char	*image_id;
5599	u64		snap_id;
5600
5601	bool		has_overlap;
5602	u64		overlap;
5603};
5604
5605static void rbd_parent_info_cleanup(struct parent_image_info *pii)
5606{
5607	kfree(pii->pool_ns);
5608	kfree(pii->image_id);
5609
5610	memset(pii, 0, sizeof(*pii));
5611}
5612
5613/*
5614 * The caller is responsible for @pii.
5615 */
5616static int decode_parent_image_spec(void **p, void *end,
5617				    struct parent_image_info *pii)
5618{
5619	u8 struct_v;
5620	u32 struct_len;
5621	int ret;
5622
5623	ret = ceph_start_decoding(p, end, 1, "ParentImageSpec",
5624				  &struct_v, &struct_len);
5625	if (ret)
5626		return ret;
5627
5628	ceph_decode_64_safe(p, end, pii->pool_id, e_inval);
5629	pii->pool_ns = ceph_extract_encoded_string(p, end, NULL, GFP_KERNEL);
5630	if (IS_ERR(pii->pool_ns)) {
5631		ret = PTR_ERR(pii->pool_ns);
5632		pii->pool_ns = NULL;
5633		return ret;
5634	}
5635	pii->image_id = ceph_extract_encoded_string(p, end, NULL, GFP_KERNEL);
5636	if (IS_ERR(pii->image_id)) {
5637		ret = PTR_ERR(pii->image_id);
5638		pii->image_id = NULL;
5639		return ret;
5640	}
5641	ceph_decode_64_safe(p, end, pii->snap_id, e_inval);
5642	return 0;
5643
5644e_inval:
5645	return -EINVAL;
5646}
5647
5648static int __get_parent_info(struct rbd_device *rbd_dev,
5649			     struct page *req_page,
5650			     struct page *reply_page,
5651			     struct parent_image_info *pii)
5652{
5653	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
5654	size_t reply_len = PAGE_SIZE;
5655	void *p, *end;
5656	int ret;
5657
5658	ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5659			     "rbd", "parent_get", CEPH_OSD_FLAG_READ,
5660			     req_page, sizeof(u64), &reply_page, &reply_len);
5661	if (ret)
5662		return ret == -EOPNOTSUPP ? 1 : ret;
5663
5664	p = page_address(reply_page);
5665	end = p + reply_len;
5666	ret = decode_parent_image_spec(&p, end, pii);
5667	if (ret)
5668		return ret;
5669
5670	ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5671			     "rbd", "parent_overlap_get", CEPH_OSD_FLAG_READ,
5672			     req_page, sizeof(u64), &reply_page, &reply_len);
5673	if (ret)
5674		return ret;
5675
5676	p = page_address(reply_page);
5677	end = p + reply_len;
5678	ceph_decode_8_safe(&p, end, pii->has_overlap, e_inval);
5679	if (pii->has_overlap)
5680		ceph_decode_64_safe(&p, end, pii->overlap, e_inval);
5681
5682	dout("%s pool_id %llu pool_ns %s image_id %s snap_id %llu has_overlap %d overlap %llu\n",
5683	     __func__, pii->pool_id, pii->pool_ns, pii->image_id, pii->snap_id,
5684	     pii->has_overlap, pii->overlap);
5685	return 0;
5686
5687e_inval:
5688	return -EINVAL;
5689}
5690
5691/*
5692 * The caller is responsible for @pii.
5693 */
5694static int __get_parent_info_legacy(struct rbd_device *rbd_dev,
5695				    struct page *req_page,
5696				    struct page *reply_page,
5697				    struct parent_image_info *pii)
5698{
5699	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
5700	size_t reply_len = PAGE_SIZE;
5701	void *p, *end;
5702	int ret;
5703
5704	ret = ceph_osdc_call(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc,
5705			     "rbd", "get_parent", CEPH_OSD_FLAG_READ,
5706			     req_page, sizeof(u64), &reply_page, &reply_len);
5707	if (ret)
5708		return ret;
5709
5710	p = page_address(reply_page);
5711	end = p + reply_len;
5712	ceph_decode_64_safe(&p, end, pii->pool_id, e_inval);
5713	pii->image_id = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
5714	if (IS_ERR(pii->image_id)) {
5715		ret = PTR_ERR(pii->image_id);
5716		pii->image_id = NULL;
5717		return ret;
5718	}
5719	ceph_decode_64_safe(&p, end, pii->snap_id, e_inval);
5720	pii->has_overlap = true;
5721	ceph_decode_64_safe(&p, end, pii->overlap, e_inval);
5722
5723	dout("%s pool_id %llu pool_ns %s image_id %s snap_id %llu has_overlap %d overlap %llu\n",
5724	     __func__, pii->pool_id, pii->pool_ns, pii->image_id, pii->snap_id,
5725	     pii->has_overlap, pii->overlap);
5726	return 0;
5727
5728e_inval:
5729	return -EINVAL;
5730}
5731
5732static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev,
5733				  struct parent_image_info *pii)
5734{
5735	struct page *req_page, *reply_page;
5736	void *p;
5737	int ret;
5738
5739	req_page = alloc_page(GFP_KERNEL);
5740	if (!req_page)
5741		return -ENOMEM;
5742
5743	reply_page = alloc_page(GFP_KERNEL);
5744	if (!reply_page) {
5745		__free_page(req_page);
5746		return -ENOMEM;
5747	}
5748
5749	p = page_address(req_page);
5750	ceph_encode_64(&p, rbd_dev->spec->snap_id);
5751	ret = __get_parent_info(rbd_dev, req_page, reply_page, pii);
5752	if (ret > 0)
5753		ret = __get_parent_info_legacy(rbd_dev, req_page, reply_page,
5754					       pii);
5755
5756	__free_page(req_page);
5757	__free_page(reply_page);
5758	return ret;
5759}
5760
5761static int rbd_dev_setup_parent(struct rbd_device *rbd_dev)
5762{
5763	struct rbd_spec *parent_spec;
5764	struct parent_image_info pii = { 0 };
5765	int ret;
5766
5767	parent_spec = rbd_spec_alloc();
5768	if (!parent_spec)
5769		return -ENOMEM;
5770
5771	ret = rbd_dev_v2_parent_info(rbd_dev, &pii);
5772	if (ret)
5773		goto out_err;
5774
5775	if (pii.pool_id == CEPH_NOPOOL || !pii.has_overlap)
5776		goto out;	/* No parent?  No problem. */
5777
5778	/* The ceph file layout needs to fit pool id in 32 bits */
5779
5780	ret = -EIO;
5781	if (pii.pool_id > (u64)U32_MAX) {
5782		rbd_warn(NULL, "parent pool id too large (%llu > %u)",
5783			(unsigned long long)pii.pool_id, U32_MAX);
5784		goto out_err;
5785	}
5786
5787	/*
5788	 * The parent won't change except when the clone is flattened,
5789	 * so we only need to record the parent image spec once.
5790	 */
5791	parent_spec->pool_id = pii.pool_id;
5792	if (pii.pool_ns && *pii.pool_ns) {
5793		parent_spec->pool_ns = pii.pool_ns;
5794		pii.pool_ns = NULL;
5795	}
5796	parent_spec->image_id = pii.image_id;
5797	pii.image_id = NULL;
5798	parent_spec->snap_id = pii.snap_id;
5799
5800	rbd_assert(!rbd_dev->parent_spec);
5801	rbd_dev->parent_spec = parent_spec;
5802	parent_spec = NULL;	/* rbd_dev now owns this */
5803
5804	/*
5805	 * Record the parent overlap.  If it's zero, issue a warning as
5806	 * we will proceed as if there is no parent.
5807	 */
5808	if (!pii.overlap)
5809		rbd_warn(rbd_dev, "clone is standalone (overlap 0)");
5810	rbd_dev->parent_overlap = pii.overlap;
5811
5812out:
5813	ret = 0;
5814out_err:
5815	rbd_parent_info_cleanup(&pii);
5816	rbd_spec_put(parent_spec);
5817	return ret;
5818}
5819
5820static int rbd_dev_v2_striping_info(struct rbd_device *rbd_dev,
5821				    u64 *stripe_unit, u64 *stripe_count)
5822{
5823	struct {
5824		__le64 stripe_unit;
5825		__le64 stripe_count;
5826	} __attribute__ ((packed)) striping_info_buf = { 0 };
5827	size_t size = sizeof (striping_info_buf);
5828	int ret;
5829
5830	ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5831				&rbd_dev->header_oloc, "get_stripe_unit_count",
5832				NULL, 0, &striping_info_buf, size);
5833	dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5834	if (ret < 0)
5835		return ret;
5836	if (ret < size)
5837		return -ERANGE;
5838
5839	*stripe_unit = le64_to_cpu(striping_info_buf.stripe_unit);
5840	*stripe_count = le64_to_cpu(striping_info_buf.stripe_count);
5841	dout("  stripe_unit = %llu stripe_count = %llu\n", *stripe_unit,
5842	     *stripe_count);
5843
5844	return 0;
5845}
5846
5847static int rbd_dev_v2_data_pool(struct rbd_device *rbd_dev, s64 *data_pool_id)
5848{
5849	__le64 data_pool_buf;
5850	int ret;
5851
5852	ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
5853				  &rbd_dev->header_oloc, "get_data_pool",
5854				  NULL, 0, &data_pool_buf,
5855				  sizeof(data_pool_buf));
5856	dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
5857	if (ret < 0)
5858		return ret;
5859	if (ret < sizeof(data_pool_buf))
5860		return -EBADMSG;
5861
5862	*data_pool_id = le64_to_cpu(data_pool_buf);
5863	dout("  data_pool_id = %lld\n", *data_pool_id);
5864	WARN_ON(*data_pool_id == CEPH_NOPOOL);
5865
5866	return 0;
5867}
5868
5869static char *rbd_dev_image_name(struct rbd_device *rbd_dev)
5870{
5871	CEPH_DEFINE_OID_ONSTACK(oid);
5872	size_t image_id_size;
5873	char *image_id;
5874	void *p;
5875	void *end;
5876	size_t size;
5877	void *reply_buf = NULL;
5878	size_t len = 0;
5879	char *image_name = NULL;
5880	int ret;
5881
5882	rbd_assert(!rbd_dev->spec->image_name);
5883
5884	len = strlen(rbd_dev->spec->image_id);
5885	image_id_size = sizeof (__le32) + len;
5886	image_id = kmalloc(image_id_size, GFP_KERNEL);
5887	if (!image_id)
5888		return NULL;
5889
5890	p = image_id;
5891	end = image_id + image_id_size;
5892	ceph_encode_string(&p, end, rbd_dev->spec->image_id, (u32)len);
5893
5894	size = sizeof (__le32) + RBD_IMAGE_NAME_LEN_MAX;
5895	reply_buf = kmalloc(size, GFP_KERNEL);
5896	if (!reply_buf)
5897		goto out;
5898
5899	ceph_oid_printf(&oid, "%s", RBD_DIRECTORY);
5900	ret = rbd_obj_method_sync(rbd_dev, &oid, &rbd_dev->header_oloc,
5901				  "dir_get_name", image_id, image_id_size,
5902				  reply_buf, size);
5903	if (ret < 0)
5904		goto out;
5905	p = reply_buf;
5906	end = reply_buf + ret;
5907
5908	image_name = ceph_extract_encoded_string(&p, end, &len, GFP_KERNEL);
5909	if (IS_ERR(image_name))
5910		image_name = NULL;
5911	else
5912		dout("%s: name is %s len is %zd\n", __func__, image_name, len);
5913out:
5914	kfree(reply_buf);
5915	kfree(image_id);
5916
5917	return image_name;
5918}
5919
5920static u64 rbd_v1_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5921{
5922	struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5923	const char *snap_name;
5924	u32 which = 0;
5925
5926	/* Skip over names until we find the one we are looking for */
5927
5928	snap_name = rbd_dev->header.snap_names;
5929	while (which < snapc->num_snaps) {
5930		if (!strcmp(name, snap_name))
5931			return snapc->snaps[which];
5932		snap_name += strlen(snap_name) + 1;
5933		which++;
5934	}
5935	return CEPH_NOSNAP;
5936}
5937
5938static u64 rbd_v2_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5939{
5940	struct ceph_snap_context *snapc = rbd_dev->header.snapc;
5941	u32 which;
5942	bool found = false;
5943	u64 snap_id;
5944
5945	for (which = 0; !found && which < snapc->num_snaps; which++) {
5946		const char *snap_name;
5947
5948		snap_id = snapc->snaps[which];
5949		snap_name = rbd_dev_v2_snap_name(rbd_dev, snap_id);
5950		if (IS_ERR(snap_name)) {
5951			/* ignore no-longer existing snapshots */
5952			if (PTR_ERR(snap_name) == -ENOENT)
5953				continue;
5954			else
5955				break;
5956		}
5957		found = !strcmp(name, snap_name);
5958		kfree(snap_name);
5959	}
5960	return found ? snap_id : CEPH_NOSNAP;
5961}
5962
5963/*
5964 * Assumes name is never RBD_SNAP_HEAD_NAME; returns CEPH_NOSNAP if
5965 * no snapshot by that name is found, or if an error occurs.
5966 */
5967static u64 rbd_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
5968{
5969	if (rbd_dev->image_format == 1)
5970		return rbd_v1_snap_id_by_name(rbd_dev, name);
5971
5972	return rbd_v2_snap_id_by_name(rbd_dev, name);
5973}
5974
5975/*
5976 * An image being mapped will have everything but the snap id.
5977 */
5978static int rbd_spec_fill_snap_id(struct rbd_device *rbd_dev)
5979{
5980	struct rbd_spec *spec = rbd_dev->spec;
5981
5982	rbd_assert(spec->pool_id != CEPH_NOPOOL && spec->pool_name);
5983	rbd_assert(spec->image_id && spec->image_name);
5984	rbd_assert(spec->snap_name);
5985
5986	if (strcmp(spec->snap_name, RBD_SNAP_HEAD_NAME)) {
5987		u64 snap_id;
5988
5989		snap_id = rbd_snap_id_by_name(rbd_dev, spec->snap_name);
5990		if (snap_id == CEPH_NOSNAP)
5991			return -ENOENT;
5992
5993		spec->snap_id = snap_id;
5994	} else {
5995		spec->snap_id = CEPH_NOSNAP;
5996	}
5997
5998	return 0;
5999}
6000
6001/*
6002 * A parent image will have all ids but none of the names.
6003 *
6004 * All names in an rbd spec are dynamically allocated.  It's OK if we
6005 * can't figure out the name for an image id.
6006 */
6007static int rbd_spec_fill_names(struct rbd_device *rbd_dev)
6008{
6009	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
6010	struct rbd_spec *spec = rbd_dev->spec;
6011	const char *pool_name;
6012	const char *image_name;
6013	const char *snap_name;
6014	int ret;
6015
6016	rbd_assert(spec->pool_id != CEPH_NOPOOL);
6017	rbd_assert(spec->image_id);
6018	rbd_assert(spec->snap_id != CEPH_NOSNAP);
6019
6020	/* Get the pool name; we have to make our own copy of this */
6021
6022	pool_name = ceph_pg_pool_name_by_id(osdc->osdmap, spec->pool_id);
6023	if (!pool_name) {
6024		rbd_warn(rbd_dev, "no pool with id %llu", spec->pool_id);
6025		return -EIO;
6026	}
6027	pool_name = kstrdup(pool_name, GFP_KERNEL);
6028	if (!pool_name)
6029		return -ENOMEM;
6030
6031	/* Fetch the image name; tolerate failure here */
6032
6033	image_name = rbd_dev_image_name(rbd_dev);
6034	if (!image_name)
6035		rbd_warn(rbd_dev, "unable to get image name");
6036
6037	/* Fetch the snapshot name */
6038
6039	snap_name = rbd_snap_name(rbd_dev, spec->snap_id);
6040	if (IS_ERR(snap_name)) {
6041		ret = PTR_ERR(snap_name);
6042		goto out_err;
6043	}
6044
6045	spec->pool_name = pool_name;
6046	spec->image_name = image_name;
6047	spec->snap_name = snap_name;
6048
6049	return 0;
6050
6051out_err:
6052	kfree(image_name);
6053	kfree(pool_name);
6054	return ret;
6055}
6056
6057static int rbd_dev_v2_snap_context(struct rbd_device *rbd_dev,
6058				   struct ceph_snap_context **psnapc)
6059{
6060	size_t size;
6061	int ret;
6062	void *reply_buf;
6063	void *p;
6064	void *end;
6065	u64 seq;
6066	u32 snap_count;
6067	struct ceph_snap_context *snapc;
6068	u32 i;
6069
6070	/*
6071	 * We'll need room for the seq value (maximum snapshot id),
6072	 * snapshot count, and array of that many snapshot ids.
6073	 * For now we have a fixed upper limit on the number we're
6074	 * prepared to receive.
6075	 */
6076	size = sizeof (__le64) + sizeof (__le32) +
6077			RBD_MAX_SNAP_COUNT * sizeof (__le64);
6078	reply_buf = kzalloc(size, GFP_KERNEL);
6079	if (!reply_buf)
6080		return -ENOMEM;
6081
6082	ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
6083				  &rbd_dev->header_oloc, "get_snapcontext",
6084				  NULL, 0, reply_buf, size);
6085	dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6086	if (ret < 0)
6087		goto out;
6088
6089	p = reply_buf;
6090	end = reply_buf + ret;
6091	ret = -ERANGE;
6092	ceph_decode_64_safe(&p, end, seq, out);
6093	ceph_decode_32_safe(&p, end, snap_count, out);
6094
6095	/*
6096	 * Make sure the reported number of snapshot ids wouldn't go
6097	 * beyond the end of our buffer.  But before checking that,
6098	 * make sure the computed size of the snapshot context we
6099	 * allocate is representable in a size_t.
6100	 */
6101	if (snap_count > (SIZE_MAX - sizeof (struct ceph_snap_context))
6102				 / sizeof (u64)) {
6103		ret = -EINVAL;
6104		goto out;
6105	}
6106	if (!ceph_has_room(&p, end, snap_count * sizeof (__le64)))
6107		goto out;
6108	ret = 0;
6109
6110	snapc = ceph_create_snap_context(snap_count, GFP_KERNEL);
6111	if (!snapc) {
6112		ret = -ENOMEM;
6113		goto out;
6114	}
6115	snapc->seq = seq;
6116	for (i = 0; i < snap_count; i++)
6117		snapc->snaps[i] = ceph_decode_64(&p);
6118
6119	*psnapc = snapc;
6120	dout("  snap context seq = %llu, snap_count = %u\n",
6121		(unsigned long long)seq, (unsigned int)snap_count);
6122out:
6123	kfree(reply_buf);
6124
6125	return ret;
6126}
6127
6128static const char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev,
6129					u64 snap_id)
6130{
6131	size_t size;
6132	void *reply_buf;
6133	__le64 snapid;
6134	int ret;
6135	void *p;
6136	void *end;
6137	char *snap_name;
6138
6139	size = sizeof (__le32) + RBD_MAX_SNAP_NAME_LEN;
6140	reply_buf = kmalloc(size, GFP_KERNEL);
6141	if (!reply_buf)
6142		return ERR_PTR(-ENOMEM);
6143
6144	snapid = cpu_to_le64(snap_id);
6145	ret = rbd_obj_method_sync(rbd_dev, &rbd_dev->header_oid,
6146				  &rbd_dev->header_oloc, "get_snapshot_name",
6147				  &snapid, sizeof(snapid), reply_buf, size);
6148	dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6149	if (ret < 0) {
6150		snap_name = ERR_PTR(ret);
6151		goto out;
6152	}
6153
6154	p = reply_buf;
6155	end = reply_buf + ret;
6156	snap_name = ceph_extract_encoded_string(&p, end, NULL, GFP_KERNEL);
6157	if (IS_ERR(snap_name))
6158		goto out;
6159
6160	dout("  snap_id 0x%016llx snap_name = %s\n",
6161		(unsigned long long)snap_id, snap_name);
6162out:
6163	kfree(reply_buf);
6164
6165	return snap_name;
6166}
6167
6168static int rbd_dev_v2_header_info(struct rbd_device *rbd_dev,
6169				  struct rbd_image_header *header,
6170				  bool first_time)
6171{
6172	int ret;
6173
6174	ret = _rbd_dev_v2_snap_size(rbd_dev, CEPH_NOSNAP,
6175				    first_time ? &header->obj_order : NULL,
6176				    &header->image_size);
6177	if (ret)
6178		return ret;
6179
6180	if (first_time) {
6181		ret = rbd_dev_v2_header_onetime(rbd_dev, header);
6182		if (ret)
6183			return ret;
6184	}
6185
6186	ret = rbd_dev_v2_snap_context(rbd_dev, &header->snapc);
6187	if (ret)
6188		return ret;
6189
6190	return 0;
6191}
6192
6193static int rbd_dev_header_info(struct rbd_device *rbd_dev,
6194			       struct rbd_image_header *header,
6195			       bool first_time)
6196{
6197	rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6198	rbd_assert(!header->object_prefix && !header->snapc);
6199
6200	if (rbd_dev->image_format == 1)
6201		return rbd_dev_v1_header_info(rbd_dev, header, first_time);
6202
6203	return rbd_dev_v2_header_info(rbd_dev, header, first_time);
6204}
6205
6206/*
6207 * Skips over white space at *buf, and updates *buf to point to the
6208 * first found non-space character (if any). Returns the length of
6209 * the token (string of non-white space characters) found.  Note
6210 * that *buf must be terminated with '\0'.
6211 */
6212static inline size_t next_token(const char **buf)
6213{
6214        /*
6215        * These are the characters that produce nonzero for
6216        * isspace() in the "C" and "POSIX" locales.
6217        */
6218	static const char spaces[] = " \f\n\r\t\v";
6219
6220        *buf += strspn(*buf, spaces);	/* Find start of token */
6221
6222	return strcspn(*buf, spaces);   /* Return token length */
6223}
6224
6225/*
6226 * Finds the next token in *buf, dynamically allocates a buffer big
6227 * enough to hold a copy of it, and copies the token into the new
6228 * buffer.  The copy is guaranteed to be terminated with '\0'.  Note
6229 * that a duplicate buffer is created even for a zero-length token.
6230 *
6231 * Returns a pointer to the newly-allocated duplicate, or a null
6232 * pointer if memory for the duplicate was not available.  If
6233 * the lenp argument is a non-null pointer, the length of the token
6234 * (not including the '\0') is returned in *lenp.
6235 *
6236 * If successful, the *buf pointer will be updated to point beyond
6237 * the end of the found token.
6238 *
6239 * Note: uses GFP_KERNEL for allocation.
6240 */
6241static inline char *dup_token(const char **buf, size_t *lenp)
6242{
6243	char *dup;
6244	size_t len;
6245
6246	len = next_token(buf);
6247	dup = kmemdup(*buf, len + 1, GFP_KERNEL);
6248	if (!dup)
6249		return NULL;
6250	*(dup + len) = '\0';
6251	*buf += len;
6252
6253	if (lenp)
6254		*lenp = len;
6255
6256	return dup;
6257}
6258
6259static int rbd_parse_param(struct fs_parameter *param,
6260			    struct rbd_parse_opts_ctx *pctx)
6261{
6262	struct rbd_options *opt = pctx->opts;
6263	struct fs_parse_result result;
6264	struct p_log log = {.prefix = "rbd"};
6265	int token, ret;
6266
6267	ret = ceph_parse_param(param, pctx->copts, NULL);
6268	if (ret != -ENOPARAM)
6269		return ret;
6270
6271	token = __fs_parse(&log, rbd_parameters, param, &result);
6272	dout("%s fs_parse '%s' token %d\n", __func__, param->key, token);
6273	if (token < 0) {
6274		if (token == -ENOPARAM)
6275			return inval_plog(&log, "Unknown parameter '%s'",
6276					  param->key);
6277		return token;
6278	}
6279
6280	switch (token) {
6281	case Opt_queue_depth:
6282		if (result.uint_32 < 1)
6283			goto out_of_range;
6284		opt->queue_depth = result.uint_32;
6285		break;
6286	case Opt_alloc_size:
6287		if (result.uint_32 < SECTOR_SIZE)
6288			goto out_of_range;
6289		if (!is_power_of_2(result.uint_32))
6290			return inval_plog(&log, "alloc_size must be a power of 2");
6291		opt->alloc_size = result.uint_32;
6292		break;
6293	case Opt_lock_timeout:
6294		/* 0 is "wait forever" (i.e. infinite timeout) */
6295		if (result.uint_32 > INT_MAX / 1000)
6296			goto out_of_range;
6297		opt->lock_timeout = msecs_to_jiffies(result.uint_32 * 1000);
6298		break;
6299	case Opt_pool_ns:
6300		kfree(pctx->spec->pool_ns);
6301		pctx->spec->pool_ns = param->string;
6302		param->string = NULL;
6303		break;
6304	case Opt_compression_hint:
6305		switch (result.uint_32) {
6306		case Opt_compression_hint_none:
6307			opt->alloc_hint_flags &=
6308			    ~(CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE |
6309			      CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE);
6310			break;
6311		case Opt_compression_hint_compressible:
6312			opt->alloc_hint_flags |=
6313			    CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE;
6314			opt->alloc_hint_flags &=
6315			    ~CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE;
6316			break;
6317		case Opt_compression_hint_incompressible:
6318			opt->alloc_hint_flags |=
6319			    CEPH_OSD_ALLOC_HINT_FLAG_INCOMPRESSIBLE;
6320			opt->alloc_hint_flags &=
6321			    ~CEPH_OSD_ALLOC_HINT_FLAG_COMPRESSIBLE;
6322			break;
6323		default:
6324			BUG();
6325		}
6326		break;
6327	case Opt_read_only:
6328		opt->read_only = true;
6329		break;
6330	case Opt_read_write:
6331		opt->read_only = false;
6332		break;
6333	case Opt_lock_on_read:
6334		opt->lock_on_read = true;
6335		break;
6336	case Opt_exclusive:
6337		opt->exclusive = true;
6338		break;
6339	case Opt_notrim:
6340		opt->trim = false;
6341		break;
6342	default:
6343		BUG();
6344	}
6345
6346	return 0;
6347
6348out_of_range:
6349	return inval_plog(&log, "%s out of range", param->key);
6350}
6351
6352/*
6353 * This duplicates most of generic_parse_monolithic(), untying it from
6354 * fs_context and skipping standard superblock and security options.
6355 */
6356static int rbd_parse_options(char *options, struct rbd_parse_opts_ctx *pctx)
6357{
6358	char *key;
6359	int ret = 0;
6360
6361	dout("%s '%s'\n", __func__, options);
6362	while ((key = strsep(&options, ",")) != NULL) {
6363		if (*key) {
6364			struct fs_parameter param = {
6365				.key	= key,
6366				.type	= fs_value_is_flag,
6367			};
6368			char *value = strchr(key, '=');
6369			size_t v_len = 0;
6370
6371			if (value) {
6372				if (value == key)
6373					continue;
6374				*value++ = 0;
6375				v_len = strlen(value);
6376				param.string = kmemdup_nul(value, v_len,
6377							   GFP_KERNEL);
6378				if (!param.string)
6379					return -ENOMEM;
6380				param.type = fs_value_is_string;
6381			}
6382			param.size = v_len;
6383
6384			ret = rbd_parse_param(&param, pctx);
6385			kfree(param.string);
6386			if (ret)
6387				break;
6388		}
6389	}
6390
6391	return ret;
6392}
6393
6394/*
6395 * Parse the options provided for an "rbd add" (i.e., rbd image
6396 * mapping) request.  These arrive via a write to /sys/bus/rbd/add,
6397 * and the data written is passed here via a NUL-terminated buffer.
6398 * Returns 0 if successful or an error code otherwise.
6399 *
6400 * The information extracted from these options is recorded in
6401 * the other parameters which return dynamically-allocated
6402 * structures:
6403 *  ceph_opts
6404 *      The address of a pointer that will refer to a ceph options
6405 *      structure.  Caller must release the returned pointer using
6406 *      ceph_destroy_options() when it is no longer needed.
6407 *  rbd_opts
6408 *	Address of an rbd options pointer.  Fully initialized by
6409 *	this function; caller must release with kfree().
6410 *  spec
6411 *	Address of an rbd image specification pointer.  Fully
6412 *	initialized by this function based on parsed options.
6413 *	Caller must release with rbd_spec_put().
6414 *
6415 * The options passed take this form:
6416 *  <mon_addrs> <options> <pool_name> <image_name> [<snap_id>]
6417 * where:
6418 *  <mon_addrs>
6419 *      A comma-separated list of one or more monitor addresses.
6420 *      A monitor address is an ip address, optionally followed
6421 *      by a port number (separated by a colon).
6422 *        I.e.:  ip1[:port1][,ip2[:port2]...]
6423 *  <options>
6424 *      A comma-separated list of ceph and/or rbd options.
6425 *  <pool_name>
6426 *      The name of the rados pool containing the rbd image.
6427 *  <image_name>
6428 *      The name of the image in that pool to map.
6429 *  <snap_id>
6430 *      An optional snapshot id.  If provided, the mapping will
6431 *      present data from the image at the time that snapshot was
6432 *      created.  The image head is used if no snapshot id is
6433 *      provided.  Snapshot mappings are always read-only.
6434 */
6435static int rbd_add_parse_args(const char *buf,
6436				struct ceph_options **ceph_opts,
6437				struct rbd_options **opts,
6438				struct rbd_spec **rbd_spec)
6439{
6440	size_t len;
6441	char *options;
6442	const char *mon_addrs;
6443	char *snap_name;
6444	size_t mon_addrs_size;
6445	struct rbd_parse_opts_ctx pctx = { 0 };
6446	int ret;
6447
6448	/* The first four tokens are required */
6449
6450	len = next_token(&buf);
6451	if (!len) {
6452		rbd_warn(NULL, "no monitor address(es) provided");
6453		return -EINVAL;
6454	}
6455	mon_addrs = buf;
6456	mon_addrs_size = len;
6457	buf += len;
6458
6459	ret = -EINVAL;
6460	options = dup_token(&buf, NULL);
6461	if (!options)
6462		return -ENOMEM;
6463	if (!*options) {
6464		rbd_warn(NULL, "no options provided");
6465		goto out_err;
6466	}
6467
6468	pctx.spec = rbd_spec_alloc();
6469	if (!pctx.spec)
6470		goto out_mem;
6471
6472	pctx.spec->pool_name = dup_token(&buf, NULL);
6473	if (!pctx.spec->pool_name)
6474		goto out_mem;
6475	if (!*pctx.spec->pool_name) {
6476		rbd_warn(NULL, "no pool name provided");
6477		goto out_err;
6478	}
6479
6480	pctx.spec->image_name = dup_token(&buf, NULL);
6481	if (!pctx.spec->image_name)
6482		goto out_mem;
6483	if (!*pctx.spec->image_name) {
6484		rbd_warn(NULL, "no image name provided");
6485		goto out_err;
6486	}
6487
6488	/*
6489	 * Snapshot name is optional; default is to use "-"
6490	 * (indicating the head/no snapshot).
6491	 */
6492	len = next_token(&buf);
6493	if (!len) {
6494		buf = RBD_SNAP_HEAD_NAME; /* No snapshot supplied */
6495		len = sizeof (RBD_SNAP_HEAD_NAME) - 1;
6496	} else if (len > RBD_MAX_SNAP_NAME_LEN) {
6497		ret = -ENAMETOOLONG;
6498		goto out_err;
6499	}
6500	snap_name = kmemdup(buf, len + 1, GFP_KERNEL);
6501	if (!snap_name)
6502		goto out_mem;
6503	*(snap_name + len) = '\0';
6504	pctx.spec->snap_name = snap_name;
6505
6506	pctx.copts = ceph_alloc_options();
6507	if (!pctx.copts)
6508		goto out_mem;
6509
6510	/* Initialize all rbd options to the defaults */
6511
6512	pctx.opts = kzalloc(sizeof(*pctx.opts), GFP_KERNEL);
6513	if (!pctx.opts)
6514		goto out_mem;
6515
6516	pctx.opts->read_only = RBD_READ_ONLY_DEFAULT;
6517	pctx.opts->queue_depth = RBD_QUEUE_DEPTH_DEFAULT;
6518	pctx.opts->alloc_size = RBD_ALLOC_SIZE_DEFAULT;
6519	pctx.opts->lock_timeout = RBD_LOCK_TIMEOUT_DEFAULT;
6520	pctx.opts->lock_on_read = RBD_LOCK_ON_READ_DEFAULT;
6521	pctx.opts->exclusive = RBD_EXCLUSIVE_DEFAULT;
6522	pctx.opts->trim = RBD_TRIM_DEFAULT;
6523
6524	ret = ceph_parse_mon_ips(mon_addrs, mon_addrs_size, pctx.copts, NULL,
6525				 ',');
6526	if (ret)
6527		goto out_err;
6528
6529	ret = rbd_parse_options(options, &pctx);
6530	if (ret)
6531		goto out_err;
6532
6533	*ceph_opts = pctx.copts;
6534	*opts = pctx.opts;
6535	*rbd_spec = pctx.spec;
6536	kfree(options);
6537	return 0;
6538
6539out_mem:
6540	ret = -ENOMEM;
6541out_err:
6542	kfree(pctx.opts);
6543	ceph_destroy_options(pctx.copts);
6544	rbd_spec_put(pctx.spec);
6545	kfree(options);
6546	return ret;
6547}
6548
6549static void rbd_dev_image_unlock(struct rbd_device *rbd_dev)
6550{
6551	down_write(&rbd_dev->lock_rwsem);
6552	if (__rbd_is_lock_owner(rbd_dev))
6553		__rbd_release_lock(rbd_dev);
6554	up_write(&rbd_dev->lock_rwsem);
6555}
6556
6557/*
6558 * If the wait is interrupted, an error is returned even if the lock
6559 * was successfully acquired.  rbd_dev_image_unlock() will release it
6560 * if needed.
6561 */
6562static int rbd_add_acquire_lock(struct rbd_device *rbd_dev)
6563{
6564	long ret;
6565
6566	if (!(rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK)) {
6567		if (!rbd_dev->opts->exclusive && !rbd_dev->opts->lock_on_read)
6568			return 0;
6569
6570		rbd_warn(rbd_dev, "exclusive-lock feature is not enabled");
6571		return -EINVAL;
6572	}
6573
6574	if (rbd_is_ro(rbd_dev))
6575		return 0;
6576
6577	rbd_assert(!rbd_is_lock_owner(rbd_dev));
6578	queue_delayed_work(rbd_dev->task_wq, &rbd_dev->lock_dwork, 0);
6579	ret = wait_for_completion_killable_timeout(&rbd_dev->acquire_wait,
6580			    ceph_timeout_jiffies(rbd_dev->opts->lock_timeout));
6581	if (ret > 0) {
6582		ret = rbd_dev->acquire_err;
6583	} else {
6584		cancel_delayed_work_sync(&rbd_dev->lock_dwork);
6585		if (!ret)
6586			ret = -ETIMEDOUT;
6587
6588		rbd_warn(rbd_dev, "failed to acquire lock: %ld", ret);
6589	}
6590	if (ret)
6591		return ret;
6592
6593	return 0;
6594}
6595
6596/*
6597 * An rbd format 2 image has a unique identifier, distinct from the
6598 * name given to it by the user.  Internally, that identifier is
6599 * what's used to specify the names of objects related to the image.
6600 *
6601 * A special "rbd id" object is used to map an rbd image name to its
6602 * id.  If that object doesn't exist, then there is no v2 rbd image
6603 * with the supplied name.
6604 *
6605 * This function will record the given rbd_dev's image_id field if
6606 * it can be determined, and in that case will return 0.  If any
6607 * errors occur a negative errno will be returned and the rbd_dev's
6608 * image_id field will be unchanged (and should be NULL).
6609 */
6610static int rbd_dev_image_id(struct rbd_device *rbd_dev)
6611{
6612	int ret;
6613	size_t size;
6614	CEPH_DEFINE_OID_ONSTACK(oid);
6615	void *response;
6616	char *image_id;
6617
6618	/*
6619	 * When probing a parent image, the image id is already
6620	 * known (and the image name likely is not).  There's no
6621	 * need to fetch the image id again in this case.  We
6622	 * do still need to set the image format though.
6623	 */
6624	if (rbd_dev->spec->image_id) {
6625		rbd_dev->image_format = *rbd_dev->spec->image_id ? 2 : 1;
6626
6627		return 0;
6628	}
6629
6630	/*
6631	 * First, see if the format 2 image id file exists, and if
6632	 * so, get the image's persistent id from it.
6633	 */
6634	ret = ceph_oid_aprintf(&oid, GFP_KERNEL, "%s%s", RBD_ID_PREFIX,
6635			       rbd_dev->spec->image_name);
6636	if (ret)
6637		return ret;
6638
6639	dout("rbd id object name is %s\n", oid.name);
6640
6641	/* Response will be an encoded string, which includes a length */
6642	size = sizeof (__le32) + RBD_IMAGE_ID_LEN_MAX;
6643	response = kzalloc(size, GFP_NOIO);
6644	if (!response) {
6645		ret = -ENOMEM;
6646		goto out;
6647	}
6648
6649	/* If it doesn't exist we'll assume it's a format 1 image */
6650
6651	ret = rbd_obj_method_sync(rbd_dev, &oid, &rbd_dev->header_oloc,
6652				  "get_id", NULL, 0,
6653				  response, size);
6654	dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
6655	if (ret == -ENOENT) {
6656		image_id = kstrdup("", GFP_KERNEL);
6657		ret = image_id ? 0 : -ENOMEM;
6658		if (!ret)
6659			rbd_dev->image_format = 1;
6660	} else if (ret >= 0) {
6661		void *p = response;
6662
6663		image_id = ceph_extract_encoded_string(&p, p + ret,
6664						NULL, GFP_NOIO);
6665		ret = PTR_ERR_OR_ZERO(image_id);
6666		if (!ret)
6667			rbd_dev->image_format = 2;
6668	}
6669
6670	if (!ret) {
6671		rbd_dev->spec->image_id = image_id;
6672		dout("image_id is %s\n", image_id);
6673	}
6674out:
6675	kfree(response);
6676	ceph_oid_destroy(&oid);
6677	return ret;
6678}
6679
6680/*
6681 * Undo whatever state changes are made by v1 or v2 header info
6682 * call.
6683 */
6684static void rbd_dev_unprobe(struct rbd_device *rbd_dev)
6685{
6686	rbd_dev_parent_put(rbd_dev);
6687	rbd_object_map_free(rbd_dev);
6688	rbd_dev_mapping_clear(rbd_dev);
6689
6690	/* Free dynamic fields from the header, then zero it out */
6691
6692	rbd_image_header_cleanup(&rbd_dev->header);
6693}
6694
6695static int rbd_dev_v2_header_onetime(struct rbd_device *rbd_dev,
6696				     struct rbd_image_header *header)
6697{
6698	int ret;
6699
6700	ret = rbd_dev_v2_object_prefix(rbd_dev, &header->object_prefix);
6701	if (ret)
6702		return ret;
6703
6704	/*
6705	 * Get the and check features for the image.  Currently the
6706	 * features are assumed to never change.
6707	 */
6708	ret = _rbd_dev_v2_snap_features(rbd_dev, CEPH_NOSNAP,
6709					rbd_is_ro(rbd_dev), &header->features);
6710	if (ret)
6711		return ret;
6712
6713	/* If the image supports fancy striping, get its parameters */
6714
6715	if (header->features & RBD_FEATURE_STRIPINGV2) {
6716		ret = rbd_dev_v2_striping_info(rbd_dev, &header->stripe_unit,
6717					       &header->stripe_count);
6718		if (ret)
6719			return ret;
6720	}
6721
6722	if (header->features & RBD_FEATURE_DATA_POOL) {
6723		ret = rbd_dev_v2_data_pool(rbd_dev, &header->data_pool_id);
6724		if (ret)
6725			return ret;
6726	}
6727
6728	return 0;
6729}
6730
6731/*
6732 * @depth is rbd_dev_image_probe() -> rbd_dev_probe_parent() ->
6733 * rbd_dev_image_probe() recursion depth, which means it's also the
6734 * length of the already discovered part of the parent chain.
6735 */
6736static int rbd_dev_probe_parent(struct rbd_device *rbd_dev, int depth)
6737{
6738	struct rbd_device *parent = NULL;
6739	int ret;
6740
6741	if (!rbd_dev->parent_spec)
6742		return 0;
6743
6744	if (++depth > RBD_MAX_PARENT_CHAIN_LEN) {
6745		pr_info("parent chain is too long (%d)\n", depth);
6746		ret = -EINVAL;
6747		goto out_err;
6748	}
6749
6750	parent = __rbd_dev_create(rbd_dev->parent_spec);
6751	if (!parent) {
6752		ret = -ENOMEM;
6753		goto out_err;
6754	}
6755
6756	/*
6757	 * Images related by parent/child relationships always share
6758	 * rbd_client and spec/parent_spec, so bump their refcounts.
6759	 */
6760	parent->rbd_client = __rbd_get_client(rbd_dev->rbd_client);
6761	parent->spec = rbd_spec_get(rbd_dev->parent_spec);
6762
6763	__set_bit(RBD_DEV_FLAG_READONLY, &parent->flags);
6764
6765	ret = rbd_dev_image_probe(parent, depth);
6766	if (ret < 0)
6767		goto out_err;
6768
6769	rbd_dev->parent = parent;
6770	atomic_set(&rbd_dev->parent_ref, 1);
6771	return 0;
6772
6773out_err:
6774	rbd_dev_unparent(rbd_dev);
6775	rbd_dev_destroy(parent);
6776	return ret;
6777}
6778
6779static void rbd_dev_device_release(struct rbd_device *rbd_dev)
6780{
6781	clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
6782	rbd_free_disk(rbd_dev);
6783	if (!single_major)
6784		unregister_blkdev(rbd_dev->major, rbd_dev->name);
6785}
6786
6787/*
6788 * rbd_dev->header_rwsem must be locked for write and will be unlocked
6789 * upon return.
6790 */
6791static int rbd_dev_device_setup(struct rbd_device *rbd_dev)
6792{
6793	int ret;
6794
6795	/* Record our major and minor device numbers. */
6796
6797	if (!single_major) {
6798		ret = register_blkdev(0, rbd_dev->name);
6799		if (ret < 0)
6800			goto err_out_unlock;
6801
6802		rbd_dev->major = ret;
6803		rbd_dev->minor = 0;
6804	} else {
6805		rbd_dev->major = rbd_major;
6806		rbd_dev->minor = rbd_dev_id_to_minor(rbd_dev->dev_id);
6807	}
6808
6809	/* Set up the blkdev mapping. */
6810
6811	ret = rbd_init_disk(rbd_dev);
6812	if (ret)
6813		goto err_out_blkdev;
6814
6815	set_capacity(rbd_dev->disk, rbd_dev->mapping.size / SECTOR_SIZE);
6816	set_disk_ro(rbd_dev->disk, rbd_is_ro(rbd_dev));
6817
6818	ret = dev_set_name(&rbd_dev->dev, "%d", rbd_dev->dev_id);
6819	if (ret)
6820		goto err_out_disk;
6821
6822	set_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
6823	up_write(&rbd_dev->header_rwsem);
6824	return 0;
6825
6826err_out_disk:
6827	rbd_free_disk(rbd_dev);
6828err_out_blkdev:
6829	if (!single_major)
6830		unregister_blkdev(rbd_dev->major, rbd_dev->name);
6831err_out_unlock:
6832	up_write(&rbd_dev->header_rwsem);
6833	return ret;
6834}
6835
6836static int rbd_dev_header_name(struct rbd_device *rbd_dev)
6837{
6838	struct rbd_spec *spec = rbd_dev->spec;
6839	int ret;
6840
6841	/* Record the header object name for this rbd image. */
6842
6843	rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6844	if (rbd_dev->image_format == 1)
6845		ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
6846				       spec->image_name, RBD_SUFFIX);
6847	else
6848		ret = ceph_oid_aprintf(&rbd_dev->header_oid, GFP_KERNEL, "%s%s",
6849				       RBD_HEADER_PREFIX, spec->image_id);
6850
6851	return ret;
6852}
6853
6854static void rbd_print_dne(struct rbd_device *rbd_dev, bool is_snap)
6855{
6856	if (!is_snap) {
6857		pr_info("image %s/%s%s%s does not exist\n",
6858			rbd_dev->spec->pool_name,
6859			rbd_dev->spec->pool_ns ?: "",
6860			rbd_dev->spec->pool_ns ? "/" : "",
6861			rbd_dev->spec->image_name);
6862	} else {
6863		pr_info("snap %s/%s%s%s@%s does not exist\n",
6864			rbd_dev->spec->pool_name,
6865			rbd_dev->spec->pool_ns ?: "",
6866			rbd_dev->spec->pool_ns ? "/" : "",
6867			rbd_dev->spec->image_name,
6868			rbd_dev->spec->snap_name);
6869	}
6870}
6871
6872static void rbd_dev_image_release(struct rbd_device *rbd_dev)
6873{
6874	if (!rbd_is_ro(rbd_dev))
6875		rbd_unregister_watch(rbd_dev);
6876
6877	rbd_dev_unprobe(rbd_dev);
6878	rbd_dev->image_format = 0;
6879	kfree(rbd_dev->spec->image_id);
6880	rbd_dev->spec->image_id = NULL;
6881}
6882
6883/*
6884 * Probe for the existence of the header object for the given rbd
6885 * device.  If this image is the one being mapped (i.e., not a
6886 * parent), initiate a watch on its header object before using that
6887 * object to get detailed information about the rbd image.
6888 *
6889 * On success, returns with header_rwsem held for write if called
6890 * with @depth == 0.
6891 */
6892static int rbd_dev_image_probe(struct rbd_device *rbd_dev, int depth)
6893{
6894	bool need_watch = !rbd_is_ro(rbd_dev);
6895	int ret;
6896
6897	/*
6898	 * Get the id from the image id object.  Unless there's an
6899	 * error, rbd_dev->spec->image_id will be filled in with
6900	 * a dynamically-allocated string, and rbd_dev->image_format
6901	 * will be set to either 1 or 2.
6902	 */
6903	ret = rbd_dev_image_id(rbd_dev);
6904	if (ret)
6905		return ret;
6906
6907	ret = rbd_dev_header_name(rbd_dev);
6908	if (ret)
6909		goto err_out_format;
6910
6911	if (need_watch) {
6912		ret = rbd_register_watch(rbd_dev);
6913		if (ret) {
6914			if (ret == -ENOENT)
6915				rbd_print_dne(rbd_dev, false);
6916			goto err_out_format;
6917		}
6918	}
6919
6920	if (!depth)
6921		down_write(&rbd_dev->header_rwsem);
6922
6923	ret = rbd_dev_header_info(rbd_dev, &rbd_dev->header, true);
6924	if (ret) {
6925		if (ret == -ENOENT && !need_watch)
6926			rbd_print_dne(rbd_dev, false);
6927		goto err_out_probe;
6928	}
6929
6930	rbd_init_layout(rbd_dev);
6931
6932	/*
6933	 * If this image is the one being mapped, we have pool name and
6934	 * id, image name and id, and snap name - need to fill snap id.
6935	 * Otherwise this is a parent image, identified by pool, image
6936	 * and snap ids - need to fill in names for those ids.
6937	 */
6938	if (!depth)
6939		ret = rbd_spec_fill_snap_id(rbd_dev);
6940	else
6941		ret = rbd_spec_fill_names(rbd_dev);
6942	if (ret) {
6943		if (ret == -ENOENT)
6944			rbd_print_dne(rbd_dev, true);
6945		goto err_out_probe;
6946	}
6947
6948	ret = rbd_dev_mapping_set(rbd_dev);
6949	if (ret)
6950		goto err_out_probe;
6951
6952	if (rbd_is_snap(rbd_dev) &&
6953	    (rbd_dev->header.features & RBD_FEATURE_OBJECT_MAP)) {
6954		ret = rbd_object_map_load(rbd_dev);
6955		if (ret)
6956			goto err_out_probe;
6957	}
6958
6959	if (rbd_dev->header.features & RBD_FEATURE_LAYERING) {
6960		ret = rbd_dev_setup_parent(rbd_dev);
6961		if (ret)
6962			goto err_out_probe;
6963	}
6964
6965	ret = rbd_dev_probe_parent(rbd_dev, depth);
6966	if (ret)
6967		goto err_out_probe;
6968
6969	dout("discovered format %u image, header name is %s\n",
6970		rbd_dev->image_format, rbd_dev->header_oid.name);
6971	return 0;
6972
6973err_out_probe:
6974	if (!depth)
6975		up_write(&rbd_dev->header_rwsem);
6976	if (need_watch)
6977		rbd_unregister_watch(rbd_dev);
6978	rbd_dev_unprobe(rbd_dev);
6979err_out_format:
6980	rbd_dev->image_format = 0;
6981	kfree(rbd_dev->spec->image_id);
6982	rbd_dev->spec->image_id = NULL;
6983	return ret;
6984}
6985
6986static void rbd_dev_update_header(struct rbd_device *rbd_dev,
6987				  struct rbd_image_header *header)
6988{
6989	rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
6990	rbd_assert(rbd_dev->header.object_prefix); /* !first_time */
6991
6992	if (rbd_dev->header.image_size != header->image_size) {
6993		rbd_dev->header.image_size = header->image_size;
6994
6995		if (!rbd_is_snap(rbd_dev)) {
6996			rbd_dev->mapping.size = header->image_size;
6997			rbd_dev_update_size(rbd_dev);
6998		}
6999	}
7000
7001	ceph_put_snap_context(rbd_dev->header.snapc);
7002	rbd_dev->header.snapc = header->snapc;
7003	header->snapc = NULL;
7004
7005	if (rbd_dev->image_format == 1) {
7006		kfree(rbd_dev->header.snap_names);
7007		rbd_dev->header.snap_names = header->snap_names;
7008		header->snap_names = NULL;
7009
7010		kfree(rbd_dev->header.snap_sizes);
7011		rbd_dev->header.snap_sizes = header->snap_sizes;
7012		header->snap_sizes = NULL;
7013	}
7014}
7015
7016static void rbd_dev_update_parent(struct rbd_device *rbd_dev,
7017				  struct parent_image_info *pii)
7018{
7019	if (pii->pool_id == CEPH_NOPOOL || !pii->has_overlap) {
7020		/*
7021		 * Either the parent never existed, or we have
7022		 * record of it but the image got flattened so it no
7023		 * longer has a parent.  When the parent of a
7024		 * layered image disappears we immediately set the
7025		 * overlap to 0.  The effect of this is that all new
7026		 * requests will be treated as if the image had no
7027		 * parent.
7028		 *
7029		 * If !pii.has_overlap, the parent image spec is not
7030		 * applicable.  It's there to avoid duplication in each
7031		 * snapshot record.
7032		 */
7033		if (rbd_dev->parent_overlap) {
7034			rbd_dev->parent_overlap = 0;
7035			rbd_dev_parent_put(rbd_dev);
7036			pr_info("%s: clone has been flattened\n",
7037				rbd_dev->disk->disk_name);
7038		}
7039	} else {
7040		rbd_assert(rbd_dev->parent_spec);
7041
7042		/*
7043		 * Update the parent overlap.  If it became zero, issue
7044		 * a warning as we will proceed as if there is no parent.
7045		 */
7046		if (!pii->overlap && rbd_dev->parent_overlap)
7047			rbd_warn(rbd_dev,
7048				 "clone has become standalone (overlap 0)");
7049		rbd_dev->parent_overlap = pii->overlap;
7050	}
7051}
7052
7053static int rbd_dev_refresh(struct rbd_device *rbd_dev)
7054{
7055	struct rbd_image_header	header = { 0 };
7056	struct parent_image_info pii = { 0 };
7057	int ret;
7058
7059	dout("%s rbd_dev %p\n", __func__, rbd_dev);
7060
7061	ret = rbd_dev_header_info(rbd_dev, &header, false);
7062	if (ret)
7063		goto out;
7064
7065	/*
7066	 * If there is a parent, see if it has disappeared due to the
7067	 * mapped image getting flattened.
7068	 */
7069	if (rbd_dev->parent) {
7070		ret = rbd_dev_v2_parent_info(rbd_dev, &pii);
7071		if (ret)
7072			goto out;
7073	}
7074
7075	down_write(&rbd_dev->header_rwsem);
7076	rbd_dev_update_header(rbd_dev, &header);
7077	if (rbd_dev->parent)
7078		rbd_dev_update_parent(rbd_dev, &pii);
7079	up_write(&rbd_dev->header_rwsem);
7080
7081out:
7082	rbd_parent_info_cleanup(&pii);
7083	rbd_image_header_cleanup(&header);
7084	return ret;
7085}
7086
7087static ssize_t do_rbd_add(const char *buf, size_t count)
7088{
7089	struct rbd_device *rbd_dev = NULL;
7090	struct ceph_options *ceph_opts = NULL;
7091	struct rbd_options *rbd_opts = NULL;
7092	struct rbd_spec *spec = NULL;
7093	struct rbd_client *rbdc;
7094	int rc;
7095
7096	if (!capable(CAP_SYS_ADMIN))
7097		return -EPERM;
7098
7099	if (!try_module_get(THIS_MODULE))
7100		return -ENODEV;
7101
7102	/* parse add command */
7103	rc = rbd_add_parse_args(buf, &ceph_opts, &rbd_opts, &spec);
7104	if (rc < 0)
7105		goto out;
7106
7107	rbdc = rbd_get_client(ceph_opts);
7108	if (IS_ERR(rbdc)) {
7109		rc = PTR_ERR(rbdc);
7110		goto err_out_args;
7111	}
7112
7113	/* pick the pool */
7114	rc = ceph_pg_poolid_by_name(rbdc->client->osdc.osdmap, spec->pool_name);
7115	if (rc < 0) {
7116		if (rc == -ENOENT)
7117			pr_info("pool %s does not exist\n", spec->pool_name);
7118		goto err_out_client;
7119	}
7120	spec->pool_id = (u64)rc;
7121
7122	rbd_dev = rbd_dev_create(rbdc, spec, rbd_opts);
7123	if (!rbd_dev) {
7124		rc = -ENOMEM;
7125		goto err_out_client;
7126	}
7127	rbdc = NULL;		/* rbd_dev now owns this */
7128	spec = NULL;		/* rbd_dev now owns this */
7129	rbd_opts = NULL;	/* rbd_dev now owns this */
7130
7131	/* if we are mapping a snapshot it will be a read-only mapping */
7132	if (rbd_dev->opts->read_only ||
7133	    strcmp(rbd_dev->spec->snap_name, RBD_SNAP_HEAD_NAME))
7134		__set_bit(RBD_DEV_FLAG_READONLY, &rbd_dev->flags);
7135
7136	rbd_dev->config_info = kstrdup(buf, GFP_KERNEL);
7137	if (!rbd_dev->config_info) {
7138		rc = -ENOMEM;
7139		goto err_out_rbd_dev;
7140	}
7141
7142	rc = rbd_dev_image_probe(rbd_dev, 0);
7143	if (rc < 0)
7144		goto err_out_rbd_dev;
7145
7146	if (rbd_dev->opts->alloc_size > rbd_dev->layout.object_size) {
7147		rbd_warn(rbd_dev, "alloc_size adjusted to %u",
7148			 rbd_dev->layout.object_size);
7149		rbd_dev->opts->alloc_size = rbd_dev->layout.object_size;
7150	}
7151
7152	rc = rbd_dev_device_setup(rbd_dev);
7153	if (rc)
7154		goto err_out_image_probe;
7155
7156	rc = rbd_add_acquire_lock(rbd_dev);
7157	if (rc)
7158		goto err_out_image_lock;
7159
7160	/* Everything's ready.  Announce the disk to the world. */
7161
7162	rc = device_add(&rbd_dev->dev);
7163	if (rc)
7164		goto err_out_image_lock;
7165
7166	rc = device_add_disk(&rbd_dev->dev, rbd_dev->disk, NULL);
7167	if (rc)
7168		goto err_out_cleanup_disk;
7169
7170	spin_lock(&rbd_dev_list_lock);
7171	list_add_tail(&rbd_dev->node, &rbd_dev_list);
7172	spin_unlock(&rbd_dev_list_lock);
7173
7174	pr_info("%s: capacity %llu features 0x%llx\n", rbd_dev->disk->disk_name,
7175		(unsigned long long)get_capacity(rbd_dev->disk) << SECTOR_SHIFT,
7176		rbd_dev->header.features);
7177	rc = count;
7178out:
7179	module_put(THIS_MODULE);
7180	return rc;
7181
7182err_out_cleanup_disk:
7183	rbd_free_disk(rbd_dev);
7184err_out_image_lock:
7185	rbd_dev_image_unlock(rbd_dev);
7186	rbd_dev_device_release(rbd_dev);
7187err_out_image_probe:
7188	rbd_dev_image_release(rbd_dev);
7189err_out_rbd_dev:
7190	rbd_dev_destroy(rbd_dev);
7191err_out_client:
7192	rbd_put_client(rbdc);
7193err_out_args:
7194	rbd_spec_put(spec);
7195	kfree(rbd_opts);
7196	goto out;
7197}
7198
7199static ssize_t add_store(const struct bus_type *bus, const char *buf, size_t count)
7200{
7201	if (single_major)
7202		return -EINVAL;
7203
7204	return do_rbd_add(buf, count);
7205}
7206
7207static ssize_t add_single_major_store(const struct bus_type *bus, const char *buf,
7208				      size_t count)
7209{
7210	return do_rbd_add(buf, count);
7211}
7212
7213static void rbd_dev_remove_parent(struct rbd_device *rbd_dev)
7214{
7215	while (rbd_dev->parent) {
7216		struct rbd_device *first = rbd_dev;
7217		struct rbd_device *second = first->parent;
7218		struct rbd_device *third;
7219
7220		/*
7221		 * Follow to the parent with no grandparent and
7222		 * remove it.
7223		 */
7224		while (second && (third = second->parent)) {
7225			first = second;
7226			second = third;
7227		}
7228		rbd_assert(second);
7229		rbd_dev_image_release(second);
7230		rbd_dev_destroy(second);
7231		first->parent = NULL;
7232		first->parent_overlap = 0;
7233
7234		rbd_assert(first->parent_spec);
7235		rbd_spec_put(first->parent_spec);
7236		first->parent_spec = NULL;
7237	}
7238}
7239
7240static ssize_t do_rbd_remove(const char *buf, size_t count)
7241{
7242	struct rbd_device *rbd_dev = NULL;
7243	int dev_id;
7244	char opt_buf[6];
7245	bool force = false;
7246	int ret;
7247
7248	if (!capable(CAP_SYS_ADMIN))
7249		return -EPERM;
7250
7251	dev_id = -1;
7252	opt_buf[0] = '\0';
7253	sscanf(buf, "%d %5s", &dev_id, opt_buf);
7254	if (dev_id < 0) {
7255		pr_err("dev_id out of range\n");
7256		return -EINVAL;
7257	}
7258	if (opt_buf[0] != '\0') {
7259		if (!strcmp(opt_buf, "force")) {
7260			force = true;
7261		} else {
7262			pr_err("bad remove option at '%s'\n", opt_buf);
7263			return -EINVAL;
7264		}
7265	}
7266
7267	ret = -ENOENT;
7268	spin_lock(&rbd_dev_list_lock);
7269	list_for_each_entry(rbd_dev, &rbd_dev_list, node) {
7270		if (rbd_dev->dev_id == dev_id) {
7271			ret = 0;
7272			break;
7273		}
7274	}
7275	if (!ret) {
7276		spin_lock_irq(&rbd_dev->lock);
7277		if (rbd_dev->open_count && !force)
7278			ret = -EBUSY;
7279		else if (test_and_set_bit(RBD_DEV_FLAG_REMOVING,
7280					  &rbd_dev->flags))
7281			ret = -EINPROGRESS;
7282		spin_unlock_irq(&rbd_dev->lock);
7283	}
7284	spin_unlock(&rbd_dev_list_lock);
7285	if (ret)
7286		return ret;
7287
7288	if (force) {
7289		/*
7290		 * Prevent new IO from being queued and wait for existing
7291		 * IO to complete/fail.
7292		 */
7293		unsigned int memflags = blk_mq_freeze_queue(rbd_dev->disk->queue);
7294
7295		blk_mark_disk_dead(rbd_dev->disk);
7296		blk_mq_unfreeze_queue(rbd_dev->disk->queue, memflags);
7297	}
7298
7299	del_gendisk(rbd_dev->disk);
7300	spin_lock(&rbd_dev_list_lock);
7301	list_del_init(&rbd_dev->node);
7302	spin_unlock(&rbd_dev_list_lock);
7303	device_del(&rbd_dev->dev);
7304
7305	rbd_dev_image_unlock(rbd_dev);
7306	rbd_dev_device_release(rbd_dev);
7307	rbd_dev_image_release(rbd_dev);
7308	rbd_dev_destroy(rbd_dev);
7309	return count;
7310}
7311
7312static ssize_t remove_store(const struct bus_type *bus, const char *buf, size_t count)
7313{
7314	if (single_major)
7315		return -EINVAL;
7316
7317	return do_rbd_remove(buf, count);
7318}
7319
7320static ssize_t remove_single_major_store(const struct bus_type *bus, const char *buf,
7321					 size_t count)
7322{
7323	return do_rbd_remove(buf, count);
7324}
7325
7326/*
7327 * create control files in sysfs
7328 * /sys/bus/rbd/...
7329 */
7330static int __init rbd_sysfs_init(void)
7331{
7332	int ret;
7333
7334	ret = device_register(&rbd_root_dev);
7335	if (ret < 0) {
7336		put_device(&rbd_root_dev);
7337		return ret;
7338	}
7339
7340	ret = bus_register(&rbd_bus_type);
7341	if (ret < 0)
7342		device_unregister(&rbd_root_dev);
7343
7344	return ret;
7345}
7346
7347static void __exit rbd_sysfs_cleanup(void)
7348{
7349	bus_unregister(&rbd_bus_type);
7350	device_unregister(&rbd_root_dev);
7351}
7352
7353static int __init rbd_slab_init(void)
7354{
7355	rbd_assert(!rbd_img_request_cache);
7356	rbd_img_request_cache = KMEM_CACHE(rbd_img_request, 0);
7357	if (!rbd_img_request_cache)
7358		return -ENOMEM;
7359
7360	rbd_assert(!rbd_obj_request_cache);
7361	rbd_obj_request_cache = KMEM_CACHE(rbd_obj_request, 0);
7362	if (!rbd_obj_request_cache)
7363		goto out_err;
7364
7365	return 0;
7366
7367out_err:
7368	kmem_cache_destroy(rbd_img_request_cache);
7369	rbd_img_request_cache = NULL;
7370	return -ENOMEM;
7371}
7372
7373static void rbd_slab_exit(void)
7374{
7375	rbd_assert(rbd_obj_request_cache);
7376	kmem_cache_destroy(rbd_obj_request_cache);
7377	rbd_obj_request_cache = NULL;
7378
7379	rbd_assert(rbd_img_request_cache);
7380	kmem_cache_destroy(rbd_img_request_cache);
7381	rbd_img_request_cache = NULL;
7382}
7383
7384static int __init rbd_init(void)
7385{
7386	int rc;
7387
7388	if (!libceph_compatible(NULL)) {
7389		rbd_warn(NULL, "libceph incompatibility (quitting)");
7390		return -EINVAL;
7391	}
7392
7393	rc = rbd_slab_init();
7394	if (rc)
7395		return rc;
7396
7397	/*
7398	 * The number of active work items is limited by the number of
7399	 * rbd devices * queue depth, so leave @max_active at default.
7400	 */
7401	rbd_wq = alloc_workqueue(RBD_DRV_NAME, WQ_MEM_RECLAIM | WQ_PERCPU, 0);
7402	if (!rbd_wq) {
7403		rc = -ENOMEM;
7404		goto err_out_slab;
7405	}
7406
7407	if (single_major) {
7408		rbd_major = register_blkdev(0, RBD_DRV_NAME);
7409		if (rbd_major < 0) {
7410			rc = rbd_major;
7411			goto err_out_wq;
7412		}
7413	}
7414
7415	rc = rbd_sysfs_init();
7416	if (rc)
7417		goto err_out_blkdev;
7418
7419	if (single_major)
7420		pr_info("loaded (major %d)\n", rbd_major);
7421	else
7422		pr_info("loaded\n");
7423
7424	return 0;
7425
7426err_out_blkdev:
7427	if (single_major)
7428		unregister_blkdev(rbd_major, RBD_DRV_NAME);
7429err_out_wq:
7430	destroy_workqueue(rbd_wq);
7431err_out_slab:
7432	rbd_slab_exit();
7433	return rc;
7434}
7435
7436static void __exit rbd_exit(void)
7437{
7438	ida_destroy(&rbd_dev_id_ida);
7439	rbd_sysfs_cleanup();
7440	if (single_major)
7441		unregister_blkdev(rbd_major, RBD_DRV_NAME);
7442	destroy_workqueue(rbd_wq);
7443	rbd_slab_exit();
7444}
7445
7446module_init(rbd_init);
7447module_exit(rbd_exit);
7448
7449MODULE_AUTHOR("Alex Elder <elder@inktank.com>");
7450MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
7451MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
7452/* following authorship retained from original osdblk.c */
7453MODULE_AUTHOR("Jeff Garzik <jeff@garzik.org>");
7454
7455MODULE_DESCRIPTION("RADOS Block Device (RBD) driver");
7456MODULE_LICENSE("GPL");