Merge branch 'for-linus' of git://git.kernel.org/pub/scm/linux/kernel/git/sage/ceph-client

Pull Ceph changes from Sage Weil: "On the RBD side, there is a conversion to blk-mq from Christoph, several long-standing bug fixes from Ilya, and some cleanup from Rickard Strandqvist. On the CephFS side there is a long list of fixes from Zheng, including improved session handling, a few IO path fixes, some dcache management correctness fixes, and several blocking while !TASK_RUNNING fixes. The core code gets a few cleanups and Chaitanya has added support for TCP_NODELAY (which has been used on the server side for ages but we somehow missed on the kernel client). There is also an update to MAINTAINERS to fix up some email addresses and reflect that Ilya and Zheng are doing most of the maintenance for RBD and CephFS these days. Do not be surprised to see a pull request come from one of them in the future if I am unavailable for some reason" * 'for-linus' of git://git.kernel.org/pub/scm/linux/kernel/git/sage/ceph-client: (27 commits) MAINTAINERS: update Ceph and RBD maintainers libceph: kfree() in put_osd() shouldn't depend on authorizer libceph: fix double __remove_osd() problem rbd: convert to blk-mq ceph: return error for traceless reply race ceph: fix dentry leaks ceph: re-send requests when MDS enters reconnecting stage ceph: show nocephx_require_signatures and notcp_nodelay options libceph: tcp_nodelay support rbd: do not treat standalone as flatten ceph: fix atomic_open snapdir ceph: properly mark empty directory as complete client: include kernel version in client metadata ceph: provide seperate {inode,file}_operations for snapdir ceph: fix request time stamp encoding ceph: fix reading inline data when i_size > PAGE_SIZE ceph: avoid block operation when !TASK_RUNNING (ceph_mdsc_close_sessions) ceph: avoid block operation when !TASK_RUNNING (ceph_get_caps) ceph: avoid block operation when !TASK_RUNNING (ceph_mdsc_sync) rbd: fix error paths in rbd_dev_refresh() ...

Linus Torvalds committed 11 years ago

Commit 4533f6e2 parents 89d3fa4 0f5417c23 files · +444 −488Browse files

Unified · Split

23 files

MAINTAINERS +4 −3
drivers/block/rbd.c +83 −110
fs/ceph/acl.c +0 −14
fs/ceph/addr.c +12 −7
fs/ceph/caps.c +74 −53
fs/ceph/dir.c +25 −8
fs/ceph/file.c +22 −15
fs/ceph/inode.c +22 −19
fs/ceph/mds_client.c +93 −34
fs/ceph/mds_client.h +2 −0
fs/ceph/snap.c +38 −16
fs/ceph/super.c +4 −0
fs/ceph/super.h +4 −1
include/linux/ceph/ceph_fs.h +1 −36
include/linux/ceph/libceph.h +2 −1
include/linux/ceph/messenger.h +3 −1
include/linux/ceph/mon_client.h +1 −8
net/ceph/ceph_common.c +15 −1
net/ceph/ceph_strings.c +0 −14
net/ceph/debugfs.c +0 −2
net/ceph/messenger.c +13 −1
net/ceph/mon_client.c +5 −134
net/ceph/osd_client.c +21 −10

MAINTAINERS+4−3

@@ -2433,7 +2433,8 @@ F: arch/powerpc/oprofile/cell · 2432 unmodified lines
2433	2433	F: arch/powerpc/platforms/cell/
2434	2434
2435	2435	CEPH DISTRIBUTED FILE SYSTEM CLIENT
2436		−M: Sage Weil <sage@inktank.com>
	2436	+M: Yan, Zheng <zyan@redhat.com>
	2437	+M: Sage Weil <sage@redhat.com>
2437	2438	L: ceph-devel@vger.kernel.org
2438	2439	W: http://ceph.com/
2439	2440	T: git git://git.kernel.org/pub/scm/linux/kernel/git/sage/ceph-client.git
@@ -7998,8 +7999,8 @@ S: Supported · 5558 unmodified lines
7998	7999	F: drivers/net/wireless/ath/wcn36xx/
7999	8000
8000	8001	RADOS BLOCK DEVICE (RBD)
8001		−M: Yehuda Sadeh <yehuda@inktank.com>
8002		−M: Sage Weil <sage@inktank.com>
	8002	+M: Ilya Dryomov <idryomov@gmail.com>
	8003	+M: Sage Weil <sage@redhat.com>
8003	8004	M: Alex Elder <elder@kernel.org>
8004	8005	M: ceph-devel@vger.kernel.org
8005	8006	W: http://ceph.com/

drivers/block/rbd.c+83−110

@@ -38,6 +38,7 @@ · 37 unmodified lines
38	38	#include <linux/kernel.h>
39	39	#include <linux/device.h>
40	40	#include <linux/module.h>
	41	+#include <linux/blk-mq.h>
41	42	#include <linux/fs.h>
42	43	#include <linux/blkdev.h>
43	44	#include <linux/slab.h>
@@ -340,9 +341,7 @@ struct rbd_device { · 296 unmodified lines
340	341
341	342	char name[DEV_NAME_LEN]; /* blkdev name, e.g. rbd3 */
342	343
343		− struct list_head rq_queue; /* incoming rq queue */
344	344	spinlock_t lock; /* queue, flags, open_count */
345		− struct work_struct rq_work;
346	345
347	346	struct rbd_image_header header;
348	347	unsigned long flags; /* possibly lock protected */
@@ -360,6 +359,9 @@ struct rbd_device { · 11 unmodified lines
360	359	atomic_t parent_ref;
361	360	struct rbd_device *parent;
362	361
	362	+ /* Block layer tags. */
	363	+ struct blk_mq_tag_set tag_set;
	364	+
363	365	/* protects updating the header */
364	366	struct rw_semaphore header_rwsem;
365	367
@@ -1817,7 +1819,8 @@ static void rbd_osd_req_callback(struct ceph_osd_request *osd_req, · 1451 unmodified lines
1817	1819
1818	1820	/*
1819	1821	* We support a 64-bit length, but ultimately it has to be
1820		− * passed to blk_end_request(), which takes an unsigned int.
	1822	+ * passed to the block layer, which just supports a 32-bit
	1823	+ * length field.
1821	1824	*/
1822	1825	obj_request->xferred = osd_req->r_reply_op_len[0];
1823	1826	rbd_assert(obj_request->xferred < (u64)UINT_MAX);
@@ -2275,7 +2278,10 @@ static bool rbd_img_obj_end_request(struct rbd_obj_request *obj_request) · 451 unmodified lines
2275	2278	more = obj_request->which < img_request->obj_request_count - 1;
2276	2279	} else {
2277	2280	rbd_assert(img_request->rq != NULL);
2278		− more = blk_end_request(img_request->rq, result, xferred);
	2281	+
	2282	+ more = blk_update_request(img_request->rq, result, xferred);
	2283	+ if (!more)
	2284	+ __blk_mq_end_request(img_request->rq, result);
2279	2285	}
2280	2286
2281	2287	return more;
@@ -3304,8 +3310,10 @@ out: · 1022 unmodified lines
3304	3310	return ret;
3305	3311	}
3306	3312
3307		−static void rbd_handle_request(struct rbd_device rbd_dev, struct request rq)
	3313	+static void rbd_queue_workfn(struct work_struct *work)
3308	3314	{
	3315	+ struct request *rq = blk_mq_rq_from_pdu(work);
	3316	+ struct rbd_device *rbd_dev = rq->q->queuedata;
3309	3317	struct rbd_img_request *img_request;
3310	3318	struct ceph_snap_context *snapc = NULL;
3311	3319	u64 offset = (u64)blk_rq_pos(rq) << SECTOR_SHIFT;
@@ -3314,6 +3322,13 @@ static void rbd_handle_request(struct rbd_device rbd_dev, struct request rq) · 2 unmodified lines
3314	3322	u64 mapping_size;
3315	3323	int result;
3316	3324
	3325	+ if (rq->cmd_type != REQ_TYPE_FS) {
	3326	+ dout("%s: non-fs request type %d\n", __func__,
	3327	+ (int) rq->cmd_type);
	3328	+ result = -EIO;
	3329	+ goto err;
	3330	+ }
	3331	+
3317	3332	if (rq->cmd_flags & REQ_DISCARD)
3318	3333	op_type = OBJ_OP_DISCARD;
3319	3334	else if (rq->cmd_flags & REQ_WRITE)
@@ -3359,6 +3374,8 @@ static void rbd_handle_request(struct rbd_device rbd_dev, struct request rq) · 39 unmodified lines
3359	3374	goto err_rq; /* Shouldn't happen */
3360	3375	}
3361	3376
	3377	+ blk_mq_start_request(rq);
	3378	+
3362	3379	down_read(&rbd_dev->header_rwsem);
3363	3380	mapping_size = rbd_dev->mapping.size;
3364	3381	if (op_type != OBJ_OP_READ) {
@@ -3404,53 +3421,18 @@ err_rq: · 39 unmodified lines
3404	3421	rbd_warn(rbd_dev, "%s %llx at %llx result %d",
3405	3422	obj_op_name(op_type), length, offset, result);
3406	3423	ceph_put_snap_context(snapc);
3407		− blk_end_request_all(rq, result);
	3424	+err:
	3425	+ blk_mq_end_request(rq, result);
3408	3426	}
3409	3427
3410		−static void rbd_request_workfn(struct work_struct *work)
	3428	+static int rbd_queue_rq(struct blk_mq_hw_ctx *hctx,
	3429	+ const struct blk_mq_queue_data *bd)
3411	3430	{
3412		− struct rbd_device *rbd_dev =
3413		− container_of(work, struct rbd_device, rq_work);
3414		− struct request rq, next;
3415		− LIST_HEAD(requests);
3416		−
3417		− spin_lock_irq(&rbd_dev->lock); /* rq->q->queue_lock */
3418		− list_splice_init(&rbd_dev->rq_queue, &requests);
3419		− spin_unlock_irq(&rbd_dev->lock);
	3431	+ struct request *rq = bd->rq;
	3432	+ struct work_struct *work = blk_mq_rq_to_pdu(rq);
3420	3433
3421		− list_for_each_entry_safe(rq, next, &requests, queuelist) {
3422		− list_del_init(&rq->queuelist);
3423		− rbd_handle_request(rbd_dev, rq);
3424		− }
3425		−}
3426		−
3427		−/*
3428		− * Called with q->queue_lock held and interrupts disabled, possibly on
3429		− * the way to schedule(). Do not sleep here!
3430		− */
3431		−static void rbd_request_fn(struct request_queue *q)
3432		−{
3433		− struct rbd_device *rbd_dev = q->queuedata;
3434		− struct request *rq;
3435		− int queued = 0;
3436		−
3437		− rbd_assert(rbd_dev);
3438		−
3439		− while ((rq = blk_fetch_request(q))) {
3440		− /* Ignore any non-FS requests that filter through. */
3441		− if (rq->cmd_type != REQ_TYPE_FS) {
3442		− dout("%s: non-fs request type %d\n", __func__,
3443		− (int) rq->cmd_type);
3444		− __blk_end_request_all(rq, 0);
3445		− continue;
3446		− }
3447		−
3448		− list_add_tail(&rq->queuelist, &rbd_dev->rq_queue);
3449		− queued++;
3450		− }
3451		−
3452		− if (queued)
3453		− queue_work(rbd_wq, &rbd_dev->rq_work);
	3434	+ queue_work(rbd_wq, work);
	3435	+ return BLK_MQ_RQ_QUEUE_OK;
3454	3436	}
3455	3437
3456	3438	/*
@@ -3511,6 +3493,7 @@ static void rbd_free_disk(struct rbd_device *rbd_dev) · 54 unmodified lines
3511	3493	del_gendisk(disk);
3512	3494	if (disk->queue)
3513	3495	blk_cleanup_queue(disk->queue);
	3496	+ blk_mq_free_tag_set(&rbd_dev->tag_set);
3514	3497	}
3515	3498	put_disk(disk);
3516	3499	}
@@ -3694,7 +3677,7 @@ static int rbd_dev_refresh(struct rbd_device *rbd_dev) · 177 unmodified lines
3694	3677
3695	3678	ret = rbd_dev_header_info(rbd_dev);
3696	3679	if (ret)
3697		− return ret;
	3680	+ goto out;
3698	3681
3699	3682	/*
3700	3683	* If there is a parent, see if it has disappeared due to the
@@ -3703,30 +3686,46 @@ static int rbd_dev_refresh(struct rbd_device *rbd_dev) · 2 unmodified lines
3703	3686	if (rbd_dev->parent) {
3704	3687	ret = rbd_dev_v2_parent_info(rbd_dev);
3705	3688	if (ret)
3706		− return ret;
	3689	+ goto out;
3707	3690	}
3708	3691
3709	3692	if (rbd_dev->spec->snap_id == CEPH_NOSNAP) {
3710		− if (rbd_dev->mapping.size != rbd_dev->header.image_size)
3711		− rbd_dev->mapping.size = rbd_dev->header.image_size;
	3693	+ rbd_dev->mapping.size = rbd_dev->header.image_size;
3712	3694	} else {
3713	3695	/* validate mapped snapshot's EXISTS flag */
3714	3696	rbd_exists_validate(rbd_dev);
3715	3697	}
3716	3698
	3699	+out:
3717	3700	up_write(&rbd_dev->header_rwsem);
3718		−
3719		− if (mapping_size != rbd_dev->mapping.size)
	3701	+ if (!ret && mapping_size != rbd_dev->mapping.size)
3720	3702	rbd_dev_update_size(rbd_dev);
3721	3703
	3704	+ return ret;
	3705	+}
	3706	+
	3707	+static int rbd_init_request(void data, struct request rq,
	3708	+ unsigned int hctx_idx, unsigned int request_idx,
	3709	+ unsigned int numa_node)
	3710	+{
	3711	+ struct work_struct *work = blk_mq_rq_to_pdu(rq);
	3712	+
	3713	+ INIT_WORK(work, rbd_queue_workfn);
3722	3714	return 0;
3723	3715	}
3724	3716
	3717	+static struct blk_mq_ops rbd_mq_ops = {
	3718	+ .queue_rq = rbd_queue_rq,
	3719	+ .map_queue = blk_mq_map_queue,
	3720	+ .init_request = rbd_init_request,
	3721	+};
	3722	+
3725	3723	static int rbd_init_disk(struct rbd_device *rbd_dev)
3726	3724	{
3727	3725	struct gendisk *disk;
3728	3726	struct request_queue *q;
3729	3727	u64 segment_size;
	3728	+ int err;
3730	3729
3731	3730	/* create gendisk info */
3732	3731	disk = alloc_disk(single_major ?
@@ -3744,10 +3743,25 @@ static int rbd_init_disk(struct rbd_device *rbd_dev) · 11 unmodified lines
3744	3743	disk->fops = &rbd_bd_ops;
3745	3744	disk->private_data = rbd_dev;
3746	3745
3747		− q = blk_init_queue(rbd_request_fn, &rbd_dev->lock);
3748		− if (!q)
	3746	+ memset(&rbd_dev->tag_set, 0, sizeof(rbd_dev->tag_set));
	3747	+ rbd_dev->tag_set.ops = &rbd_mq_ops;
	3748	+ rbd_dev->tag_set.queue_depth = BLKDEV_MAX_RQ;
	3749	+ rbd_dev->tag_set.numa_node = NUMA_NO_NODE;
	3750	+ rbd_dev->tag_set.flags =
	3751	+ BLK_MQ_F_SHOULD_MERGE \| BLK_MQ_F_SG_MERGE;
	3752	+ rbd_dev->tag_set.nr_hw_queues = 1;
	3753	+ rbd_dev->tag_set.cmd_size = sizeof(struct work_struct);
	3754	+
	3755	+ err = blk_mq_alloc_tag_set(&rbd_dev->tag_set);
	3756	+ if (err)
3749	3757	goto out_disk;
3750	3758
	3759	+ q = blk_mq_init_queue(&rbd_dev->tag_set);
	3760	+ if (IS_ERR(q)) {
	3761	+ err = PTR_ERR(q);
	3762	+ goto out_tag_set;
	3763	+ }
	3764	+
3751	3765	/* We use the default size, but let's be explicit about it. */
3752	3766	blk_queue_physical_block_size(q, SECTOR_SIZE);
3753	3767
@@ -3773,10 +3787,11 @@ static int rbd_init_disk(struct rbd_device *rbd_dev) · 19 unmodified lines
3773	3787	rbd_dev->disk = disk;
3774	3788
3775	3789	return 0;
	3790	+out_tag_set:
	3791	+ blk_mq_free_tag_set(&rbd_dev->tag_set);
3776	3792	out_disk:
3777	3793	put_disk(disk);
3778		−
3779		− return -ENOMEM;
	3794	+ return err;
3780	3795	}
3781	3796
3782	3797	/*
@@ -4033,8 +4048,6 @@ static struct rbd_device rbd_dev_create(struct rbd_client rbdc, · 250 unmodified lines
4033	4048	return NULL;
4034	4049
4035	4050	spin_lock_init(&rbd_dev->lock);
4036		− INIT_LIST_HEAD(&rbd_dev->rq_queue);
4037		− INIT_WORK(&rbd_dev->rq_work, rbd_request_workfn);
4038	4051	rbd_dev->flags = 0;
4039	4052	atomic_set(&rbd_dev->parent_ref, 0);
4040	4053	INIT_LIST_HEAD(&rbd_dev->node);
@@ -4274,32 +4287,22 @@ static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev) · 233 unmodified lines
4274	4287	}
4275	4288
4276	4289	/*
4277		− * We always update the parent overlap. If it's zero we
4278		− * treat it specially.
	4290	+ * We always update the parent overlap. If it's zero we issue
	4291	+ * a warning, as we will proceed as if there was no parent.
4279	4292	*/
4280		− rbd_dev->parent_overlap = overlap;
4281	4293	if (!overlap) {
4282		−
4283		− /* A null parent_spec indicates it's the initial probe */
4284		−
4285	4294	if (parent_spec) {
4286		− /*
4287		− * The overlap has become zero, so the clone
4288		− * must have been resized down to 0 at some
4289		− * point. Treat this the same as a flatten.
4290		− */
4291		− rbd_dev_parent_put(rbd_dev);
4292		− pr_info("%s: clone image now standalone\n",
4293		− rbd_dev->disk->disk_name);
	4295	+ /* refresh, careful to warn just once */
	4296	+ if (rbd_dev->parent_overlap)
	4297	+ rbd_warn(rbd_dev,
	4298	+ "clone now standalone (overlap became 0)");
4294	4299	} else {
4295		− /*
4296		− * For the initial probe, if we find the
4297		− * overlap is zero we just pretend there was
4298		− * no parent image.
4299		− */
4300		− rbd_warn(rbd_dev, "ignoring parent with overlap 0");
	4300	+ /* initial probe */
	4301	+ rbd_warn(rbd_dev, "clone is standalone (overlap 0)");
4301	4302	}
4302	4303	}
	4304	+ rbd_dev->parent_overlap = overlap;
	4305	+
4303	4306	out:
4304	4307	ret = 0;
4305	4308	out_err:
@@ -4770,36 +4773,6 @@ static inline size_t next_token(const char **buf) · 464 unmodified lines
4770	4773	return strcspn(buf, spaces); / Return token length */
4771	4774	}
4772	4775
4773		−/*
4774		− * Finds the next token in *buf, and if the provided token buffer is
4775		− * big enough, copies the found token into it. The result, if
4776		− * copied, is guaranteed to be terminated with '\0'. Note that *buf
4777		− * must be terminated with '\0' on entry.
4778		− *
4779		− * Returns the length of the token found (not including the '\0').
4780		− * Return value will be 0 if no token is found, and it will be >=
4781		− * token_size if the token would not fit.
4782		− *
4783		− * The *buf pointer will be updated to point beyond the end of the
4784		− * found token. Note that this occurs even if the token buffer is
4785		− * too small to hold it.
4786		− */
4787		−static inline size_t copy_token(const char **buf,
4788		− char *token,
4789		− size_t token_size)
4790		−{
4791		− size_t len;
4792		−
4793		− len = next_token(buf);
4794		− if (len < token_size) {
4795		− memcpy(token, *buf, len);
4796		− *(token + len) = '\0';
4797		− }
4798		− *buf += len;
4799		−
4800		− return len;
4801		−}
4802		−
4803	4776	/*
4804	4777	* Finds the next token in *buf, dynamically allocates a buffer big
4805	4778	* enough to hold a copy of it, and copies the token into the new

fs/ceph/acl.c+0−14

@@ -40,20 +40,6 @@ static inline void ceph_set_cached_acl(struct inode *inode, · 39 unmodified lines
40	40	spin_unlock(&ci->i_ceph_lock);
41	41	}
42	42
43		−static inline struct posix_acl ceph_get_cached_acl(struct inode inode,
44		− int type)
45		−{
46		− struct ceph_inode_info *ci = ceph_inode(inode);
47		− struct posix_acl *acl = ACL_NOT_CACHED;
48		−
49		− spin_lock(&ci->i_ceph_lock);
50		− if (__ceph_caps_issued_mask(ci, CEPH_CAP_XATTR_SHARED, 0))
51		− acl = get_cached_acl(inode, type);
52		− spin_unlock(&ci->i_ceph_lock);
53		−
54		− return acl;
55		−}
56		−
57	43	struct posix_acl ceph_get_acl(struct inode inode, int type)
58	44	{
59	45	int size;

fs/ceph/addr.c+12−7

@@ -196,17 +196,22 @@ static int readpage_nounlock(struct file filp, struct page page) · 195 unmodified lines
196	196	u64 len = PAGE_CACHE_SIZE;
197	197
198	198	if (off >= i_size_read(inode)) {
199		− zero_user_segment(page, err, PAGE_CACHE_SIZE);
	199	+ zero_user_segment(page, 0, PAGE_CACHE_SIZE);
200	200	SetPageUptodate(page);
201	201	return 0;
202	202	}
203	203
204		− /*
205		− * Uptodate inline data should have been added into page cache
206		− * while getting Fcr caps.
207		− */
208		− if (ci->i_inline_version != CEPH_INLINE_NONE)
209		− return -EINVAL;
	204	+ if (ci->i_inline_version != CEPH_INLINE_NONE) {
	205	+ /*
	206	+ * Uptodate inline data should have been added
	207	+ * into page cache while getting Fcr caps.
	208	+ */
	209	+ if (off == 0)
	210	+ return -EINVAL;
	211	+ zero_user_segment(page, 0, PAGE_CACHE_SIZE);
	212	+ SetPageUptodate(page);
	213	+ return 0;
	214	+ }
210	215
211	216	err = ceph_readpage_from_fscache(inode, page);
212	217	if (err == 0)

fs/ceph/caps.c+74−53

