mds_client.c 101 KB
Newer Older
1
#include <linux/ceph/ceph_debug.h>
Sage Weil's avatar
Sage Weil committed
2

3
#include <linux/fs.h>
Sage Weil's avatar
Sage Weil committed
4
#include <linux/wait.h>
5
#include <linux/slab.h>
6
#include <linux/gfp.h>
Sage Weil's avatar
Sage Weil committed
7
#include <linux/sched.h>
8 9
#include <linux/debugfs.h>
#include <linux/seq_file.h>
10
#include <linux/utsname.h>
11
#include <linux/ratelimit.h>
Sage Weil's avatar
Sage Weil committed
12 13

#include "super.h"
14 15
#include "mds_client.h"

16
#include <linux/ceph/ceph_features.h>
17 18 19 20 21
#include <linux/ceph/messenger.h>
#include <linux/ceph/decode.h>
#include <linux/ceph/pagelist.h>
#include <linux/ceph/auth.h>
#include <linux/ceph/debugfs.h>
Sage Weil's avatar
Sage Weil committed
22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

/*
 * A cluster of MDS (metadata server) daemons is responsible for
 * managing the file system namespace (the directory hierarchy and
 * inodes) and for coordinating shared access to storage.  Metadata is
 * partitioning hierarchically across a number of servers, and that
 * partition varies over time as the cluster adjusts the distribution
 * in order to balance load.
 *
 * The MDS client is primarily responsible to managing synchronous
 * metadata requests for operations like open, unlink, and so forth.
 * If there is a MDS failure, we find out about it when we (possibly
 * request and) receive a new MDS map, and can resubmit affected
 * requests.
 *
 * For the most part, though, we take advantage of a lossless
 * communications channel to the MDS, and do not need to worry about
 * timing out or resubmitting requests.
 *
 * We maintain a stateful "session" with each MDS we interact with.
 * Within each session, we sent periodic heartbeat messages to ensure
 * any capabilities or leases we have been issues remain valid.  If
 * the session times out and goes stale, our leases and capabilities
 * are no longer valid.
 */

48
struct ceph_reconnect_state {
49
	int nr_caps;
50 51 52 53
	struct ceph_pagelist *pagelist;
	bool flock;
};

Sage Weil's avatar
Sage Weil committed
54 55 56
static void __wake_requests(struct ceph_mds_client *mdsc,
			    struct list_head *head);

57
static const struct ceph_connection_operations mds_con_ops;
Sage Weil's avatar
Sage Weil committed
58 59 60 61 62 63 64 65 66 67


/*
 * mds reply parsing
 */

/*
 * parse individual inode info
 */
static int parse_reply_info_in(void **p, void *end,
68
			       struct ceph_mds_reply_info_in *info,
69
			       u64 features)
Sage Weil's avatar
Sage Weil committed
70 71 72 73 74 75 76 77 78 79 80 81 82
{
	int err = -EIO;

	info->in = *p;
	*p += sizeof(struct ceph_mds_reply_inode) +
		sizeof(*info->in->fragtree.splits) *
		le32_to_cpu(info->in->fragtree.nsplits);

	ceph_decode_32_safe(p, end, info->symlink_len, bad);
	ceph_decode_need(p, end, info->symlink_len, bad);
	info->symlink = *p;
	*p += info->symlink_len;

83 84 85 86 87 88
	if (features & CEPH_FEATURE_DIRLAYOUTHASH)
		ceph_decode_copy_safe(p, end, &info->dir_layout,
				      sizeof(info->dir_layout), bad);
	else
		memset(&info->dir_layout, 0, sizeof(info->dir_layout));

Sage Weil's avatar
Sage Weil committed
89 90 91 92
	ceph_decode_32_safe(p, end, info->xattr_len, bad);
	ceph_decode_need(p, end, info->xattr_len, bad);
	info->xattr_data = *p;
	*p += info->xattr_len;
93 94 95 96 97 98 99 100 101 102

	if (features & CEPH_FEATURE_MDS_INLINE_DATA) {
		ceph_decode_64_safe(p, end, info->inline_version, bad);
		ceph_decode_32_safe(p, end, info->inline_len, bad);
		ceph_decode_need(p, end, info->inline_len, bad);
		info->inline_data = *p;
		*p += info->inline_len;
	} else
		info->inline_version = CEPH_INLINE_NONE;

Sage Weil's avatar
Sage Weil committed
103 104 105 106 107 108 109 110 111 112
	return 0;
bad:
	return err;
}

/*
 * parse a normal reply, which may contain a (dir+)dentry and/or a
 * target inode.
 */
static int parse_reply_info_trace(void **p, void *end,
113
				  struct ceph_mds_reply_info_parsed *info,
114
				  u64 features)
Sage Weil's avatar
Sage Weil committed
115 116 117 118
{
	int err;

	if (info->head->is_dentry) {
119
		err = parse_reply_info_in(p, end, &info->diri, features);
Sage Weil's avatar
Sage Weil committed
120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139
		if (err < 0)
			goto out_bad;

		if (unlikely(*p + sizeof(*info->dirfrag) > end))
			goto bad;
		info->dirfrag = *p;
		*p += sizeof(*info->dirfrag) +
			sizeof(u32)*le32_to_cpu(info->dirfrag->ndist);
		if (unlikely(*p > end))
			goto bad;

		ceph_decode_32_safe(p, end, info->dname_len, bad);
		ceph_decode_need(p, end, info->dname_len, bad);
		info->dname = *p;
		*p += info->dname_len;
		info->dlease = *p;
		*p += sizeof(*info->dlease);
	}

	if (info->head->is_target) {
140
		err = parse_reply_info_in(p, end, &info->targeti, features);
Sage Weil's avatar
Sage Weil committed
141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159
		if (err < 0)
			goto out_bad;
	}

	if (unlikely(*p != end))
		goto bad;
	return 0;

bad:
	err = -EIO;
out_bad:
	pr_err("problem parsing mds trace %d\n", err);
	return err;
}

/*
 * parse readdir results
 */
static int parse_reply_info_dir(void **p, void *end,
160
				struct ceph_mds_reply_info_parsed *info,
161
				u64 features)
Sage Weil's avatar
Sage Weil committed
162 163 164 165 166 167 168 169 170 171 172 173 174
{
	u32 num, i = 0;
	int err;

	info->dir_dir = *p;
	if (*p + sizeof(*info->dir_dir) > end)
		goto bad;
	*p += sizeof(*info->dir_dir) +
		sizeof(u32)*le32_to_cpu(info->dir_dir->ndist);
	if (*p > end)
		goto bad;

	ceph_decode_need(p, end, sizeof(num) + 2, bad);
175 176 177
	num = ceph_decode_32(p);
	info->dir_end = ceph_decode_8(p);
	info->dir_complete = ceph_decode_8(p);
Sage Weil's avatar
Sage Weil committed
178 179 180
	if (num == 0)
		goto done;

181
	BUG_ON(!info->dir_in);
Sage Weil's avatar
Sage Weil committed
182 183 184
	info->dir_dname = (void *)(info->dir_in + num);
	info->dir_dname_len = (void *)(info->dir_dname + num);
	info->dir_dlease = (void *)(info->dir_dname_len + num);
185 186 187 188 189 190
	if ((unsigned long)(info->dir_dlease + num) >
	    (unsigned long)info->dir_in + info->dir_buf_size) {
		pr_err("dir contents are larger than expected\n");
		WARN_ON(1);
		goto bad;
	}
Sage Weil's avatar
Sage Weil committed
191

192
	info->dir_nr = num;
Sage Weil's avatar
Sage Weil committed
193 194 195
	while (num) {
		/* dentry */
		ceph_decode_need(p, end, sizeof(u32)*2, bad);
196
		info->dir_dname_len[i] = ceph_decode_32(p);
Sage Weil's avatar
Sage Weil committed
197 198 199 200 201 202 203 204 205
		ceph_decode_need(p, end, info->dir_dname_len[i], bad);
		info->dir_dname[i] = *p;
		*p += info->dir_dname_len[i];
		dout("parsed dir dname '%.*s'\n", info->dir_dname_len[i],
		     info->dir_dname[i]);
		info->dir_dlease[i] = *p;
		*p += sizeof(struct ceph_mds_reply_lease);

		/* inode */
206
		err = parse_reply_info_in(p, end, &info->dir_in[i], features);
Sage Weil's avatar
Sage Weil committed
207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224
		if (err < 0)
			goto out_bad;
		i++;
		num--;
	}

done:
	if (*p != end)
		goto bad;
	return 0;

bad:
	err = -EIO;
out_bad:
	pr_err("problem parsing dir contents %d\n", err);
	return err;
}

225 226 227 228
/*
 * parse fcntl F_GETLK results
 */
static int parse_reply_info_filelock(void **p, void *end,
229
				     struct ceph_mds_reply_info_parsed *info,
230
				     u64 features)
231 232 233 234 235 236 237 238 239 240 241 242 243 244 245
{
	if (*p + sizeof(*info->filelock_reply) > end)
		goto bad;

	info->filelock_reply = *p;
	*p += sizeof(*info->filelock_reply);

	if (unlikely(*p != end))
		goto bad;
	return 0;

bad:
	return -EIO;
}

246 247 248 249 250
/*
 * parse create results
 */
static int parse_reply_info_create(void **p, void *end,
				  struct ceph_mds_reply_info_parsed *info,
251
				  u64 features)
252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269
{
	if (features & CEPH_FEATURE_REPLY_CREATE_INODE) {
		if (*p == end) {
			info->has_create_ino = false;
		} else {
			info->has_create_ino = true;
			info->ino = ceph_decode_64(p);
		}
	}

	if (unlikely(*p != end))
		goto bad;
	return 0;

bad:
	return -EIO;
}

270 271 272 273
/*
 * parse extra results
 */
static int parse_reply_info_extra(void **p, void *end,
274
				  struct ceph_mds_reply_info_parsed *info,
275
				  u64 features)
276
{
277 278 279
	u32 op = le32_to_cpu(info->head->op);

	if (op == CEPH_MDS_OP_GETFILELOCK)
280
		return parse_reply_info_filelock(p, end, info, features);
281
	else if (op == CEPH_MDS_OP_READDIR || op == CEPH_MDS_OP_LSSNAP)
282
		return parse_reply_info_dir(p, end, info, features);
283
	else if (op == CEPH_MDS_OP_CREATE)
284 285 286
		return parse_reply_info_create(p, end, info, features);
	else
		return -EIO;
287 288
}

Sage Weil's avatar
Sage Weil committed
289 290 291 292
/*
 * parse entire mds reply
 */
static int parse_reply_info(struct ceph_msg *msg,
293
			    struct ceph_mds_reply_info_parsed *info,
294
			    u64 features)
Sage Weil's avatar
Sage Weil committed
295 296 297 298 299 300 301 302 303 304 305 306
{
	void *p, *end;
	u32 len;
	int err;

	info->head = msg->front.iov_base;
	p = msg->front.iov_base + sizeof(struct ceph_mds_reply_head);
	end = p + msg->front.iov_len - sizeof(struct ceph_mds_reply_head);

	/* trace */
	ceph_decode_32_safe(&p, end, len, bad);
	if (len > 0) {
307
		ceph_decode_need(&p, end, len, bad);
308
		err = parse_reply_info_trace(&p, p+len, info, features);
Sage Weil's avatar
Sage Weil committed
309 310 311 312
		if (err < 0)
			goto out_bad;
	}

313
	/* extra */
Sage Weil's avatar
Sage Weil committed
314 315
	ceph_decode_32_safe(&p, end, len, bad);
	if (len > 0) {
316
		ceph_decode_need(&p, end, len, bad);
317
		err = parse_reply_info_extra(&p, p+len, info, features);
Sage Weil's avatar
Sage Weil committed
318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340
		if (err < 0)
			goto out_bad;
	}

	/* snap blob */
	ceph_decode_32_safe(&p, end, len, bad);
	info->snapblob_len = len;
	info->snapblob = p;
	p += len;

	if (p != end)
		goto bad;
	return 0;

bad:
	err = -EIO;
out_bad:
	pr_err("mds parse_reply err %d\n", err);
	return err;
}

static void destroy_reply_info(struct ceph_mds_reply_info_parsed *info)
{
341 342 343
	if (!info->dir_in)
		return;
	free_pages((unsigned long)info->dir_in, get_order(info->dir_buf_size));
Sage Weil's avatar
Sage Weil committed
344 345 346 347 348 349
}


/*
 * sessions
 */
350
const char *ceph_session_state_name(int s)
Sage Weil's avatar
Sage Weil committed
351 352 353 354 355 356 357
{
	switch (s) {
	case CEPH_MDS_SESSION_NEW: return "new";
	case CEPH_MDS_SESSION_OPENING: return "opening";
	case CEPH_MDS_SESSION_OPEN: return "open";
	case CEPH_MDS_SESSION_HUNG: return "hung";
	case CEPH_MDS_SESSION_CLOSING: return "closing";
358
	case CEPH_MDS_SESSION_RESTARTING: return "restarting";
Sage Weil's avatar
Sage Weil committed
359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379
	case CEPH_MDS_SESSION_RECONNECTING: return "reconnecting";
	default: return "???";
	}
}

static struct ceph_mds_session *get_session(struct ceph_mds_session *s)
{
	if (atomic_inc_not_zero(&s->s_ref)) {
		dout("mdsc get_session %p %d -> %d\n", s,
		     atomic_read(&s->s_ref)-1, atomic_read(&s->s_ref));
		return s;
	} else {
		dout("mdsc get_session %p 0 -- FAIL", s);
		return NULL;
	}
}

void ceph_put_mds_session(struct ceph_mds_session *s)
{
	dout("mdsc put_session %p %d -> %d\n", s,
	     atomic_read(&s->s_ref), atomic_read(&s->s_ref)-1);
380
	if (atomic_dec_and_test(&s->s_ref)) {
381
		if (s->s_auth.authorizer)
382 383 384
			ceph_auth_destroy_authorizer(
				s->s_mdsc->fsc->client->monc.auth,
				s->s_auth.authorizer);
Sage Weil's avatar
Sage Weil committed
385
		kfree(s);
386
	}
Sage Weil's avatar
Sage Weil committed
387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412
}

/*
 * called under mdsc->mutex
 */
struct ceph_mds_session *__ceph_lookup_mds_session(struct ceph_mds_client *mdsc,
						   int mds)
{
	struct ceph_mds_session *session;

	if (mds >= mdsc->max_sessions || mdsc->sessions[mds] == NULL)
		return NULL;
	session = mdsc->sessions[mds];
	dout("lookup_mds_session %p %d\n", session,
	     atomic_read(&session->s_ref));
	get_session(session);
	return session;
}

static bool __have_session(struct ceph_mds_client *mdsc, int mds)
{
	if (mds >= mdsc->max_sessions)
		return false;
	return mdsc->sessions[mds];
}

413 414 415 416 417 418 419 420 421
static int __verify_registered_session(struct ceph_mds_client *mdsc,
				       struct ceph_mds_session *s)
{
	if (s->s_mds >= mdsc->max_sessions ||
	    mdsc->sessions[s->s_mds] != s)
		return -ENOENT;
	return 0;
}

Sage Weil's avatar
Sage Weil committed
422 423 424 425 426 427 428 429 430
/*
 * create+register a new session for given mds.
 * called under mdsc->mutex.
 */
static struct ceph_mds_session *register_session(struct ceph_mds_client *mdsc,
						 int mds)
{
	struct ceph_mds_session *s;

431 432 433
	if (mds >= mdsc->mdsmap->m_max_mds)
		return ERR_PTR(-EINVAL);

Sage Weil's avatar
Sage Weil committed
434
	s = kzalloc(sizeof(*s), GFP_NOFS);
435 436
	if (!s)
		return ERR_PTR(-ENOMEM);
Sage Weil's avatar
Sage Weil committed
437 438 439 440 441 442 443
	s->s_mdsc = mdsc;
	s->s_mds = mds;
	s->s_state = CEPH_MDS_SESSION_NEW;
	s->s_ttl = 0;
	s->s_seq = 0;
	mutex_init(&s->s_mutex);

444
	ceph_con_init(&s->s_con, s, &mds_con_ops, &mdsc->fsc->client->msgr);
Sage Weil's avatar
Sage Weil committed
445

446
	spin_lock_init(&s->s_gen_ttl_lock);
Sage Weil's avatar
Sage Weil committed
447
	s->s_cap_gen = 0;
448
	s->s_cap_ttl = jiffies - 1;
449 450

	spin_lock_init(&s->s_cap_lock);
Sage Weil's avatar
Sage Weil committed
451 452 453 454
	s->s_renew_requested = 0;
	s->s_renew_seq = 0;
	INIT_LIST_HEAD(&s->s_caps);
	s->s_nr_caps = 0;
455
	s->s_trim_caps = 0;
Sage Weil's avatar
Sage Weil committed
456 457 458 459
	atomic_set(&s->s_ref, 1);
	INIT_LIST_HEAD(&s->s_waiting);
	INIT_LIST_HEAD(&s->s_unsafe);
	s->s_num_cap_releases = 0;
460
	s->s_cap_reconnect = 0;
461
	s->s_cap_iterator = NULL;
Sage Weil's avatar
Sage Weil committed
462 463 464 465 466 467 468 469 470 471 472 473
	INIT_LIST_HEAD(&s->s_cap_releases);
	INIT_LIST_HEAD(&s->s_cap_flushing);
	INIT_LIST_HEAD(&s->s_cap_snaps_flushing);

	dout("register_session mds%d\n", mds);
	if (mds >= mdsc->max_sessions) {
		int newmax = 1 << get_count_order(mds+1);
		struct ceph_mds_session **sa;

		dout("register_session realloc to %d\n", newmax);
		sa = kcalloc(newmax, sizeof(void *), GFP_NOFS);
		if (sa == NULL)
474
			goto fail_realloc;
Sage Weil's avatar
Sage Weil committed
475 476 477 478 479 480 481 482 483
		if (mdsc->sessions) {
			memcpy(sa, mdsc->sessions,
			       mdsc->max_sessions * sizeof(void *));
			kfree(mdsc->sessions);
		}
		mdsc->sessions = sa;
		mdsc->max_sessions = newmax;
	}
	mdsc->sessions[mds] = s;
484
	atomic_inc(&mdsc->num_sessions);
Sage Weil's avatar
Sage Weil committed
485
	atomic_inc(&s->s_ref);  /* one ref to sessions[], one to caller */
486

487 488
	ceph_con_open(&s->s_con, CEPH_ENTITY_TYPE_MDS, mds,
		      ceph_mdsmap_get_addr(mdsc->mdsmap, mds));
489

Sage Weil's avatar
Sage Weil committed
490
	return s;
491 492 493 494

fail_realloc:
	kfree(s);
	return ERR_PTR(-ENOMEM);
Sage Weil's avatar
Sage Weil committed
495 496 497 498 499
}

/*
 * called under mdsc->mutex
 */
500
static void __unregister_session(struct ceph_mds_client *mdsc,
501
			       struct ceph_mds_session *s)
Sage Weil's avatar
Sage Weil committed
502
{
503 504
	dout("__unregister_session mds%d %p\n", s->s_mds, s);
	BUG_ON(mdsc->sessions[s->s_mds] != s);
505 506 507
	mdsc->sessions[s->s_mds] = NULL;
	ceph_con_close(&s->s_con);
	ceph_put_mds_session(s);
508
	atomic_dec(&mdsc->num_sessions);
Sage Weil's avatar
Sage Weil committed
509 510 511 512 513 514 515 516 517 518 519 520 521 522 523
}

/*
 * drop session refs in request.
 *
 * should be last request ref, or hold mdsc->mutex
 */
static void put_request_session(struct ceph_mds_request *req)
{
	if (req->r_session) {
		ceph_put_mds_session(req->r_session);
		req->r_session = NULL;
	}
}

524
void ceph_mdsc_release_request(struct kref *kref)
Sage Weil's avatar
Sage Weil committed
525
{
526 527 528
	struct ceph_mds_request *req = container_of(kref,
						    struct ceph_mds_request,
						    r_kref);
529
	destroy_reply_info(&req->r_reply_info);
530 531
	if (req->r_request)
		ceph_msg_put(req->r_request);
532
	if (req->r_reply)
533 534
		ceph_msg_put(req->r_reply);
	if (req->r_inode) {
535
		ceph_put_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
536 537 538
		iput(req->r_inode);
	}
	if (req->r_locked_dir)
539
		ceph_put_cap_refs(ceph_inode(req->r_locked_dir), CEPH_CAP_PIN);
540
	iput(req->r_target_inode);
541 542
	if (req->r_dentry)
		dput(req->r_dentry);
543 544 545
	if (req->r_old_dentry)
		dput(req->r_old_dentry);
	if (req->r_old_dentry_dir) {
546 547 548 549 550 551 552 553 554
		/*
		 * track (and drop pins for) r_old_dentry_dir
		 * separately, since r_old_dentry's d_parent may have
		 * changed between the dir mutex being dropped and
		 * this request being freed.
		 */
		ceph_put_cap_refs(ceph_inode(req->r_old_dentry_dir),
				  CEPH_CAP_PIN);
		iput(req->r_old_dentry_dir);
Sage Weil's avatar
Sage Weil committed
555
	}
556 557
	kfree(req->r_path1);
	kfree(req->r_path2);
558 559
	if (req->r_pagelist)
		ceph_pagelist_release(req->r_pagelist);
560
	put_request_session(req);
561
	ceph_unreserve_caps(req->r_mdsc, &req->r_caps_reservation);
562
	kfree(req);
Sage Weil's avatar
Sage Weil committed
563 564 565 566 567 568 569 570 571 572 573
}

/*
 * lookup session, bump ref if found.
 *
 * called under mdsc->mutex.
 */
static struct ceph_mds_request *__lookup_request(struct ceph_mds_client *mdsc,
					     u64 tid)
{
	struct ceph_mds_request *req;
574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609
	struct rb_node *n = mdsc->request_tree.rb_node;

	while (n) {
		req = rb_entry(n, struct ceph_mds_request, r_node);
		if (tid < req->r_tid)
			n = n->rb_left;
		else if (tid > req->r_tid)
			n = n->rb_right;
		else {
			ceph_mdsc_get_request(req);
			return req;
		}
	}
	return NULL;
}

static void __insert_request(struct ceph_mds_client *mdsc,
			     struct ceph_mds_request *new)
{
	struct rb_node **p = &mdsc->request_tree.rb_node;
	struct rb_node *parent = NULL;
	struct ceph_mds_request *req = NULL;

	while (*p) {
		parent = *p;
		req = rb_entry(parent, struct ceph_mds_request, r_node);
		if (new->r_tid < req->r_tid)
			p = &(*p)->rb_left;
		else if (new->r_tid > req->r_tid)
			p = &(*p)->rb_right;
		else
			BUG();
	}

	rb_link_node(&new->r_node, parent, p);
	rb_insert_color(&new->r_node, &mdsc->request_tree);
Sage Weil's avatar
Sage Weil committed
610 611 612 613 614 615 616 617 618 619 620 621 622 623
}

/*
 * Register an in-flight request, and assign a tid.  Link to directory
 * are modifying (if any).
 *
 * Called under mdsc->mutex.
 */
static void __register_request(struct ceph_mds_client *mdsc,
			       struct ceph_mds_request *req,
			       struct inode *dir)
{
	req->r_tid = ++mdsc->last_tid;
	if (req->r_num_caps)
624 625
		ceph_reserve_caps(mdsc, &req->r_caps_reservation,
				  req->r_num_caps);
Sage Weil's avatar
Sage Weil committed
626 627
	dout("__register_request %p tid %lld\n", req, req->r_tid);
	ceph_mdsc_get_request(req);
628
	__insert_request(mdsc, req);
Sage Weil's avatar
Sage Weil committed
629

630 631 632
	req->r_uid = current_fsuid();
	req->r_gid = current_fsgid();

633 634 635
	if (mdsc->oldest_tid == 0 && req->r_op != CEPH_MDS_OP_SETFILELOCK)
		mdsc->oldest_tid = req->r_tid;

Sage Weil's avatar
Sage Weil committed
636
	if (dir) {
637
		ihold(dir);
Sage Weil's avatar
Sage Weil committed
638 639 640 641 642 643 644 645
		req->r_unsafe_dir = dir;
	}
}

static void __unregister_request(struct ceph_mds_client *mdsc,
				 struct ceph_mds_request *req)
{
	dout("__unregister_request %p tid %lld\n", req, req->r_tid);
646

647 648 649
	/* Never leave an unregistered request on an unsafe list! */
	list_del_init(&req->r_unsafe_item);

650 651 652 653 654 655 656 657 658 659 660 661 662 663
	if (req->r_tid == mdsc->oldest_tid) {
		struct rb_node *p = rb_next(&req->r_node);
		mdsc->oldest_tid = 0;
		while (p) {
			struct ceph_mds_request *next_req =
				rb_entry(p, struct ceph_mds_request, r_node);
			if (next_req->r_op != CEPH_MDS_OP_SETFILELOCK) {
				mdsc->oldest_tid = next_req->r_tid;
				break;
			}
			p = rb_next(p);
		}
	}

664
	rb_erase(&req->r_node, &mdsc->request_tree);
665
	RB_CLEAR_NODE(&req->r_node);
Sage Weil's avatar
Sage Weil committed
666

667
	if (req->r_unsafe_dir && req->r_got_unsafe) {
Sage Weil's avatar
Sage Weil committed
668 669 670 671
		struct ceph_inode_info *ci = ceph_inode(req->r_unsafe_dir);
		spin_lock(&ci->i_unsafe_lock);
		list_del_init(&req->r_unsafe_dir_item);
		spin_unlock(&ci->i_unsafe_lock);
672
	}
673 674 675 676 677 678
	if (req->r_target_inode && req->r_got_unsafe) {
		struct ceph_inode_info *ci = ceph_inode(req->r_target_inode);
		spin_lock(&ci->i_unsafe_lock);
		list_del_init(&req->r_unsafe_target_item);
		spin_unlock(&ci->i_unsafe_lock);
	}
679

680
	if (req->r_unsafe_dir) {
681 682
		iput(req->r_unsafe_dir);
		req->r_unsafe_dir = NULL;
Sage Weil's avatar
Sage Weil committed
683
	}
684

685 686
	complete_all(&req->r_safe_completion);

687
	ceph_mdsc_put_request(req);
Sage Weil's avatar
Sage Weil committed
688 689 690 691 692 693 694 695 696 697
}

/*
 * Choose mds to send request to next.  If there is a hint set in the
 * request (e.g., due to a prior forward hint from the mds), use that.
 * Otherwise, consult frag tree and/or caps to identify the
 * appropriate mds.  If all else fails, choose randomly.
 *
 * Called under mdsc->mutex.
 */
698
static struct dentry *get_nonsnap_parent(struct dentry *dentry)
699
{
700 701 702 703 704 705
	/*
	 * we don't need to worry about protecting the d_parent access
	 * here because we never renaming inside the snapped namespace
	 * except to resplice to another snapdir, and either the old or new
	 * result is a valid result.
	 */
706
	while (!IS_ROOT(dentry) && ceph_snap(d_inode(dentry)) != CEPH_NOSNAP)
707 708 709 710
		dentry = dentry->d_parent;
	return dentry;
}

Sage Weil's avatar
Sage Weil committed
711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740
static int __choose_mds(struct ceph_mds_client *mdsc,
			struct ceph_mds_request *req)
{
	struct inode *inode;
	struct ceph_inode_info *ci;
	struct ceph_cap *cap;
	int mode = req->r_direct_mode;
	int mds = -1;
	u32 hash = req->r_direct_hash;
	bool is_hash = req->r_direct_is_hash;

	/*
	 * is there a specific mds we should try?  ignore hint if we have
	 * no session and the mds is not up (active or recovering).
	 */
	if (req->r_resend_mds >= 0 &&
	    (__have_session(mdsc, req->r_resend_mds) ||
	     ceph_mdsmap_get_state(mdsc->mdsmap, req->r_resend_mds) > 0)) {
		dout("choose_mds using resend_mds mds%d\n",
		     req->r_resend_mds);
		return req->r_resend_mds;
	}

	if (mode == USE_RANDOM_MDS)
		goto random;

	inode = NULL;
	if (req->r_inode) {
		inode = req->r_inode;
	} else if (req->r_dentry) {
741 742
		/* ignore race with rename; old or new d_parent is okay */
		struct dentry *parent = req->r_dentry->d_parent;
743
		struct inode *dir = d_inode(parent);
744

745
		if (dir->i_sb != mdsc->fsc->sb) {
746
			/* not this fs! */
747
			inode = d_inode(req->r_dentry);
748 749 750
		} else if (ceph_snap(dir) != CEPH_NOSNAP) {
			/* direct snapped/virtual snapdir requests
			 * based on parent dir inode */
751
			struct dentry *dn = get_nonsnap_parent(parent);
752
			inode = d_inode(dn);
753
			dout("__choose_mds using nonsnap parent %p\n", inode);
Yan, Zheng's avatar
Yan, Zheng committed
754
		} else {
755
			/* dentry target */
756
			inode = d_inode(req->r_dentry);
Yan, Zheng's avatar
Yan, Zheng committed
757 758 759 760 761 762
			if (!inode || mode == USE_AUTH_MDS) {
				/* dir + name */
				inode = dir;
				hash = ceph_dentry_hash(dir, req->r_dentry);
				is_hash = true;
			}
Sage Weil's avatar
Sage Weil committed
763 764
		}
	}
765

Sage Weil's avatar
Sage Weil committed
766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787
	dout("__choose_mds %p is_hash=%d (%d) mode %d\n", inode, (int)is_hash,
	     (int)hash, mode);
	if (!inode)
		goto random;
	ci = ceph_inode(inode);

	if (is_hash && S_ISDIR(inode->i_mode)) {
		struct ceph_inode_frag frag;
		int found;

		ceph_choose_frag(ci, hash, &frag, &found);
		if (found) {
			if (mode == USE_ANY_MDS && frag.ndist > 0) {
				u8 r;

				/* choose a random replica */
				get_random_bytes(&r, 1);
				r %= frag.ndist;
				mds = frag.dist[r];
				dout("choose_mds %p %llx.%llx "
				     "frag %u mds%d (%d/%d)\n",
				     inode, ceph_vinop(inode),
788
				     frag.frag, mds,
Sage Weil's avatar
Sage Weil committed
789
				     (int)r, frag.ndist);
790 791 792
				if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=
				    CEPH_MDS_STATE_ACTIVE)
					return mds;
Sage Weil's avatar
Sage Weil committed
793 794 795 796 797 798 799 800 801 802 803 804
			}

			/* since this file/dir wasn't known to be
			 * replicated, then we want to look for the
			 * authoritative mds. */
			mode = USE_AUTH_MDS;
			if (frag.mds >= 0) {
				/* choose auth mds */
				mds = frag.mds;
				dout("choose_mds %p %llx.%llx "
				     "frag %u mds%d (auth)\n",
				     inode, ceph_vinop(inode), frag.frag, mds);
805 806 807
				if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=
				    CEPH_MDS_STATE_ACTIVE)
					return mds;
Sage Weil's avatar
Sage Weil committed
808 809 810 811
			}
		}
	}

812
	spin_lock(&ci->i_ceph_lock);
Sage Weil's avatar
Sage Weil committed
813 814 815 816 817 818
	cap = NULL;
	if (mode == USE_AUTH_MDS)
		cap = ci->i_auth_cap;
	if (!cap && !RB_EMPTY_ROOT(&ci->i_caps))
		cap = rb_entry(rb_first(&ci->i_caps), struct ceph_cap, ci_node);
	if (!cap) {
819
		spin_unlock(&ci->i_ceph_lock);
Sage Weil's avatar
Sage Weil committed
820 821 822 823 824 825
		goto random;
	}
	mds = cap->session->s_mds;
	dout("choose_mds %p %llx.%llx mds%d (%scap %p)\n",
	     inode, ceph_vinop(inode), mds,
	     cap == ci->i_auth_cap ? "auth " : "", cap);
826
	spin_unlock(&ci->i_ceph_lock);
Sage Weil's avatar
Sage Weil committed
827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843
	return mds;

random:
	mds = ceph_mdsmap_get_random_mds(mdsc->mdsmap);
	dout("choose_mds chose random mds%d\n", mds);
	return mds;
}


/*
 * session messages
 */
static struct ceph_msg *create_session_msg(u32 op, u64 seq)
{
	struct ceph_msg *msg;
	struct ceph_mds_session_head *h;

844 845
	msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h), GFP_NOFS,
			   false);
846
	if (!msg) {
Sage Weil's avatar
Sage Weil committed
847
		pr_err("create_session_msg ENOMEM creating msg\n");
848
		return NULL;
Sage Weil's avatar
Sage Weil committed
849 850 851 852
	}
	h = msg->front.iov_base;
	h->op = cpu_to_le32(op);
	h->seq = cpu_to_le64(seq);
853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870

	return msg;
}

/*
 * session message, specialization for CEPH_SESSION_REQUEST_OPEN
 * to include additional client metadata fields.
 */
static struct ceph_msg *create_session_open_msg(struct ceph_mds_client *mdsc, u64 seq)
{
	struct ceph_msg *msg;
	struct ceph_mds_session_head *h;
	int i = -1;
	int metadata_bytes = 0;
	int metadata_key_count = 0;
	struct ceph_options *opt = mdsc->fsc->client->options;
	void *p;

871
	const char* metadata[][2] = {
872
		{"hostname", utsname()->nodename},
873
		{"kernel_version", utsname()->release},
874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899
		{"entity_id", opt->name ? opt->name : ""},
		{NULL, NULL}
	};

	/* Calculate serialized length of metadata */
	metadata_bytes = 4;  /* map length */
	for (i = 0; metadata[i][0] != NULL; ++i) {
		metadata_bytes += 8 + strlen(metadata[i][0]) +
			strlen(metadata[i][1]);
		metadata_key_count++;
	}

	/* Allocate the message */
	msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h) + metadata_bytes,
			   GFP_NOFS, false);
	if (!msg) {
		pr_err("create_session_msg ENOMEM creating msg\n");
		return NULL;
	}
	h = msg->front.iov_base;
	h->op = cpu_to_le32(CEPH_SESSION_REQUEST_OPEN);
	h->seq = cpu_to_le64(seq);

	/*
	 * Serialize client metadata into waiting buffer space, using
	 * the format that userspace expects for map<string, string>
John Spray's avatar
John Spray committed
900 901
	 *
	 * ClientSession messages with metadata are v2
902
	 */
John Spray's avatar
John Spray committed
903 904
	msg->hdr.version = cpu_to_le16(2);
	msg->hdr.compat_version = cpu_to_le16(1);
905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924

	/* The write pointer, following the session_head structure */
	p = msg->front.iov_base + sizeof(*h);

	/* Number of entries in the map */
	ceph_encode_32(&p, metadata_key_count);

	/* Two length-prefixed strings for each entry in the map */
	for (i = 0; metadata[i][0] != NULL; ++i) {
		size_t const key_len = strlen(metadata[i][0]);
		size_t const val_len = strlen(metadata[i][1]);

		ceph_encode_32(&p, key_len);
		memcpy(p, metadata[i][0], key_len);
		p += key_len;
		ceph_encode_32(&p, val_len);
		memcpy(p, metadata[i][1], val_len);
		p += val_len;
	}

Sage Weil's avatar
Sage Weil committed
925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947
	return msg;
}

/*
 * send session open request.
 *
 * called under mdsc->mutex
 */
static int __open_session(struct ceph_mds_client *mdsc,
			  struct ceph_mds_session *session)
{
	struct ceph_msg *msg;
	int mstate;
	int mds = session->s_mds;

	/* wait for mds to go active? */
	mstate = ceph_mdsmap_get_state(mdsc->mdsmap, mds);
	dout("open_session to mds%d (%s)\n", mds,
	     ceph_mds_state_name(mstate));
	session->s_state = CEPH_MDS_SESSION_OPENING;
	session->s_renew_requested = jiffies;

	/* send connect message */
948
	msg = create_session_open_msg(mdsc, session->s_seq);
949 950
	if (!msg)
		return -ENOMEM;
Sage Weil's avatar
Sage Weil committed
951 952 953 954
	ceph_con_send(&session->s_con, msg);
	return 0;
}

955 956 957 958 959
/*
 * open sessions for any export targets for the given mds
 *
 * called under mdsc->mutex
 */
960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991
static struct ceph_mds_session *
__open_export_target_session(struct ceph_mds_client *mdsc, int target)
{
	struct ceph_mds_session *session;

	session = __ceph_lookup_mds_session(mdsc, target);
	if (!session) {
		session = register_session(mdsc, target);
		if (IS_ERR(session))
			return session;
	}
	if (session->s_state == CEPH_MDS_SESSION_NEW ||
	    session->s_state == CEPH_MDS_SESSION_CLOSING)
		__open_session(mdsc, session);

	return session;
}

struct ceph_mds_session *
ceph_mdsc_open_export_target_session(struct ceph_mds_client *mdsc, int target)
{
	struct ceph_mds_session *session;

	dout("open_export_target_session to mds%d\n", target);

	mutex_lock(&mdsc->mutex);
	session = __open_export_target_session(mdsc, target);
	mutex_unlock(&mdsc->mutex);

	return session;
}

992 993 994 995 996 997 998 999 1000
static void __open_export_target_sessions(struct ceph_mds_client *mdsc,
					  struct ceph_mds_session *session)
{
	struct ceph_mds_info *mi;
	struct ceph_mds_session *ts;
	int i, mds = session->s_mds;

	if (mds >= mdsc->mdsmap->m_max_mds)
		return;
1001

1002 1003 1004 1005 1006
	mi = &mdsc->mdsmap->m_info[mds];
	dout("open_export_target_sessions for mds%d (%d targets)\n",
	     session->s_mds, mi->num_export_targets);

	for (i = 0; i < mi->num_export_targets; i++) {
1007 1008 1009
		ts = __open_export_target_session(mdsc, mi->export_targets[i]);
		if (!IS_ERR(ts))
			ceph_put_mds_session(ts);
1010 1011 1012
	}
}

1013 1014 1015 1016 1017 1018 1019 1020
void ceph_mdsc_open_export_target_sessions(struct ceph_mds_client *mdsc,
					   struct ceph_mds_session *session)
{
	mutex_lock(&mdsc->mutex);
	__open_export_target_sessions(mdsc, session);
	mutex_unlock(&mdsc->mutex);
}

Sage Weil's avatar
Sage Weil committed
1021 1022 1023 1024
/*
 * session caps
 */

1025 1026 1027 1028
/* caller holds s_cap_lock, we drop it */
static void cleanup_cap_releases(struct ceph_mds_client *mdsc,
				 struct ceph_mds_session *session)
	__releases(session->s_cap_lock)
Sage Weil's avatar
Sage Weil committed
1029
{
1030 1031 1032 1033
	LIST_HEAD(tmp_list);
	list_splice_init(&session->s_cap_releases, &tmp_list);
	session->s_num_cap_releases = 0;
	spin_unlock(&session->s_cap_lock);
Sage Weil's avatar
Sage Weil committed
1034

1035 1036 1037 1038 1039 1040 1041 1042
	dout("cleanup_cap_releases mds%d\n", session->s_mds);
	while (!list_empty(&tmp_list)) {
		struct ceph_cap *cap;
		/* zero out the in-progress message */
		cap = list_first_entry(&tmp_list,
					struct ceph_cap, session_caps);
		list_del(&cap->session_caps);
		ceph_put_cap(mdsc, cap);
Sage Weil's avatar
Sage Weil committed
1043 1044 1045
	}
}

1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056
static void cleanup_session_requests(struct ceph_mds_client *mdsc,
				     struct ceph_mds_session *session)
{
	struct ceph_mds_request *req;
	struct rb_node *p;

	dout("cleanup_session_requests mds%d\n", session->s_mds);
	mutex_lock(&mdsc->mutex);
	while (!list_empty(&session->s_unsafe)) {
		req = list_first_entry(&session->s_unsafe,
				       struct ceph_mds_request, r_unsafe_item);
1057 1058
		pr_warn_ratelimited(" dropping unsafe request %llu\n",
				    req->r_tid);
1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072
		__unregister_request(mdsc, req);
	}
	/* zero r_attempts, so kick_requests() will re-send requests */
	p = rb_first(&mdsc->request_tree);
	while (p) {
		req = rb_entry(p, struct ceph_mds_request, r_node);
		p = rb_next(p);
		if (req->r_session &&
		    req->r_session->s_mds == session->s_mds)
			req->r_attempts = 0;
	}
	mutex_unlock(&mdsc->mutex);
}

Sage Weil's avatar
Sage Weil committed
1073
/*
Sage Weil's avatar
Sage Weil committed
1074 1075
 * Helper to safely iterate over all caps associated with a session, with
 * special care taken to handle a racing __ceph_remove_cap().
Sage Weil's avatar
Sage Weil committed
1076
 *
Sage Weil's avatar
Sage Weil committed
1077
 * Caller must hold session s_mutex.
Sage Weil's avatar
Sage Weil committed
1078 1079 1080 1081 1082
 */
static int iterate_session_caps(struct ceph_mds_session *session,
				 int (*cb)(struct inode *, struct ceph_cap *,
					    void *), void *arg)
{
1083 1084 1085 1086
	struct list_head *p;
	struct ceph_cap *cap;
	struct inode *inode, *last_inode = NULL;
	struct ceph_cap *old_cap = NULL;
Sage Weil's avatar
Sage Weil committed
1087 1088 1089 1090
	int ret;

	dout("iterate_session_caps %p mds%d\n", session, session->s_mds);
	spin_lock(&session->s_cap_lock);
1091 1092 1093
	p = session->s_caps.next;
	while (p != &session->s_caps) {
		cap = list_entry(p, struct ceph_cap, session_caps);
Sage Weil's avatar
Sage Weil committed
1094
		inode = igrab(&cap->ci->vfs_inode);
1095 1096
		if (!inode) {
			p = p->next;
Sage Weil's avatar
Sage Weil committed
1097
			continue;
1098 1099
		}
		session->s_cap_iterator = cap;
Sage Weil's avatar
Sage Weil committed
1100
		spin_unlock(&session->s_cap_lock);
1101 1102 1103 1104 1105 1106

		if (last_inode) {
			iput(last_inode);
			last_inode = NULL;
		}
		if (old_cap) {
1107
			ceph_put_cap(session->s_mdsc, old_cap);
1108 1109 1110
			old_cap = NULL;
		}

Sage Weil's avatar
Sage Weil committed
1111
		ret = cb(inode, cap, arg);
1112 1113
		last_inode = inode;

Sage Weil's avatar
Sage Weil committed
1114
		spin_lock(&session->s_cap_lock);
1115 1116 1117 1118 1119
		p = p->next;
		if (cap->ci == NULL) {
			dout("iterate_session_caps  finishing cap %p removal\n",
			     cap);
			BUG_ON(cap->session != session);
1120
			cap->session = NULL;
1121 1122
			list_del_init(&cap->session_caps);
			session->s_nr_caps--;
1123 1124 1125 1126 1127 1128 1129
			if (cap->queue_release) {
				list_add_tail(&cap->session_caps,
					      &session->s_cap_releases);
				session->s_num_cap_releases++;
			} else {
				old_cap = cap;  /* put_cap it w/o locks held */
			}
1130
		}
1131 1132
		if (ret < 0)
			goto out;
Sage Weil's avatar
Sage Weil committed
1133
	}
1134 1135
	ret = 0;
out:
1136
	session->s_cap_iterator = NULL;
Sage Weil's avatar
Sage Weil committed
1137
	spin_unlock(&session->s_cap_lock);
1138

1139
	iput(last_inode);
1140
	if (old_cap)
1141
		ceph_put_cap(session->s_mdsc, old_cap);
1142

1143
	return ret;
Sage Weil's avatar
Sage Weil committed
1144 1145 1146
}