@@ -577,7 +577,6 @@ void ceph_add_cap(struct inode *inode, · 576 unmodified lines
577	577	struct ceph_snap_realm *realm = ceph_lookup_snap_realm(mdsc,
578	578	realmino);
579	579	if (realm) {
580		− ceph_get_snap_realm(mdsc, realm);
581	580	spin_lock(&realm->inodes_with_caps_lock);
582	581	ci->i_snap_realm = realm;
583	582	list_add(&ci->i_snap_realm_item,
@@ -1451,8 +1450,8 @@ static int __mark_caps_flushing(struct inode *inode, · 867 unmodified lines
1451	1450	spin_lock(&mdsc->cap_dirty_lock);
1452	1451	list_del_init(&ci->i_dirty_item);
1453	1452
1454		− ci->i_cap_flush_seq = ++mdsc->cap_flush_seq;
1455	1453	if (list_empty(&ci->i_flushing_item)) {
	1454	+ ci->i_cap_flush_seq = ++mdsc->cap_flush_seq;
1456	1455	list_add_tail(&ci->i_flushing_item, &session->s_cap_flushing);
1457	1456	mdsc->num_cap_flushing++;
1458	1457	dout(" inode %p now flushing seq %lld\n", inode,
@@ -2073,17 +2072,16 @@ static void __take_cap_refs(struct ceph_inode_info *ci, int got) · 614 unmodified lines
2073	2072	* requested from the MDS.
2074	2073	*/
2075	2074	static int try_get_cap_refs(struct ceph_inode_info *ci, int need, int want,
2076		− loff_t endoff, int got, struct page *pinned_page,
2077		− int check_max, int err)
	2075	+ loff_t endoff, int got, int check_max, int *err)
2078	2076	{
2079	2077	struct inode *inode = &ci->vfs_inode;
2080	2078	int ret = 0;
2081		− int have, implemented, _got = 0;
	2079	+ int have, implemented;
2082	2080	int file_wanted;
2083	2081
2084	2082	dout("get_cap_refs %p need %s want %s\n", inode,
2085	2083	ceph_cap_string(need), ceph_cap_string(want));
2086		−again:
	2084	+
2087	2085	spin_lock(&ci->i_ceph_lock);
2088	2086
2089	2087	/* make sure file is actually open */
@@ -2138,50 +2136,34 @@ again: · 48 unmodified lines
2138	2136	inode, ceph_cap_string(have), ceph_cap_string(not),
2139	2137	ceph_cap_string(revoking));
2140	2138	if ((revoking & not) == 0) {
2141		− _got = need \| (have & want);
2142		− __take_cap_refs(ci, _got);
	2139	+ *got = need \| (have & want);
	2140	+ __take_cap_refs(ci, *got);
2143	2141	ret = 1;
2144	2142	}
2145	2143	} else {
	2144	+ int session_readonly = false;
	2145	+ if ((need & CEPH_CAP_FILE_WR) && ci->i_auth_cap) {
	2146	+ struct ceph_mds_session *s = ci->i_auth_cap->session;
	2147	+ spin_lock(&s->s_cap_lock);
	2148	+ session_readonly = s->s_readonly;
	2149	+ spin_unlock(&s->s_cap_lock);
	2150	+ }
	2151	+ if (session_readonly) {
	2152	+ dout("get_cap_refs %p needed %s but mds%d readonly\n",
	2153	+ inode, ceph_cap_string(need), ci->i_auth_cap->mds);
	2154	+ *err = -EROFS;
	2155	+ ret = 1;
	2156	+ goto out_unlock;
	2157	+ }
	2158	+
2146	2159	dout("get_cap_refs %p have %s needed %s\n", inode,
2147	2160	ceph_cap_string(have), ceph_cap_string(need));
2148	2161	}
2149	2162	out_unlock:
2150	2163	spin_unlock(&ci->i_ceph_lock);
2151	2164
2152		− if (ci->i_inline_version != CEPH_INLINE_NONE &&
2153		− (_got & (CEPH_CAP_FILE_CACHE\|CEPH_CAP_FILE_LAZYIO)) &&
2154		− i_size_read(inode) > 0) {
2155		− int ret1;
2156		− struct page *page = find_get_page(inode->i_mapping, 0);
2157		− if (page) {
2158		− if (PageUptodate(page)) {
2159		− *pinned_page = page;
2160		− goto out;
2161		− }
2162		− page_cache_release(page);
2163		− }
2164		− /*
2165		− * drop cap refs first because getattr while holding
2166		− * caps refs can cause deadlock.
2167		− */
2168		− ceph_put_cap_refs(ci, _got);
2169		− _got = 0;
2170		−
2171		− /* getattr request will bring inline data into page cache */
2172		− ret1 = __ceph_do_getattr(inode, NULL,
2173		− CEPH_STAT_CAP_INLINE_DATA, true);
2174		− if (ret1 >= 0) {
2175		− ret = 0;
2176		− goto again;
2177		− }
2178		− *err = ret1;
2179		− ret = 1;
2180		− }
2181		−out:
2182	2165	dout("get_cap_refs %p ret %d got %s\n", inode,
2183		− ret, ceph_cap_string(_got));
2184		− *got = _got;
	2166	+ ret, ceph_cap_string(*got));
2185	2167	return ret;
2186	2168	}
2187	2169
@@ -2221,22 +2203,52 @@ static void check_max_size(struct inode *inode, loff_t endoff) · 33 unmodified lines
2221	2203	int ceph_get_caps(struct ceph_inode_info *ci, int need, int want,
2222	2204	loff_t endoff, int got, struct page *pinned_page)
2223	2205	{
2224		− int check_max, ret, err;
	2206	+ int _got, check_max, ret, err = 0;
2225	2207
2226	2208	retry:
2227	2209	if (endoff > 0)
2228	2210	check_max_size(&ci->vfs_inode, endoff);
	2211	+ _got = 0;
2229	2212	check_max = 0;
2230		− err = 0;
2231	2213	ret = wait_event_interruptible(ci->i_cap_wq,
2232		− try_get_cap_refs(ci, need, want, endoff,
2233		− got, pinned_page,
2234		− &check_max, &err));
	2214	+ try_get_cap_refs(ci, need, want, endoff,
	2215	+ &_got, &check_max, &err));
2235	2216	if (err)
2236	2217	ret = err;
	2218	+ if (ret < 0)
	2219	+ return ret;
	2220	+
2237	2221	if (check_max)
2238	2222	goto retry;
2239		− return ret;
	2223	+
	2224	+ if (ci->i_inline_version != CEPH_INLINE_NONE &&
	2225	+ (_got & (CEPH_CAP_FILE_CACHE\|CEPH_CAP_FILE_LAZYIO)) &&
	2226	+ i_size_read(&ci->vfs_inode) > 0) {
	2227	+ struct page *page = find_get_page(ci->vfs_inode.i_mapping, 0);
	2228	+ if (page) {
	2229	+ if (PageUptodate(page)) {
	2230	+ *pinned_page = page;
	2231	+ goto out;
	2232	+ }
	2233	+ page_cache_release(page);
	2234	+ }
	2235	+ /*
	2236	+ * drop cap refs first because getattr while holding
	2237	+ * caps refs can cause deadlock.
	2238	+ */
	2239	+ ceph_put_cap_refs(ci, _got);
	2240	+ _got = 0;
	2241	+
	2242	+ /* getattr request will bring inline data into page cache */
	2243	+ ret = __ceph_do_getattr(&ci->vfs_inode, NULL,
	2244	+ CEPH_STAT_CAP_INLINE_DATA, true);
	2245	+ if (ret < 0)
	2246	+ return ret;
	2247	+ goto retry;
	2248	+ }
	2249	+out:
	2250	+ *got = _got;
	2251	+ return 0;
2240	2252	}
2241	2253
2242	2254	/*
@@ -2432,13 +2444,13 @@ static void invalidate_aliases(struct inode *inode) · 189 unmodified lines
2432	2444	*/
2433	2445	static void handle_cap_grant(struct ceph_mds_client *mdsc,
2434	2446	struct inode inode, struct ceph_mds_caps grant,
2435		− void *snaptrace, int snaptrace_len,
2436	2447	u64 inline_version,
2437	2448	void *inline_data, int inline_len,
2438	2449	struct ceph_buffer *xattr_buf,
2439	2450	struct ceph_mds_session *session,
2440	2451	struct ceph_cap *cap, int issued)
2441	2452	__releases(ci->i_ceph_lock)
	2453	+ __releases(mdsc->snap_rwsem)
2442	2454	{
2443	2455	struct ceph_inode_info *ci = ceph_inode(inode);
2444	2456	int mds = session->s_mds;
@@ -2639,10 +2651,6 @@ static void handle_cap_grant(struct ceph_mds_client *mdsc, · 194 unmodified lines
2639	2651	spin_unlock(&ci->i_ceph_lock);
2640	2652
2641	2653	if (le32_to_cpu(grant->op) == CEPH_CAP_OP_IMPORT) {
2642		− down_write(&mdsc->snap_rwsem);
2643		− ceph_update_snap_trace(mdsc, snaptrace,
2644		− snaptrace + snaptrace_len, false);
2645		− downgrade_write(&mdsc->snap_rwsem);
2646	2654	kick_flushing_inode_caps(mdsc, session, inode);
2647	2655	up_read(&mdsc->snap_rwsem);
2648	2656	if (newcaps & ~issued)
@@ -3052,6 +3060,7 @@ void ceph_handle_caps(struct ceph_mds_session *session, · 403 unmodified lines
3052	3060	struct ceph_cap *cap;
3053	3061	struct ceph_mds_caps *h;
3054	3062	struct ceph_mds_cap_peer *peer = NULL;
	3063	+ struct ceph_snap_realm *realm;
3055	3064	int mds = session->s_mds;
3056	3065	int op, issued;
3057	3066	u32 seq, mseq;
@@ -3153,11 +3162,23 @@ void ceph_handle_caps(struct ceph_mds_session *session, · 95 unmodified lines
3153	3162	goto done_unlocked;
3154	3163
3155	3164	case CEPH_CAP_OP_IMPORT:
	3165	+ realm = NULL;
	3166	+ if (snaptrace_len) {
	3167	+ down_write(&mdsc->snap_rwsem);
	3168	+ ceph_update_snap_trace(mdsc, snaptrace,
	3169	+ snaptrace + snaptrace_len,
	3170	+ false, &realm);
	3171	+ downgrade_write(&mdsc->snap_rwsem);
	3172	+ } else {
	3173	+ down_read(&mdsc->snap_rwsem);
	3174	+ }
3156	3175	handle_cap_import(mdsc, inode, h, peer, session,
3157	3176	&cap, &issued);
3158		− handle_cap_grant(mdsc, inode, h, snaptrace, snaptrace_len,
	3177	+ handle_cap_grant(mdsc, inode, h,
3159	3178	inline_version, inline_data, inline_len,
3160	3179	msg->middle, session, cap, issued);
	3180	+ if (realm)
	3181	+ ceph_put_snap_realm(mdsc, realm);
3161	3182	goto done_unlocked;
3162	3183	}
3163	3184
@@ -3177,7 +3198,7 @@ void ceph_handle_caps(struct ceph_mds_session *session, · 13 unmodified lines
3177	3198	case CEPH_CAP_OP_GRANT:
3178	3199	__ceph_caps_issued(ci, &issued);
3179	3200	issued \|= __ceph_caps_dirty(ci);
3180		− handle_cap_grant(mdsc, inode, h, NULL, 0,
	3201	+ handle_cap_grant(mdsc, inode, h,
3181	3202	inline_version, inline_data, inline_len,
3182	3203	msg->middle, session, cap, issued);
3183	3204	goto done_unlocked;

fs/ceph/dir.c+25−8

@@ -26,8 +26,6 @@ · 25 unmodified lines
26	26	* point by name.
27	27	*/
28	28
29		−const struct inode_operations ceph_dir_iops;
30		−const struct file_operations ceph_dir_fops;
31	29	const struct dentry_operations ceph_dentry_ops;
32	30
33	31	/*
@@ -672,13 +670,17 @@ int ceph_handle_notrace_create(struct inode dir, struct dentry dentry) · 638 unmodified lines
672	670	/*
673	671	* We created the item, then did a lookup, and found
674	672	* it was already linked to another inode we already
675		− * had in our cache (and thus got spliced). Link our
676		− * dentry to that inode, but don't hash it, just in
677		− * case the VFS wants to dereference it.
	673	+ * had in our cache (and thus got spliced). To not
	674	+ * confuse VFS (especially when inode is a directory),
	675	+ * we don't link our dentry to that inode, return an
	676	+ * error instead.
	677	+ *
	678	+ * This event should be rare and it happens only when
	679	+ * we talk to old MDS. Recent MDS does not send traceless
	680	+ * reply for request that creates new inode.
678	681	*/
679		− BUG_ON(!result->d_inode);
680		− d_instantiate(dentry, result->d_inode);
681		− return 0;
	682	+ d_drop(result);
	683	+ return -ESTALE;
682	684	}
683	685	return PTR_ERR(result);
684	686	}
@@ -1335,6 +1337,13 @@ const struct file_operations ceph_dir_fops = { · 650 unmodified lines
1335	1337	.fsync = ceph_dir_fsync,
1336	1338	};
1337	1339
	1340	+const struct file_operations ceph_snapdir_fops = {
	1341	+ .iterate = ceph_readdir,
	1342	+ .llseek = ceph_dir_llseek,
	1343	+ .open = ceph_open,
	1344	+ .release = ceph_release,
	1345	+};
	1346	+
1338	1347	const struct inode_operations ceph_dir_iops = {
1339	1348	.lookup = ceph_lookup,
1340	1349	.permission = ceph_permission,
@@ -1357,6 +1366,14 @@ const struct inode_operations ceph_dir_iops = { · 16 unmodified lines
1357	1366	.atomic_open = ceph_atomic_open,
1358	1367	};
1359	1368
	1369	+const struct inode_operations ceph_snapdir_iops = {
	1370	+ .lookup = ceph_lookup,
	1371	+ .permission = ceph_permission,
	1372	+ .getattr = ceph_getattr,
	1373	+ .mkdir = ceph_mkdir,
	1374	+ .rmdir = ceph_unlink,
	1375	+};
	1376	+
1360	1377	const struct dentry_operations ceph_dentry_ops = {
1361	1378	.d_revalidate = ceph_d_revalidate,
1362	1379	.d_release = ceph_d_release,

fs/ceph/file.c+22−15

@@ -275,10 +275,10 @@ int ceph_atomic_open(struct inode dir, struct dentry dentry, · 274 unmodified lines
275	275	err = ceph_mdsc_do_request(mdsc,
276	276	(flags & (O_CREAT\|O_TRUNC)) ? dir : NULL,
277	277	req);
	278	+ err = ceph_handle_snapdir(req, dentry, err);
278	279	if (err)
279	280	goto out_req;
280	281
281		− err = ceph_handle_snapdir(req, dentry, err);
282	282	if (err == 0 && (flags & O_CREAT) && !req->r_reply_info.head->is_dentry)
283	283	err = ceph_handle_notrace_create(dir, dentry);
284	284
@@ -392,13 +392,14 @@ more: · 107 unmodified lines
392	392	if (ret >= 0) {
393	393	int didpages;
394	394	if (was_short && (pos + ret < inode->i_size)) {
395		− u64 tmp = min(this_len - ret,
396		− inode->i_size - pos - ret);
	395	+ int zlen = min(this_len - ret,
	396	+ inode->i_size - pos - ret);
	397	+ int zoff = (o_direct ? buf_align : io_align) +
	398	+ read + ret;
397	399	dout(" zero gap %llu to %llu\n",
398		− pos + ret, pos + ret + tmp);
399		− ceph_zero_page_vector_range(page_align + read + ret,
400		− tmp, pages);
401		− ret += tmp;
	400	+ pos + ret, pos + ret + zlen);
	401	+ ceph_zero_page_vector_range(zoff, zlen, pages);
	402	+ ret += zlen;
402	403	}
403	404
404	405	didpages = (page_align + ret) >> PAGE_CACHE_SHIFT;
@@ -878,28 +879,34 @@ again: · 473 unmodified lines
878	879
879	880	i_size = i_size_read(inode);
880	881	if (retry_op == READ_INLINE) {
881		− /* does not support inline data > PAGE_SIZE */
882		− if (i_size > PAGE_CACHE_SIZE) {
883		− ret = -EIO;
884		− } else if (iocb->ki_pos < i_size) {
	882	+ BUG_ON(ret > 0 \|\| read > 0);
	883	+ if (iocb->ki_pos < i_size &&
	884	+ iocb->ki_pos < PAGE_CACHE_SIZE) {
885	885	loff_t end = min_t(loff_t, i_size,
886	886	iocb->ki_pos + len);
	887	+ end = min_t(loff_t, end, PAGE_CACHE_SIZE);
887	888	if (statret < end)
888	889	zero_user_segment(page, statret, end);
889	890	ret = copy_page_to_iter(page,
890	891	iocb->ki_pos & ~PAGE_MASK,
891	892	end - iocb->ki_pos, to);
892	893	iocb->ki_pos += ret;
893		− } else {
894		− ret = 0;
	894	+ read += ret;
	895	+ }
	896	+ if (iocb->ki_pos < i_size && read < len) {
	897	+ size_t zlen = min_t(size_t, len - read,
	898	+ i_size - iocb->ki_pos);
	899	+ ret = iov_iter_zero(zlen, to);
	900	+ iocb->ki_pos += ret;
	901	+ read += ret;
895	902	}
896	903	__free_pages(page, 0);
897		− return ret;
	904	+ return read;
898	905	}
899	906
900	907	/* hit EOF or hole? */
901	908	if (retry_op == CHECK_EOF && iocb->ki_pos < i_size &&
902		− ret < len) {
	909	+ ret < len) {
903	910	dout("sync_read hit hole, ppos %lld < size %lld"
904	911	", reading more\n", iocb->ki_pos,
905	912	inode->i_size);

fs/ceph/inode.c+22−19

@@ -82,8 +82,8 @@ struct inode ceph_get_snapdir(struct inode parent) · 81 unmodified lines
82	82	inode->i_mode = parent->i_mode;
83	83	inode->i_uid = parent->i_uid;
84	84	inode->i_gid = parent->i_gid;
85		− inode->i_op = &ceph_dir_iops;
86		− inode->i_fop = &ceph_dir_fops;
	85	+ inode->i_op = &ceph_snapdir_iops;
	86	+ inode->i_fop = &ceph_snapdir_fops;
87	87	ci->i_snap_caps = CEPH_CAP_PIN; /* so we can open */
88	88	ci->i_rbytes = 0;
89	89	return inode;
@@ -838,30 +838,31 @@ static int fill_inode(struct inode inode, struct page locked_page, · 748 unmodified lines
838	838	ceph_vinop(inode), inode->i_mode);
839	839	}
840	840
841		− /* set dir completion flag? */
842		− if (S_ISDIR(inode->i_mode) &&
843		− ci->i_files == 0 && ci->i_subdirs == 0 &&
844		− ceph_snap(inode) == CEPH_NOSNAP &&
845		− (le32_to_cpu(info->cap.caps) & CEPH_CAP_FILE_SHARED) &&
846		− (issued & CEPH_CAP_FILE_EXCL) == 0 &&
847		− !__ceph_dir_is_complete(ci)) {
848		− dout(" marking %p complete (empty)\n", inode);
849		− __ceph_dir_set_complete(ci, atomic_read(&ci->i_release_count),
850		− ci->i_ordered_count);
851		− }
852		−
853	841	/* were we issued a capability? */
854	842	if (info->cap.caps) {
855	843	if (ceph_snap(inode) == CEPH_NOSNAP) {
	844	+ unsigned caps = le32_to_cpu(info->cap.caps);
856	845	ceph_add_cap(inode, session,
857	846	le64_to_cpu(info->cap.cap_id),
858		− cap_fmode,
859		− le32_to_cpu(info->cap.caps),
	847	+ cap_fmode, caps,
860	848	le32_to_cpu(info->cap.wanted),
861	849	le32_to_cpu(info->cap.seq),
862	850	le32_to_cpu(info->cap.mseq),
863	851	le64_to_cpu(info->cap.realm),
864	852	info->cap.flags, &new_cap);
	853	+
	854	+ /* set dir completion flag? */
	855	+ if (S_ISDIR(inode->i_mode) &&
	856	+ ci->i_files == 0 && ci->i_subdirs == 0 &&
	857	+ (caps & CEPH_CAP_FILE_SHARED) &&
	858	+ (issued & CEPH_CAP_FILE_EXCL) == 0 &&
	859	+ !__ceph_dir_is_complete(ci)) {
	860	+ dout(" marking %p complete (empty)\n", inode);
	861	+ __ceph_dir_set_complete(ci,
	862	+ atomic_read(&ci->i_release_count),
	863	+ ci->i_ordered_count);
	864	+ }
	865	+
865	866	wake = true;
866	867	} else {
867	868	dout(" %p got snap_caps %s\n", inode,
@@ -1446,12 +1447,14 @@ retry_lookup: · 578 unmodified lines
1446	1447	}
1447	1448
1448	1449	if (!dn->d_inode) {
1449		− dn = splice_dentry(dn, in, NULL);
1450		− if (IS_ERR(dn)) {
1451		− err = PTR_ERR(dn);
	1450	+ struct dentry *realdn = splice_dentry(dn, in, NULL);
	1451	+ if (IS_ERR(realdn)) {
	1452	+ err = PTR_ERR(realdn);
	1453	+ d_drop(dn);
1452	1454	dn = NULL;
1453	1455	goto next_item;
1454	1456	}
	1457	+ dn = realdn;
1455	1458	}
1456	1459
1457	1460	di = dn->d_fsdata;

fs/ceph/mds_client.c+93−34

@@ -480,6 +480,7 @@ static struct ceph_mds_session register_session(struct ceph_mds_client mdsc, · 479 unmodified lines
480	480	mdsc->max_sessions = newmax;
481	481	}
482	482	mdsc->sessions[mds] = s;
	483	+ atomic_inc(&mdsc->num_sessions);
483	484	atomic_inc(&s->s_ref); /* one ref to sessions[], one to caller */
484	485
485	486	ceph_con_open(&s->s_con, CEPH_ENTITY_TYPE_MDS, mds,
@@ -503,6 +504,7 @@ static void __unregister_session(struct ceph_mds_client *mdsc, · 17 unmodified lines
503	504	mdsc->sessions[s->s_mds] = NULL;
504	505	ceph_con_close(&s->s_con);
505	506	ceph_put_mds_session(s);
	507	+ atomic_dec(&mdsc->num_sessions);
506	508	}
507	509
508	510	/*
@@ -842,8 +844,9 @@ static struct ceph_msg create_session_open_msg(struct ceph_mds_client mdsc, u6 · 333 unmodified lines
842	844	struct ceph_options *opt = mdsc->fsc->client->options;
843	845	void *p;
844	846
845		− const char* metadata[3][2] = {
	847	+ const char* metadata[][2] = {
846	848	{"hostname", utsname()->nodename},
	849	+ {"kernel_version", utsname()->release},
847	850	{"entity_id", opt->name ? opt->name : ""},
848	851	{NULL, NULL}
849	852	};
@@ -1464,19 +1467,33 @@ out_unlocked: · 614 unmodified lines
1464	1467	return err;
1465	1468	}
1466	1469
	1470	+static int check_cap_flush(struct inode *inode, u64 want_flush_seq)
	1471	+{
	1472	+ struct ceph_inode_info *ci = ceph_inode(inode);
	1473	+ int ret;
	1474	+ spin_lock(&ci->i_ceph_lock);
	1475	+ if (ci->i_flushing_caps)
	1476	+ ret = ci->i_cap_flush_seq >= want_flush_seq;
	1477	+ else
	1478	+ ret = 1;
	1479	+ spin_unlock(&ci->i_ceph_lock);
	1480	+ return ret;
	1481	+}
	1482	+
1467	1483	/*
1468	1484	* flush all dirty inode data to disk.
1469	1485	*
1470	1486	* returns true if we've flushed through want_flush_seq
1471	1487	*/
1472		−static int check_cap_flush(struct ceph_mds_client *mdsc, u64 want_flush_seq)
	1488	+static void wait_caps_flush(struct ceph_mds_client *mdsc, u64 want_flush_seq)
1473	1489	{
1474		− int mds, ret = 1;
	1490	+ int mds;
1475	1491
1476	1492	dout("check_cap_flush want %lld\n", want_flush_seq);
1477	1493	mutex_lock(&mdsc->mutex);
1478		− for (mds = 0; ret && mds < mdsc->max_sessions; mds++) {
	1494	+ for (mds = 0; mds < mdsc->max_sessions; mds++) {
1479	1495	struct ceph_mds_session *session = mdsc->sessions[mds];
	1496	+ struct inode *inode = NULL;
1480	1497
1481	1498	if (!session)
1482	1499	continue;
@@ -1489,29 +1506,29 @@ static int check_cap_flush(struct ceph_mds_client *mdsc, u64 want_flush_seq) · 6 unmodified lines
1489	1506	list_entry(session->s_cap_flushing.next,
1490	1507	struct ceph_inode_info,
1491	1508	i_flushing_item);
1492		− struct inode *inode = &ci->vfs_inode;
1493	1509
1494		− spin_lock(&ci->i_ceph_lock);
1495		− if (ci->i_cap_flush_seq <= want_flush_seq) {
	1510	+ if (!check_cap_flush(&ci->vfs_inode, want_flush_seq)) {
1496	1511	dout("check_cap_flush still flushing %p "
1497		− "seq %lld <= %lld to mds%d\n", inode,
1498		− ci->i_cap_flush_seq, want_flush_seq,
1499		− session->s_mds);
1500		− ret = 0;
	1512	+ "seq %lld <= %lld to mds%d\n",
	1513	+ &ci->vfs_inode, ci->i_cap_flush_seq,
	1514	+ want_flush_seq, session->s_mds);
	1515	+ inode = igrab(&ci->vfs_inode);
1501	1516	}
1502		− spin_unlock(&ci->i_ceph_lock);
1503	1517	}
1504	1518	mutex_unlock(&session->s_mutex);
1505	1519	ceph_put_mds_session(session);
1506	1520
1507		− if (!ret)
1508		− return ret;
	1521	+ if (inode) {
	1522	+ wait_event(mdsc->cap_flushing_wq,
	1523	+ check_cap_flush(inode, want_flush_seq));
	1524	+ iput(inode);
	1525	+ }
	1526	+
1509	1527	mutex_lock(&mdsc->mutex);
1510	1528	}
1511	1529
1512	1530	mutex_unlock(&mdsc->mutex);
1513	1531	dout("check_cap_flush ok, flushed thru %lld\n", want_flush_seq);
1514		− return ret;
1515	1532	}
1516	1533
1517	1534	/*
@@ -1923,7 +1940,11 @@ static struct ceph_msg create_request_message(struct ceph_mds_client mdsc, · 405 unmodified lines
1923	1940	head->num_releases = cpu_to_le16(releases);
1924	1941
1925	1942	/* time stamp */
1926		− ceph_encode_copy(&p, &req->r_stamp, sizeof(req->r_stamp));
	1943	+ {
	1944	+ struct ceph_timespec ts;
	1945	+ ceph_encode_timespec(&ts, &req->r_stamp);
	1946	+ ceph_encode_copy(&p, &ts, sizeof(ts));
	1947	+ }
1927	1948
1928	1949	BUG_ON(p > end);
1929	1950	msg->front.iov_len = p - msg->front.iov_base;
@@ -2012,7 +2033,11 @@ static int __prepare_send_request(struct ceph_mds_client *mdsc, · 82 unmodified lines
2012	2033
2013	2034	/* time stamp */
2014	2035	p = msg->front.iov_base + req->r_request_release_offset;
2015		− ceph_encode_copy(&p, &req->r_stamp, sizeof(req->r_stamp));
	2036	+ {
	2037	+ struct ceph_timespec ts;
	2038	+ ceph_encode_timespec(&ts, &req->r_stamp);
	2039	+ ceph_encode_copy(&p, &ts, sizeof(ts));
	2040	+ }
2016	2041
2017	2042	msg->front.iov_len = p - msg->front.iov_base;
2018	2043	msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
@@ -2159,6 +2184,8 @@ static void kick_requests(struct ceph_mds_client *mdsc, int mds) · 140 unmodified lines
2159	2184	p = rb_next(p);
2160	2185	if (req->r_got_unsafe)
2161	2186	continue;
	2187	+ if (req->r_attempts > 0)
	2188	+ continue; /* only new requests */
2162	2189	if (req->r_session &&
2163	2190	req->r_session->s_mds == mds) {
2164	2191	dout(" kicking tid %llu\n", req->r_tid);
@@ -2286,6 +2313,7 @@ static void handle_reply(struct ceph_mds_session session, struct ceph_msg msg) · 121 unmodified lines
2286	2313	struct ceph_mds_request *req;
2287	2314	struct ceph_mds_reply_head *head = msg->front.iov_base;
2288	2315	struct ceph_mds_reply_info_parsed rinfo; / parsed reply info */
	2316	+ struct ceph_snap_realm *realm;
2289	2317	u64 tid;
2290	2318	int err, result;
2291	2319	int mds = session->s_mds;
@@ -2401,11 +2429,13 @@ static void handle_reply(struct ceph_mds_session session, struct ceph_msg msg) · 109 unmodified lines
2401	2429	}
2402	2430
2403	2431	/* snap trace */
	2432	+ realm = NULL;
2404	2433	if (rinfo->snapblob_len) {
2405	2434	down_write(&mdsc->snap_rwsem);
2406	2435	ceph_update_snap_trace(mdsc, rinfo->snapblob,
2407		− rinfo->snapblob + rinfo->snapblob_len,
2408		− le32_to_cpu(head->op) == CEPH_MDS_OP_RMSNAP);
	2436	+ rinfo->snapblob + rinfo->snapblob_len,
	2437	+ le32_to_cpu(head->op) == CEPH_MDS_OP_RMSNAP,
	2438	+ &realm);
2409	2439	downgrade_write(&mdsc->snap_rwsem);
2410	2440	} else {
2411	2441	down_read(&mdsc->snap_rwsem);
@@ -2423,6 +2453,8 @@ static void handle_reply(struct ceph_mds_session session, struct ceph_msg msg) · 11 unmodified lines
2423	2453	mutex_unlock(&req->r_fill_mutex);
2424	2454
2425	2455	up_read(&mdsc->snap_rwsem);
	2456	+ if (realm)
	2457	+ ceph_put_snap_realm(mdsc, realm);
2426	2458	out_err:
2427	2459	mutex_lock(&mdsc->mutex);
2428	2460	if (!req->r_aborted) {
@@ -2487,6 +2519,7 @@ static void handle_forward(struct ceph_mds_client *mdsc, · 58 unmodified lines
2487	2519	dout("forward tid %llu to mds%d (we resend)\n", tid, next_mds);
2488	2520	BUG_ON(req->r_err);
2489	2521	BUG_ON(req->r_got_result);
	2522	+ req->r_attempts = 0;
2490	2523	req->r_num_fwd = fwd_seq;
2491	2524	req->r_resend_mds = next_mds;
2492	2525	put_request_session(req);
@@ -2580,6 +2613,14 @@ static void handle_session(struct ceph_mds_session *session, · 87 unmodified lines
2580	2613	send_flushmsg_ack(mdsc, session, seq);
2581	2614	break;
2582	2615
	2616	+ case CEPH_SESSION_FORCE_RO:
	2617	+ dout("force_session_readonly %p\n", session);
	2618	+ spin_lock(&session->s_cap_lock);
	2619	+ session->s_readonly = true;
	2620	+ spin_unlock(&session->s_cap_lock);
	2621	+ wake_up_session_caps(session, 0);
	2622	+ break;
	2623	+
2583	2624	default:
2584	2625	pr_err("mdsc_handle_session bad op %d mds%d\n", op, mds);
2585	2626	WARN_ON(1);
@@ -2610,6 +2651,7 @@ static void replay_unsafe_requests(struct ceph_mds_client *mdsc, · 24 unmodified lines
2610	2651	struct ceph_mds_session *session)
2611	2652	{
2612	2653	struct ceph_mds_request req, nreq;
	2654	+ struct rb_node *p;
2613	2655	int err;
2614	2656
2615	2657	dout("replay_unsafe_requests mds%d\n", session->s_mds);
@@ -2622,6 +2664,28 @@ static void replay_unsafe_requests(struct ceph_mds_client *mdsc, · 6 unmodified lines
2622	2664	ceph_con_send(&session->s_con, req->r_request);
2623	2665	}
2624	2666	}
	2667	+
	2668	+ /*
	2669	+ * also re-send old requests when MDS enters reconnect stage. So that MDS
	2670	+ * can process completed request in clientreplay stage.
	2671	+ */
	2672	+ p = rb_first(&mdsc->request_tree);
	2673	+ while (p) {
	2674	+ req = rb_entry(p, struct ceph_mds_request, r_node);
	2675	+ p = rb_next(p);
	2676	+ if (req->r_got_unsafe)
	2677	+ continue;
	2678	+ if (req->r_attempts == 0)
	2679	+ continue; /* only old requests */
	2680	+ if (req->r_session &&
	2681	+ req->r_session->s_mds == session->s_mds) {
	2682	+ err = __prepare_send_request(mdsc, req, session->s_mds);
	2683	+ if (!err) {
	2684	+ ceph_msg_get(req->r_request);
	2685	+ ceph_con_send(&session->s_con, req->r_request);
	2686	+ }
	2687	+ }
	2688	+ }
2625	2689	mutex_unlock(&mdsc->mutex);
2626	2690	}
2627	2691
@@ -2787,6 +2851,8 @@ static void send_mds_reconnect(struct ceph_mds_client *mdsc, · 159 unmodified lines
2787	2851	spin_unlock(&session->s_gen_ttl_lock);
2788	2852
2789	2853	spin_lock(&session->s_cap_lock);
	2854	+ /* don't know if session is readonly */
	2855	+ session->s_readonly = 0;
2790	2856	/*
2791	2857	* notify __ceph_remove_cap() that we are composing cap reconnect.
2792	2858	* If a cap get released before being added to the cap reconnect,
@@ -2933,9 +2999,6 @@ static void check_new_map(struct ceph_mds_client *mdsc, · 140 unmodified lines
2933	2999	mutex_unlock(&s->s_mutex);
2934	3000	s->s_state = CEPH_MDS_SESSION_RESTARTING;
2935	3001	}
2936		−
2937		− /* kick any requests waiting on the recovering mds */
2938		− kick_requests(mdsc, i);
2939	3002	} else if (oldstate == newstate) {
2940	3003	continue; /* nothing new with this mds */
2941	3004	}
@@ -3295,6 +3358,7 @@ int ceph_mdsc_init(struct ceph_fs_client *fsc) · 353 unmodified lines
3295	3358	init_waitqueue_head(&mdsc->session_close_wq);
3296	3359	INIT_LIST_HEAD(&mdsc->waiting_for_map);
3297	3360	mdsc->sessions = NULL;
	3361	+ atomic_set(&mdsc->num_sessions, 0);
3298	3362	mdsc->max_sessions = 0;
3299	3363	mdsc->stopping = 0;
3300	3364	init_rwsem(&mdsc->snap_rwsem);
@@ -3428,14 +3492,17 @@ void ceph_mdsc_sync(struct ceph_mds_client *mdsc) · 127 unmodified lines
3428	3492	dout("sync\n");
3429	3493	mutex_lock(&mdsc->mutex);
3430	3494	want_tid = mdsc->last_tid;
3431		− want_flush = mdsc->cap_flush_seq;
3432	3495	mutex_unlock(&mdsc->mutex);
3433		− dout("sync want tid %lld flush_seq %lld\n", want_tid, want_flush);
3434	3496
3435	3497	ceph_flush_dirty_caps(mdsc);
	3498	+ spin_lock(&mdsc->cap_dirty_lock);
	3499	+ want_flush = mdsc->cap_flush_seq;
	3500	+ spin_unlock(&mdsc->cap_dirty_lock);
	3501	+
	3502	+ dout("sync want tid %lld flush_seq %lld\n", want_tid, want_flush);
3436	3503
3437	3504	wait_unsafe_requests(mdsc, want_tid);
3438		− wait_event(mdsc->cap_flushing_wq, check_cap_flush(mdsc, want_flush));
	3505	+ wait_caps_flush(mdsc, want_flush);
3439	3506	}
3440	3507
3441	3508	/*
@@ -3443,17 +3510,9 @@ void ceph_mdsc_sync(struct ceph_mds_client *mdsc) · 1 unmodified line
3443	3510	*/
3444	3511	static bool done_closing_sessions(struct ceph_mds_client *mdsc)
3445	3512	{
3446		− int i, n = 0;
3447		−
3448	3513	if (mdsc->fsc->mount_state == CEPH_MOUNT_SHUTDOWN)
3449	3514	return true;
3450		−
3451		− mutex_lock(&mdsc->mutex);
3452		− for (i = 0; i < mdsc->max_sessions; i++)
3453		− if (mdsc->sessions[i])
3454		− n++;
3455		− mutex_unlock(&mdsc->mutex);
3456		− return n == 0;
	3515	+ return atomic_read(&mdsc->num_sessions) == 0;
3457	3516	}
3458	3517
3459	3518	/*

fs/ceph/mds_client.h+2−0

@@ -137,6 +137,7 @@ struct ceph_mds_session { · 136 unmodified lines
137	137	int s_nr_caps, s_trim_caps;
138	138	int s_num_cap_releases;
139	139	int s_cap_reconnect;
	140	+ int s_readonly;
140	141	struct list_head s_cap_releases; /* waiting cap_release messages */
141	142	struct list_head s_cap_releases_done; /* ready to send */
142	143	struct ceph_cap *s_cap_iterator;
@@ -272,6 +273,7 @@ struct ceph_mds_client { · 129 unmodified lines
272	273	struct list_head waiting_for_map;
273	274
274	275	struct ceph_mds_session *sessions; / NULL for mds if no session */
	276	+ atomic_t num_sessions;
275	277	int max_sessions; /* len of s_mds_sessions */
276	278	int stopping; /* true if shutting down */
277	279

fs/ceph/snap.c+38−16

@@ -70,13 +70,11 @@ void ceph_get_snap_realm(struct ceph_mds_client *mdsc, · 69 unmodified lines
70	70	* safe. we do need to protect against concurrent empty list
71	71	* additions, however.
72	72	*/
73		− if (atomic_read(&realm->nref) == 0) {
	73	+ if (atomic_inc_return(&realm->nref) == 1) {
74	74	spin_lock(&mdsc->snap_empty_lock);
75	75	list_del_init(&realm->empty_item);
76	76	spin_unlock(&mdsc->snap_empty_lock);
77	77	}
78		−
79		− atomic_inc(&realm->nref);
80	78	}
81	79
82	80	static void __insert_snap_realm(struct rb_root *root,
@@ -116,7 +114,7 @@ static struct ceph_snap_realm *ceph_create_snap_realm( · 33 unmodified lines
116	114	if (!realm)
117	115	return ERR_PTR(-ENOMEM);
118	116
119		− atomic_set(&realm->nref, 0); /* tree does not take a ref */
	117	+ atomic_set(&realm->nref, 1); /* for caller */
120	118	realm->ino = ino;
121	119	INIT_LIST_HEAD(&realm->children);
122	120	INIT_LIST_HEAD(&realm->child_item);
@@ -134,8 +132,8 @@ static struct ceph_snap_realm *ceph_create_snap_realm( · 11 unmodified lines
134	132	*
135	133	* caller must hold snap_rwsem for write.
136	134	*/
137		−struct ceph_snap_realm ceph_lookup_snap_realm(struct ceph_mds_client mdsc,
138		− u64 ino)
	135	+static struct ceph_snap_realm __lookup_snap_realm(struct ceph_mds_client mdsc,
	136	+ u64 ino)
139	137	{
140	138	struct rb_node *n = mdsc->snap_realms.rb_node;
141	139	struct ceph_snap_realm *r;
@@ -154,6 +152,16 @@ struct ceph_snap_realm ceph_lookup_snap_realm(struct ceph_mds_client mdsc, · 12 unmodified lines
154	152	return NULL;
155	153	}
156	154
	155	+struct ceph_snap_realm ceph_lookup_snap_realm(struct ceph_mds_client mdsc,
	156	+ u64 ino)
	157	+{
	158	+ struct ceph_snap_realm *r;
	159	+ r = __lookup_snap_realm(mdsc, ino);
	160	+ if (r)
	161	+ ceph_get_snap_realm(mdsc, r);
	162	+ return r;
	163	+}
	164	+
157	165	static void __put_snap_realm(struct ceph_mds_client *mdsc,
158	166	struct ceph_snap_realm *realm);
159	167
@@ -273,7 +281,6 @@ static int adjust_snap_realm_parent(struct ceph_mds_client *mdsc, · 113 unmodified lines
273	281	}
274	282	realm->parent_ino = parentino;
275	283	realm->parent = parent;
276		− ceph_get_snap_realm(mdsc, parent);
277	284	list_add(&realm->child_item, &parent->children);
278	285	return 1;
279	286	}
@@ -631,12 +638,14 @@ static void queue_realm_cap_snaps(struct ceph_snap_realm *realm) · 351 unmodified lines
631	638	* Caller must hold snap_rwsem for write.
632	639	*/
633	640	int ceph_update_snap_trace(struct ceph_mds_client *mdsc,
634		− void p, void e, bool deletion)
	641	+ void p, void e, bool deletion,
	642	+ struct ceph_snap_realm **realm_ret)
635	643	{
636	644	struct ceph_mds_snap_realm ri; / encoded */
637	645	__le64 snaps; / encoded */
638	646	__le64 prior_parent_snaps; / encoded */
639		− struct ceph_snap_realm *realm;
	647	+ struct ceph_snap_realm *realm = NULL;
	648	+ struct ceph_snap_realm *first_realm = NULL;
640	649	int invalidate = 0;
641	650	int err = -ENOMEM;
642	651	LIST_HEAD(dirty_realms);
@@ -704,13 +713,18 @@ more: · 61 unmodified lines
704	713	dout("done with %llx %p, invalidated=%d, %p %p\n", realm->ino,
705	714	realm, invalidate, p, e);
706	715
707		− if (p < e)
708		− goto more;
709		−
710	716	/* invalidate when we reach the _end_ (root) of the trace */
711		− if (invalidate)
	717	+ if (invalidate && p >= e)
712	718	rebuild_snap_realms(realm);
713	719
	720	+ if (!first_realm)
	721	+ first_realm = realm;
	722	+ else
	723	+ ceph_put_snap_realm(mdsc, realm);
	724	+
	725	+ if (p < e)
	726	+ goto more;
	727	+
714	728	/*
715	729	* queue cap snaps _after_ we've built the new snap contexts,
716	730	* so that i_head_snapc can be set appropriately.
@@ -721,12 +735,21 @@ more: · 4 unmodified lines
721	735	queue_realm_cap_snaps(realm);
722	736	}
723	737
	738	+ if (realm_ret)
	739	+ *realm_ret = first_realm;
	740	+ else
	741	+ ceph_put_snap_realm(mdsc, first_realm);
	742	+
724	743	__cleanup_empty_realms(mdsc);
725	744	return 0;
726	745
727	746	bad:
728	747	err = -EINVAL;
729	748	fail:
	749	+ if (realm && !IS_ERR(realm))
	750	+ ceph_put_snap_realm(mdsc, realm);
	751	+ if (first_realm)
	752	+ ceph_put_snap_realm(mdsc, first_realm);
730	753	pr_err("update_snap_trace error %d\n", err);
731	754	return err;
732	755	}
@@ -844,7 +867,6 @@ void ceph_handle_snap(struct ceph_mds_client *mdsc, · 111 unmodified lines
844	867	if (IS_ERR(realm))
845	868	goto out;
846	869	}
847		− ceph_get_snap_realm(mdsc, realm);
848	870
849	871	dout("splitting snap_realm %llx %p\n", realm->ino, realm);
850	872	for (i = 0; i < num_split_inos; i++) {
@@ -905,7 +927,7 @@ skip_inode: · 54 unmodified lines
905	927	/* we may have taken some of the old realm's children. */
906	928	for (i = 0; i < num_split_realms; i++) {
907	929	struct ceph_snap_realm *child =
908		− ceph_lookup_snap_realm(mdsc,
	930	+ __lookup_snap_realm(mdsc,
909	931	le64_to_cpu(split_realms[i]));
910	932	if (!child)
911	933	continue;
@@ -918,7 +940,7 @@ skip_inode: · 6 unmodified lines
918	940	* snap, we can avoid queueing cap_snaps.
919	941	*/
920	942	ceph_update_snap_trace(mdsc, p, e,
921		− op == CEPH_SNAP_OP_DESTROY);
	943	+ op == CEPH_SNAP_OP_DESTROY, NULL);
922	944
923	945	if (op == CEPH_SNAP_OP_SPLIT)
924	946	/* we took a reference when we created the realm, above */

fs/ceph/super.c+4−0

@@ -414,6 +414,10 @@ static int ceph_show_options(struct seq_file m, struct dentry root) · 413 unmodified lines
414	414	seq_puts(m, ",noshare");
415	415	if (opt->flags & CEPH_OPT_NOCRC)
416	416	seq_puts(m, ",nocrc");
	417	+ if (opt->flags & CEPH_OPT_NOMSGAUTH)
	418	+ seq_puts(m, ",nocephx_require_signatures");
	419	+ if ((opt->flags & CEPH_OPT_TCP_NODELAY) == 0)
	420	+ seq_puts(m, ",notcp_nodelay");
417	421
418	422	if (opt->name)
419	423	seq_printf(m, ",name=%s", opt->name);

fs/ceph/super.h+4−1

@@ -693,7 +693,8 @@ extern void ceph_get_snap_realm(struct ceph_mds_client *mdsc, · 692 unmodified lines
693	693	extern void ceph_put_snap_realm(struct ceph_mds_client *mdsc,
694	694	struct ceph_snap_realm *realm);
695	695	extern int ceph_update_snap_trace(struct ceph_mds_client *m,
696		− void p, void e, bool deletion);
	696	+ void p, void e, bool deletion,
	697	+ struct ceph_snap_realm **realm_ret);
697	698	extern void ceph_handle_snap(struct ceph_mds_client *mdsc,
698	699	struct ceph_mds_session *session,
699	700	struct ceph_msg *msg);
@@ -892,7 +893,9 @@ extern void ceph_fill_inline_data(struct inode inode, struct page locked_page, · 192 unmodified lines
892	893	int ceph_uninline_data(struct file filp, struct page locked_page);
893	894	/* dir.c */
894	895	extern const struct file_operations ceph_dir_fops;
	896	+extern const struct file_operations ceph_snapdir_fops;
895	897	extern const struct inode_operations ceph_dir_iops;
	898	+extern const struct inode_operations ceph_snapdir_iops;
896	899	extern const struct dentry_operations ceph_dentry_ops, ceph_snap_dentry_ops,
897	900	ceph_snapdir_dentry_ops;
898	901

include/linux/ceph/ceph_fs.h+1−36

@@ -158,17 +158,6 @@ enum { · 157 unmodified lines
158	158	};
159	159
160	160
161		−/* pool operations */
162		−enum {
163		− POOL_OP_CREATE = 0x01,
164		− POOL_OP_DELETE = 0x02,
165		− POOL_OP_AUID_CHANGE = 0x03,
166		− POOL_OP_CREATE_SNAP = 0x11,
167		− POOL_OP_DELETE_SNAP = 0x12,
168		− POOL_OP_CREATE_UNMANAGED_SNAP = 0x21,
169		− POOL_OP_DELETE_UNMANAGED_SNAP = 0x22,
170		−};
171		−
172	161	struct ceph_mon_request_header {
173	162	__le64 have_version;
174	163	__le16 session_mon;
@@ -191,31 +180,6 @@ struct ceph_mon_statfs_reply { · 16 unmodified lines
191	180	struct ceph_statfs st;
192	181	} __attribute__ ((packed));
193	182
194		−const char *ceph_pool_op_name(int op);
195		−
196		−struct ceph_mon_poolop {
197		− struct ceph_mon_request_header monhdr;
198		− struct ceph_fsid fsid;
199		− __le32 pool;
200		− __le32 op;
201		− __le64 auid;
202		− __le64 snapid;
203		− __le32 name_len;
204		−} __attribute__ ((packed));
205		−
206		−struct ceph_mon_poolop_reply {
207		− struct ceph_mon_request_header monhdr;
208		− struct ceph_fsid fsid;
209		− __le32 reply_code;
210		− __le32 epoch;
211		− char has_data;
212		− char data[0];
213		−} __attribute__ ((packed));
214		−
215		−struct ceph_mon_unmanaged_snap {
216		− __le64 snapid;
217		−} __attribute__ ((packed));
218		−
219	183	struct ceph_osd_getmap {
220	184	struct ceph_mon_request_header monhdr;
221	185	struct ceph_fsid fsid;
@@ -307,6 +271,7 @@ enum { · 85 unmodified lines
307	271	CEPH_SESSION_RECALL_STATE,
308	272	CEPH_SESSION_FLUSHMSG,
309	273	CEPH_SESSION_FLUSHMSG_ACK,
	274	+ CEPH_SESSION_FORCE_RO,
310	275	};
311	276
312	277	extern const char *ceph_session_op_name(int op);

include/linux/ceph/libceph.h+2−1

@@ -30,8 +30,9 @@ · 29 unmodified lines
30	30	#define CEPH_OPT_MYIP (1<<2) /* specified my ip */
31	31	#define CEPH_OPT_NOCRC (1<<3) /* no data crc on writes */
32	32	#define CEPH_OPT_NOMSGAUTH (1<<4) /* not require cephx message signature */
	33	+#define CEPH_OPT_TCP_NODELAY (1<<5) /* TCP_NODELAY on TCP sockets */
33	34
34		−#define CEPH_OPT_DEFAULT (0)
	35	+#define CEPH_OPT_DEFAULT (CEPH_OPT_TCP_NODELAY)
35	36
36	37	#define ceph_set_opt(client, opt) \
37	38	(client)->options->flags \|= CEPH_OPT_##opt;

include/linux/ceph/messenger.h+3−1

@@ -57,6 +57,7 @@ struct ceph_messenger { · 56 unmodified lines
57	57
58	58	atomic_t stopping;
59	59	bool nocrc;
	60	+ bool tcp_nodelay;
60	61
61	62	/*
62	63	* the global_seq counts connections i (attempt to) initiate
@@ -264,7 +265,8 @@ extern void ceph_messenger_init(struct ceph_messenger *msgr, · 201 unmodified lines
264	265	struct ceph_entity_addr *myaddr,
265	266	u64 supported_features,
266	267	u64 required_features,
267		− bool nocrc);
	268	+ bool nocrc,
	269	+ bool tcp_nodelay);
268	270
269	271	extern void ceph_con_init(struct ceph_connection con, void private,
270	272	const struct ceph_connection_operations *ops,

include/linux/ceph/mon_client.h+1−8

@@ -40,7 +40,7 @@ struct ceph_mon_request { · 39 unmodified lines
40	40	};
41	41
42	42	/*
43		− * ceph_mon_generic_request is being used for the statfs, poolop and
	43	+ * ceph_mon_generic_request is being used for the statfs and
44	44	* mon_get_version requests which are being done a bit differently
45	45	* because we need to get data back to the caller
46	46	*/
@@ -50,7 +50,6 @@ struct ceph_mon_generic_request { · 3 unmodified lines
50	50	struct rb_node node;
51	51	int result;
52	52	void *buf;
53		− int buf_len;
54	53	struct completion completion;
55	54	struct ceph_msg request; / original request */
56	55	struct ceph_msg reply; / and reply */
@@ -117,10 +116,4 @@ extern int ceph_monc_open_session(struct ceph_mon_client *monc); · 60 unmodified lines
117	116
118	117	extern int ceph_monc_validate_auth(struct ceph_mon_client *monc);
119	118
120		−extern int ceph_monc_create_snapid(struct ceph_mon_client *monc,
121		− u32 pool, u64 *snapid);
122		−
123		−extern int ceph_monc_delete_snapid(struct ceph_mon_client *monc,
124		− u32 pool, u64 snapid);
125		−
126	119	#endif

net/ceph/ceph_common.c+15−1

@@ -239,6 +239,8 @@ enum { · 238 unmodified lines
239	239	Opt_nocrc,
240	240	Opt_cephx_require_signatures,
241	241	Opt_nocephx_require_signatures,
	242	+ Opt_tcp_nodelay,
	243	+ Opt_notcp_nodelay,
242	244	};
243	245
244	246	static match_table_t opt_tokens = {
@@ -259,6 +261,8 @@ static match_table_t opt_tokens = { · 14 unmodified lines
259	261	{Opt_nocrc, "nocrc"},
260	262	{Opt_cephx_require_signatures, "cephx_require_signatures"},
261	263	{Opt_nocephx_require_signatures, "nocephx_require_signatures"},
	264	+ {Opt_tcp_nodelay, "tcp_nodelay"},
	265	+ {Opt_notcp_nodelay, "notcp_nodelay"},
262	266	{-1, NULL}
263	267	};
264	268
@@ -457,6 +461,7 @@ ceph_parse_options(char options, const char dev_name, · 192 unmodified lines
457	461	case Opt_nocrc:
458	462	opt->flags \|= CEPH_OPT_NOCRC;
459	463	break;
	464	+
460	465	case Opt_cephx_require_signatures:
461	466	opt->flags &= ~CEPH_OPT_NOMSGAUTH;
462	467	break;
@@ -464,6 +469,13 @@ ceph_parse_options(char options, const char dev_name, · 1 unmodified line
464	469	opt->flags \|= CEPH_OPT_NOMSGAUTH;
465	470	break;
466	471
	472	+ case Opt_tcp_nodelay:
	473	+ opt->flags \|= CEPH_OPT_TCP_NODELAY;
	474	+ break;
	475	+ case Opt_notcp_nodelay:
	476	+ opt->flags &= ~CEPH_OPT_TCP_NODELAY;
	477	+ break;
	478	+
467	479	default:
468	480	BUG_ON(token);
469	481	}
@@ -518,10 +530,12 @@ struct ceph_client ceph_create_client(struct ceph_options opt, void *private, · 48 unmodified lines
518	530	/* msgr */
519	531	if (ceph_test_opt(client, MYIP))
520	532	myaddr = &client->options->my_addr;
	533	+
521	534	ceph_messenger_init(&client->msgr, myaddr,
522	535	client->supported_features,
523	536	client->required_features,
524		− ceph_test_opt(client, NOCRC));
	537	+ ceph_test_opt(client, NOCRC),
	538	+ ceph_test_opt(client, TCP_NODELAY));
525	539
526	540	/* subsystems */
527	541	err = ceph_monc_init(&client->monc, client);

net/ceph/ceph_strings.c+0−14

@@ -42,17 +42,3 @@ const char *ceph_osd_state_name(int s) · 41 unmodified lines
42	42	return "???";
43	43	}
44	44	}
45		−
46		−const char *ceph_pool_op_name(int op)
47		−{
48		− switch (op) {
49		− case POOL_OP_CREATE: return "create";
50		− case POOL_OP_DELETE: return "delete";
51		− case POOL_OP_AUID_CHANGE: return "auid change";
52		− case POOL_OP_CREATE_SNAP: return "create snap";
53		− case POOL_OP_DELETE_SNAP: return "delete snap";
54		− case POOL_OP_CREATE_UNMANAGED_SNAP: return "create unmanaged snap";
55		− case POOL_OP_DELETE_UNMANAGED_SNAP: return "delete unmanaged snap";
56		− }
57		− return "???";
58		−}

net/ceph/debugfs.c+0−2

@@ -127,8 +127,6 @@ static int monc_show(struct seq_file s, void p) · 126 unmodified lines
127	127	op = le16_to_cpu(req->request->hdr.type);
128	128	if (op == CEPH_MSG_STATFS)
129	129	seq_printf(s, "%llu statfs\n", req->tid);
130		− else if (op == CEPH_MSG_POOLOP)
131		− seq_printf(s, "%llu poolop\n", req->tid);
132	130	else if (op == CEPH_MSG_MON_GET_VERSION)
133	131	seq_printf(s, "%llu mon_get_version", req->tid);
134	132	else

net/ceph/messenger.c+13−1

@@ -510,6 +510,16 @@ static int ceph_tcp_connect(struct ceph_connection *con) · 509 unmodified lines
510	510	return ret;
511	511	}
512	512
	513	+ if (con->msgr->tcp_nodelay) {
	514	+ int optval = 1;
	515	+
	516	+ ret = kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY,
	517	+ (char *)&optval, sizeof(optval));
	518	+ if (ret)
	519	+ pr_err("kernel_setsockopt(TCP_NODELAY) failed: %d",
	520	+ ret);
	521	+ }
	522	+
513	523	sk_set_memalloc(sock->sk);
514	524
515	525	con->sock = sock;
@@ -2922,7 +2932,8 @@ void ceph_messenger_init(struct ceph_messenger *msgr, · 2406 unmodified lines
2922	2932	struct ceph_entity_addr *myaddr,
2923	2933	u64 supported_features,
2924	2934	u64 required_features,
2925		− bool nocrc)
	2935	+ bool nocrc,
	2936	+ bool tcp_nodelay)
2926	2937	{
2927	2938	msgr->supported_features = supported_features;
2928	2939	msgr->required_features = required_features;
@@ -2937,6 +2948,7 @@ void ceph_messenger_init(struct ceph_messenger *msgr, · 8 unmodified lines
2937	2948	get_random_bytes(&msgr->inst.addr.nonce, sizeof(msgr->inst.addr.nonce));
2938	2949	encode_my_addr(msgr);
2939	2950	msgr->nocrc = nocrc;
	2951	+ msgr->tcp_nodelay = tcp_nodelay;
2940	2952
2941	2953	atomic_set(&msgr->stopping, 0);
2942	2954

net/ceph/mon_client.c+5−134

@@ -410,7 +410,7 @@ out_unlocked: · 409 unmodified lines
410	410	}
411	411
412	412	/*
413		− * generic requests (e.g., statfs, poolop)
	413	+ * generic requests (currently statfs, mon_get_version)
414	414	*/
415	415	static struct ceph_mon_generic_request *__lookup_generic_req(
416	416	struct ceph_mon_client *monc, u64 tid)
@@ -569,7 +569,7 @@ static void handle_statfs_reply(struct ceph_mon_client *monc, · 152 unmodified lines
569	569	return;
570	570
571	571	bad:
572		− pr_err("corrupt generic reply, tid %llu\n", tid);
	572	+ pr_err("corrupt statfs reply, tid %llu\n", tid);
573	573	ceph_msg_dump(msg);
574	574	}
575	575
@@ -588,7 +588,6 @@ int ceph_monc_do_statfs(struct ceph_mon_client monc, struct ceph_statfs buf) · 12 unmodified lines
588	588
589	589	kref_init(&req->kref);
590	590	req->buf = buf;
591		− req->buf_len = sizeof(*buf);
592	591	init_completion(&req->completion);
593	592
594	593	err = -ENOMEM;
@@ -611,7 +610,7 @@ int ceph_monc_do_statfs(struct ceph_mon_client monc, struct ceph_statfs buf) · 16 unmodified lines
611	610	err = do_generic_request(monc, req);
612	611
613	612	out:
614		− kref_put(&req->kref, release_generic_request);
	613	+ put_generic_request(req);
615	614	return err;
616	615	}
617	616	EXPORT_SYMBOL(ceph_monc_do_statfs);
@@ -647,7 +646,7 @@ static void handle_get_version_reply(struct ceph_mon_client *monc, · 29 unmodified lines
647	646
648	647	return;
649	648	bad:
650		− pr_err("corrupt mon_get_version reply\n");
	649	+ pr_err("corrupt mon_get_version reply, tid %llu\n", tid);
651	650	ceph_msg_dump(msg);
652	651	}
653	652
@@ -670,7 +669,6 @@ int ceph_monc_do_get_version(struct ceph_mon_client monc, const char what, · 16 unmodified lines
670	669
671	670	kref_init(&req->kref);
672	671	req->buf = newest;
673		− req->buf_len = sizeof(*newest);
674	672	init_completion(&req->completion);
675	673
676	674	req->request = ceph_msg_new(CEPH_MSG_MON_GET_VERSION,
@@ -701,133 +699,11 @@ int ceph_monc_do_get_version(struct ceph_mon_client monc, const char what, · 24 unmodified lines
701	699
702	700	mutex_unlock(&monc->mutex);
703	701	out:
704		− kref_put(&req->kref, release_generic_request);
	702	+ put_generic_request(req);
705	703	return err;
706	704	}
707	705	EXPORT_SYMBOL(ceph_monc_do_get_version);
708	706
709		−/*
710		− * pool ops
711		− */
712		−static int get_poolop_reply_buf(const char *src, size_t src_len,
713		− char *dst, size_t dst_len)
714		−{
715		− u32 buf_len;
716		−
717		− if (src_len != sizeof(u32) + dst_len)
718		− return -EINVAL;
719		−
720		− buf_len = le32_to_cpu((__le32 )src);
721		− if (buf_len != dst_len)
722		− return -EINVAL;
723		−
724		− memcpy(dst, src + sizeof(u32), dst_len);
725		− return 0;
726		−}
727		−
728		−static void handle_poolop_reply(struct ceph_mon_client *monc,
729		− struct ceph_msg *msg)
730		−{
731		− struct ceph_mon_generic_request *req;
732		− struct ceph_mon_poolop_reply *reply = msg->front.iov_base;
733		− u64 tid = le64_to_cpu(msg->hdr.tid);
734		−
735		− if (msg->front.iov_len < sizeof(*reply))
736		− goto bad;
737		− dout("handle_poolop_reply %p tid %llu\n", msg, tid);
738		−
739		− mutex_lock(&monc->mutex);
740		− req = __lookup_generic_req(monc, tid);
741		− if (req) {
742		− if (req->buf_len &&
743		− get_poolop_reply_buf(msg->front.iov_base + sizeof(*reply),
744		− msg->front.iov_len - sizeof(*reply),
745		− req->buf, req->buf_len) < 0) {
746		− mutex_unlock(&monc->mutex);
747		− goto bad;
748		− }
749		− req->result = le32_to_cpu(reply->reply_code);
750		− get_generic_request(req);
751		− }
752		− mutex_unlock(&monc->mutex);
753		− if (req) {
754		− complete(&req->completion);
755		− put_generic_request(req);
756		− }
757		− return;
758		−
759		−bad:
760		− pr_err("corrupt generic reply, tid %llu\n", tid);
761		− ceph_msg_dump(msg);
762		−}
763		−
764		−/*
765		− * Do a synchronous pool op.
766		− */
767		−static int do_poolop(struct ceph_mon_client *monc, u32 op,
768		− u32 pool, u64 snapid,
769		− char *buf, int len)
770		−{
771		− struct ceph_mon_generic_request *req;
772		− struct ceph_mon_poolop *h;
773		− int err;
774		−
775		− req = kzalloc(sizeof(*req), GFP_NOFS);
776		− if (!req)
777		− return -ENOMEM;
778		−
779		− kref_init(&req->kref);
780		− req->buf = buf;
781		− req->buf_len = len;
782		− init_completion(&req->completion);
783		−
784		− err = -ENOMEM;
785		− req->request = ceph_msg_new(CEPH_MSG_POOLOP, sizeof(*h), GFP_NOFS,
786		− true);
787		− if (!req->request)
788		− goto out;
789		− req->reply = ceph_msg_new(CEPH_MSG_POOLOP_REPLY, 1024, GFP_NOFS,
790		− true);
791		− if (!req->reply)
792		− goto out;
793		−
794		− /* fill out request */
795		− req->request->hdr.version = cpu_to_le16(2);
796		− h = req->request->front.iov_base;
797		− h->monhdr.have_version = 0;
798		− h->monhdr.session_mon = cpu_to_le16(-1);
799		− h->monhdr.session_mon_tid = 0;
800		− h->fsid = monc->monmap->fsid;
801		− h->pool = cpu_to_le32(pool);
802		− h->op = cpu_to_le32(op);
803		− h->auid = 0;
804		− h->snapid = cpu_to_le64(snapid);
805		− h->name_len = 0;
806		−
807		− err = do_generic_request(monc, req);
808		−
809		−out:
810		− kref_put(&req->kref, release_generic_request);
811		− return err;
812		−}
813		−
814		−int ceph_monc_create_snapid(struct ceph_mon_client *monc,
815		− u32 pool, u64 *snapid)
816		−{
817		− return do_poolop(monc, POOL_OP_CREATE_UNMANAGED_SNAP,
818		− pool, 0, (char )snapid, sizeof(snapid));
819		−
820		−}
821		−EXPORT_SYMBOL(ceph_monc_create_snapid);
822		−
823		−int ceph_monc_delete_snapid(struct ceph_mon_client *monc,
824		− u32 pool, u64 snapid)
825		−{
826		− return do_poolop(monc, POOL_OP_CREATE_UNMANAGED_SNAP,
827		− pool, snapid, NULL, 0);
828		−
829		−}
830		−
831	707	/*
832	708	* Resend pending generic requests.
833	709	*/
@@ -1112,10 +988,6 @@ static void dispatch(struct ceph_connection con, struct ceph_msg msg) · 278 unmodified lines
1112	988	handle_get_version_reply(monc, msg);
1113	989	break;
1114	990
1115		− case CEPH_MSG_POOLOP_REPLY:
1116		− handle_poolop_reply(monc, msg);
1117		− break;
1118		−
1119	991	case CEPH_MSG_MON_MAP:
1120	992	ceph_monc_handle_map(monc, msg);
1121	993	break;
@@ -1154,7 +1026,6 @@ static struct ceph_msg mon_alloc_msg(struct ceph_connection con, · 32 unmodified lines
1154	1026	case CEPH_MSG_MON_SUBSCRIBE_ACK:
1155	1027	m = ceph_msg_get(monc->m_subscribe_ack);
1156	1028	break;
1157		− case CEPH_MSG_POOLOP_REPLY:
1158	1029	case CEPH_MSG_STATFS_REPLY:
1159	1030	return get_generic_reply(con, hdr, skip);
1160	1031	case CEPH_MSG_AUTH_REPLY:

net/ceph/osd_client.c+21−10

@@ -1035,10 +1035,11 @@ static void put_osd(struct ceph_osd *osd) · 1034 unmodified lines
1035	1035	{
1036	1036	dout("put_osd %p %d -> %d\n", osd, atomic_read(&osd->o_ref),
1037	1037	atomic_read(&osd->o_ref) - 1);
1038		− if (atomic_dec_and_test(&osd->o_ref) && osd->o_auth.authorizer) {
	1038	+ if (atomic_dec_and_test(&osd->o_ref)) {
1039	1039	struct ceph_auth_client *ac = osd->o_osdc->client->monc.auth;
1040	1040
1041		− ceph_auth_destroy_authorizer(ac, osd->o_auth.authorizer);
	1041	+ if (osd->o_auth.authorizer)
	1042	+ ceph_auth_destroy_authorizer(ac, osd->o_auth.authorizer);
1042	1043	kfree(osd);
1043	1044	}
1044	1045	}
@@ -1048,14 +1049,24 @@ static void put_osd(struct ceph_osd *osd) · 3 unmodified lines
1048	1049	*/
1049	1050	static void __remove_osd(struct ceph_osd_client osdc, struct ceph_osd osd)
1050	1051	{
1051		− dout("__remove_osd %p\n", osd);
	1052	+ dout("%s %p osd%d\n", __func__, osd, osd->o_osd);
1052	1053	WARN_ON(!list_empty(&osd->o_requests));
1053	1054	WARN_ON(!list_empty(&osd->o_linger_requests));
1054	1055
1055		− rb_erase(&osd->o_node, &osdc->osds);
1056	1056	list_del_init(&osd->o_osd_lru);
1057		− ceph_con_close(&osd->o_con);
1058		− put_osd(osd);
	1057	+ rb_erase(&osd->o_node, &osdc->osds);
	1058	+ RB_CLEAR_NODE(&osd->o_node);
	1059	+}
	1060	+
	1061	+static void remove_osd(struct ceph_osd_client osdc, struct ceph_osd osd)
	1062	+{
	1063	+ dout("%s %p osd%d\n", __func__, osd, osd->o_osd);
	1064	+
	1065	+ if (!RB_EMPTY_NODE(&osd->o_node)) {
	1066	+ ceph_con_close(&osd->o_con);
	1067	+ __remove_osd(osdc, osd);
	1068	+ put_osd(osd);
	1069	+ }
1059	1070	}
1060	1071
1061	1072	static void remove_all_osds(struct ceph_osd_client *osdc)
@@ -1065,7 +1076,7 @@ static void remove_all_osds(struct ceph_osd_client *osdc) · 3 unmodified lines
1065	1076	while (!RB_EMPTY_ROOT(&osdc->osds)) {
1066	1077	struct ceph_osd *osd = rb_entry(rb_first(&osdc->osds),
1067	1078	struct ceph_osd, o_node);
1068		− __remove_osd(osdc, osd);
	1079	+ remove_osd(osdc, osd);
1069	1080	}
1070	1081	mutex_unlock(&osdc->request_mutex);
1071	1082	}
@@ -1106,7 +1117,7 @@ static void remove_old_osds(struct ceph_osd_client *osdc) · 34 unmodified lines
1106	1117	list_for_each_entry_safe(osd, nosd, &osdc->osd_lru, o_osd_lru) {
1107	1118	if (time_before(jiffies, osd->lru_ttl))
1108	1119	break;
1109		− __remove_osd(osdc, osd);
	1120	+ remove_osd(osdc, osd);
1110	1121	}
1111	1122	mutex_unlock(&osdc->request_mutex);
1112	1123	}
@@ -1121,8 +1132,7 @@ static int __reset_osd(struct ceph_osd_client osdc, struct ceph_osd osd) · 8 unmodified lines
1121	1132	dout("__reset_osd %p osd%d\n", osd, osd->o_osd);
1122	1133	if (list_empty(&osd->o_requests) &&
1123	1134	list_empty(&osd->o_linger_requests)) {
1124		− __remove_osd(osdc, osd);
1125		−
	1135	+ remove_osd(osdc, osd);
1126	1136	return -ENODEV;
1127	1137	}
1128	1138
@@ -1926,6 +1936,7 @@ static void reset_changed_osds(struct ceph_osd_client *osdc) · 797 unmodified lines
1926	1936	{
1927	1937	struct rb_node p, n;
1928	1938
	1939	+ dout("%s %p\n", __func__, osdc);
1929	1940	for (p = rb_first(&osdc->osds); p; p = n) {
1930	1941	struct ceph_osd *osd = rb_entry(p, struct ceph_osd, o_node);
1931	1942