static int remove_session_caps_cb(struct inode *inode, struct ceph_cap *cap,
1147
				  void *arg)
Sage Weil's avatar
Sage Weil committed
1148 1149
{
	struct ceph_inode_info *ci = ceph_inode(inode);
1150
	LIST_HEAD(to_remove);
1151 1152
	int drop = 0;

Sage Weil's avatar
Sage Weil committed
1153 1154
	dout("removing cap %p, ci is %p, inode is %p\n",
	     cap, ci, &ci->vfs_inode);
1155
	spin_lock(&ci->i_ceph_lock);
1156
	__ceph_remove_cap(cap, false);
1157
	if (!ci->i_auth_cap) {
1158
		struct ceph_cap_flush *cf;
1159
		struct ceph_mds_client *mdsc =
1160
			ceph_sb_to_client(inode->i_sb)->mdsc;
1161

1162 1163 1164 1165 1166 1167 1168 1169 1170
		while (true) {
			struct rb_node *n = rb_first(&ci->i_cap_flush_tree);
			if (!n)
				break;
			cf = rb_entry(n, struct ceph_cap_flush, i_node);
			rb_erase(&cf->i_node, &ci->i_cap_flush_tree);
			list_add(&cf->list, &to_remove);
		}

1171
		spin_lock(&mdsc->cap_dirty_lock);
1172 1173 1174 1175

		list_for_each_entry(cf, &to_remove, list)
			rb_erase(&cf->g_node, &mdsc->cap_flush_tree);

1176
		if (!list_empty(&ci->i_dirty_item)) {
1177 1178
			pr_warn_ratelimited(
				" dropping dirty %s state for %p %lld\n",
1179 1180 1181 1182 1183 1184 1185
				ceph_cap_string(ci->i_dirty_caps),
				inode, ceph_ino(inode));
			ci->i_dirty_caps = 0;
			list_del_init(&ci->i_dirty_item);
			drop = 1;
		}
		if (!list_empty(&ci->i_flushing_item)) {
1186 1187
			pr_warn_ratelimited(
				" dropping dirty+flushing %s state for %p %lld\n",
1188 1189 1190 1191 1192 1193 1194 1195
				ceph_cap_string(ci->i_flushing_caps),
				inode, ceph_ino(inode));
			ci->i_flushing_caps = 0;
			list_del_init(&ci->i_flushing_item);
			mdsc->num_cap_flushing--;
			drop = 1;
		}
		spin_unlock(&mdsc->cap_dirty_lock);
1196

1197 1198 1199 1200
		if (!ci->i_dirty_caps && ci->i_prealloc_cap_flush) {
			list_add(&ci->i_prealloc_cap_flush->list, &to_remove);
			ci->i_prealloc_cap_flush = NULL;
		}
1201
	}
1202
	spin_unlock(&ci->i_ceph_lock);
1203 1204 1205 1206 1207
	while (!list_empty(&to_remove)) {
		struct ceph_cap_flush *cf;
		cf = list_first_entry(&to_remove,
				      struct ceph_cap_flush, list);
		list_del(&cf->list);
1208
		ceph_free_cap_flush(cf);
1209
	}
1210 1211
	while (drop--)
		iput(inode);
Sage Weil's avatar
Sage Weil committed
1212 1213 1214 1215 1216 1217 1218 1219 1220 1221
	return 0;
}

/*
 * caller must hold session s_mutex
 */
static void remove_session_caps(struct ceph_mds_session *session)
{
	dout("remove_session_caps on %p\n", session);
	iterate_session_caps(session, remove_session_caps_cb, NULL);
1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244

	spin_lock(&session->s_cap_lock);
	if (session->s_nr_caps > 0) {
		struct super_block *sb = session->s_mdsc->fsc->sb;
		struct inode *inode;
		struct ceph_cap *cap, *prev = NULL;
		struct ceph_vino vino;
		/*
		 * iterate_session_caps() skips inodes that are being
		 * deleted, we need to wait until deletions are complete.
		 * __wait_on_freeing_inode() is designed for the job,
		 * but it is not exported, so use lookup inode function
		 * to access it.
		 */
		while (!list_empty(&session->s_caps)) {
			cap = list_entry(session->s_caps.next,
					 struct ceph_cap, session_caps);
			if (cap == prev)
				break;
			prev = cap;
			vino = cap->ci->i_vino;
			spin_unlock(&session->s_cap_lock);

1245
			inode = ceph_find_inode(sb, vino);
1246 1247 1248 1249 1250
			iput(inode);

			spin_lock(&session->s_cap_lock);
		}
	}
1251 1252 1253

	// drop cap expires and unlock s_cap_lock
	cleanup_cap_releases(session->s_mdsc, session);
1254

Sage Weil's avatar
Sage Weil committed
1255
	BUG_ON(session->s_nr_caps > 0);
1256
	BUG_ON(!list_empty(&session->s_cap_flushing));
Sage Weil's avatar
Sage Weil committed
1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267
}

/*
 * wake up any threads waiting on this session's caps.  if the cap is
 * old (didn't get renewed on the client reconnect), remove it now.
 *
 * caller must hold s_mutex.
 */
static int wake_up_session_cb(struct inode *inode, struct ceph_cap *cap,
			      void *arg)
{
1268 1269
	struct ceph_inode_info *ci = ceph_inode(inode);

1270
	wake_up_all(&ci->i_cap_wq);
1271
	if (arg) {
1272
		spin_lock(&ci->i_ceph_lock);
1273 1274
		ci->i_wanted_max_size = 0;
		ci->i_requested_max_size = 0;
1275
		spin_unlock(&ci->i_ceph_lock);
1276
	}
Sage Weil's avatar
Sage Weil committed
1277 1278 1279
	return 0;
}

1280 1281
static void wake_up_session_caps(struct ceph_mds_session *session,
				 int reconnect)
Sage Weil's avatar
Sage Weil committed
1282 1283
{
	dout("wake_up_session_caps %p mds%d\n", session, session->s_mds);
1284 1285
	iterate_session_caps(session, wake_up_session_cb,
			     (void *)(unsigned long)reconnect);
Sage Weil's avatar
Sage Weil committed
1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302
}

/*
 * Send periodic message to MDS renewing all currently held caps.  The
 * ack will reset the expiration for all caps from this session.
 *
 * caller holds s_mutex
 */
static int send_renew_caps(struct ceph_mds_client *mdsc,
			   struct ceph_mds_session *session)
{
	struct ceph_msg *msg;
	int state;

	if (time_after_eq(jiffies, session->s_cap_ttl) &&
	    time_after_eq(session->s_cap_ttl, session->s_renew_requested))
		pr_info("mds%d caps stale\n", session->s_mds);
1303
	session->s_renew_requested = jiffies;
Sage Weil's avatar
Sage Weil committed
1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317

	/* do not try to renew caps until a recovering mds has reconnected
	 * with its clients. */
	state = ceph_mdsmap_get_state(mdsc->mdsmap, session->s_mds);
	if (state < CEPH_MDS_STATE_RECONNECT) {
		dout("send_renew_caps ignoring mds%d (%s)\n",
		     session->s_mds, ceph_mds_state_name(state));
		return 0;
	}

	dout("send_renew_caps to mds%d (%s)\n", session->s_mds,
		ceph_mds_state_name(state));
	msg = create_session_msg(CEPH_SESSION_REQUEST_RENEWCAPS,
				 ++session->s_renew_seq);
1318 1319
	if (!msg)
		return -ENOMEM;
Sage Weil's avatar
Sage Weil committed
1320 1321 1322 1323
	ceph_con_send(&session->s_con, msg);
	return 0;
}

1324 1325 1326 1327 1328 1329
static int send_flushmsg_ack(struct ceph_mds_client *mdsc,
			     struct ceph_mds_session *session, u64 seq)
{
	struct ceph_msg *msg;

	dout("send_flushmsg_ack to mds%d (%s)s seq %lld\n",
1330
	     session->s_mds, ceph_session_state_name(session->s_state), seq);
1331 1332 1333 1334 1335 1336 1337 1338
	msg = create_session_msg(CEPH_SESSION_FLUSHMSG_ACK, seq);
	if (!msg)
		return -ENOMEM;
	ceph_con_send(&session->s_con, msg);
	return 0;
}


Sage Weil's avatar
Sage Weil committed
1339 1340
/*
 * Note new cap ttl, and any transition from stale -> not stale (fresh?).
1341 1342
 *
 * Called under session->s_mutex
Sage Weil's avatar
Sage Weil committed
1343 1344 1345 1346 1347 1348 1349 1350
 */
static void renewed_caps(struct ceph_mds_client *mdsc,
			 struct ceph_mds_session *session, int is_renew)
{
	int was_stale;
	int wake = 0;

	spin_lock(&session->s_cap_lock);
1351
	was_stale = is_renew && time_after_eq(jiffies, session->s_cap_ttl);
Sage Weil's avatar
Sage Weil committed
1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369

	session->s_cap_ttl = session->s_renew_requested +
		mdsc->mdsmap->m_session_timeout*HZ;

	if (was_stale) {
		if (time_before(jiffies, session->s_cap_ttl)) {
			pr_info("mds%d caps renewed\n", session->s_mds);
			wake = 1;
		} else {
			pr_info("mds%d caps still stale\n", session->s_mds);
		}
	}
	dout("renewed_caps mds%d ttl now %lu, was %s, now %s\n",
	     session->s_mds, session->s_cap_ttl, was_stale ? "stale" : "fresh",
	     time_before(jiffies, session->s_cap_ttl) ? "stale" : "fresh");
	spin_unlock(&session->s_cap_lock);

	if (wake)
1370
		wake_up_session_caps(session, 0);
Sage Weil's avatar
Sage Weil committed
1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381
}

/*
 * send a session close request
 */
static int request_close_session(struct ceph_mds_client *mdsc,
				 struct ceph_mds_session *session)
{
	struct ceph_msg *msg;

	dout("request_close_session mds%d state %s seq %lld\n",
1382
	     session->s_mds, ceph_session_state_name(session->s_state),
Sage Weil's avatar
Sage Weil committed
1383 1384
	     session->s_seq);
	msg = create_session_msg(CEPH_SESSION_REQUEST_CLOSE, session->s_seq);
1385 1386 1387 1388
	if (!msg)
		return -ENOMEM;
	ceph_con_send(&session->s_con, msg);
	return 0;
Sage Weil's avatar
Sage Weil committed
1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416
}

/*
 * Called with s_mutex held.
 */
static int __close_session(struct ceph_mds_client *mdsc,
			 struct ceph_mds_session *session)
{
	if (session->s_state >= CEPH_MDS_SESSION_CLOSING)
		return 0;
	session->s_state = CEPH_MDS_SESSION_CLOSING;
	return request_close_session(mdsc, session);
}

/*
 * Trim old(er) caps.
 *
 * Because we can't cache an inode without one or more caps, we do
 * this indirectly: if a cap is unused, we prune its aliases, at which
 * point the inode will hopefully get dropped to.
 *
 * Yes, this is a bit sloppy.  Our only real goal here is to respond to
 * memory pressure from the MDS, though, so it needn't be perfect.
 */
static int trim_caps_cb(struct inode *inode, struct ceph_cap *cap, void *arg)
{
	struct ceph_mds_session *session = arg;
	struct ceph_inode_info *ci = ceph_inode(inode);
Yan, Zheng's avatar
Yan, Zheng committed
1417
	int used, wanted, oissued, mine;
Sage Weil's avatar
Sage Weil committed
1418 1419 1420 1421

	if (session->s_trim_caps <= 0)
		return -1;

1422
	spin_lock(&ci->i_ceph_lock);
Sage Weil's avatar
Sage Weil committed
1423 1424
	mine = cap->issued | cap->implemented;
	used = __ceph_caps_used(ci);
Yan, Zheng's avatar
Yan, Zheng committed
1425
	wanted = __ceph_caps_file_wanted(ci);
Sage Weil's avatar
Sage Weil committed
1426 1427
	oissued = __ceph_caps_issued_other(ci, cap);

Yan, Zheng's avatar
Yan, Zheng committed
1428
	dout("trim_caps_cb %p cap %p mine %s oissued %s used %s wanted %s\n",
Sage Weil's avatar
Sage Weil committed
1429
	     inode, cap, ceph_cap_string(mine), ceph_cap_string(oissued),
Yan, Zheng's avatar
Yan, Zheng committed
1430 1431
	     ceph_cap_string(used), ceph_cap_string(wanted));
	if (cap == ci->i_auth_cap) {
1432 1433
		if (ci->i_dirty_caps || ci->i_flushing_caps ||
		    !list_empty(&ci->i_cap_snaps))
Yan, Zheng's avatar
Yan, Zheng committed
1434 1435 1436 1437
			goto out;
		if ((used | wanted) & CEPH_CAP_ANY_WR)
			goto out;
	}
1438 1439 1440 1441 1442 1443 1444
	/* The inode has cached pages, but it's no longer used.
	 * we can safely drop it */
	if (wanted == 0 && used == CEPH_CAP_FILE_CACHE &&
	    !(oissued & CEPH_CAP_FILE_CACHE)) {
	  used = 0;
	  oissued = 0;
	}
Yan, Zheng's avatar
Yan, Zheng committed
1445
	if ((used | wanted) & ~oissued & mine)
Sage Weil's avatar
Sage Weil committed
1446 1447 1448 1449 1450
		goto out;   /* we need these caps */

	session->s_trim_caps--;
	if (oissued) {
		/* we aren't the only cap.. just remove us */
1451
		__ceph_remove_cap(cap, true);
Sage Weil's avatar
Sage Weil committed
1452
	} else {
1453
		/* try dropping referring dentries */
1454
		spin_unlock(&ci->i_ceph_lock);
Sage Weil's avatar
Sage Weil committed
1455 1456 1457 1458 1459 1460 1461
		d_prune_aliases(inode);
		dout("trim_caps_cb %p cap %p  pruned, count now %d\n",
		     inode, cap, atomic_read(&inode->i_count));
		return 0;
	}

out:
1462
	spin_unlock(&ci->i_ceph_lock);
Sage Weil's avatar
Sage Weil committed
1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482
	return 0;
}

/*
 * Trim session cap count down to some max number.
 */
static int trim_caps(struct ceph_mds_client *mdsc,
		     struct ceph_mds_session *session,
		     int max_caps)
{
	int trim_caps = session->s_nr_caps - max_caps;

	dout("trim_caps mds%d start: %d / %d, trim %d\n",
	     session->s_mds, session->s_nr_caps, max_caps, trim_caps);
	if (trim_caps > 0) {
		session->s_trim_caps = trim_caps;
		iterate_session_caps(session, trim_caps_cb, session);
		dout("trim_caps mds%d done: %d / %d, trimmed %d\n",
		     session->s_mds, session->s_nr_caps, max_caps,
			trim_caps - session->s_trim_caps);
1483
		session->s_trim_caps = 0;
Sage Weil's avatar
Sage Weil committed
1484
	}
1485 1486

	ceph_send_cap_releases(mdsc, session);
Sage Weil's avatar
Sage Weil committed
1487 1488 1489
	return 0;
}

1490 1491
static int check_capsnap_flush(struct ceph_inode_info *ci,
			       u64 want_snap_seq)
1492
{
1493
	int ret = 1;
1494
	spin_lock(&ci->i_ceph_lock);
1495 1496 1497 1498
	if (want_snap_seq > 0 && !list_empty(&ci->i_cap_snaps)) {
		struct ceph_cap_snap *capsnap =
			list_first_entry(&ci->i_cap_snaps,
					 struct ceph_cap_snap, ci_item);
1499
		ret = capsnap->follows >= want_snap_seq;
1500
	}
1501
	spin_unlock(&ci->i_ceph_lock);
1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521
	return ret;
}

static int check_caps_flush(struct ceph_mds_client *mdsc,
			    u64 want_flush_tid)
{
	struct rb_node *n;
	struct ceph_cap_flush *cf;
	int ret = 1;

	spin_lock(&mdsc->cap_dirty_lock);
	n = rb_first(&mdsc->cap_flush_tree);
	cf = n ? rb_entry(n, struct ceph_cap_flush, g_node) : NULL;
	if (cf && cf->tid <= want_flush_tid) {
		dout("check_caps_flush still flushing tid %llu <= %llu\n",
		     cf->tid, want_flush_tid);
		ret = 0;
	}
	spin_unlock(&mdsc->cap_dirty_lock);
	return ret;
1522 1523
}

Sage Weil's avatar
Sage Weil committed
1524 1525 1526
/*
 * flush all dirty inode data to disk.
 *
1527
 * returns true if we've flushed through want_flush_tid
Sage Weil's avatar
Sage Weil committed
1528
 */
1529
static void wait_caps_flush(struct ceph_mds_client *mdsc,
1530
			    u64 want_flush_tid, u64 want_snap_seq)
Sage Weil's avatar
Sage Weil committed
1531
{
1532
	int mds;
Sage Weil's avatar
Sage Weil committed
1533

1534 1535
	dout("check_caps_flush want %llu snap want %llu\n",
	     want_flush_tid, want_snap_seq);
Sage Weil's avatar
Sage Weil committed
1536
	mutex_lock(&mdsc->mutex);
1537
	for (mds = 0; mds < mdsc->max_sessions; ) {
Sage Weil's avatar
Sage Weil committed
1538
		struct ceph_mds_session *session = mdsc->sessions[mds];
1539
		struct inode *inode = NULL;
Sage Weil's avatar
Sage Weil committed
1540

1541 1542
		if (!session) {
			mds++;
Sage Weil's avatar
Sage Weil committed
1543
			continue;
1544
		}
Sage Weil's avatar
Sage Weil committed
1545 1546 1547 1548
		get_session(session);
		mutex_unlock(&mdsc->mutex);

		mutex_lock(&session->s_mutex);
1549 1550 1551 1552 1553 1554
		if (!list_empty(&session->s_cap_snaps_flushing)) {
			struct ceph_cap_snap *capsnap =
				list_first_entry(&session->s_cap_snaps_flushing,
						 struct ceph_cap_snap,
						 flushing_item);
			struct ceph_inode_info *ci = capsnap->ci;
1555
			if (!check_capsnap_flush(ci, want_snap_seq)) {
1556 1557 1558 1559
				dout("check_cap_flush still flushing snap %p "
				     "follows %lld <= %lld to mds%d\n",
				     &ci->vfs_inode, capsnap->follows,
				     want_snap_seq, mds);
1560
				inode = igrab(&ci->vfs_inode);
Sage Weil's avatar
Sage Weil committed
1561 1562 1563 1564 1565
			}
		}
		mutex_unlock(&session->s_mutex);
		ceph_put_mds_session(session);

1566
		if (inode) {
1567
			wait_event(mdsc->cap_flushing_wq,
1568 1569 1570 1571
				   check_capsnap_flush(ceph_inode(inode),
						       want_snap_seq));
			iput(inode);
		} else {
1572
			mds++;
1573
		}
1574

Sage Weil's avatar
Sage Weil committed
1575 1576 1577
		mutex_lock(&mdsc->mutex);
	}
	mutex_unlock(&mdsc->mutex);
1578 1579 1580 1581 1582

	wait_event(mdsc->cap_flushing_wq,
		   check_caps_flush(mdsc, want_flush_tid));

	dout("check_caps_flush ok, flushed thru %llu\n", want_flush_tid);
Sage Weil's avatar
Sage Weil committed
1583 1584 1585 1586 1587
}

/*
 * called under s_mutex
 */
1588 1589
void ceph_send_cap_releases(struct ceph_mds_client *mdsc,
			    struct ceph_mds_session *session)
Sage Weil's avatar
Sage Weil committed
1590
{
1591 1592 1593 1594 1595 1596
	struct ceph_msg *msg = NULL;
	struct ceph_mds_cap_release *head;
	struct ceph_mds_cap_item *item;
	struct ceph_cap *cap;
	LIST_HEAD(tmp_list);
	int num_cap_releases;
Sage Weil's avatar
Sage Weil committed
1597

1598
	spin_lock(&session->s_cap_lock);
1599 1600 1601 1602
again:
	list_splice_init(&session->s_cap_releases, &tmp_list);
	num_cap_releases = session->s_num_cap_releases;
	session->s_num_cap_releases = 0;
Sage Weil's avatar
Sage Weil committed
1603
	spin_unlock(&session->s_cap_lock);
1604

1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618
	while (!list_empty(&tmp_list)) {
		if (!msg) {
			msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPRELEASE,
					PAGE_CACHE_SIZE, GFP_NOFS, false);
			if (!msg)
				goto out_err;
			head = msg->front.iov_base;
			head->num = cpu_to_le32(0);
			msg->front.iov_len = sizeof(*head);
		}
		cap = list_first_entry(&tmp_list, struct ceph_cap,
					session_caps);
		list_del(&cap->session_caps);
		num_cap_releases--;
1619

1620
		head = msg->front.iov_base;
1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636
		le32_add_cpu(&head->num, 1);
		item = msg->front.iov_base + msg->front.iov_len;
		item->ino = cpu_to_le64(cap->cap_ino);
		item->cap_id = cpu_to_le64(cap->cap_id);
		item->migrate_seq = cpu_to_le32(cap->mseq);
		item->seq = cpu_to_le32(cap->issue_seq);
		msg->front.iov_len += sizeof(*item);

		ceph_put_cap(mdsc, cap);

		if (le32_to_cpu(head->num) == CEPH_CAPS_PER_RELEASE) {
			msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
			dout("send_cap_releases mds%d %p\n", session->s_mds, msg);
			ceph_con_send(&session->s_con, msg);
			msg = NULL;
		}
1637
	}
1638

1639
	BUG_ON(num_cap_releases != 0);
1640

1641 1642 1643 1644 1645 1646 1647 1648 1649
	spin_lock(&session->s_cap_lock);
	if (!list_empty(&session->s_cap_releases))
		goto again;
	spin_unlock(&session->s_cap_lock);

	if (msg) {
		msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
		dout("send_cap_releases mds%d %p\n", session->s_mds, msg);
		ceph_con_send(&session->s_con, msg);
1650
	}
1651 1652 1653 1654 1655 1656 1657 1658
	return;
out_err:
	pr_err("send_cap_releases mds%d, failed to allocate message\n",
		session->s_mds);
	spin_lock(&session->s_cap_lock);
	list_splice(&tmp_list, &session->s_cap_releases);
	session->s_num_cap_releases += num_cap_releases;
	spin_unlock(&session->s_cap_lock);
1659 1660
}

Sage Weil's avatar
Sage Weil committed
1661 1662 1663 1664
/*
 * requests
 */

1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682
int ceph_alloc_readdir_reply_buffer(struct ceph_mds_request *req,
				    struct inode *dir)
{
	struct ceph_inode_info *ci = ceph_inode(dir);
	struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;
	struct ceph_mount_options *opt = req->r_mdsc->fsc->mount_options;
	size_t size = sizeof(*rinfo->dir_in) + sizeof(*rinfo->dir_dname_len) +
		      sizeof(*rinfo->dir_dname) + sizeof(*rinfo->dir_dlease);
	int order, num_entries;

	spin_lock(&ci->i_ceph_lock);
	num_entries = ci->i_files + ci->i_subdirs;
	spin_unlock(&ci->i_ceph_lock);
	num_entries = max(num_entries, 1);
	num_entries = min(num_entries, opt->max_readdir);

	order = get_order(size * num_entries);
	while (order >= 0) {
1683 1684
		rinfo->dir_in = (void*)__get_free_pages(GFP_KERNEL |
							__GFP_NOWARN,
1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698 1699 1700 1701 1702
							order);
		if (rinfo->dir_in)
			break;
		order--;
	}
	if (!rinfo->dir_in)
		return -ENOMEM;

	num_entries = (PAGE_SIZE << order) / size;
	num_entries = min(num_entries, opt->max_readdir);

	rinfo->dir_buf_size = PAGE_SIZE << order;
	req->r_num_caps = num_entries + 1;
	req->r_args.readdir.max_entries = cpu_to_le32(num_entries);
	req->r_args.readdir.max_bytes = cpu_to_le32(opt->max_readdir_bytes);
	return 0;
}

Sage Weil's avatar
Sage Weil committed
1703 1704 1705 1706 1707 1708 1709 1710 1711 1712 1713
/*
 * Create an mds request.
 */
struct ceph_mds_request *
ceph_mdsc_create_request(struct ceph_mds_client *mdsc, int op, int mode)
{
	struct ceph_mds_request *req = kzalloc(sizeof(*req), GFP_NOFS);

	if (!req)
		return ERR_PTR(-ENOMEM);

1714
	mutex_init(&req->r_fill_mutex);
1715
	req->r_mdsc = mdsc;
Sage Weil's avatar
Sage Weil committed
1716 1717 1718
	req->r_started = jiffies;
	req->r_resend_mds = -1;
	INIT_LIST_HEAD(&req->r_unsafe_dir_item);
1719
	INIT_LIST_HEAD(&req->r_unsafe_target_item);
Sage Weil's avatar
Sage Weil committed
1720
	req->r_fmode = -1;
1721
	kref_init(&req->r_kref);
Sage Weil's avatar
Sage Weil committed
1722 1723 1724 1725 1726
	INIT_LIST_HEAD(&req->r_wait);
	init_completion(&req->r_completion);
	init_completion(&req->r_safe_completion);
	INIT_LIST_HEAD(&req->r_unsafe_item);

1727 1728
	req->r_stamp = CURRENT_TIME;

Sage Weil's avatar
Sage Weil committed
1729 1730 1731 1732 1733 1734
	req->r_op = op;
	req->r_direct_mode = mode;
	return req;
}

/*
1735
 * return oldest (lowest) request, tid in request tree, 0 if none.
Sage Weil's avatar
Sage Weil committed
1736 1737 1738
 *
 * called under mdsc->mutex.
 */
1739 1740 1741 1742 1743 1744 1745 1746
static struct ceph_mds_request *__get_oldest_req(struct ceph_mds_client *mdsc)
{
	if (RB_EMPTY_ROOT(&mdsc->request_tree))
		return NULL;
	return rb_entry(rb_first(&mdsc->request_tree),
			struct ceph_mds_request, r_node);
}

1747
static inline  u64 __get_oldest_tid(struct ceph_mds_client *mdsc)
Sage Weil's avatar
Sage Weil committed
1748
{
1749
	return mdsc->oldest_tid;
Sage Weil's avatar
Sage Weil committed
1750 1751 1752 1753 1754 1755 1756 1757 1758 1759 1760 1761 1762 1763 1764 1765 1766 1767
}

/*
 * Build a dentry's path.  Allocate on heap; caller must kfree.  Based
 * on build_path_from_dentry in fs/cifs/dir.c.
 *
 * If @stop_on_nosnap, generate path relative to the first non-snapped
 * inode.
 *
 * Encode hidden .snap dirs as a double /, i.e.
 *   foo/.snap/bar -> foo//bar
 */
char *ceph_mdsc_build_path(struct dentry *dentry, int *plen, u64 *base,
			   int stop_on_nosnap)
{
	struct dentry *temp;
	char *path;
	int len, pos;
1768
	unsigned seq;
Sage Weil's avatar
Sage Weil committed
1769 1770 1771 1772 1773 1774

	if (dentry == NULL)
		return ERR_PTR(-EINVAL);

retry:
	len = 0;
1775 1776
	seq = read_seqbegin(&rename_lock);
	rcu_read_lock();
Sage Weil's avatar
Sage Weil committed
1777
	for (temp = dentry; !IS_ROOT(temp);) {
1778
		struct inode *inode = d_inode(temp);
Sage Weil's avatar
Sage Weil committed
1779 1780 1781 1782 1783 1784 1785 1786 1787
		if (inode && ceph_snap(inode) == CEPH_SNAPDIR)
			len++;  /* slash only */
		else if (stop_on_nosnap && inode &&
			 ceph_snap(inode) == CEPH_NOSNAP)
			break;
		else
			len += 1 + temp->d_name.len;
		temp = temp->d_parent;
	}
1788
	rcu_read_unlock();
Sage Weil's avatar
Sage Weil committed
1789 1790 1791 1792 1793 1794 1795 1796
	if (len)
		len--;  /* no leading '/' */

	path = kmalloc(len+1, GFP_NOFS);
	if (path == NULL)
		return ERR_PTR(-ENOMEM);
	pos = len;
	path[pos] = 0;	/* trailing null */
1797
	rcu_read_lock();
Sage Weil's avatar
Sage Weil committed
1798
	for (temp = dentry; !IS_ROOT(temp) && pos != 0; ) {
1799
		struct inode *inode;
Sage Weil's avatar
Sage Weil committed
1800

1801
		spin_lock(&temp->d_lock);
1802
		inode = d_inode(temp);
Sage Weil's avatar
Sage Weil committed
1803
		if (inode && ceph_snap(inode) == CEPH_SNAPDIR) {
1804
			dout("build_path path+%d: %p SNAPDIR\n",
Sage Weil's avatar
Sage Weil committed
1805 1806 1807
			     pos, temp);
		} else if (stop_on_nosnap && inode &&
			   ceph_snap(inode) == CEPH_NOSNAP) {
1808
			spin_unlock(&temp->d_lock);
Sage Weil's avatar
Sage Weil committed
1809 1810 1811
			break;
		} else {
			pos -= temp->d_name.len;
1812 1813
			if (pos < 0) {
				spin_unlock(&temp->d_lock);
Sage Weil's avatar
Sage Weil committed
1814
				break;
1815
			}
Sage Weil's avatar
Sage Weil committed
1816 1817 1818
			strncpy(path + pos, temp->d_name.name,
				temp->d_name.len);
		}
1819
		spin_unlock(&temp->d_lock);
Sage Weil's avatar
Sage Weil committed
1820 1821 1822 1823
		if (pos)
			path[--pos] = '/';
		temp = temp->d_parent;
	}
1824 1825
	rcu_read_unlock();
	if (pos != 0 || read_seqretry(&rename_lock, seq)) {
1826
		pr_err("build_path did not end path lookup where "
Sage Weil's avatar
Sage Weil committed
1827 1828 1829 1830 1831 1832 1833 1834 1835
		       "expected, namelen is %d, pos is %d\n", len, pos);
		/* presumably this is only possible if racing with a
		   rename of one of the parent directories (we can not
		   lock the dentries above us to prevent this, but
		   retrying should be harmless) */
		kfree(path);
		goto retry;
	}

1836
	*base = ceph_ino(d_inode(temp));
Sage Weil's avatar
Sage Weil committed
1837
	*plen = len;
1838
	dout("build_path on %p %d built %llx '%.*s'\n",
Al Viro's avatar
Al Viro committed
1839
	     dentry, d_count(dentry), *base, len, path);
Sage Weil's avatar
Sage Weil committed
1840 1841 1842 1843 1844 1845 1846 1847
	return path;
}

static int build_dentry_path(struct dentry *dentry,
			     const char **ppath, int *ppathlen, u64 *pino,
			     int *pfreepath)
{
	char *path;
1848
	struct inode *dir;
Sage Weil's avatar
Sage Weil committed
1849

1850 1851 1852 1853 1854
	rcu_read_lock();
	dir = d_inode_rcu(dentry->d_parent);
	if (dir && ceph_snap(dir) == CEPH_NOSNAP) {
		*pino = ceph_ino(dir);
		rcu_read_unlock();
Sage Weil's avatar
Sage Weil committed
1855 1856 1857 1858
		*ppath = dentry->d_name.name;
		*ppathlen = dentry->d_name.len;
		return 0;
	}
1859
	rcu_read_unlock();
Sage Weil's avatar
Sage Weil committed
1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877 1878 1879 1880 1881 1882 1883 1884 1885 1886 1887 1888 1889 1890 1891 1892 1893 1894 1895 1896 1897 1898 1899 1900 1901 1902 1903 1904 1905 1906 1907 1908
	path = ceph_mdsc_build_path(dentry, ppathlen, pino, 1);
	if (IS_ERR(path))
		return PTR_ERR(path);
	*ppath = path;
	*pfreepath = 1;
	return 0;
}

static int build_inode_path(struct inode *inode,
			    const char **ppath, int *ppathlen, u64 *pino,
			    int *pfreepath)
{
	struct dentry *dentry;
	char *path;

	if (ceph_snap(inode) == CEPH_NOSNAP) {
		*pino = ceph_ino(inode);
		*ppathlen = 0;
		return 0;
	}
	dentry = d_find_alias(inode);
	path = ceph_mdsc_build_path(dentry, ppathlen, pino, 1);
	dput(dentry);
	if (IS_ERR(path))
		return PTR_ERR(path);
	*ppath = path;
	*pfreepath = 1;
	return 0;
}

/*
 * request arguments may be specified via an inode *, a dentry *, or
 * an explicit ino+path.
 */
static int set_request_path_attr(struct inode *rinode, struct dentry *rdentry,
				  const char *rpath, u64 rino,
				  const char **ppath, int *pathlen,
				  u64 *ino, int *freepath)
{
	int r = 0;

	if (rinode) {
		r = build_inode_path(rinode, ppath, pathlen, ino, freepath);
		dout(" inode %p %llx.%llx\n", rinode, ceph_ino(rinode),
		     ceph_snap(rinode));
	} else if (rdentry) {
		r = build_dentry_path(rdentry, ppath, pathlen, ino, freepath);
		dout(" dentry %p %llx/%.*s\n", rdentry, *ino, *pathlen,
		     *ppath);
1909
	} else if (rpath || rino) {
Sage Weil's avatar
Sage Weil committed
1910 1911
		*ino = rino;
		*ppath = rpath;
1912
		*pathlen = rpath ? strlen(rpath) : 0;
Sage Weil's avatar
Sage Weil committed
1913 1914 1915 1916 1917 1918 1919 1920 1921 1922 1923
		dout(" path %.*s\n", *pathlen, rpath);
	}

	return r;
}

/*
 * called under mdsc->mutex
 */
static struct ceph_msg *create_request_message(struct ceph_mds_client *mdsc,
					       struct ceph_mds_request *req,
1924
					       int mds, bool drop_cap_releases)
Sage Weil's avatar
Sage Weil committed
1925 1926 1927 1928 1929 1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954
{
	struct ceph_msg *msg;
	struct ceph_mds_request_head *head;
	const char *path1 = NULL;
	const char *path2 = NULL;
	u64 ino1 = 0, ino2 = 0;
	int pathlen1 = 0, pathlen2 = 0;
	int freepath1 = 0, freepath2 = 0;
	int len;
	u16 releases;
	void *p, *end;
	int ret;

	ret = set_request_path_attr(req->r_inode, req->r_dentry,
			      req->r_path1, req->r_ino1.ino,
			      &path1, &pathlen1, &ino1, &freepath1);
	if (ret < 0) {
		msg = ERR_PTR(ret);
		goto out;
	}

	ret = set_request_path_attr(NULL, req->r_old_dentry,
			      req->r_path2, req->r_ino2.ino,
			      &path2, &pathlen2, &ino2, &freepath2);
	if (ret < 0) {
		msg = ERR_PTR(ret);
		goto out_free1;
	}

	len = sizeof(*head) +
1955
		pathlen1 + pathlen2 + 2*(1 + sizeof(u32) + sizeof(u64)) +
1956
		sizeof(struct ceph_timespec);
Sage Weil's avatar
Sage Weil committed
1957 1958 1959 1960 1961 1962 1963 1964 1965 1966

	/* calculate (max) length for cap releases */
	len += sizeof(struct ceph_mds_request_release) *
		(!!req->r_inode_drop + !!req->r_dentry_drop +
		 !!req->r_old_inode_drop + !!req->r_old_dentry_drop);
	if (req->r_dentry_drop)
		len += req->r_dentry->d_name.len;
	if (req->r_old_dentry_drop)
		len += req->r_old_dentry->d_name.len;

1967
	msg = ceph_msg_new(CEPH_MSG_CLIENT_REQUEST, len, GFP_NOFS, false);
1968 1969
	if (!msg) {
		msg = ERR_PTR(-ENOMEM);
Sage Weil's avatar
Sage Weil committed
1970
		goto out_free2;
1971
	}
Sage Weil's avatar
Sage Weil committed
1972

John Spray's avatar
John Spray committed
1973
	msg->hdr.version = cpu_to_le16(2);
1974 1975
	msg->hdr.tid = cpu_to_le64(req->r_tid);

Sage Weil's avatar
Sage Weil committed
1976 1977 1978 1979 1980 1981
	head = msg->front.iov_base;
	p = msg->front.iov_base + sizeof(*head);
	end = msg->front.iov_base + msg->front.iov_len;

	head->mdsmap_epoch = cpu_to_le32(mdsc->mdsmap->m_epoch);
	head->op = cpu_to_le32(req->r_op);
1982 1983
	head->caller_uid = cpu_to_le32(from_kuid(&init_user_ns, req->r_uid));
	head->caller_gid = cpu_to_le32(from_kgid(&init_user_ns, req->r_gid));
Sage Weil's avatar
Sage Weil committed
1984 1985 1986 1987 1988
	head->args = req->r_args;

	ceph_encode_filepath(&p, end, ino1, path1);
	ceph_encode_filepath(&p, end, ino2, path2);

1989 1990 1991
	/* make note of release offset, in case we need to replay */
	req->r_request_release_offset = p - msg->front.iov_base;

Sage Weil's avatar
Sage Weil committed
1992 1993 1994 1995
	/* cap releases */
	releases = 0;
	if (req->r_inode_drop)
		releases += ceph_encode_inode_release(&p,
1996
		      req->r_inode ? req->r_inode : d_inode(req->r_dentry),
Sage Weil's avatar
Sage Weil committed
1997 1998 1999 2000 2001 2002 2003 2004 2005
		      mds, req->r_inode_drop, req->r_inode_unless, 0);
	if (req->r_dentry_drop)
		releases += ceph_encode_dentry_release(&p, req->r_dentry,
		       mds, req->r_dentry_drop, req->r_dentry_unless);
	if (req->r_old_dentry_drop)
		releases += ceph_encode_dentry_release(&p, req->r_old_dentry,
		       mds, req->r_old_dentry_drop, req->r_old_dentry_unless);
	if (req->r_old_inode_drop)
		releases += ceph_encode_inode_release(&p,
2006
		      d_inode(req->r_old_dentry),
Sage Weil's avatar
Sage Weil committed
2007
		      mds, req->r_old_inode_drop, req->r_old_inode_unless, 0);
2008 2009 2010 2011 2012 2013

	if (drop_cap_releases) {
		releases = 0;
		p = msg->front.iov_base + req->r_request_release_offset;
	}

Sage Weil's avatar
Sage Weil committed
2014 2015
	head->num_releases = cpu_to_le16(releases);

2016
	/* time stamp */
2017 2018 2019 2020 2021
	{
		struct ceph_timespec ts;
		ceph_encode_timespec(&ts, &req->r_stamp);
		ceph_encode_copy(&p, &ts, sizeof(ts));
	}
2022

Sage Weil's avatar
Sage Weil committed
2023 2024 2025 2026
	BUG_ON(p > end);
	msg->front.iov_len = p - msg->front.iov_base;
	msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);

2027 2028 2029 2030 2031 2032 2033
	if (req->r_pagelist) {
		struct ceph_pagelist *pagelist = req->r_pagelist;
		atomic_inc(&pagelist->refcnt);
		ceph_msg_data_add_pagelist(msg, pagelist);
		msg->hdr.data_len = cpu_to_le32(pagelist->length);
	} else {
		msg->hdr.data_len = 0;
2034
	}
2035

Sage Weil's avatar
Sage Weil committed
2036 2037 2038 2039 2040 2041 2042 2043 2044 2045 2046 2047 2048 2049 2050 2051 2052 2053 2054 2055 2056 2057
	msg->hdr.data_off = cpu_to_le16(0);

out_free2:
	if (freepath2)
		kfree((char *)path2);
out_free1:
	if (freepath1)
		kfree((char *)path1);
out:
	return msg;
}

/*
 * called under mdsc->mutex if error, under no mutex if
 * success.
 */
static void complete_request(struct ceph_mds_client *mdsc,
			     struct ceph_mds_request *req)
{
	if (req->r_callback)
		req->r_callback(mdsc, req);
	else
2058
		complete_all(&req->r_completion);
Sage Weil's avatar
Sage Weil committed
2059 2060 2061 2062 2063 2064 2065
}

/*
 * called under mdsc->mutex
 */
static int __prepare_send_request(struct ceph_mds_client *mdsc,
				  struct ceph_mds_request *req,
2066
				  int mds, bool drop_cap_releases)
Sage Weil's avatar
Sage Weil committed
2067 2068 2069 2070 2071 2072
{
	struct ceph_mds_request_head *rhead;
	struct ceph_msg *msg;
	int flags = 0;

	req->r_attempts++;
2073 2074 2075 2076 2077 2078 2079 2080 2081
	if (req->r_inode) {
		struct ceph_cap *cap =
			ceph_get_cap_for_mds(ceph_inode(req->r_inode), mds);

		if (cap)
			req->r_sent_on_mseq = cap->mseq;
		else
			req->r_sent_on_mseq = -1;
	}
Sage Weil's avatar
Sage Weil committed
2082 2083 2084
	dout("prepare_send_request %p tid %lld %s (attempt %d)\n", req,
	     req->r_tid, ceph_mds_op_name(req->r_op), req->r_attempts);

2085
	if (req->r_got_unsafe) {
2086
		void *p;
2087 2088 2089 2090 2091 2092 2093 2094 2095 2096 2097 2098 2099 2100 2101 2102 2103
		/*
		 * Replay.  Do not regenerate message (and rebuild
		 * paths, etc.); just use the original message.
		 * Rebuilding paths will break for renames because
		 * d_move mangles the src name.
		 */
		msg = req->r_request;
		rhead = msg->front.iov_base;

		flags = le32_to_cpu(rhead->flags);
		flags |= CEPH_MDS_FLAG_REPLAY;
		rhead->flags = cpu_to_le32(flags);

		if (req->r_target_inode)
			rhead->ino = cpu_to_le64(ceph_ino(req->r_target_inode));

		rhead->num_retry = req->r_attempts - 1;
2104 2105 2106

		/* remove cap/dentry releases from message */
		rhead->num_releases = 0;
2107 2108 2109

		/* time stamp */
		p = msg->front.iov_base + req->r_request_release_offset;
2110 2111 2112 2113 2114
		{
			struct ceph_timespec ts;
			ceph_encode_timespec(&ts, &req->r_stamp);
			ceph_encode_copy(&p, &ts, sizeof(ts));
		}
2115 2116 2117

		msg->front.iov_len = p - msg->front.iov_base;
		msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
2118 2119 2120
		return 0;
	}

Sage Weil's avatar
Sage Weil committed
2121 2122 2123 2124
	if (req->r_request) {
		ceph_msg_put(req->r_request);
		req->r_request = NULL;
	}
2125
	msg = create_request_message(mdsc, req, mds, drop_cap_releases);
Sage Weil's avatar
Sage Weil committed
2126
	if (IS_ERR(msg)) {
2127
		req->r_err = PTR_ERR(msg);
2128
		return PTR_ERR(msg);
Sage Weil's avatar
Sage Weil committed
2129 2130 2131 2132 2133 2134 2135 2136 2137 2138 2139 2140
	}
	req->r_request = msg;

	rhead = msg->front.iov_base;
	rhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
	if (req->r_got_unsafe)
		flags |= CEPH_MDS_FLAG_REPLAY;
	if (req->r_locked_dir)
		flags |= CEPH_MDS_FLAG_WANT_DENTRY;
	rhead->flags = cpu_to_le32(flags);
	rhead->num_fwd = req->r_num_fwd;
	rhead->num_retry = req->r_attempts - 1;
2141
	rhead->ino = 0;
Sage Weil's avatar
Sage Weil committed
2142 2143 2144 2145 2146 2147 2148 2149 2150 2151 2152 2153 2154

	dout(" r_locked_dir = %p\n", req->r_locked_dir);
	return 0;
}

/*
 * send request, or put it on the appropriate wait list.
 */
static int __do_request(struct ceph_mds_client *mdsc,
			struct ceph_mds_request *req)
{
	struct ceph_mds_session *session = NULL;
	int mds = -1;
2155
	int err = 0;
Sage Weil's avatar
Sage Weil committed
2156

2157 2158 2159
	if (req->r_err || req->r_got_result) {
		if (req->r_aborted)
			__unregister_request(mdsc, req);
Sage Weil's avatar
Sage Weil committed
2160
		goto out;
2161
	}
Sage Weil's avatar
Sage Weil committed
2162 2163 2164 2165 2166 2167 2168

	if (req->r_timeout &&
	    time_after_eq(jiffies, req->r_started + req->r_timeout)) {
		dout("do_request timed out\n");
		err = -EIO;
		goto finish;
	}
2169 2170 2171 2172 2173
	if (ACCESS_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) {
		dout("do_request forced umount\n");
		err = -EIO;
		goto finish;
	}
Sage Weil's avatar
Sage Weil committed
2174

2175 2176
	put_request_session(req);

Sage Weil's avatar
Sage Weil committed
2177 2178 2179 2180 2181 2182 2183 2184 2185 2186
	mds = __choose_mds(mdsc, req);
	if (mds < 0 ||
	    ceph_mdsmap_get_state(mdsc->mdsmap, mds) < CEPH_MDS_STATE_ACTIVE) {
		dout("do_request no mds or not active, waiting for map\n");
		list_add(&req->r_wait, &mdsc->waiting_for_map);
		goto out;
	}

	/* get, open session */
	session = __ceph_lookup_mds_session(mdsc, mds);
2187
	if (!session) {
Sage Weil's avatar
Sage Weil committed
2188
		session = register_session(mdsc, mds);
2189 2190 2191 2192 2193
		if (IS_ERR(session)) {
			err = PTR_ERR(session);
			goto finish;
		}
	}
2194 2195
	req->r_session = get_session(session);

Sage Weil's avatar
Sage Weil committed
2196
	dout("do_request mds%d session %p state %s\n", mds, session,
2197
	     ceph_session_state_name(session->s_state));
Sage Weil's avatar
Sage Weil committed
2198 2199 2200 2201 2202 2203 2204 2205 2206 2207 2208 2209 2210 2211 2212
	if (session->s_state != CEPH_MDS_SESSION_OPEN &&
	    session->s_state != CEPH_MDS_SESSION_HUNG) {
		if (session->s_state == CEPH_MDS_SESSION_NEW ||
		    session->s_state == CEPH_MDS_SESSION_CLOSING)
			__open_session(mdsc, session);
		list_add(&req->r_wait, &session->s_waiting);
		goto out_session;
	}

	/* send request */
	req->r_resend_mds = -1;   /* forget any previous mds hint */

	if (req->r_request_started == 0)   /* note request start time */
		req->r_request_started = jiffies;

2213
	err = __prepare_send_request(mdsc, req, mds, false);
Sage Weil's avatar
Sage Weil committed
2214 2215 2216 2217 2218 2219 2220
	if (!err) {
		ceph_msg_get(req->r_request);
		ceph_con_send(&session->s_con, req->r_request);
	}

out_session:
	ceph_put_mds_session(session);
2221 2222 2223 2224 2225 2226 2227
finish:
	if (err) {
		dout("__do_request early error %d\n", err);
		req->r_err = err;
		complete_request(mdsc, req);
		__unregister_request(mdsc, req);
	}
Sage Weil's avatar
Sage Weil committed
2228 2229 2230 2231 2232 2233 2234 2235 2236 2237
out:
	return err;
}

/*
 * called under mdsc->mutex
 */
static void __wake_requests(struct ceph_mds_client *mdsc,
			    struct list_head *head)
{
2238 2239 2240 2241
	struct ceph_mds_request *req;
	LIST_HEAD(tmp_list);

	list_splice_init(head, &tmp_list);
Sage Weil's avatar
Sage Weil committed
2242

2243 2244 2245
	while (!list_empty(&tmp_list)) {
		req = list_entry(tmp_list.next,
				 struct ceph_mds_request, r_wait);
Sage Weil's avatar
Sage Weil committed
2246
		list_del_init(&req->r_wait);
Sage Weil's avatar
Sage Weil committed
2247
		dout(" wake request %p tid %llu\n", req, req->r_tid);
Sage Weil's avatar
Sage Weil committed
2248 2249 2250 2251 2252 2253
		__do_request(mdsc, req);
	}
}

/*
 * Wake up threads with requests pending for @mds, so that they can
2254
 * resubmit their requests to a possibly different mds.
Sage Weil's avatar
Sage Weil committed
2255
 */
2256
static void kick_requests(struct ceph_mds_client *mdsc, int mds)
Sage Weil's avatar
Sage Weil committed
2257
{
2258
	struct ceph_mds_request *req;
Yan, Zheng's avatar
Yan, Zheng committed
2259
	struct rb_node *p = rb_first(&mdsc->request_tree);
Sage Weil's avatar
Sage Weil committed
2260 2261

	dout("kick_requests mds%d\n", mds);
Yan, Zheng's avatar
Yan, Zheng committed
2262
	while (p) {
2263
		req = rb_entry(p, struct ceph_mds_request, r_node);
Yan, Zheng's avatar
Yan, Zheng committed
2264
		p = rb_next(p);
2265 2266
		if (req->r_got_unsafe)
			continue;
2267 2268
		if (req->r_attempts > 0)
			continue; /* only new requests */
2269 2270 2271
		if (req->r_session &&
		    req->r_session->s_mds == mds) {
			dout(" kicking tid %llu\n", req->r_tid);
2272
			list_del_init(&req->r_wait);
2273
			__do_request(mdsc, req);
Sage Weil's avatar
Sage Weil committed
2274 2275 2276 2277 2278 2279 2280 2281 2282 2283 2284 2285 2286 2287 2288 2289 2290 2291 2292 2293 2294 2295 2296 2297 2298 2299 2300 2301 2302 2303 2304
		}
	}
}

void ceph_mdsc_submit_request(struct ceph_mds_client *mdsc,
			      struct ceph_mds_request *req)
{
	dout("submit_request on %p\n", req);
	mutex_lock(&mdsc->mutex);
	__register_request(mdsc, req, NULL);
	__do_request(mdsc, req);
	mutex_unlock(&mdsc->mutex);
}

/*
 * Synchrously perform an mds request.  Take care of all of the
 * session setup, forwarding, retry details.
 */
int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
			 struct inode *dir,
			 struct ceph_mds_request *req)
{
	int err;

	dout("do_request on %p\n", req);

	/* take CAP_PIN refs for r_inode, r_locked_dir, r_old_dentry */
	if (req->r_inode)
		ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
	if (req->r_locked_dir)
		ceph_get_cap_refs(ceph_inode(req->r_locked_dir), CEPH_CAP_PIN);
2305
	if (req->r_old_dentry_dir)
2306 2307
		ceph_get_cap_refs(ceph_inode(req->r_old_dentry_dir),
				  CEPH_CAP_PIN);
Sage Weil's avatar
Sage Weil committed
2308 2309 2310 2311 2312 2313

	/* issue */
	mutex_lock(&mdsc->mutex);
	__register_request(mdsc, req, dir);
	__do_request(mdsc, req);

2314 2315 2316
	if (req->r_err) {
		err = req->r_err;
		goto out;
Sage Weil's avatar
Sage Weil committed
2317 2318
	}

2319 2320 2321
	/* wait */
	mutex_unlock(&mdsc->mutex);
	dout("do_request waiting\n");
2322
	if (!req->r_timeout && req->r_wait_for_completion) {
2323
		err = req->r_wait_for_completion(mdsc, req);
2324
	} else {
2325 2326 2327 2328 2329 2330 2331 2332 2333
		long timeleft = wait_for_completion_killable_timeout(
					&req->r_completion,
					ceph_timeout_jiffies(req->r_timeout));
		if (timeleft > 0)
			err = 0;
		else if (!timeleft)
			err = -EIO;  /* timed out */
		else
			err = timeleft;  /* killed */
2334 2335 2336
	}
	dout("do_request waited, got %d\n", err);
	mutex_lock(&mdsc->mutex);
2337

2338 2339 2340 2341 2342
	/* only abort if we didn't race with a real reply */
	if (req->r_got_result) {
		err = le32_to_cpu(req->r_reply_info.head->result);
	} else if (err < 0) {
		dout("aborted request %lld with %d\n", req->r_tid, err);
2343 2344 2345 2346 2347 2348 2349

		/*
		 * ensure we aren't running concurrently with
		 * ceph_fill_trace or ceph_readdir_prepopulate, which
		 * rely on locks (dir mutex) held by our caller.
		 */
		mutex_lock(&req->r_fill_mutex);
2350 2351
		req->r_err = err;
		req->r_aborted = true;
2352
		mutex_unlock(&req->r_fill_mutex);
2353

2354
		if (req->r_locked_dir &&
2355 2356
		    (req->r_op & CEPH_MDS_OP_WRITE))
			ceph_invalidate_dir_request(req);
Sage Weil's avatar
Sage Weil committed
2357
	} else {
2358
		err = req->r_err;
Sage Weil's avatar
Sage Weil committed
2359 2360
	}

2361 2362
out:
	mutex_unlock(&mdsc->mutex);
Sage Weil's avatar
Sage Weil committed
2363 2364 2365 2366
	dout("do_request %p done, result %d\n", req, err);
	return err;
}

2367
/*
2368
 * Invalidate dir's completeness, dentry lease state on an aborted MDS
2369 2370 2371 2372 2373 2374
 * namespace request.
 */
void ceph_invalidate_dir_request(struct ceph_mds_request *req)
{
	struct inode *inode = req->r_locked_dir;

2375
	dout("invalidate_dir_request %p (complete, lease(s))\n", inode);
2376

2377
	ceph_dir_clear_complete(inode);
2378 2379 2380 2381 2382 2383
	if (req->r_dentry)
		ceph_invalidate_dentry_lease(req->r_dentry);
	if (req->r_old_dentry)
		ceph_invalidate_dentry_lease(req->r_old_dentry);
}

Sage Weil's avatar
Sage Weil committed
2384 2385 2386 2387 2388 2389 2390 2391 2392 2393 2394 2395 2396
/*
 * Handle mds reply.
 *
 * We take the session mutex and parse and process the reply immediately.
 * This preserves the logical ordering of replies, capabilities, etc., sent
 * by the MDS as they are applied to our local cache.
 */
static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
{
	struct ceph_mds_client *mdsc = session->s_mdsc;
	struct ceph_mds_request *req;
	struct ceph_mds_reply_head *head = msg->front.iov_base;
	struct ceph_mds_reply_info_parsed *rinfo;  /* parsed reply info */
2397
	struct ceph_snap_realm *realm;
Sage Weil's avatar
Sage Weil committed
2398 2399
	u64 tid;
	int err, result;
2400
	int mds = session->s_mds;
Sage Weil's avatar
Sage Weil committed
2401 2402 2403

	if (msg->front.iov_len < sizeof(*head)) {
		pr_err("mdsc_handle_reply got corrupt (short) reply\n");
2404
		ceph_msg_dump(msg);
Sage Weil's avatar
Sage Weil committed
2405 2406 2407 2408
		return;
	}

	/* get request, session */
2409
	tid = le64_to_cpu(msg->hdr.tid);
Sage Weil's avatar
Sage Weil committed
2410 2411 2412 2413 2414 2415 2416 2417 2418 2419
	mutex_lock(&mdsc->mutex);
	req = __lookup_request(mdsc, tid);
	if (!req) {
		dout("handle_reply on unknown tid %llu\n", tid);
		mutex_unlock(&mdsc->mutex);
		return;
	}
	dout("handle_reply %p\n", req);

	/* correct session? */
2420
	if (req->r_session != session) {
Sage Weil's avatar
Sage Weil committed
2421 2422 2423 2424 2425 2426 2427 2428 2429 2430
		pr_err("mdsc_handle_reply got %llu on session mds%d"
		       " not mds%d\n", tid, session->s_mds,
		       req->r_session ? req->r_session->s_mds : -1);
		mutex_unlock(&mdsc->mutex);
		goto out;
	}

	/* dup? */
	if ((req->r_got_unsafe && !head->safe) ||
	    (req->r_got_safe && head->safe)) {
2431
		pr_warn("got a dup %s reply on %llu from mds%d\n",
Sage Weil's avatar
Sage Weil committed
2432 2433 2434 2435
			   head->safe ? "safe" : "unsafe", tid, mds);
		mutex_unlock(&mdsc->mutex);
		goto out;
	}
2436
	if (req->r_got_safe) {
2437
		pr_warn("got unsafe after safe on %llu from mds%d\n",
2438 2439 2440 2441
			   tid, mds);
		mutex_unlock(&mdsc->mutex);
		goto out;
	}
Sage Weil's avatar
Sage Weil committed
2442 2443 2444 2445

	result = le32_to_cpu(head->result);

	/*
2446 2447 2448 2449 2450
	 * Handle an ESTALE
	 * if we're not talking to the authority, send to them
	 * if the authority has changed while we weren't looking,
	 * send to new authority
	 * Otherwise we just have to return an ESTALE
Sage Weil's avatar
Sage Weil committed
2451 2452
	 */
	if (result == -ESTALE) {
2453
		dout("got ESTALE on request %llu", req->r_tid);
2454
		req->r_resend_mds = -1;
Yan, Zheng's avatar
Yan, Zheng committed
2455
		if (req->r_direct_mode != USE_AUTH_MDS) {
2456 2457
			dout("not using auth, setting for that now");
			req->r_direct_mode = USE_AUTH_MDS;
Sage Weil's avatar
Sage Weil committed
2458 2459 2460
			__do_request(mdsc, req);
			mutex_unlock(&mdsc->mutex);
			goto out;
2461
		} else  {
Yan, Zheng's avatar
Yan, Zheng committed
2462 2463 2464
			int mds = __choose_mds(mdsc, req);
			if (mds >= 0 && mds != req->r_session->s_mds) {
				dout("but auth changed, so resending");
2465 2466 2467 2468
				__do_request(mdsc, req);
				mutex_unlock(&mdsc->mutex);
				goto out;
			}
Sage Weil's avatar
Sage Weil committed
2469
		}
2470
		dout("have to return ESTALE on request %llu", req->r_tid);
Sage Weil's avatar
Sage Weil committed
2471 2472
	}

2473

Sage Weil's avatar
Sage Weil committed
2474 2475 2476 2477 2478 2479 2480 2481 2482 2483 2484 2485 2486 2487 2488
	if (head->safe) {
		req->r_got_safe = true;
		__unregister_request(mdsc, req);

		if (req->r_got_unsafe) {
			/*
			 * We already handled the unsafe response, now do the
			 * cleanup.  No need to examine the response; the MDS
			 * doesn't include any result info in the safe
			 * response.  And even if it did, there is nothing
			 * useful we could do with a revised return value.
			 */
			dout("got safe reply %llu, mds%d\n", tid, mds);

			/* last unsafe request during umount? */
2489
			if (mdsc->stopping && !__get_oldest_req(mdsc))
2490
				complete_all(&mdsc->safe_umount_waiters);
Sage Weil's avatar
Sage Weil committed
2491 2492 2493
			mutex_unlock(&mdsc->mutex);
			goto out;
		}
2494
	} else {
Sage Weil's avatar
Sage Weil committed
2495 2496
		req->r_got_unsafe = true;
		list_add_tail(&req->r_unsafe_item, &req->r_session->s_unsafe);
2497 2498 2499 2500 2501 2502 2503 2504
		if (req->r_unsafe_dir) {
			struct ceph_inode_info *ci =
					ceph_inode(req->r_unsafe_dir);
			spin_lock(&ci->i_unsafe_lock);
			list_add_tail(&req->r_unsafe_dir_item,
				      &ci->i_unsafe_dirops);
			spin_unlock(&ci->i_unsafe_lock);
		}
Sage Weil's avatar
Sage Weil committed
2505 2506 2507 2508
	}

	dout("handle_reply tid %lld result %d\n", tid, result);
	rinfo = &req->r_reply_info;
2509
	err = parse_reply_info(msg, rinfo, session->s_con.peer_features);
Sage Weil's avatar
Sage Weil committed
2510 2511 2512 2513
	mutex_unlock(&mdsc->mutex);

	mutex_lock(&session->s_mutex);
	if (err < 0) {
2514
		pr_err("mdsc_handle_reply got corrupt reply mds%d(tid:%lld)\n", mds, tid);
2515
		ceph_msg_dump(msg);
Sage Weil's avatar
Sage Weil committed
2516 2517 2518 2519
		goto out_err;
	}

	/* snap trace */
2520
	realm = NULL;
Sage Weil's avatar
Sage Weil committed
2521 2522 2523
	if (rinfo->snapblob_len) {
		down_write(&mdsc->snap_rwsem);
		ceph_update_snap_trace(mdsc, rinfo->snapblob,
2524 2525 2526
				rinfo->snapblob + rinfo->snapblob_len,
				le32_to_cpu(head->op) == CEPH_MDS_OP_RMSNAP,
				&realm);
Sage Weil's avatar
Sage Weil committed
2527 2528 2529 2530 2531 2532
		downgrade_write(&mdsc->snap_rwsem);
	} else {
		down_read(&mdsc->snap_rwsem);
	}

	/* insert trace into our cache */
2533
	mutex_lock(&req->r_fill_mutex);
2534
	err = ceph_fill_trace(mdsc->fsc->sb, req, req->r_session);
Sage Weil's avatar
Sage Weil committed
2535
	if (err == 0) {
2536
		if (result == 0 && (req->r_op == CEPH_MDS_OP_READDIR ||
2537
				    req->r_op == CEPH_MDS_OP_LSSNAP))
Sage Weil's avatar
Sage Weil committed
2538
			ceph_readdir_prepopulate(req, req->r_session);
2539
		ceph_unreserve_caps(mdsc, &req->r_caps_reservation);
Sage Weil's avatar
Sage Weil committed
2540
	}
2541
	mutex_unlock(&req->r_fill_mutex);
Sage Weil's avatar
Sage Weil committed
2542 2543

	up_read(&mdsc->snap_rwsem);
2544 2545
	if (realm)
		ceph_put_snap_realm(mdsc, realm);
2546 2547 2548 2549 2550 2551 2552

	if (err == 0 && req->r_got_unsafe && req->r_target_inode) {
		struct ceph_inode_info *ci = ceph_inode(req->r_target_inode);
		spin_lock(&ci->i_unsafe_lock);
		list_add_tail(&req->r_unsafe_target_item, &ci->i_unsafe_iops);
		spin_unlock(&ci->i_unsafe_lock);
	}
Sage Weil's avatar
Sage Weil committed
2553
out_err:
2554 2555 2556 2557 2558
	mutex_lock(&mdsc->mutex);
	if (!req->r_aborted) {
		if (err) {
			req->r_err = err;
		} else {
2559
			req->r_reply =  ceph_msg_get(msg);
2560 2561
			req->r_got_result = true;
		}
Sage Weil's avatar
Sage Weil committed
2562
	} else {
2563
		dout("reply arrived after request %lld was aborted\n", tid);
Sage Weil's avatar
Sage Weil committed
2564
	}
2565
	mutex_unlock(&mdsc->mutex);
Sage Weil's avatar
Sage Weil committed
2566 2567 2568 2569 2570 2571 2572 2573 2574 2575 2576 2577 2578 2579 2580

	mutex_unlock(&session->s_mutex);

	/* kick calling process */
	complete_request(mdsc, req);
out:
	ceph_mdsc_put_request(req);
	return;
}



/*
 * handle mds notification that our request has been forwarded.
 */
2581 2582 2583
static void handle_forward(struct ceph_mds_client *mdsc,
			   struct ceph_mds_session *session,
			   struct ceph_msg *msg)
Sage Weil's avatar
Sage Weil committed
2584 2585
{
	struct ceph_mds_request *req;
2586
	u64 tid = le64_to_cpu(msg->hdr.tid);
Sage Weil's avatar
Sage Weil committed
2587 2588 2589 2590 2591 2592
	u32 next_mds;
	u32 fwd_seq;
	int err = -EINVAL;
	void *p = msg->front.iov_base;
	void *end = p + msg->front.iov_len;

2593
	ceph_decode_need(&p, end, 2*sizeof(u32), bad);
2594 2595
	next_mds = ceph_decode_32(&p);
	fwd_seq = ceph_decode_32(&p);
Sage Weil's avatar
Sage Weil committed
2596 2597 2598 2599

	mutex_lock(&mdsc->mutex);
	req = __lookup_request(mdsc, tid);
	if (!req) {
2600
		dout("forward tid %llu to mds%d - req dne\n", tid, next_mds);
Sage Weil's avatar
Sage Weil committed
2601 2602 2603
		goto out;  /* dup reply? */
	}

2604 2605 2606 2607 2608
	if (req->r_aborted) {
		dout("forward tid %llu aborted, unregistering\n", tid);
		__unregister_request(mdsc, req);
	} else if (fwd_seq <= req->r_num_fwd) {
		dout("forward tid %llu to mds%d - old seq %d <= %d\n",
Sage Weil's avatar
Sage Weil committed
2609 2610 2611
		     tid, next_mds, req->r_num_fwd, fwd_seq);
	} else {
		/* resend. forward race not possible; mds would drop */
2612 2613 2614
		dout("forward tid %llu to mds%d (we resend)\n", tid, next_mds);
		BUG_ON(req->r_err);
		BUG_ON(req->r_got_result);
2615
		req->r_attempts = 0;
Sage Weil's avatar
Sage Weil committed
2616 2617 2618 2619 2620 2621 2622 2623 2624 2625 2626 2627 2628 2629 2630 2631 2632 2633 2634 2635 2636 2637 2638
		req->r_num_fwd = fwd_seq;
		req->r_resend_mds = next_mds;
		put_request_session(req);
		__do_request(mdsc, req);
	}
	ceph_mdsc_put_request(req);
out:
	mutex_unlock(&mdsc->mutex);
	return;

bad:
	pr_err("mdsc_handle_forward decode error err=%d\n", err);
}

/*
 * handle a mds session control message
 */
static void handle_session(struct ceph_mds_session *session,
			   struct ceph_msg *msg)
{
	struct ceph_mds_client *mdsc = session->s_mdsc;
	u32 op;
	u64 seq;
2639
	int mds = session->s_mds;
Sage Weil's avatar
Sage Weil committed
2640 2641 2642 2643 2644 2645 2646 2647 2648 2649
	struct ceph_mds_session_head *h = msg->front.iov_base;
	int wake = 0;

	/* decode */
	if (msg->front.iov_len != sizeof(*h))
		goto bad;
	op = le32_to_cpu(h->op);
	seq = le64_to_cpu(h->seq);

	mutex_lock(&mdsc->mutex);
2650 2651
	if (op == CEPH_SESSION_CLOSE)
		__unregister_session(mdsc, session);
Sage Weil's avatar
Sage Weil committed
2652 2653 2654 2655 2656 2657 2658 2659
	/* FIXME: this ttl calculation is generous */
	session->s_ttl = jiffies + HZ*mdsc->mdsmap->m_session_autoclose;
	mutex_unlock(&mdsc->mutex);

	mutex_lock(&session->s_mutex);

	dout("handle_session mds%d %s %p state %s seq %llu\n",
	     mds, ceph_session_op_name(op), session,
2660
	     ceph_session_state_name(session->s_state), seq);
Sage Weil's avatar
Sage Weil committed
2661 2662 2663 2664 2665 2666 2667 2668

	if (session->s_state == CEPH_MDS_SESSION_HUNG) {
		session->s_state = CEPH_MDS_SESSION_OPEN;
		pr_info("mds%d came back\n", session->s_mds);
	}

	switch (op) {
	case CEPH_SESSION_OPEN:
2669 2670
		if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
			pr_info("mds%d reconnect success\n", session->s_mds);
Sage Weil's avatar
Sage Weil committed
2671 2672 2673 2674 2675 2676 2677 2678 2679 2680 2681 2682 2683
		session->s_state = CEPH_MDS_SESSION_OPEN;
		renewed_caps(mdsc, session, 0);
		wake = 1;
		if (mdsc->stopping)
			__close_session(mdsc, session);
		break;

	case CEPH_SESSION_RENEWCAPS:
		if (session->s_renew_seq == seq)
			renewed_caps(mdsc, session, 1);
		break;

	case CEPH_SESSION_CLOSE:
2684 2685
		if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
			pr_info("mds%d reconnect denied\n", session->s_mds);
2686
		cleanup_session_requests(mdsc, session);
Sage Weil's avatar
Sage Weil committed
2687
		remove_session_caps(session);
2688
		wake = 2; /* for good measure */
2689
		wake_up_all(&mdsc->session_close_wq);
Sage Weil's avatar
Sage Weil committed
2690 2691 2692 2693 2694
		break;

	case CEPH_SESSION_STALE:
		pr_info("mds%d caps went stale, renewing\n",
			session->s_mds);
2695
		spin_lock(&session->s_gen_ttl_lock);
Sage Weil's avatar
Sage Weil committed
2696
		session->s_cap_gen++;
2697
		session->s_cap_ttl = jiffies - 1;
2698
		spin_unlock(&session->s_gen_ttl_lock);
Sage Weil's avatar
Sage Weil committed
2699 2700 2701 2702 2703 2704 2705
		send_renew_caps(mdsc, session);
		break;

	case CEPH_SESSION_RECALL_STATE:
		trim_caps(mdsc, session, le32_to_cpu(h->max_caps));
		break;

2706 2707 2708 2709
	case CEPH_SESSION_FLUSHMSG:
		send_flushmsg_ack(mdsc, session, seq);
		break;

2710 2711 2712 2713 2714 2715 2716 2717
	case CEPH_SESSION_FORCE_RO:
		dout("force_session_readonly %p\n", session);
		spin_lock(&session->s_cap_lock);
		session->s_readonly = true;
		spin_unlock(&session->s_cap_lock);
		wake_up_session_caps(session, 0);
		break;

Sage Weil's avatar
Sage Weil committed
2718 2719 2720 2721 2722 2723 2724 2725 2726
	default:
		pr_err("mdsc_handle_session bad op %d mds%d\n", op, mds);
		WARN_ON(1);
	}

	mutex_unlock(&session->s_mutex);
	if (wake) {
		mutex_lock(&mdsc->mutex);
		__wake_requests(mdsc, &session->s_waiting);
2727 2728
		if (wake == 2)
			kick_requests(mdsc, mds);
Sage Weil's avatar
Sage Weil committed
2729 2730 2731 2732 2733 2734 2735
		mutex_unlock(&mdsc->mutex);
	}
	return;

bad:
	pr_err("mdsc_handle_session corrupt message mds%d len %d\n", mds,
	       (int)msg->front.iov_len);
2736
	ceph_msg_dump(msg);
Sage Weil's avatar
Sage Weil committed
2737 2738 2739 2740 2741 2742 2743 2744 2745 2746 2747
	return;
}


/*
 * called under session->mutex.
 */
static void replay_unsafe_requests(struct ceph_mds_client *mdsc,
				   struct ceph_mds_session *session)
{
	struct ceph_mds_request *req, *nreq;
2748
	struct rb_node *p;
Sage Weil's avatar
Sage Weil committed
2749 2750 2751 2752 2753 2754
	int err;

	dout("replay_unsafe_requests mds%d\n", session->s_mds);

	mutex_lock(&mdsc->mutex);
	list_for_each_entry_safe(req, nreq, &session->s_unsafe, r_unsafe_item) {
2755
		err = __prepare_send_request(mdsc, req, session->s_mds, true);
Sage Weil's avatar
Sage Weil committed
2756 2757 2758 2759 2760
		if (!err) {
			ceph_msg_get(req->r_request);
			ceph_con_send(&session->s_con, req->r_request);
		}
	}
2761 2762 2763 2764 2765 2766 2767 2768 2769 2770 2771 2772 2773 2774 2775

	/*
	 * also re-send old requests when MDS enters reconnect stage. So that MDS
	 * can process completed request in clientreplay stage.
	 */
	p = rb_first(&mdsc->request_tree);
	while (p) {
		req = rb_entry(p, struct ceph_mds_request, r_node);
		p = rb_next(p);
		if (req->r_got_unsafe)
			continue;
		if (req->r_attempts == 0)
			continue; /* only old requests */
		if (req->r_session &&
		    req->r_session->s_mds == session->s_mds) {
2776 2777
			err = __prepare_send_request(mdsc, req,
						     session->s_mds, true);
2778 2779 2780 2781 2782 2783
			if (!err) {
				ceph_msg_get(req->r_request);
				ceph_con_send(&session->s_con, req->r_request);
			}
		}
	}
Sage Weil's avatar
Sage Weil committed
2784 2785 2786 2787 2788 2789 2790 2791 2792
	mutex_unlock(&mdsc->mutex);
}

/*
 * Encode information about a cap for a reconnect with the MDS.
 */
static int encode_caps_cb(struct inode *inode, struct ceph_cap *cap,
			  void *arg)
{
2793 2794 2795 2796 2797
	union {
		struct ceph_mds_cap_reconnect v2;
		struct ceph_mds_cap_reconnect_v1 v1;
	} rec;
	size_t reclen;
Sage Weil's avatar
Sage Weil committed
2798
	struct ceph_inode_info *ci;
2799 2800
	struct ceph_reconnect_state *recon_state = arg;
	struct ceph_pagelist *pagelist = recon_state->pagelist;
Sage Weil's avatar
Sage Weil committed
2801 2802 2803 2804 2805 2806 2807 2808 2809 2810
	char *path;
	int pathlen, err;
	u64 pathbase;
	struct dentry *dentry;

	ci = cap->ci;

	dout(" adding %p ino %llx.%llx cap %p %lld %s\n",
	     inode, ceph_vinop(inode), cap, cap->cap_id,
	     ceph_cap_string(cap->issued));
2811 2812 2813
	err = ceph_pagelist_encode_64(pagelist, ceph_ino(inode));
	if (err)
		return err;
Sage Weil's avatar
Sage Weil committed
2814 2815 2816 2817 2818 2819

	dentry = d_find_alias(inode);
	if (dentry) {
		path = ceph_mdsc_build_path(dentry, &pathlen, &pathbase, 0);
		if (IS_ERR(path)) {
			err = PTR_ERR(path);
2820
			goto out_dput;
Sage Weil's avatar
Sage Weil committed
2821 2822 2823 2824 2825
		}
	} else {
		path = NULL;
		pathlen = 0;
	}
2826 2827
	err = ceph_pagelist_encode_string(pagelist, path, pathlen);
	if (err)
2828
		goto out_free;
Sage Weil's avatar
Sage Weil committed
2829

2830
	spin_lock(&ci->i_ceph_lock);
Sage Weil's avatar
Sage Weil committed
2831 2832
	cap->seq = 0;        /* reset cap seq */
	cap->issue_seq = 0;  /* and issue_seq */
2833
	cap->mseq = 0;       /* and migrate_seq */
2834
	cap->cap_gen = cap->session->s_cap_gen;
2835 2836 2837 2838 2839 2840 2841 2842 2843 2844 2845 2846 2847 2848 2849 2850 2851 2852 2853 2854

	if (recon_state->flock) {
		rec.v2.cap_id = cpu_to_le64(cap->cap_id);
		rec.v2.wanted = cpu_to_le32(__ceph_caps_wanted(ci));
		rec.v2.issued = cpu_to_le32(cap->issued);
		rec.v2.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);
		rec.v2.pathbase = cpu_to_le64(pathbase);
		rec.v2.flock_len = 0;
		reclen = sizeof(rec.v2);
	} else {
		rec.v1.cap_id = cpu_to_le64(cap->cap_id);
		rec.v1.wanted = cpu_to_le32(__ceph_caps_wanted(ci));
		rec.v1.issued = cpu_to_le32(cap->issued);
		rec.v1.size = cpu_to_le64(inode->i_size);
		ceph_encode_timespec(&rec.v1.mtime, &inode->i_mtime);
		ceph_encode_timespec(&rec.v1.atime, &inode->i_atime);
		rec.v1.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);
		rec.v1.pathbase = cpu_to_le64(pathbase);
		reclen = sizeof(rec.v1);
	}
2855
	spin_unlock(&ci->i_ceph_lock);
Sage Weil's avatar
Sage Weil committed
2856

2857 2858
	if (recon_state->flock) {
		int num_fcntl_locks, num_flock_locks;
2859 2860 2861 2862 2863 2864 2865 2866 2867 2868 2869 2870 2871 2872 2873 2874 2875 2876 2877 2878 2879 2880 2881 2882 2883 2884 2885 2886 2887 2888 2889
		struct ceph_filelock *flocks;

encode_again:
		ceph_count_locks(inode, &num_fcntl_locks, &num_flock_locks);
		flocks = kmalloc((num_fcntl_locks+num_flock_locks) *
				 sizeof(struct ceph_filelock), GFP_NOFS);
		if (!flocks) {
			err = -ENOMEM;
			goto out_free;
		}
		err = ceph_encode_locks_to_buffer(inode, flocks,
						  num_fcntl_locks,
						  num_flock_locks);
		if (err) {
			kfree(flocks);
			if (err == -ENOSPC)
				goto encode_again;
			goto out_free;
		}
		/*
		 * number of encoded locks is stable, so copy to pagelist
		 */
		rec.v2.flock_len = cpu_to_le32(2*sizeof(u32) +
				    (num_fcntl_locks+num_flock_locks) *
				    sizeof(struct ceph_filelock));
		err = ceph_pagelist_append(pagelist, &rec, reclen);
		if (!err)
			err = ceph_locks_to_pagelist(flocks, pagelist,
						     num_fcntl_locks,
						     num_flock_locks);
		kfree(flocks);
2890 2891
	} else {
		err = ceph_pagelist_append(pagelist, &rec, reclen);
2892
	}
2893 2894

	recon_state->nr_caps++;
2895
out_free:
Sage Weil's avatar
Sage Weil committed
2896
	kfree(path);
2897
out_dput:
Sage Weil's avatar
Sage Weil committed
2898
	dput(dentry);
2899
	return err;
Sage Weil's avatar
Sage Weil committed
2900 2901 2902 2903 2904 2905 2906 2907 2908 2909 2910 2911 2912 2913 2914
}


/*
 * If an MDS fails and recovers, clients need to reconnect in order to
 * reestablish shared state.  This includes all caps issued through
 * this session _and_ the snap_realm hierarchy.  Because it's not
 * clear which snap realms the mds cares about, we send everything we
 * know about.. that ensures we'll then get any new info the
 * recovering MDS might have.
 *
 * This is a relatively heavyweight operation, but it's rare.
 *
 * called with mdsc->mutex held.
 */
2915 2916
static void send_mds_reconnect(struct ceph_mds_client *mdsc,
			       struct ceph_mds_session *session)
Sage Weil's avatar
Sage Weil committed
2917 2918
{
	struct ceph_msg *reply;
Sage Weil's avatar
Sage Weil committed
2919
	struct rb_node *p;
2920
	int mds = session->s_mds;
2921
	int err = -ENOMEM;
2922
	int s_nr_caps;
2923
	struct ceph_pagelist *pagelist;
2924
	struct ceph_reconnect_state recon_state;
Sage Weil's avatar
Sage Weil committed
2925

2926
	pr_info("mds%d reconnect start\n", mds);
Sage Weil's avatar
Sage Weil committed
2927

2928 2929 2930 2931 2932
	pagelist = kmalloc(sizeof(*pagelist), GFP_NOFS);
	if (!pagelist)
		goto fail_nopagelist;
	ceph_pagelist_init(pagelist);

2933
	reply = ceph_msg_new(CEPH_MSG_CLIENT_RECONNECT, 0, GFP_NOFS, false);
2934
	if (!reply)
2935 2936
		goto fail_nomsg;

2937 2938 2939
	mutex_lock(&session->s_mutex);
	session->s_state = CEPH_MDS_SESSION_RECONNECTING;
	session->s_seq = 0;
Sage Weil's avatar
Sage Weil committed
2940 2941

	dout("session %p state %s\n", session,
2942
	     ceph_session_state_name(session->s_state));
Sage Weil's avatar
Sage Weil committed
2943

2944 2945 2946 2947 2948
	spin_lock(&session->s_gen_ttl_lock);
	session->s_cap_gen++;
	spin_unlock(&session->s_gen_ttl_lock);

	spin_lock(&session->s_cap_lock);
2949 2950
	/* don't know if session is readonly */
	session->s_readonly = 0;
2951 2952 2953 2954 2955 2956
	/*
	 * notify __ceph_remove_cap() that we are composing cap reconnect.
	 * If a cap get released before being added to the cap reconnect,
	 * __ceph_remove_cap() should skip queuing cap release.
	 */
	session->s_cap_reconnect = 1;
2957
	/* drop old cap expires; we're about to reestablish that state */
2958
	cleanup_cap_releases(mdsc, session);
2959

2960
	/* trim unused caps to reduce MDS's cache rejoin time */
2961 2962
	if (mdsc->fsc->sb->s_root)
		shrink_dcache_parent(mdsc->fsc->sb->s_root);
2963 2964 2965 2966 2967 2968 2969 2970 2971 2972 2973

	ceph_con_close(&session->s_con);
	ceph_con_open(&session->s_con,
		      CEPH_ENTITY_TYPE_MDS, mds,
		      ceph_mdsmap_get_addr(mdsc->mdsmap, mds));

	/* replay unsafe requests */
	replay_unsafe_requests(mdsc, session);

	down_read(&mdsc->snap_rwsem);

Sage Weil's avatar
Sage Weil committed
2974
	/* traverse this session's caps */
2975 2976
	s_nr_caps = session->s_nr_caps;
	err = ceph_pagelist_encode_32(pagelist, s_nr_caps);
2977 2978
	if (err)
		goto fail;
2979

2980
	recon_state.nr_caps = 0;
2981 2982 2983
	recon_state.pagelist = pagelist;
	recon_state.flock = session->s_con.peer_features & CEPH_FEATURE_FLOCK;
	err = iterate_session_caps(session, encode_caps_cb, &recon_state);
Sage Weil's avatar
Sage Weil committed
2984
	if (err < 0)
2985
		goto fail;
Sage Weil's avatar
Sage Weil committed
2986

2987 2988 2989 2990
	spin_lock(&session->s_cap_lock);
	session->s_cap_reconnect = 0;
	spin_unlock(&session->s_cap_lock);

Sage Weil's avatar
Sage Weil committed
2991 2992 2993 2994 2995
	/*
	 * snaprealms.  we provide mds with the ino, seq (version), and
	 * parent for all of our realms.  If the mds has any newer info,
	 * it will tell us.
	 */
Sage Weil's avatar
Sage Weil committed
2996 2997 2998
	for (p = rb_first(&mdsc->snap_realms); p; p = rb_next(p)) {
		struct ceph_snap_realm *realm =
			rb_entry(p, struct ceph_snap_realm, node);
2999
		struct ceph_mds_snaprealm_reconnect sr_rec;
Sage Weil's avatar
Sage Weil committed
3000 3001 3002

		dout(" adding snap realm %llx seq %lld parent %llx\n",
		     realm->ino, realm->seq, realm->parent_ino);
3003 3004 3005 3006 3007 3008
		sr_rec.ino = cpu_to_le64(realm->ino);
		sr_rec.seq = cpu_to_le64(realm->seq);
		sr_rec.parent = cpu_to_le64(realm->parent_ino);
		err = ceph_pagelist_append(pagelist, &sr_rec, sizeof(sr_rec));
		if (err)
			goto fail;
Sage Weil's avatar
Sage Weil committed
3009 3010
	}

3011 3012
	if (recon_state.flock)
		reply->hdr.version = cpu_to_le16(2);
3013 3014 3015 3016 3017 3018 3019 3020

	/* raced with cap release? */
	if (s_nr_caps != recon_state.nr_caps) {
		struct page *page = list_first_entry(&pagelist->head,
						     struct page, lru);
		__le32 *addr = kmap_atomic(page);
		*addr = cpu_to_le32(recon_state.nr_caps);
		kunmap_atomic(addr);
3021
	}
3022 3023 3024

	reply->hdr.data_len = cpu_to_le32(pagelist->length);
	ceph_msg_data_add_pagelist(reply, pagelist);
3025 3026 3027

	ceph_early_kick_flushing_caps(mdsc, session);

Sage Weil's avatar
Sage Weil committed
3028 3029
	ceph_con_send(&session->s_con, reply);

3030 3031 3032 3033 3034 3035
	mutex_unlock(&session->s_mutex);

	mutex_lock(&mdsc->mutex);
	__wake_requests(mdsc, &session->s_waiting);
	mutex_unlock(&mdsc->mutex);

Sage Weil's avatar
Sage Weil committed
3036 3037 3038
	up_read(&mdsc->snap_rwsem);
	return;

3039
fail:
Sage Weil's avatar
Sage Weil committed
3040
	ceph_msg_put(reply);
3041 3042
	up_read(&mdsc->snap_rwsem);
	mutex_unlock(&session->s_mutex);
3043 3044 3045
fail_nomsg:
	ceph_pagelist_release(pagelist);
fail_nopagelist:
3046 3047
	pr_err("error %d preparing reconnect for mds%d\n", err, mds);
	return;
Sage Weil's avatar
Sage Weil committed
3048 3049 3050 3051 3052 3053 3054 3055 3056 3057 3058 3059 3060 3061 3062 3063 3064 3065 3066 3067 3068 3069 3070 3071 3072 3073 3074
}


/*
 * compare old and new mdsmaps, kicking requests
 * and closing out old connections as necessary
 *
 * called under mdsc->mutex.
 */
static void check_new_map(struct ceph_mds_client *mdsc,
			  struct ceph_mdsmap *newmap,
			  struct ceph_mdsmap *oldmap)
{
	int i;
	int oldstate, newstate;
	struct ceph_mds_session *s;

	dout("check_new_map new %u old %u\n",
	     newmap->m_epoch, oldmap->m_epoch);

	for (i = 0; i < oldmap->m_max_mds && i < mdsc->max_sessions; i++) {
		if (mdsc->sessions[i] == NULL)
			continue;
		s = mdsc->sessions[i];
		oldstate = ceph_mdsmap_get_state(oldmap, i);
		newstate = ceph_mdsmap_get_state(newmap, i);

3075
		dout("check_new_map mds%d state %s%s -> %s%s (session %s)\n",
Sage Weil's avatar
Sage Weil committed
3076
		     i, ceph_mds_state_name(oldstate),
3077
		     ceph_mdsmap_is_laggy(oldmap, i) ? " (laggy)" : "",
Sage Weil's avatar
Sage Weil committed
3078
		     ceph_mds_state_name(newstate),
3079
		     ceph_mdsmap_is_laggy(newmap, i) ? " (laggy)" : "",
3080
		     ceph_session_state_name(s->s_state));
Sage Weil's avatar
Sage Weil committed
3081

3082 3083
		if (i >= newmap->m_max_mds ||
		    memcmp(ceph_mdsmap_get_addr(oldmap, i),
Sage Weil's avatar
Sage Weil committed
3084 3085 3086 3087 3088 3089
			   ceph_mdsmap_get_addr(newmap, i),
			   sizeof(struct ceph_entity_addr))) {
			if (s->s_state == CEPH_MDS_SESSION_OPENING) {
				/* the session never opened, just close it
				 * out now */
				__wake_requests(mdsc, &s->s_waiting);
3090
				__unregister_session(mdsc, s);
Sage Weil's avatar
Sage Weil committed
3091 3092 3093 3094 3095 3096 3097 3098 3099 3100 3101 3102 3103 3104 3105 3106 3107
			} else {
				/* just close it */
				mutex_unlock(&mdsc->mutex);
				mutex_lock(&s->s_mutex);
				mutex_lock(&mdsc->mutex);
				ceph_con_close(&s->s_con);
				mutex_unlock(&s->s_mutex);
				s->s_state = CEPH_MDS_SESSION_RESTARTING;
			}
		} else if (oldstate == newstate) {
			continue;  /* nothing new with this mds */
		}

		/*
		 * send reconnect?
		 */
		if (s->s_state == CEPH_MDS_SESSION_RESTARTING &&
3108 3109 3110 3111 3112
		    newstate >= CEPH_MDS_STATE_RECONNECT) {
			mutex_unlock(&mdsc->mutex);
			send_mds_reconnect(mdsc, s);
			mutex_lock(&mdsc->mutex);
		}
Sage Weil's avatar
Sage Weil committed
3113 3114

		/*
3115
		 * kick request on any mds that has gone active.
Sage Weil's avatar
Sage Weil committed
3116 3117 3118
		 */
		if (oldstate < CEPH_MDS_STATE_ACTIVE &&
		    newstate >= CEPH_MDS_STATE_ACTIVE) {
3119 3120 3121 3122
			if (oldstate != CEPH_MDS_STATE_CREATING &&
			    oldstate != CEPH_MDS_STATE_STARTING)
				pr_info("mds%d recovery completed\n", s->s_mds);
			kick_requests(mdsc, i);
Sage Weil's avatar
Sage Weil committed
3123
			ceph_kick_flushing_caps(mdsc, s);
3124
			wake_up_session_caps(s, 1);
Sage Weil's avatar
Sage Weil committed
3125 3126
		}
	}
3127 3128 3129 3130 3131 3132 3133 3134 3135 3136 3137 3138 3139 3140 3141

	for (i = 0; i < newmap->m_max_mds && i < mdsc->max_sessions; i++) {
		s = mdsc->sessions[i];
		if (!s)
			continue;
		if (!ceph_mdsmap_is_laggy(newmap, i))
			continue;
		if (s->s_state == CEPH_MDS_SESSION_OPEN ||
		    s->s_state == CEPH_MDS_SESSION_HUNG ||
		    s->s_state == CEPH_MDS_SESSION_CLOSING) {
			dout(" connecting to export targets of laggy mds%d\n",
			     i);
			__open_export_target_sessions(mdsc, s);
		}
	}
Sage Weil's avatar
Sage Weil committed
3142 3143 3144 3145 3146 3147 3148 3149 3150 3151 3152 3153 3154 3155 3156 3157 3158 3159 3160
}



/*
 * leases
 */

/*
 * caller must hold session s_mutex, dentry->d_lock
 */
void __ceph_mdsc_drop_dentry_lease(struct dentry *dentry)
{
	struct ceph_dentry_info *di = ceph_dentry(dentry);

	ceph_put_mds_session(di->lease_session);
	di->lease_session = NULL;
}

3161 3162 3163
static void handle_lease(struct ceph_mds_client *mdsc,
			 struct ceph_mds_session *session,
			 struct ceph_msg *msg)
Sage Weil's avatar
Sage Weil committed
3164
{
3165
	struct super_block *sb = mdsc->fsc->sb;
Sage Weil's avatar
Sage Weil committed
3166 3167 3168
	struct inode *inode;
	struct dentry *parent, *dentry;
	struct ceph_dentry_info *di;
3169
	int mds = session->s_mds;
Sage Weil's avatar
Sage Weil committed
3170
	struct ceph_mds_lease *h = msg->front.iov_base;
3171
	u32 seq;
Sage Weil's avatar
Sage Weil committed
3172 3173 3174 3175 3176 3177 3178 3179 3180 3181 3182
	struct ceph_vino vino;
	struct qstr dname;
	int release = 0;

	dout("handle_lease from mds%d\n", mds);

	/* decode */
	if (msg->front.iov_len < sizeof(*h) + sizeof(u32))
		goto bad;
	vino.ino = le64_to_cpu(h->ino);
	vino.snap = CEPH_NOSNAP;
3183
	seq = le32_to_cpu(h->seq);
Sage Weil's avatar
Sage Weil committed
3184 3185 3186 3187 3188 3189 3190
	dname.name = (void *)h + sizeof(*h) + sizeof(u32);
	dname.len = msg->front.iov_len - sizeof(*h) - sizeof(u32);
	if (dname.len != get_unaligned_le32(h+1))
		goto bad;

	/* lookup inode */
	inode = ceph_find_inode(sb, vino);
Sage Weil's avatar
Sage Weil committed
3191 3192
	dout("handle_lease %s, ino %llx %p %.*s\n",
	     ceph_lease_op_name(h->action), vino.ino, inode,
3193
	     dname.len, dname.name);
3194 3195 3196 3197

	mutex_lock(&session->s_mutex);
	session->s_seq++;

Sage Weil's avatar
Sage Weil committed
3198 3199 3200 3201 3202 3203 3204 3205 3206 3207 3208 3209 3210 3211 3212 3213 3214 3215 3216 3217 3218 3219
	if (inode == NULL) {
		dout("handle_lease no inode %llx\n", vino.ino);
		goto release;
	}

	/* dentry */
	parent = d_find_alias(inode);
	if (!parent) {
		dout("no parent dentry on inode %p\n", inode);
		WARN_ON(1);
		goto release;  /* hrm... */
	}
	dname.hash = full_name_hash(dname.name, dname.len);
	dentry = d_lookup(parent, &dname);
	dput(parent);
	if (!dentry)
		goto release;

	spin_lock(&dentry->d_lock);
	di = ceph_dentry(dentry);
	switch (h->action) {
	case CEPH_MDS_LEASE_REVOKE:
3220
		if (di->lease_session == session) {
3221 3222
			if (ceph_seq_cmp(di->lease_seq, seq) > 0)
				h->seq = cpu_to_le32(di->lease_seq);
Sage Weil's avatar
Sage Weil committed
3223 3224 3225 3226 3227 3228
			__ceph_mdsc_drop_dentry_lease(dentry);
		}
		release = 1;
		break;

	case CEPH_MDS_LEASE_RENEW:
3229
		if (di->lease_session == session &&
Sage Weil's avatar
Sage Weil committed
3230 3231 3232 3233
		    di->lease_gen == session->s_cap_gen &&
		    di->lease_renew_from &&
		    di->lease_renew_after == 0) {
			unsigned long duration =
3234
				msecs_to_jiffies(le32_to_cpu(h->duration_ms));
Sage Weil's avatar
Sage Weil committed
3235

3236
			di->lease_seq = seq;
Sage Weil's avatar
Sage Weil committed
3237 3238 3239 3240 3241 3242 3243 3244 3245 3246 3247 3248 3249 3250 3251 3252 3253 3254 3255 3256 3257 3258 3259 3260 3261 3262
			dentry->d_time = di->lease_renew_from + duration;
			di->lease_renew_after = di->lease_renew_from +
				(duration >> 1);
			di->lease_renew_from = 0;
		}
		break;
	}
	spin_unlock(&dentry->d_lock);
	dput(dentry);

	if (!release)
		goto out;

release:
	/* let's just reuse the same message */
	h->action = CEPH_MDS_LEASE_REVOKE_ACK;
	ceph_msg_get(msg);
	ceph_con_send(&session->s_con, msg);

out:
	iput(inode);
	mutex_unlock(&session->s_mutex);
	return;

bad:
	pr_err("corrupt lease message\n");
3263
	ceph_msg_dump(msg);
Sage Weil's avatar
Sage Weil committed
3264 3265 3266 3267 3268 3269 3270 3271 3272 3273 3274 3275 3276 3277 3278 3279 3280
}

void ceph_mdsc_lease_send_msg(struct ceph_mds_session *session,
			      struct inode *inode,
			      struct dentry *dentry, char action,
			      u32 seq)
{
	struct ceph_msg *msg;
	struct ceph_mds_lease *lease;
	int len = sizeof(*lease) + sizeof(u32);
	int dnamelen = 0;

	dout("lease_send_msg inode %p dentry %p %s to mds%d\n",
	     inode, dentry, ceph_lease_op_name(action), session->s_mds);
	dnamelen = dentry->d_name.len;
	len += dnamelen;

3281
	msg = ceph_msg_new(CEPH_MSG_CLIENT_LEASE, len, GFP_NOFS, false);
3282
	if (!msg)
Sage Weil's avatar
Sage Weil committed
3283 3284 3285 3286 3287 3288 3289 3290 3291 3292 3293 3294 3295 3296 3297 3298 3299 3300 3301 3302 3303 3304 3305 3306
		return;
	lease = msg->front.iov_base;
	lease->action = action;
	lease->ino = cpu_to_le64(ceph_vino(inode).ino);
	lease->first = lease->last = cpu_to_le64(ceph_vino(inode).snap);
	lease->seq = cpu_to_le32(seq);
	put_unaligned_le32(dnamelen, lease + 1);
	memcpy((void *)(lease + 1) + 4, dentry->d_name.name, dnamelen);

	/*
	 * if this is a preemptive lease RELEASE, no need to
	 * flush request stream, since the actual request will
	 * soon follow.
	 */
	msg->more_to_follow = (action == CEPH_MDS_LEASE_RELEASE);

	ceph_con_send(&session->s_con, msg);
}

/*
 * Preemptively release a lease we expect to invalidate anyway.
 * Pass @inode always, @dentry is optional.
 */
void ceph_mdsc_lease_release(struct ceph_mds_client *mdsc, struct inode *inode,
Sage Weil's avatar
Sage Weil committed
3307
			     struct dentry *dentry)
Sage Weil's avatar
Sage Weil committed
3308 3309 3310 3311 3312 3313 3314 3315 3316 3317 3318 3319 3320 3321 3322 3323
{
	struct ceph_dentry_info *di;
	struct ceph_mds_session *session;
	u32 seq;

	BUG_ON(inode == NULL);
	BUG_ON(dentry == NULL);

	/* is dentry lease valid? */
	spin_lock(&dentry->d_lock);
	di = ceph_dentry(dentry);
	if (!di || !di->lease_session ||
	    di->lease_session->s_mds < 0 ||
	    di->lease_gen != di->lease_session->s_cap_gen ||
	    !time_before(jiffies, dentry->d_time)) {
		dout("lease_release inode %p dentry %p -- "
Sage Weil's avatar
Sage Weil committed
3324 3325
		     "no lease\n",
		     inode, dentry);
Sage Weil's avatar
Sage Weil committed
3326 3327 3328 3329 3330 3331 3332 3333 3334 3335
		spin_unlock(&dentry->d_lock);
		return;
	}

	/* we do have a lease on this dentry; note mds and seq */
	session = ceph_get_mds_session(di->lease_session);
	seq = di->lease_seq;
	__ceph_mdsc_drop_dentry_lease(dentry);
	spin_unlock(&dentry->d_lock);

Sage Weil's avatar
Sage Weil committed
3336 3337
	dout("lease_release inode %p dentry %p to mds%d\n",
	     inode, dentry, session->s_mds);
Sage Weil's avatar
Sage Weil committed
3338 3339 3340 3341 3342 3343 3344 3345 3346 3347 3348 3349 3350 3351 3352 3353 3354 3355 3356 3357 3358 3359 3360 3361 3362 3363 3364 3365 3366 3367 3368 3369 3370 3371 3372 3373 3374 3375 3376 3377 3378 3379 3380 3381 3382 3383 3384 3385
	ceph_mdsc_lease_send_msg(session, inode, dentry,
				 CEPH_MDS_LEASE_RELEASE, seq);
	ceph_put_mds_session(session);
}

/*
 * drop all leases (and dentry refs) in preparation for umount
 */
static void drop_leases(struct ceph_mds_client *mdsc)
{
	int i;

	dout("drop_leases\n");
	mutex_lock(&mdsc->mutex);
	for (i = 0; i < mdsc->max_sessions; i++) {
		struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);
		if (!s)
			continue;
		mutex_unlock(&mdsc->mutex);
		mutex_lock(&s->s_mutex);
		mutex_unlock(&s->s_mutex);
		ceph_put_mds_session(s);
		mutex_lock(&mdsc->mutex);
	}
	mutex_unlock(&mdsc->mutex);
}



/*
 * delayed work -- periodically trim expired leases, renew caps with mds
 */
static void schedule_delayed(struct ceph_mds_client *mdsc)
{
	int delay = 5;
	unsigned hz = round_jiffies_relative(HZ * delay);
	schedule_delayed_work(&mdsc->delayed_work, hz);
}

static void delayed_work(struct work_struct *work)
{
	int i;
	struct ceph_mds_client *mdsc =
		container_of(work, struct ceph_mds_client, delayed_work.work);
	int renew_interval;
	int renew_caps;

	dout("mdsc delayed_work\n");
3386
	ceph_check_delayed_caps(mdsc);
Sage Weil's avatar
Sage Weil committed
3387 3388 3389 3390 3391 3392 3393 3394 3395 3396 3397 3398 3399 3400 3401 3402 3403 3404 3405 3406 3407 3408 3409 3410 3411 3412 3413 3414 3415 3416 3417 3418 3419 3420 3421 3422 3423

	mutex_lock(&mdsc->mutex);
	renew_interval = mdsc->mdsmap->m_session_timeout >> 2;
	renew_caps = time_after_eq(jiffies, HZ*renew_interval +
				   mdsc->last_renew_caps);
	if (renew_caps)
		mdsc->last_renew_caps = jiffies;

	for (i = 0; i < mdsc->max_sessions; i++) {
		struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);
		if (s == NULL)
			continue;
		if (s->s_state == CEPH_MDS_SESSION_CLOSING) {
			dout("resending session close request for mds%d\n",
			     s->s_mds);
			request_close_session(mdsc, s);
			ceph_put_mds_session(s);
			continue;
		}
		if (s->s_ttl && time_after(jiffies, s->s_ttl)) {
			if (s->s_state == CEPH_MDS_SESSION_OPEN) {
				s->s_state = CEPH_MDS_SESSION_HUNG;
				pr_info("mds%d hung\n", s->s_mds);
			}
		}
		if (s->s_state < CEPH_MDS_SESSION_OPEN) {
			/* this mds is failed or recovering, just wait */
			ceph_put_mds_session(s);
			continue;
		}
		mutex_unlock(&mdsc->mutex);

		mutex_lock(&s->s_mutex);
		if (renew_caps)
			send_renew_caps(mdsc, s);
		else
			ceph_con_keepalive(&s->s_con);
3424 3425
		if (s->s_state == CEPH_MDS_SESSION_OPEN ||
		    s->s_state == CEPH_MDS_SESSION_HUNG)
3426
			ceph_send_cap_releases(mdsc, s);
Sage Weil's avatar
Sage Weil committed
3427 3428 3429 3430 3431 3432 3433 3434 3435 3436
		mutex_unlock(&s->s_mutex);
		ceph_put_mds_session(s);

		mutex_lock(&mdsc->mutex);
	}
	mutex_unlock(&mdsc->mutex);

	schedule_delayed(mdsc);
}

3437
int ceph_mdsc_init(struct ceph_fs_client *fsc)
Sage Weil's avatar
Sage Weil committed
3438 3439

{
3440 3441 3442 3443 3444 3445 3446
	struct ceph_mds_client *mdsc;

	mdsc = kzalloc(sizeof(struct ceph_mds_client), GFP_NOFS);
	if (!mdsc)
		return -ENOMEM;
	mdsc->fsc = fsc;
	fsc->mdsc = mdsc;
Sage Weil's avatar
Sage Weil committed
3447 3448
	mutex_init(&mdsc->mutex);
	mdsc->mdsmap = kzalloc(sizeof(*mdsc->mdsmap), GFP_NOFS);
3449 3450
	if (mdsc->mdsmap == NULL) {
		kfree(mdsc);
3451
		return -ENOMEM;
3452
	}
3453

Sage Weil's avatar
Sage Weil committed
3454
	init_completion(&mdsc->safe_umount_waiters);
3455
	init_waitqueue_head(&mdsc->session_close_wq);
Sage Weil's avatar
Sage Weil committed
3456 3457
	INIT_LIST_HEAD(&mdsc->waiting_for_map);
	mdsc->sessions = NULL;
3458
	atomic_set(&mdsc->num_sessions, 0);
Sage Weil's avatar
Sage Weil committed
3459 3460
	mdsc->max_sessions = 0;
	mdsc->stopping = 0;
3461
	mdsc->last_snap_seq = 0;
Sage Weil's avatar
Sage Weil committed
3462
	init_rwsem(&mdsc->snap_rwsem);
Sage Weil's avatar
Sage Weil committed
3463
	mdsc->snap_realms = RB_ROOT;
Sage Weil's avatar
Sage Weil committed
3464 3465 3466
	INIT_LIST_HEAD(&mdsc->snap_empty);
	spin_lock_init(&mdsc->snap_empty_lock);
	mdsc->last_tid = 0;
3467
	mdsc->oldest_tid = 0;
3468
	mdsc->request_tree = RB_ROOT;
Sage Weil's avatar
Sage Weil committed
3469 3470 3471 3472 3473 3474
	INIT_DELAYED_WORK(&mdsc->delayed_work, delayed_work);
	mdsc->last_renew_caps = jiffies;
	INIT_LIST_HEAD(&mdsc->cap_delay_list);
	spin_lock_init(&mdsc->cap_delay_lock);
	INIT_LIST_HEAD(&mdsc->snap_flush_list);
	spin_lock_init(&mdsc->snap_flush_lock);
3475
	mdsc->last_cap_flush_tid = 1;
3476
	mdsc->cap_flush_tree = RB_ROOT;
Sage Weil's avatar
Sage Weil committed
3477
	INIT_LIST_HEAD(&mdsc->cap_dirty);
3478
	INIT_LIST_HEAD(&mdsc->cap_dirty_migrating);
Sage Weil's avatar
Sage Weil committed
3479 3480 3481 3482 3483
	mdsc->num_cap_flushing = 0;
	spin_lock_init(&mdsc->cap_dirty_lock);
	init_waitqueue_head(&mdsc->cap_flushing_wq);
	spin_lock_init(&mdsc->dentry_lru_lock);
	INIT_LIST_HEAD(&mdsc->dentry_lru);
3484

3485
	ceph_caps_init(mdsc);
3486
	ceph_adjust_min_caps(mdsc, fsc->min_caps);
3487

3488 3489 3490
	init_rwsem(&mdsc->pool_perm_rwsem);
	mdsc->pool_perm_tree = RB_ROOT;

3491
	return 0;
Sage Weil's avatar
Sage Weil committed
3492 3493 3494 3495 3496 3497 3498 3499
}

/*
 * Wait for safe replies on open mds requests.  If we time out, drop
 * all requests from the tree to avoid dangling dentry refs.
 */
static void wait_requests(struct ceph_mds_client *mdsc)
{
3500
	struct ceph_options *opts = mdsc->fsc->client->options;
Sage Weil's avatar
Sage Weil committed
3501 3502 3503
	struct ceph_mds_request *req;

	mutex_lock(&mdsc->mutex);
3504
	if (__get_oldest_req(mdsc)) {
Sage Weil's avatar
Sage Weil committed
3505
		mutex_unlock(&mdsc->mutex);
3506

Sage Weil's avatar
Sage Weil committed
3507 3508
		dout("wait_requests waiting for requests\n");
		wait_for_completion_timeout(&mdsc->safe_umount_waiters,
3509
				    ceph_timeout_jiffies(opts->mount_timeout));
Sage Weil's avatar
Sage Weil committed
3510 3511

		/* tear down remaining requests */
3512 3513
		mutex_lock(&mdsc->mutex);
		while ((req = __get_oldest_req(mdsc))) {
Sage Weil's avatar
Sage Weil committed
3514 3515
			dout("wait_requests timed out on tid %llu\n",
			     req->r_tid);
3516
			__unregister_request(mdsc, req);
Sage Weil's avatar
Sage Weil committed
3517 3518 3519 3520 3521 3522 3523 3524 3525 3526 3527 3528 3529 3530 3531 3532
		}
	}
	mutex_unlock(&mdsc->mutex);
	dout("wait_requests done\n");
}

/*
 * called before mount is ro, and before dentries are torn down.
 * (hmm, does this still race with new lookups?)
 */
void ceph_mdsc_pre_umount(struct ceph_mds_client *mdsc)
{
	dout("pre_umount\n");
	mdsc->stopping = 1;

	drop_leases(mdsc);
3533
	ceph_flush_dirty_caps(mdsc);
Sage Weil's avatar
Sage Weil committed
3534
	wait_requests(mdsc);
3535 3536 3537 3538 3539 3540

	/*
	 * wait for reply handlers to drop their request refs and
	 * their inode/dcache refs
	 */
	ceph_msgr_flush();
Sage Weil's avatar
Sage Weil committed
3541 3542 3543 3544 3545 3546 3547
}

/*
 * wait for all write mds requests to flush.
 */
static void wait_unsafe_requests(struct ceph_mds_client *mdsc, u64 want_tid)
{
3548
	struct ceph_mds_request *req = NULL, *nextreq;
3549
	struct rb_node *n;
Sage Weil's avatar
Sage Weil committed
3550 3551 3552

	mutex_lock(&mdsc->mutex);
	dout("wait_unsafe_requests want %lld\n", want_tid);
3553
restart:
3554 3555
	req = __get_oldest_req(mdsc);
	while (req && req->r_tid <= want_tid) {
3556 3557 3558 3559 3560 3561
		/* find next request */
		n = rb_next(&req->r_node);
		if (n)
			nextreq = rb_entry(n, struct ceph_mds_request, r_node);
		else
			nextreq = NULL;
3562 3563
		if (req->r_op != CEPH_MDS_OP_SETFILELOCK &&
		    (req->r_op & CEPH_MDS_OP_WRITE)) {
3564 3565
			/* write op */
			ceph_mdsc_get_request(req);
3566 3567
			if (nextreq)
				ceph_mdsc_get_request(nextreq);
3568 3569 3570 3571 3572 3573
			mutex_unlock(&mdsc->mutex);
			dout("wait_unsafe_requests  wait on %llu (want %llu)\n",
			     req->r_tid, want_tid);
			wait_for_completion(&req->r_safe_completion);
			mutex_lock(&mdsc->mutex);
			ceph_mdsc_put_request(req);
3574 3575 3576 3577 3578 3579 3580 3581
			if (!nextreq)
				break;  /* next dne before, so we're done! */
			if (RB_EMPTY_NODE(&nextreq->r_node)) {
				/* next request was removed from tree */
				ceph_mdsc_put_request(nextreq);
				goto restart;
			}
			ceph_mdsc_put_request(nextreq);  /* won't go away */
3582
		}
3583
		req = nextreq;
Sage Weil's avatar
Sage Weil committed
3584 3585 3586 3587 3588 3589 3590
	}
	mutex_unlock(&mdsc->mutex);
	dout("wait_unsafe_requests done\n");
}

void ceph_mdsc_sync(struct ceph_mds_client *mdsc)
{
3591
	u64 want_tid, want_flush, want_snap;
Sage Weil's avatar
Sage Weil committed
3592

3593
	if (ACCESS_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN)
3594 3595
		return;

Sage Weil's avatar
Sage Weil committed
3596 3597 3598 3599 3600
	dout("sync\n");
	mutex_lock(&mdsc->mutex);
	want_tid = mdsc->last_tid;
	mutex_unlock(&mdsc->mutex);

3601
	ceph_flush_dirty_caps(mdsc);
3602
	spin_lock(&mdsc->cap_dirty_lock);
3603
	want_flush = mdsc->last_cap_flush_tid;
3604 3605
	spin_unlock(&mdsc->cap_dirty_lock);

3606 3607 3608 3609 3610 3611
	down_read(&mdsc->snap_rwsem);
	want_snap = mdsc->last_snap_seq;
	up_read(&mdsc->snap_rwsem);

	dout("sync want tid %lld flush_seq %lld snap_seq %lld\n",
	     want_tid, want_flush, want_snap);
Sage Weil's avatar
Sage Weil committed
3612 3613

	wait_unsafe_requests(mdsc, want_tid);
3614
	wait_caps_flush(mdsc, want_flush, want_snap);
Sage Weil's avatar
Sage Weil committed
3615 3616
}

3617 3618 3619
/*
 * true if all sessions are closed, or we force unmount
 */
3620
static bool done_closing_sessions(struct ceph_mds_client *mdsc)
3621
{
3622
	if (ACCESS_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN)
3623
		return true;
3624
	return atomic_read(&mdsc->num_sessions) == 0;
3625
}
Sage Weil's avatar
Sage Weil committed
3626 3627 3628 3629 3630 3631

/*
 * called after sb is ro.
 */
void ceph_mdsc_close_sessions(struct ceph_mds_client *mdsc)
{
3632
	struct ceph_options *opts = mdsc->fsc->client->options;
Sage Weil's avatar
Sage Weil committed
3633 3634 3635 3636 3637 3638
	struct ceph_mds_session *session;
	int i;

	dout("close_sessions\n");

	/* close sessions */
3639 3640 3641 3642 3643
	mutex_lock(&mdsc->mutex);
	for (i = 0; i < mdsc->max_sessions; i++) {
		session = __ceph_lookup_mds_session(mdsc, i);
		if (!session)
			continue;
Sage Weil's avatar
Sage Weil committed
3644
		mutex_unlock(&mdsc->mutex);
3645 3646 3647 3648
		mutex_lock(&session->s_mutex);
		__close_session(mdsc, session);
		mutex_unlock(&session->s_mutex);
		ceph_put_mds_session(session);
Sage Weil's avatar
Sage Weil committed
3649 3650
		mutex_lock(&mdsc->mutex);
	}
3651 3652 3653 3654
	mutex_unlock(&mdsc->mutex);

	dout("waiting for sessions to close\n");
	wait_event_timeout(mdsc->session_close_wq, done_closing_sessions(mdsc),
3655
			   ceph_timeout_jiffies(opts->mount_timeout));
Sage Weil's avatar
Sage Weil committed
3656 3657

	/* tear down remaining sessions */
3658
	mutex_lock(&mdsc->mutex);
Sage Weil's avatar
Sage Weil committed
3659 3660 3661
	for (i = 0; i < mdsc->max_sessions; i++) {
		if (mdsc->sessions[i]) {
			session = get_session(mdsc->sessions[i]);
3662
			__unregister_session(mdsc, session);
Sage Weil's avatar
Sage Weil committed
3663 3664 3665 3666 3667 3668 3669 3670 3671 3672 3673 3674 3675 3676 3677 3678 3679 3680
			mutex_unlock(&mdsc->mutex);
			mutex_lock(&session->s_mutex);
			remove_session_caps(session);
			mutex_unlock(&session->s_mutex);
			ceph_put_mds_session(session);
			mutex_lock(&mdsc->mutex);
		}
	}
	WARN_ON(!list_empty(&mdsc->cap_delay_list));
	mutex_unlock(&mdsc->mutex);

	ceph_cleanup_empty_realms(mdsc);

	cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */

	dout("stopped\n");
}

3681 3682 3683 3684 3685 3686 3687 3688 3689 3690 3691 3692 3693 3694 3695 3696 3697 3698 3699 3700 3701 3702 3703 3704 3705 3706 3707 3708
void ceph_mdsc_force_umount(struct ceph_mds_client *mdsc)
{
	struct ceph_mds_session *session;
	int mds;

	dout("force umount\n");

	mutex_lock(&mdsc->mutex);
	for (mds = 0; mds < mdsc->max_sessions; mds++) {
		session = __ceph_lookup_mds_session(mdsc, mds);
		if (!session)
			continue;
		mutex_unlock(&mdsc->mutex);
		mutex_lock(&session->s_mutex);
		__close_session(mdsc, session);
		if (session->s_state == CEPH_MDS_SESSION_CLOSING) {
			cleanup_session_requests(mdsc, session);
			remove_session_caps(session);
		}
		mutex_unlock(&session->s_mutex);
		ceph_put_mds_session(session);
		mutex_lock(&mdsc->mutex);
		kick_requests(mdsc, mds);
	}
	__wake_requests(mdsc, &mdsc->waiting_for_map);
	mutex_unlock(&mdsc->mutex);
}

3709
static void ceph_mdsc_stop(struct ceph_mds_client *mdsc)
Sage Weil's avatar
Sage Weil committed
3710 3711 3712 3713 3714 3715
{
	dout("stop\n");
	cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */
	if (mdsc->mdsmap)
		ceph_mdsmap_destroy(mdsc->mdsmap);
	kfree(mdsc->sessions);
3716
	ceph_caps_finalize(mdsc);
3717
	ceph_pool_perm_destroy(mdsc);
Sage Weil's avatar
Sage Weil committed
3718 3719
}

3720 3721 3722 3723
void ceph_mdsc_destroy(struct ceph_fs_client *fsc)
{
	struct ceph_mds_client *mdsc = fsc->mdsc;

3724
	dout("mdsc_destroy %p\n", mdsc);
3725
	ceph_mdsc_stop(mdsc);
3726 3727 3728 3729

	/* flush out any connection work with references to us */
	ceph_msgr_flush();

3730 3731
	fsc->mdsc = NULL;
	kfree(mdsc);
3732
	dout("mdsc_destroy %p done\n", mdsc);
3733 3734
}

Sage Weil's avatar
Sage Weil committed
3735 3736 3737 3738 3739 3740 3741 3742 3743 3744 3745 3746 3747 3748 3749 3750

/*
 * handle mds map update.
 */
void ceph_mdsc_handle_map(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
{
	u32 epoch;
	u32 maplen;
	void *p = msg->front.iov_base;
	void *end = p + msg->front.iov_len;
	struct ceph_mdsmap *newmap, *oldmap;
	struct ceph_fsid fsid;
	int err = -EINVAL;

	ceph_decode_need(&p, end, sizeof(fsid)+2*sizeof(u32), bad);
	ceph_decode_copy(&p, &fsid, sizeof(fsid));
3751
	if (ceph_check_fsid(mdsc->fsc->client, &fsid) < 0)
3752
		return;
3753 3754
	epoch = ceph_decode_32(&p);
	maplen = ceph_decode_32(&p);
Sage Weil's avatar
Sage Weil committed
3755 3756 3757
	dout("handle_map epoch %u len %d\n", epoch, (int)maplen);

	/* do we need it? */
3758
	ceph_monc_got_mdsmap(&mdsc->fsc->client->monc, epoch);
Sage Weil's avatar
Sage Weil committed
3759 3760 3761 3762 3763 3764 3765 3766 3767 3768 3769 3770 3771 3772 3773 3774 3775 3776 3777 3778 3779 3780 3781
	mutex_lock(&mdsc->mutex);
	if (mdsc->mdsmap && epoch <= mdsc->mdsmap->m_epoch) {
		dout("handle_map epoch %u <= our %u\n",
		     epoch, mdsc->mdsmap->m_epoch);
		mutex_unlock(&mdsc->mutex);
		return;
	}

	newmap = ceph_mdsmap_decode(&p, end);
	if (IS_ERR(newmap)) {
		err = PTR_ERR(newmap);
		goto bad_unlock;
	}

	/* swap into place */
	if (mdsc->mdsmap) {
		oldmap = mdsc->mdsmap;
		mdsc->mdsmap = newmap;
		check_new_map(mdsc, newmap, oldmap);
		ceph_mdsmap_destroy(oldmap);
	} else {
		mdsc->mdsmap = newmap;  /* first mds map */
	}
3782
	mdsc->fsc->sb->s_maxbytes = mdsc->mdsmap->m_max_file_size;
Sage Weil's avatar
Sage Weil committed
3783 3784 3785 3786 3787 3788 3789 3790 3791 3792 3793 3794 3795 3796 3797 3798 3799 3800 3801

	__wake_requests(mdsc, &mdsc->waiting_for_map);

	mutex_unlock(&mdsc->mutex);
	schedule_delayed(mdsc);
	return;

bad_unlock:
	mutex_unlock(&mdsc->mutex);
bad:
	pr_err("error decoding mdsmap %d\n", err);
	return;
}

static struct ceph_connection *con_get(struct ceph_connection *con)
{
	struct ceph_mds_session *s = con->private;

	if (get_session(s)) {
3802
		dout("mdsc con_get %p ok (%d)\n", s, atomic_read(&s->s_ref));
Sage Weil's avatar
Sage Weil committed
3803 3804 3805 3806 3807 3808 3809 3810 3811 3812
		return con;
	}
	dout("mdsc con_get %p FAIL\n", s);
	return NULL;
}

static void con_put(struct ceph_connection *con)
{
	struct ceph_mds_session *s = con->private;

3813
	dout("mdsc con_put %p (%d)\n", s, atomic_read(&s->s_ref) - 1);
Sage Weil's avatar
Sage Weil committed
3814 3815 3816 3817 3818 3819 3820 3821 3822 3823
	ceph_put_mds_session(s);
}

/*
 * if the client is unresponsive for long enough, the mds will kill
 * the session entirely.
 */
static void peer_reset(struct ceph_connection *con)
{
	struct ceph_mds_session *s = con->private;
3824
	struct ceph_mds_client *mdsc = s->s_mdsc;
Sage Weil's avatar
Sage Weil committed
3825

3826
	pr_warn("mds%d closed our session\n", s->s_mds);
3827
	send_mds_reconnect(mdsc, s);
Sage Weil's avatar
Sage Weil committed
3828 3829 3830 3831 3832 3833 3834 3835
}

static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
{
	struct ceph_mds_session *s = con->private;
	struct ceph_mds_client *mdsc = s->s_mdsc;
	int type = le16_to_cpu(msg->hdr.type);

3836 3837 3838 3839 3840 3841 3842
	mutex_lock(&mdsc->mutex);
	if (__verify_registered_session(mdsc, s) < 0) {
		mutex_unlock(&mdsc->mutex);
		goto out;
	}
	mutex_unlock(&mdsc->mutex);

Sage Weil's avatar
Sage Weil committed
3843 3844 3845 3846 3847 3848 3849 3850 3851 3852 3853
	switch (type) {
	case CEPH_MSG_MDS_MAP:
		ceph_mdsc_handle_map(mdsc, msg);
		break;
	case CEPH_MSG_CLIENT_SESSION:
		handle_session(s, msg);
		break;
	case CEPH_MSG_CLIENT_REPLY:
		handle_reply(s, msg);
		break;
	case CEPH_MSG_CLIENT_REQUEST_FORWARD:
3854
		handle_forward(mdsc, s, msg);
Sage Weil's avatar
Sage Weil committed
3855 3856 3857 3858 3859
		break;
	case CEPH_MSG_CLIENT_CAPS:
		ceph_handle_caps(s, msg);
		break;
	case CEPH_MSG_CLIENT_SNAP:
3860
		ceph_handle_snap(mdsc, s, msg);
Sage Weil's avatar
Sage Weil committed
3861 3862
		break;
	case CEPH_MSG_CLIENT_LEASE:
3863
		handle_lease(mdsc, s, msg);
Sage Weil's avatar
Sage Weil committed
3864 3865 3866 3867 3868 3869
		break;

	default:
		pr_err("received unknown message type %d %s\n", type,
		       ceph_msg_type_name(type));
	}
3870
out:
Sage Weil's avatar
Sage Weil committed
3871 3872 3873
	ceph_msg_put(msg);
}

3874 3875 3876
/*
 * authentication
 */
3877 3878 3879 3880 3881 3882

/*
 * Note: returned pointer is the address of a structure that's
 * managed separately.  Caller must *not* attempt to free it.
 */
static struct ceph_auth_handshake *get_authorizer(struct ceph_connection *con,
3883
					int *proto, int force_new)
3884 3885 3886
{
	struct ceph_mds_session *s = con->private;
	struct ceph_mds_client *mdsc = s->s_mdsc;
3887
	struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
3888
	struct ceph_auth_handshake *auth = &s->s_auth;
3889

3890
	if (force_new && auth->authorizer) {
3891
		ceph_auth_destroy_authorizer(ac, auth->authorizer);
3892
		auth->authorizer = NULL;
3893
	}
3894 3895 3896
	if (!auth->authorizer) {
		int ret = ceph_auth_create_authorizer(ac, CEPH_ENTITY_TYPE_MDS,
						      auth);
3897 3898
		if (ret)
			return ERR_PTR(ret);
3899 3900 3901
	} else {
		int ret = ceph_auth_update_authorizer(ac, CEPH_ENTITY_TYPE_MDS,
						      auth);
3902
		if (ret)
3903
			return ERR_PTR(ret);
3904 3905
	}
	*proto = ac->protocol;
3906

3907
	return auth;
3908 3909 3910 3911 3912 3913 3914
}


static int verify_authorizer_reply(struct ceph_connection *con, int len)
{
	struct ceph_mds_session *s = con->private;
	struct ceph_mds_client *mdsc = s->s_mdsc;
3915
	struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
3916

3917
	return ceph_auth_verify_authorizer_reply(ac, s->s_auth.authorizer, len);
3918 3919
}

3920 3921 3922 3923
static int invalidate_authorizer(struct ceph_connection *con)
{
	struct ceph_mds_session *s = con->private;
	struct ceph_mds_client *mdsc = s->s_mdsc;
3924
	struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
3925

3926
	ceph_auth_invalidate_authorizer(ac, CEPH_ENTITY_TYPE_MDS);
3927

3928
	return ceph_monc_validate_auth(&mdsc->fsc->client->monc);
3929 3930
}

3931 3932 3933 3934 3935 3936 3937 3938 3939 3940 3941 3942 3943 3944 3945 3946 3947 3948 3949 3950 3951
static struct ceph_msg *mds_alloc_msg(struct ceph_connection *con,
				struct ceph_msg_header *hdr, int *skip)
{
	struct ceph_msg *msg;
	int type = (int) le16_to_cpu(hdr->type);
	int front_len = (int) le32_to_cpu(hdr->front_len);

	if (con->in_msg)
		return con->in_msg;

	*skip = 0;
	msg = ceph_msg_new(type, front_len, GFP_NOFS, false);
	if (!msg) {
		pr_err("unable to allocate msg type %d len %d\n",
		       type, front_len);
		return NULL;
	}

	return msg;
}

3952
static int mds_sign_message(struct ceph_msg *msg)
3953
{
3954
       struct ceph_mds_session *s = msg->con->private;
3955
       struct ceph_auth_handshake *auth = &s->s_auth;
3956

3957 3958 3959
       return ceph_auth_sign_message(auth, msg);
}

3960
static int mds_check_message_signature(struct ceph_msg *msg)
3961
{
3962
       struct ceph_mds_session *s = msg->con->private;
3963
       struct ceph_auth_handshake *auth = &s->s_auth;
3964

3965 3966 3967
       return ceph_auth_check_message_signature(auth, msg);
}

3968
static const struct ceph_connection_operations mds_con_ops = {
Sage Weil's avatar
Sage Weil committed
3969 3970 3971
	.get = con_get,
	.put = con_put,
	.dispatch = dispatch,
3972 3973
	.get_authorizer = get_authorizer,
	.verify_authorizer_reply = verify_authorizer_reply,
3974
	.invalidate_authorizer = invalidate_authorizer,
Sage Weil's avatar
Sage Weil committed
3975
	.peer_reset = peer_reset,
3976
	.alloc_msg = mds_alloc_msg,
3977 3978
	.sign_message = mds_sign_message,
	.check_message_signature = mds_check_message_signature,
Sage Weil's avatar
Sage Weil committed
3979 3980 3981
};

/* eof */