caps.c

   1// SPDX-License-Identifier: GPL-2.0
   2#include <linux/ceph/ceph_debug.h>
   3
   4#include <linux/fs.h>
   5#include <linux/kernel.h>
   6#include <linux/sched/signal.h>
   7#include <linux/slab.h>
   8#include <linux/vmalloc.h>
   9#include <linux/wait.h>
  10#include <linux/writeback.h>
  11#include <linux/iversion.h>
  12#include <linux/filelock.h>
  13#include <linux/jiffies.h>
  14
  15#include "super.h"
  16#include "mds_client.h"
  17#include "cache.h"
  18#include "crypto.h"
  19#include <linux/ceph/decode.h>
  20#include <linux/ceph/messenger.h>
  21#include <trace/events/ceph.h>
  22
  23/*
  24 * Capability management
  25 *
  26 * The Ceph metadata servers control client access to inode metadata
  27 * and file data by issuing capabilities, granting clients permission
  28 * to read and/or write both inode field and file data to OSDs
  29 * (storage nodes).  Each capability consists of a set of bits
  30 * indicating which operations are allowed.
  31 *
  32 * If the client holds a *_SHARED cap, the client has a coherent value
  33 * that can be safely read from the cached inode.
  34 *
  35 * In the case of a *_EXCL (exclusive) or FILE_WR capabilities, the
  36 * client is allowed to change inode attributes (e.g., file size,
  37 * mtime), note its dirty state in the ceph_cap, and asynchronously
  38 * flush that metadata change to the MDS.
  39 *
  40 * In the event of a conflicting operation (perhaps by another
  41 * client), the MDS will revoke the conflicting client capabilities.
  42 *
  43 * In order for a client to cache an inode, it must hold a capability
  44 * with at least one MDS server.  When inodes are released, release
  45 * notifications are batched and periodically sent en masse to the MDS
  46 * cluster to release server state.
  47 */
  48
  49static u64 __get_oldest_flush_tid(struct ceph_mds_client *mdsc);
  50static void __kick_flushing_caps(struct ceph_mds_client *mdsc,
  51				 struct ceph_mds_session *session,
  52				 struct ceph_inode_info *ci,
  53				 u64 oldest_flush_tid);
  54
  55/*
  56 * Generate readable cap strings for debugging output.
  57 */
  58#define MAX_CAP_STR 20
  59static char cap_str[MAX_CAP_STR][40];
  60static DEFINE_SPINLOCK(cap_str_lock);
  61static int last_cap_str;
  62
  63static char *gcap_string(char *s, int c)
  64{
  65	if (c & CEPH_CAP_GSHARED)
  66		*s++ = 's';
  67	if (c & CEPH_CAP_GEXCL)
  68		*s++ = 'x';
  69	if (c & CEPH_CAP_GCACHE)
  70		*s++ = 'c';
  71	if (c & CEPH_CAP_GRD)
  72		*s++ = 'r';
  73	if (c & CEPH_CAP_GWR)
  74		*s++ = 'w';
  75	if (c & CEPH_CAP_GBUFFER)
  76		*s++ = 'b';
  77	if (c & CEPH_CAP_GWREXTEND)
  78		*s++ = 'a';
  79	if (c & CEPH_CAP_GLAZYIO)
  80		*s++ = 'l';
  81	return s;
  82}
  83
  84const char *ceph_cap_string(int caps)
  85{
  86	int i;
  87	char *s;
  88	int c;
  89
  90	spin_lock(&cap_str_lock);
  91	i = last_cap_str++;
  92	if (last_cap_str == MAX_CAP_STR)
  93		last_cap_str = 0;
  94	spin_unlock(&cap_str_lock);
  95
  96	s = cap_str[i];
  97
  98	if (caps & CEPH_CAP_PIN)
  99		*s++ = 'p';
 100
 101	c = (caps >> CEPH_CAP_SAUTH) & 3;
 102	if (c) {
 103		*s++ = 'A';
 104		s = gcap_string(s, c);
 105	}
 106
 107	c = (caps >> CEPH_CAP_SLINK) & 3;
 108	if (c) {
 109		*s++ = 'L';
 110		s = gcap_string(s, c);
 111	}
 112
 113	c = (caps >> CEPH_CAP_SXATTR) & 3;
 114	if (c) {
 115		*s++ = 'X';
 116		s = gcap_string(s, c);
 117	}
 118
 119	c = caps >> CEPH_CAP_SFILE;
 120	if (c) {
 121		*s++ = 'F';
 122		s = gcap_string(s, c);
 123	}
 124
 125	if (s == cap_str[i])
 126		*s++ = '-';
 127	*s = 0;
 128	return cap_str[i];
 129}
 130
 131void ceph_caps_init(struct ceph_mds_client *mdsc)
 132{
 133	INIT_LIST_HEAD(&mdsc->caps_list);
 134	spin_lock_init(&mdsc->caps_list_lock);
 135}
 136
 137void ceph_caps_finalize(struct ceph_mds_client *mdsc)
 138{
 139	struct ceph_cap *cap;
 140
 141	spin_lock(&mdsc->caps_list_lock);
 142	while (!list_empty(&mdsc->caps_list)) {
 143		cap = list_first_entry(&mdsc->caps_list,
 144				       struct ceph_cap, caps_item);
 145		list_del(&cap->caps_item);
 146		kmem_cache_free(ceph_cap_cachep, cap);
 147	}
 148	mdsc->caps_total_count = 0;
 149	mdsc->caps_avail_count = 0;
 150	mdsc->caps_use_count = 0;
 151	mdsc->caps_reserve_count = 0;
 152	mdsc->caps_min_count = 0;
 153	spin_unlock(&mdsc->caps_list_lock);
 154}
 155
 156void ceph_adjust_caps_max_min(struct ceph_mds_client *mdsc,
 157			      struct ceph_mount_options *fsopt)
 158{
 159	spin_lock(&mdsc->caps_list_lock);
 160	mdsc->caps_min_count = fsopt->max_readdir;
 161	if (mdsc->caps_min_count < 1024)
 162		mdsc->caps_min_count = 1024;
 163	mdsc->caps_use_max = fsopt->caps_max;
 164	if (mdsc->caps_use_max > 0 &&
 165	    mdsc->caps_use_max < mdsc->caps_min_count)
 166		mdsc->caps_use_max = mdsc->caps_min_count;
 167	spin_unlock(&mdsc->caps_list_lock);
 168}
 169
 170static void __ceph_unreserve_caps(struct ceph_mds_client *mdsc, int nr_caps)
 171{
 172	struct ceph_cap *cap;
 173	int i;
 174
 175	if (nr_caps) {
 176		BUG_ON(mdsc->caps_reserve_count < nr_caps);
 177		mdsc->caps_reserve_count -= nr_caps;
 178		if (mdsc->caps_avail_count >=
 179		    mdsc->caps_reserve_count + mdsc->caps_min_count) {
 180			mdsc->caps_total_count -= nr_caps;
 181			for (i = 0; i < nr_caps; i++) {
 182				cap = list_first_entry(&mdsc->caps_list,
 183					struct ceph_cap, caps_item);
 184				list_del(&cap->caps_item);
 185				kmem_cache_free(ceph_cap_cachep, cap);
 186			}
 187		} else {
 188			mdsc->caps_avail_count += nr_caps;
 189		}
 190
 191		doutc(mdsc->fsc->client,
 192		      "caps %d = %d used + %d resv + %d avail\n",
 193		      mdsc->caps_total_count, mdsc->caps_use_count,
 194		      mdsc->caps_reserve_count, mdsc->caps_avail_count);
 195		BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
 196						 mdsc->caps_reserve_count +
 197						 mdsc->caps_avail_count);
 198	}
 199}
 200
 201/*
 202 * Called under mdsc->mutex.
 203 */
 204int ceph_reserve_caps(struct ceph_mds_client *mdsc,
 205		      struct ceph_cap_reservation *ctx, int need)
 206{
 207	struct ceph_client *cl = mdsc->fsc->client;
 208	int i, j;
 209	struct ceph_cap *cap;
 210	int have;
 211	int alloc = 0;
 212	int max_caps;
 213	int err = 0;
 214	bool trimmed = false;
 215	struct ceph_mds_session *s;
 216	LIST_HEAD(newcaps);
 217
 218	doutc(cl, "ctx=%p need=%d\n", ctx, need);
 219
 220	/* first reserve any caps that are already allocated */
 221	spin_lock(&mdsc->caps_list_lock);
 222	if (mdsc->caps_avail_count >= need)
 223		have = need;
 224	else
 225		have = mdsc->caps_avail_count;
 226	mdsc->caps_avail_count -= have;
 227	mdsc->caps_reserve_count += have;
 228	BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
 229					 mdsc->caps_reserve_count +
 230					 mdsc->caps_avail_count);
 231	spin_unlock(&mdsc->caps_list_lock);
 232
 233	for (i = have; i < need; ) {
 234		cap = kmem_cache_alloc(ceph_cap_cachep, GFP_NOFS);
 235		if (cap) {
 236			list_add(&cap->caps_item, &newcaps);
 237			alloc++;
 238			i++;
 239			continue;
 240		}
 241
 242		if (!trimmed) {
 243			for (j = 0; j < mdsc->max_sessions; j++) {
 244				s = __ceph_lookup_mds_session(mdsc, j);
 245				if (!s)
 246					continue;
 247				mutex_unlock(&mdsc->mutex);
 248
 249				mutex_lock(&s->s_mutex);
 250				max_caps = s->s_nr_caps - (need - i);
 251				ceph_trim_caps(mdsc, s, max_caps);
 252				mutex_unlock(&s->s_mutex);
 253
 254				ceph_put_mds_session(s);
 255				mutex_lock(&mdsc->mutex);
 256			}
 257			trimmed = true;
 258
 259			spin_lock(&mdsc->caps_list_lock);
 260			if (mdsc->caps_avail_count) {
 261				int more_have;
 262				if (mdsc->caps_avail_count >= need - i)
 263					more_have = need - i;
 264				else
 265					more_have = mdsc->caps_avail_count;
 266
 267				i += more_have;
 268				have += more_have;
 269				mdsc->caps_avail_count -= more_have;
 270				mdsc->caps_reserve_count += more_have;
 271
 272			}
 273			spin_unlock(&mdsc->caps_list_lock);
 274
 275			continue;
 276		}
 277
 278		pr_warn_client(cl, "ctx=%p ENOMEM need=%d got=%d\n", ctx, need,
 279			       have + alloc);
 280		err = -ENOMEM;
 281		break;
 282	}
 283
 284	if (!err) {
 285		BUG_ON(have + alloc != need);
 286		ctx->count = need;
 287		ctx->used = 0;
 288	}
 289
 290	spin_lock(&mdsc->caps_list_lock);
 291	mdsc->caps_total_count += alloc;
 292	mdsc->caps_reserve_count += alloc;
 293	list_splice(&newcaps, &mdsc->caps_list);
 294
 295	BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
 296					 mdsc->caps_reserve_count +
 297					 mdsc->caps_avail_count);
 298
 299	if (err)
 300		__ceph_unreserve_caps(mdsc, have + alloc);
 301
 302	spin_unlock(&mdsc->caps_list_lock);
 303
 304	doutc(cl, "ctx=%p %d = %d used + %d resv + %d avail\n", ctx,
 305	      mdsc->caps_total_count, mdsc->caps_use_count,
 306	      mdsc->caps_reserve_count, mdsc->caps_avail_count);
 307	return err;
 308}
 309
 310void ceph_unreserve_caps(struct ceph_mds_client *mdsc,
 311			 struct ceph_cap_reservation *ctx)
 312{
 313	struct ceph_client *cl = mdsc->fsc->client;
 314	bool reclaim = false;
 315	if (!ctx->count)
 316		return;
 317
 318	doutc(cl, "ctx=%p count=%d\n", ctx, ctx->count);
 319	spin_lock(&mdsc->caps_list_lock);
 320	__ceph_unreserve_caps(mdsc, ctx->count);
 321	ctx->count = 0;
 322
 323	if (mdsc->caps_use_max > 0 &&
 324	    mdsc->caps_use_count > mdsc->caps_use_max)
 325		reclaim = true;
 326	spin_unlock(&mdsc->caps_list_lock);
 327
 328	if (reclaim)
 329		ceph_reclaim_caps_nr(mdsc, ctx->used);
 330}
 331
 332struct ceph_cap *ceph_get_cap(struct ceph_mds_client *mdsc,
 333			      struct ceph_cap_reservation *ctx)
 334{
 335	struct ceph_client *cl = mdsc->fsc->client;
 336	struct ceph_cap *cap = NULL;
 337
 338	/* temporary, until we do something about cap import/export */
 339	if (!ctx) {
 340		cap = kmem_cache_alloc(ceph_cap_cachep, GFP_NOFS);
 341		if (cap) {
 342			spin_lock(&mdsc->caps_list_lock);
 343			mdsc->caps_use_count++;
 344			mdsc->caps_total_count++;
 345			spin_unlock(&mdsc->caps_list_lock);
 346		} else {
 347			spin_lock(&mdsc->caps_list_lock);
 348			if (mdsc->caps_avail_count) {
 349				BUG_ON(list_empty(&mdsc->caps_list));
 350
 351				mdsc->caps_avail_count--;
 352				mdsc->caps_use_count++;
 353				cap = list_first_entry(&mdsc->caps_list,
 354						struct ceph_cap, caps_item);
 355				list_del(&cap->caps_item);
 356
 357				BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
 358				       mdsc->caps_reserve_count + mdsc->caps_avail_count);
 359			}
 360			spin_unlock(&mdsc->caps_list_lock);
 361		}
 362
 363		return cap;
 364	}
 365
 366	spin_lock(&mdsc->caps_list_lock);
 367	doutc(cl, "ctx=%p (%d) %d = %d used + %d resv + %d avail\n", ctx,
 368	      ctx->count, mdsc->caps_total_count, mdsc->caps_use_count,
 369	      mdsc->caps_reserve_count, mdsc->caps_avail_count);
 370	BUG_ON(!ctx->count);
 371	BUG_ON(ctx->count > mdsc->caps_reserve_count);
 372	BUG_ON(list_empty(&mdsc->caps_list));
 373
 374	ctx->count--;
 375	ctx->used++;
 376	mdsc->caps_reserve_count--;
 377	mdsc->caps_use_count++;
 378
 379	cap = list_first_entry(&mdsc->caps_list, struct ceph_cap, caps_item);
 380	list_del(&cap->caps_item);
 381
 382	BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
 383	       mdsc->caps_reserve_count + mdsc->caps_avail_count);
 384	spin_unlock(&mdsc->caps_list_lock);
 385	return cap;
 386}
 387
 388void ceph_put_cap(struct ceph_mds_client *mdsc, struct ceph_cap *cap)
 389{
 390	struct ceph_client *cl = mdsc->fsc->client;
 391
 392	spin_lock(&mdsc->caps_list_lock);
 393	doutc(cl, "%p %d = %d used + %d resv + %d avail\n", cap,
 394	      mdsc->caps_total_count, mdsc->caps_use_count,
 395	      mdsc->caps_reserve_count, mdsc->caps_avail_count);
 396	mdsc->caps_use_count--;
 397	/*
 398	 * Keep some preallocated caps around (ceph_min_count), to
 399	 * avoid lots of free/alloc churn.
 400	 */
 401	if (mdsc->caps_avail_count >= mdsc->caps_reserve_count +
 402				      mdsc->caps_min_count) {
 403		mdsc->caps_total_count--;
 404		kmem_cache_free(ceph_cap_cachep, cap);
 405	} else {
 406		mdsc->caps_avail_count++;
 407		list_add(&cap->caps_item, &mdsc->caps_list);
 408	}
 409
 410	BUG_ON(mdsc->caps_total_count != mdsc->caps_use_count +
 411	       mdsc->caps_reserve_count + mdsc->caps_avail_count);
 412	spin_unlock(&mdsc->caps_list_lock);
 413}
 414
 415void ceph_reservation_status(struct ceph_fs_client *fsc,
 416			     int *total, int *avail, int *used, int *reserved,
 417			     int *min)
 418{
 419	struct ceph_mds_client *mdsc = fsc->mdsc;
 420
 421	spin_lock(&mdsc->caps_list_lock);
 422
 423	if (total)
 424		*total = mdsc->caps_total_count;
 425	if (avail)
 426		*avail = mdsc->caps_avail_count;
 427	if (used)
 428		*used = mdsc->caps_use_count;
 429	if (reserved)
 430		*reserved = mdsc->caps_reserve_count;
 431	if (min)
 432		*min = mdsc->caps_min_count;
 433
 434	spin_unlock(&mdsc->caps_list_lock);
 435}
 436
 437/*
 438 * Find ceph_cap for given mds, if any.
 439 *
 440 * Called with i_ceph_lock held.
 441 */
 442struct ceph_cap *__get_cap_for_mds(struct ceph_inode_info *ci, int mds)
 443{
 444	struct ceph_cap *cap;
 445	struct rb_node *n = ci->i_caps.rb_node;
 446
 447	while (n) {
 448		cap = rb_entry(n, struct ceph_cap, ci_node);
 449		if (mds < cap->mds)
 450			n = n->rb_left;
 451		else if (mds > cap->mds)
 452			n = n->rb_right;
 453		else
 454			return cap;
 455	}
 456	return NULL;
 457}
 458
 459struct ceph_cap *ceph_get_cap_for_mds(struct ceph_inode_info *ci, int mds)
 460{
 461	struct ceph_cap *cap;
 462
 463	spin_lock(&ci->i_ceph_lock);
 464	cap = __get_cap_for_mds(ci, mds);
 465	spin_unlock(&ci->i_ceph_lock);
 466	return cap;
 467}
 468
 469/*
 470 * Called under i_ceph_lock.
 471 */
 472static void __insert_cap_node(struct ceph_inode_info *ci,
 473			      struct ceph_cap *new)
 474{
 475	struct rb_node **p = &ci->i_caps.rb_node;
 476	struct rb_node *parent = NULL;
 477	struct ceph_cap *cap = NULL;
 478
 479	while (*p) {
 480		parent = *p;
 481		cap = rb_entry(parent, struct ceph_cap, ci_node);
 482		if (new->mds < cap->mds)
 483			p = &(*p)->rb_left;
 484		else if (new->mds > cap->mds)
 485			p = &(*p)->rb_right;
 486		else
 487			BUG();
 488	}
 489
 490	rb_link_node(&new->ci_node, parent, p);
 491	rb_insert_color(&new->ci_node, &ci->i_caps);
 492}
 493
 494/*
 495 * (re)set cap hold timeouts, which control the delayed release
 496 * of unused caps back to the MDS.  Should be called on cap use.
 497 */
 498static void __cap_set_timeouts(struct ceph_mds_client *mdsc,
 499			       struct ceph_inode_info *ci)
 500{
 501	struct inode *inode = &ci->netfs.inode;
 502	struct ceph_mount_options *opt = mdsc->fsc->mount_options;
 503
 504	ci->i_hold_caps_max = round_jiffies(jiffies +
 505					    opt->caps_wanted_delay_max * HZ);
 506	doutc(mdsc->fsc->client, "%p %llx.%llx %lu\n", inode,
 507	      ceph_vinop(inode), ci->i_hold_caps_max - jiffies);
 508}
 509
 510/*
 511 * (Re)queue cap at the end of the delayed cap release list.
 512 *
 513 * If I_FLUSH is set, leave the inode at the front of the list.
 514 *
 515 * Caller holds i_ceph_lock
 516 *    -> we take mdsc->cap_delay_lock
 517 */
 518static void __cap_delay_requeue(struct ceph_mds_client *mdsc,
 519				struct ceph_inode_info *ci)
 520{
 521	struct inode *inode = &ci->netfs.inode;
 522
 523	doutc(mdsc->fsc->client, "%p %llx.%llx flags 0x%lx at %lu\n",
 524	      inode, ceph_vinop(inode), ci->i_ceph_flags,
 525	      ci->i_hold_caps_max);
 526	if (!mdsc->stopping) {
 527		spin_lock(&mdsc->cap_delay_lock);
 528		if (!list_empty(&ci->i_cap_delay_list)) {
 529			if (ci->i_ceph_flags & CEPH_I_FLUSH)
 530				goto no_change;
 531			list_del_init(&ci->i_cap_delay_list);
 532		}
 533		__cap_set_timeouts(mdsc, ci);
 534		list_add_tail(&ci->i_cap_delay_list, &mdsc->cap_delay_list);
 535no_change:
 536		spin_unlock(&mdsc->cap_delay_lock);
 537	}
 538}
 539
 540/*
 541 * Queue an inode for immediate writeback.  Mark inode with I_FLUSH,
 542 * indicating we should send a cap message to flush dirty metadata
 543 * asap, and move to the front of the delayed cap list.
 544 */
 545static void __cap_delay_requeue_front(struct ceph_mds_client *mdsc,
 546				      struct ceph_inode_info *ci)
 547{
 548	struct inode *inode = &ci->netfs.inode;
 549
 550	doutc(mdsc->fsc->client, "%p %llx.%llx\n", inode, ceph_vinop(inode));
 551	spin_lock(&mdsc->cap_delay_lock);
 552	ci->i_ceph_flags |= CEPH_I_FLUSH;
 553	if (!list_empty(&ci->i_cap_delay_list))
 554		list_del_init(&ci->i_cap_delay_list);
 555	list_add(&ci->i_cap_delay_list, &mdsc->cap_delay_list);
 556	spin_unlock(&mdsc->cap_delay_lock);
 557}
 558
 559/*
 560 * Cancel delayed work on cap.
 561 *
 562 * Caller must hold i_ceph_lock.
 563 */
 564static void __cap_delay_cancel(struct ceph_mds_client *mdsc,
 565			       struct ceph_inode_info *ci)
 566{
 567	struct inode *inode = &ci->netfs.inode;
 568
 569	doutc(mdsc->fsc->client, "%p %llx.%llx\n", inode, ceph_vinop(inode));
 570	if (list_empty(&ci->i_cap_delay_list))
 571		return;
 572	spin_lock(&mdsc->cap_delay_lock);
 573	list_del_init(&ci->i_cap_delay_list);
 574	spin_unlock(&mdsc->cap_delay_lock);
 575}
 576
 577/* Common issue checks for add_cap, handle_cap_grant. */
 578static void __check_cap_issue(struct ceph_inode_info *ci, struct ceph_cap *cap,
 579			      unsigned issued)
 580{
 581	struct inode *inode = &ci->netfs.inode;
 582	struct ceph_client *cl = ceph_inode_to_client(inode);
 583
 584	unsigned had = __ceph_caps_issued(ci, NULL);
 585
 586	lockdep_assert_held(&ci->i_ceph_lock);
 587
 588	/*
 589	 * Each time we receive FILE_CACHE anew, we increment
 590	 * i_rdcache_gen.
 591	 */
 592	if (S_ISREG(ci->netfs.inode.i_mode) &&
 593	    (issued & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) &&
 594	    (had & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) == 0) {
 595		ci->i_rdcache_gen++;
 596	}
 597
 598	/*
 599	 * If FILE_SHARED is newly issued, mark dir not complete. We don't
 600	 * know what happened to this directory while we didn't have the cap.
 601	 * If FILE_SHARED is being revoked, also mark dir not complete. It
 602	 * stops on-going cached readdir.
 603	 */
 604	if ((issued & CEPH_CAP_FILE_SHARED) != (had & CEPH_CAP_FILE_SHARED)) {
 605		if (issued & CEPH_CAP_FILE_SHARED)
 606			atomic_inc(&ci->i_shared_gen);
 607		if (S_ISDIR(ci->netfs.inode.i_mode)) {
 608			doutc(cl, " marking %p NOT complete\n", inode);
 609			__ceph_dir_clear_complete(ci);
 610		}
 611	}
 612
 613	/* Wipe saved layout if we're losing DIR_CREATE caps */
 614	if (S_ISDIR(ci->netfs.inode.i_mode) && (had & CEPH_CAP_DIR_CREATE) &&
 615		!(issued & CEPH_CAP_DIR_CREATE)) {
 616	     ceph_put_string(rcu_dereference_raw(ci->i_cached_layout.pool_ns));
 617	     memset(&ci->i_cached_layout, 0, sizeof(ci->i_cached_layout));
 618	}
 619}
 620
 621/**
 622 * change_auth_cap_ses - move inode to appropriate lists when auth caps change
 623 * @ci: inode to be moved
 624 * @session: new auth caps session
 625 */
 626void change_auth_cap_ses(struct ceph_inode_info *ci,
 627			 struct ceph_mds_session *session)
 628{
 629	lockdep_assert_held(&ci->i_ceph_lock);
 630
 631	if (list_empty(&ci->i_dirty_item) && list_empty(&ci->i_flushing_item))
 632		return;
 633
 634	spin_lock(&session->s_mdsc->cap_dirty_lock);
 635	if (!list_empty(&ci->i_dirty_item))
 636		list_move(&ci->i_dirty_item, &session->s_cap_dirty);
 637	if (!list_empty(&ci->i_flushing_item))
 638		list_move_tail(&ci->i_flushing_item, &session->s_cap_flushing);
 639	spin_unlock(&session->s_mdsc->cap_dirty_lock);
 640}
 641
 642/*
 643 * Add a capability under the given MDS session.
 644 *
 645 * Caller should hold session snap_rwsem (read) and ci->i_ceph_lock
 646 *
 647 * @fmode is the open file mode, if we are opening a file, otherwise
 648 * it is < 0.  (This is so we can atomically add the cap and add an
 649 * open file reference to it.)
 650 */
 651void ceph_add_cap(struct inode *inode,
 652		  struct ceph_mds_session *session, u64 cap_id,
 653		  unsigned issued, unsigned wanted,
 654		  unsigned seq, unsigned mseq, u64 realmino, int flags,
 655		  struct ceph_cap **new_cap)
 656{
 657	struct ceph_mds_client *mdsc = ceph_inode_to_fs_client(inode)->mdsc;
 658	struct ceph_client *cl = ceph_inode_to_client(inode);
 659	struct ceph_inode_info *ci = ceph_inode(inode);
 660	struct ceph_cap *cap;
 661	int mds = session->s_mds;
 662	int actual_wanted;
 663	u32 gen;
 664
 665	lockdep_assert_held(&ci->i_ceph_lock);
 666
 667	doutc(cl, "%p %llx.%llx mds%d cap %llx %s seq %d\n", inode,
 668	      ceph_vinop(inode), session->s_mds, cap_id,
 669	      ceph_cap_string(issued), seq);
 670
 671	gen = atomic_read(&session->s_cap_gen);
 672
 673	cap = __get_cap_for_mds(ci, mds);
 674	if (!cap) {
 675		cap = *new_cap;
 676		*new_cap = NULL;
 677
 678		cap->issued = 0;
 679		cap->implemented = 0;
 680		cap->mds = mds;
 681		cap->mds_wanted = 0;
 682		cap->mseq = 0;
 683
 684		cap->ci = ci;
 685		__insert_cap_node(ci, cap);
 686
 687		/* add to session cap list */
 688		cap->session = session;
 689		spin_lock(&session->s_cap_lock);
 690		list_add_tail(&cap->session_caps, &session->s_caps);
 691		session->s_nr_caps++;
 692		atomic64_inc(&mdsc->metric.total_caps);
 693		spin_unlock(&session->s_cap_lock);
 694	} else {
 695		spin_lock(&session->s_cap_lock);
 696		list_move_tail(&cap->session_caps, &session->s_caps);
 697		spin_unlock(&session->s_cap_lock);
 698
 699		if (cap->cap_gen < gen)
 700			cap->issued = cap->implemented = CEPH_CAP_PIN;
 701
 702		/*
 703		 * auth mds of the inode changed. we received the cap export
 704		 * message, but still haven't received the cap import message.
 705		 * handle_cap_export() updated the new auth MDS' cap.
 706		 *
 707		 * "ceph_seq_cmp(seq, cap->seq) <= 0" means we are processing
 708		 * a message that was send before the cap import message. So
 709		 * don't remove caps.
 710		 */
 711		if (ceph_seq_cmp(seq, cap->seq) <= 0) {
 712			WARN_ON(cap != ci->i_auth_cap);
 713			WARN_ON(cap->cap_id != cap_id);
 714			seq = cap->seq;
 715			mseq = cap->mseq;
 716			issued |= cap->issued;
 717			flags |= CEPH_CAP_FLAG_AUTH;
 718		}
 719	}
 720
 721	if (!ci->i_snap_realm ||
 722	    ((flags & CEPH_CAP_FLAG_AUTH) &&
 723	     realmino != (u64)-1 && ci->i_snap_realm->ino != realmino)) {
 724		/*
 725		 * add this inode to the appropriate snap realm
 726		 */
 727		struct ceph_snap_realm *realm = ceph_lookup_snap_realm(mdsc,
 728							       realmino);
 729		if (realm)
 730			ceph_change_snap_realm(inode, realm);
 731		else
 732			WARN(1, "%s: couldn't find snap realm 0x%llx (ino 0x%llx oldrealm 0x%llx)\n",
 733			     __func__, realmino, ci->i_vino.ino,
 734			     ci->i_snap_realm ? ci->i_snap_realm->ino : 0);
 735	}
 736
 737	__check_cap_issue(ci, cap, issued);
 738
 739	/*
 740	 * If we are issued caps we don't want, or the mds' wanted
 741	 * value appears to be off, queue a check so we'll release
 742	 * later and/or update the mds wanted value.
 743	 */
 744	actual_wanted = __ceph_caps_wanted(ci);
 745	if ((wanted & ~actual_wanted) ||
 746	    (issued & ~actual_wanted & CEPH_CAP_ANY_WR)) {
 747		doutc(cl, "issued %s, mds wanted %s, actual %s, queueing\n",
 748		      ceph_cap_string(issued), ceph_cap_string(wanted),
 749		      ceph_cap_string(actual_wanted));
 750		__cap_delay_requeue(mdsc, ci);
 751	}
 752
 753	if (flags & CEPH_CAP_FLAG_AUTH) {
 754		if (!ci->i_auth_cap ||
 755		    ceph_seq_cmp(ci->i_auth_cap->mseq, mseq) < 0) {
 756			if (ci->i_auth_cap &&
 757			    ci->i_auth_cap->session != cap->session)
 758				change_auth_cap_ses(ci, cap->session);
 759			ci->i_auth_cap = cap;
 760			cap->mds_wanted = wanted;
 761		}
 762	} else {
 763		WARN_ON(ci->i_auth_cap == cap);
 764	}
 765
 766	doutc(cl, "inode %p %llx.%llx cap %p %s now %s seq %d mds%d\n",
 767	      inode, ceph_vinop(inode), cap, ceph_cap_string(issued),
 768	      ceph_cap_string(issued|cap->issued), seq, mds);
 769	cap->cap_id = cap_id;
 770	cap->issued = issued;
 771	cap->implemented |= issued;
 772	if (ceph_seq_cmp(mseq, cap->mseq) > 0)
 773		cap->mds_wanted = wanted;
 774	else
 775		cap->mds_wanted |= wanted;
 776	cap->seq = seq;
 777	cap->issue_seq = seq;
 778	cap->mseq = mseq;
 779	cap->cap_gen = gen;
 780	wake_up_all(&ci->i_cap_wq);
 781}
 782
 783/*
 784 * Return true if cap has not timed out and belongs to the current
 785 * generation of the MDS session (i.e. has not gone 'stale' due to
 786 * us losing touch with the mds).
 787 */
 788static int __cap_is_valid(struct ceph_cap *cap)
 789{
 790	struct inode *inode = &cap->ci->netfs.inode;
 791	struct ceph_client *cl = cap->session->s_mdsc->fsc->client;
 792	unsigned long ttl;
 793	u32 gen;
 794
 795	gen = atomic_read(&cap->session->s_cap_gen);
 796	ttl = cap->session->s_cap_ttl;
 797
 798	if (cap->cap_gen < gen || time_after_eq(jiffies, ttl)) {
 799		doutc(cl, "%p %llx.%llx cap %p issued %s but STALE (gen %u vs %u)\n",
 800		      inode, ceph_vinop(inode), cap,
 801		      ceph_cap_string(cap->issued), cap->cap_gen, gen);
 802		return 0;
 803	}
 804
 805	return 1;
 806}
 807
 808/*
 809 * Return set of valid cap bits issued to us.  Note that caps time
 810 * out, and may be invalidated in bulk if the client session times out
 811 * and session->s_cap_gen is bumped.
 812 */
 813int __ceph_caps_issued(struct ceph_inode_info *ci, int *implemented)
 814{
 815	struct inode *inode = &ci->netfs.inode;
 816	struct ceph_client *cl = ceph_inode_to_client(inode);
 817	int have = ci->i_snap_caps;
 818	struct ceph_cap *cap;
 819	struct rb_node *p;
 820
 821	if (implemented)
 822		*implemented = 0;
 823	for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
 824		cap = rb_entry(p, struct ceph_cap, ci_node);
 825		if (!__cap_is_valid(cap))
 826			continue;
 827		doutc(cl, "%p %llx.%llx cap %p issued %s\n", inode,
 828		      ceph_vinop(inode), cap, ceph_cap_string(cap->issued));
 829		have |= cap->issued;
 830		if (implemented)
 831			*implemented |= cap->implemented;
 832	}
 833	/*
 834	 * exclude caps issued by non-auth MDS, but are been revoking
 835	 * by the auth MDS. The non-auth MDS should be revoking/exporting
 836	 * these caps, but the message is delayed.
 837	 */
 838	if (ci->i_auth_cap) {
 839		cap = ci->i_auth_cap;
 840		have &= ~cap->implemented | cap->issued;
 841	}
 842	return have;
 843}
 844
 845/*
 846 * Get cap bits issued by caps other than @ocap
 847 */
 848int __ceph_caps_issued_other(struct ceph_inode_info *ci, struct ceph_cap *ocap)
 849{
 850	int have = ci->i_snap_caps;
 851	struct ceph_cap *cap;
 852	struct rb_node *p;
 853
 854	for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
 855		cap = rb_entry(p, struct ceph_cap, ci_node);
 856		if (cap == ocap)
 857			continue;
 858		if (!__cap_is_valid(cap))
 859			continue;
 860		have |= cap->issued;
 861	}
 862	return have;
 863}
 864
 865/*
 866 * Move a cap to the end of the LRU (oldest caps at list head, newest
 867 * at list tail).
 868 */
 869static void __touch_cap(struct ceph_cap *cap)
 870{
 871	struct inode *inode = &cap->ci->netfs.inode;
 872	struct ceph_mds_session *s = cap->session;
 873	struct ceph_client *cl = s->s_mdsc->fsc->client;
 874
 875	spin_lock(&s->s_cap_lock);
 876	if (!s->s_cap_iterator) {
 877		doutc(cl, "%p %llx.%llx cap %p mds%d\n", inode,
 878		      ceph_vinop(inode), cap, s->s_mds);
 879		list_move_tail(&cap->session_caps, &s->s_caps);
 880	} else {
 881		doutc(cl, "%p %llx.%llx cap %p mds%d NOP, iterating over caps\n",
 882		      inode, ceph_vinop(inode), cap, s->s_mds);
 883	}
 884	spin_unlock(&s->s_cap_lock);
 885}
 886
 887/*
 888 * Check if we hold the given mask.  If so, move the cap(s) to the
 889 * front of their respective LRUs.  (This is the preferred way for
 890 * callers to check for caps they want.)
 891 */
 892int __ceph_caps_issued_mask(struct ceph_inode_info *ci, int mask, int touch)
 893{
 894	struct inode *inode = &ci->netfs.inode;
 895	struct ceph_client *cl = ceph_inode_to_client(inode);
 896	struct ceph_cap *cap;
 897	struct rb_node *p;
 898	int have = ci->i_snap_caps;
 899
 900	if ((have & mask) == mask) {
 901		doutc(cl, "mask %p %llx.%llx snap issued %s (mask %s)\n",
 902		      inode, ceph_vinop(inode), ceph_cap_string(have),
 903		      ceph_cap_string(mask));
 904		return 1;
 905	}
 906
 907	for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
 908		cap = rb_entry(p, struct ceph_cap, ci_node);
 909		if (!__cap_is_valid(cap))
 910			continue;
 911		if ((cap->issued & mask) == mask) {
 912			doutc(cl, "mask %p %llx.%llx cap %p issued %s (mask %s)\n",
 913			      inode, ceph_vinop(inode), cap,
 914			      ceph_cap_string(cap->issued),
 915			      ceph_cap_string(mask));
 916			if (touch)
 917				__touch_cap(cap);
 918			return 1;
 919		}
 920
 921		/* does a combination of caps satisfy mask? */
 922		have |= cap->issued;
 923		if ((have & mask) == mask) {
 924			doutc(cl, "mask %p %llx.%llx combo issued %s (mask %s)\n",
 925			      inode, ceph_vinop(inode),
 926			      ceph_cap_string(cap->issued),
 927			      ceph_cap_string(mask));
 928			if (touch) {
 929				struct rb_node *q;
 930
 931				/* touch this + preceding caps */
 932				__touch_cap(cap);
 933				for (q = rb_first(&ci->i_caps); q != p;
 934				     q = rb_next(q)) {
 935					cap = rb_entry(q, struct ceph_cap,
 936						       ci_node);
 937					if (!__cap_is_valid(cap))
 938						continue;
 939					if (cap->issued & mask)
 940						__touch_cap(cap);
 941				}
 942			}
 943			return 1;
 944		}
 945	}
 946
 947	return 0;
 948}
 949
 950int __ceph_caps_issued_mask_metric(struct ceph_inode_info *ci, int mask,
 951				   int touch)
 952{
 953	struct ceph_fs_client *fsc = ceph_sb_to_fs_client(ci->netfs.inode.i_sb);
 954	int r;
 955
 956	r = __ceph_caps_issued_mask(ci, mask, touch);
 957	if (r)
 958		ceph_update_cap_hit(&fsc->mdsc->metric);
 959	else
 960		ceph_update_cap_mis(&fsc->mdsc->metric);
 961	return r;
 962}
 963
 964/*
 965 * Return true if mask caps are currently being revoked by an MDS.
 966 */
 967int __ceph_caps_revoking_other(struct ceph_inode_info *ci,
 968			       struct ceph_cap *ocap, int mask)
 969{
 970	struct ceph_cap *cap;
 971	struct rb_node *p;
 972
 973	for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
 974		cap = rb_entry(p, struct ceph_cap, ci_node);
 975		if (cap != ocap &&
 976		    (cap->implemented & ~cap->issued & mask))
 977			return 1;
 978	}
 979	return 0;
 980}
 981
 982int __ceph_caps_used(struct ceph_inode_info *ci)
 983{
 984	int used = 0;
 985	if (ci->i_pin_ref)
 986		used |= CEPH_CAP_PIN;
 987	if (ci->i_rd_ref)
 988		used |= CEPH_CAP_FILE_RD;
 989	if (ci->i_rdcache_ref ||
 990	    (S_ISREG(ci->netfs.inode.i_mode) &&
 991	     ci->netfs.inode.i_data.nrpages))
 992		used |= CEPH_CAP_FILE_CACHE;
 993	if (ci->i_wr_ref)
 994		used |= CEPH_CAP_FILE_WR;
 995	if (ci->i_wb_ref || ci->i_wrbuffer_ref)
 996		used |= CEPH_CAP_FILE_BUFFER;
 997	if (ci->i_fx_ref)
 998		used |= CEPH_CAP_FILE_EXCL;
 999	return used;
1000}
1001
1002#define FMODE_WAIT_BIAS 1000
1003
1004/*
1005 * wanted, by virtue of open file modes
1006 */
1007int __ceph_caps_file_wanted(struct ceph_inode_info *ci)
1008{
1009	const int PIN_SHIFT = ffs(CEPH_FILE_MODE_PIN);
1010	const int RD_SHIFT = ffs(CEPH_FILE_MODE_RD);
1011	const int WR_SHIFT = ffs(CEPH_FILE_MODE_WR);
1012	const int LAZY_SHIFT = ffs(CEPH_FILE_MODE_LAZY);
1013	struct ceph_mount_options *opt =
1014		ceph_inode_to_fs_client(&ci->netfs.inode)->mount_options;
1015	unsigned long used_cutoff = jiffies - opt->caps_wanted_delay_max * HZ;
1016	unsigned long idle_cutoff = jiffies - opt->caps_wanted_delay_min * HZ;
1017
1018	if (S_ISDIR(ci->netfs.inode.i_mode)) {
1019		int want = 0;
1020
1021		/* use used_cutoff here, to keep dir's wanted caps longer */
1022		if (ci->i_nr_by_mode[RD_SHIFT] > 0 ||
1023		    time_after(ci->i_last_rd, used_cutoff))
1024			want |= CEPH_CAP_ANY_SHARED;
1025
1026		if (ci->i_nr_by_mode[WR_SHIFT] > 0 ||
1027		    time_after(ci->i_last_wr, used_cutoff)) {
1028			want |= CEPH_CAP_ANY_SHARED | CEPH_CAP_FILE_EXCL;
1029			if (opt->flags & CEPH_MOUNT_OPT_ASYNC_DIROPS)
1030				want |= CEPH_CAP_ANY_DIR_OPS;
1031		}
1032
1033		if (want || ci->i_nr_by_mode[PIN_SHIFT] > 0)
1034			want |= CEPH_CAP_PIN;
1035
1036		return want;
1037	} else {
1038		int bits = 0;
1039
1040		if (ci->i_nr_by_mode[RD_SHIFT] > 0) {
1041			if (ci->i_nr_by_mode[RD_SHIFT] >= FMODE_WAIT_BIAS ||
1042			    time_after(ci->i_last_rd, used_cutoff))
1043				bits |= 1 << RD_SHIFT;
1044		} else if (time_after(ci->i_last_rd, idle_cutoff)) {
1045			bits |= 1 << RD_SHIFT;
1046		}
1047
1048		if (ci->i_nr_by_mode[WR_SHIFT] > 0) {
1049			if (ci->i_nr_by_mode[WR_SHIFT] >= FMODE_WAIT_BIAS ||
1050			    time_after(ci->i_last_wr, used_cutoff))
1051				bits |= 1 << WR_SHIFT;
1052		} else if (time_after(ci->i_last_wr, idle_cutoff)) {
1053			bits |= 1 << WR_SHIFT;
1054		}
1055
1056		/* check lazyio only when read/write is wanted */
1057		if ((bits & (CEPH_FILE_MODE_RDWR << 1)) &&
1058		    ci->i_nr_by_mode[LAZY_SHIFT] > 0)
1059			bits |= 1 << LAZY_SHIFT;
1060
1061		return bits ? ceph_caps_for_mode(bits >> 1) : 0;
1062	}
1063}
1064
1065/*
1066 * wanted, by virtue of open file modes AND cap refs (buffered/cached data)
1067 */
1068int __ceph_caps_wanted(struct ceph_inode_info *ci)
1069{
1070	int w = __ceph_caps_file_wanted(ci) | __ceph_caps_used(ci);
1071	if (S_ISDIR(ci->netfs.inode.i_mode)) {
1072		/* we want EXCL if holding caps of dir ops */
1073		if (w & CEPH_CAP_ANY_DIR_OPS)
1074			w |= CEPH_CAP_FILE_EXCL;
1075	} else {
1076		/* we want EXCL if dirty data */
1077		if (w & CEPH_CAP_FILE_BUFFER)
1078			w |= CEPH_CAP_FILE_EXCL;
1079	}
1080	return w;
1081}
1082
1083/*
1084 * Return caps we have registered with the MDS(s) as 'wanted'.
1085 */
1086int __ceph_caps_mds_wanted(struct ceph_inode_info *ci, bool check)
1087{
1088	struct ceph_cap *cap;
1089	struct rb_node *p;
1090	int mds_wanted = 0;
1091
1092	for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
1093		cap = rb_entry(p, struct ceph_cap, ci_node);
1094		if (check && !__cap_is_valid(cap))
1095			continue;
1096		if (cap == ci->i_auth_cap)
1097			mds_wanted |= cap->mds_wanted;
1098		else
1099			mds_wanted |= (cap->mds_wanted & ~CEPH_CAP_ANY_FILE_WR);
1100	}
1101	return mds_wanted;
1102}
1103
1104int ceph_is_any_caps(struct inode *inode)
1105{
1106	struct ceph_inode_info *ci = ceph_inode(inode);
1107	int ret;
1108
1109	spin_lock(&ci->i_ceph_lock);
1110	ret = __ceph_is_any_real_caps(ci);
1111	spin_unlock(&ci->i_ceph_lock);
1112
1113	return ret;
1114}
1115
1116/*
1117 * Remove a cap.  Take steps to deal with a racing iterate_session_caps.
1118 *
1119 * caller should hold i_ceph_lock.
1120 * caller will not hold session s_mutex if called from destroy_inode.
1121 */
1122void __ceph_remove_cap(struct ceph_cap *cap, bool queue_release)
1123{
1124	struct ceph_mds_session *session = cap->session;
1125	struct ceph_client *cl = session->s_mdsc->fsc->client;
1126	struct ceph_inode_info *ci = cap->ci;
1127	struct inode *inode = &ci->netfs.inode;
1128	struct ceph_mds_client *mdsc;
1129	int removed = 0;
1130
1131	/* 'ci' being NULL means the remove have already occurred */
1132	if (!ci) {
1133		doutc(cl, "inode is NULL\n");
1134		return;
1135	}
1136
1137	lockdep_assert_held(&ci->i_ceph_lock);
1138
1139	doutc(cl, "%p from %p %llx.%llx\n", cap, inode, ceph_vinop(inode));
1140
1141	mdsc = ceph_inode_to_fs_client(&ci->netfs.inode)->mdsc;
1142
1143	/* remove from inode's cap rbtree, and clear auth cap */
1144	rb_erase(&cap->ci_node, &ci->i_caps);
1145	if (ci->i_auth_cap == cap)
1146		ci->i_auth_cap = NULL;
1147
1148	/* remove from session list */
1149	spin_lock(&session->s_cap_lock);
1150	if (session->s_cap_iterator == cap) {
1151		/* not yet, we are iterating over this very cap */
1152		doutc(cl, "delaying %p removal from session %p\n", cap,
1153		      cap->session);
1154	} else {
1155		list_del_init(&cap->session_caps);
1156		session->s_nr_caps--;
1157		atomic64_dec(&mdsc->metric.total_caps);
1158		cap->session = NULL;
1159		removed = 1;
1160	}
1161	/* protect backpointer with s_cap_lock: see iterate_session_caps */
1162	cap->ci = NULL;
1163
1164	/*
1165	 * s_cap_reconnect is protected by s_cap_lock. no one changes
1166	 * s_cap_gen while session is in the reconnect state.
1167	 */
1168	if (queue_release &&
1169	    (!session->s_cap_reconnect ||
1170	     cap->cap_gen == atomic_read(&session->s_cap_gen))) {
1171		cap->queue_release = 1;
1172		if (removed) {
1173			__ceph_queue_cap_release(session, cap);
1174			removed = 0;
1175		}
1176	} else {
1177		cap->queue_release = 0;
1178	}
1179	cap->cap_ino = ci->i_vino.ino;
1180
1181	spin_unlock(&session->s_cap_lock);
1182
1183	if (removed)
1184		ceph_put_cap(mdsc, cap);
1185
1186	if (!__ceph_is_any_real_caps(ci)) {
1187		/* when reconnect denied, we remove session caps forcibly,
1188		 * i_wr_ref can be non-zero. If there are ongoing write,
1189		 * keep i_snap_realm.
1190		 */
1191		if (ci->i_wr_ref == 0 && ci->i_snap_realm)
1192			ceph_change_snap_realm(&ci->netfs.inode, NULL);
1193
1194		__cap_delay_cancel(mdsc, ci);
1195	}
1196}
1197
1198void ceph_remove_cap(struct ceph_mds_client *mdsc, struct ceph_cap *cap,
1199		     bool queue_release)
1200{
1201	struct ceph_inode_info *ci = cap->ci;
1202	struct ceph_fs_client *fsc;
1203
1204	/* 'ci' being NULL means the remove have already occurred */
1205	if (!ci) {
1206		doutc(mdsc->fsc->client, "inode is NULL\n");
1207		return;
1208	}
1209
1210	lockdep_assert_held(&ci->i_ceph_lock);
1211
1212	fsc = ceph_inode_to_fs_client(&ci->netfs.inode);
1213	WARN_ON_ONCE(ci->i_auth_cap == cap &&
1214		     !list_empty(&ci->i_dirty_item) &&
1215		     !fsc->blocklisted &&
1216		     !ceph_inode_is_shutdown(&ci->netfs.inode));
1217
1218	__ceph_remove_cap(cap, queue_release);
1219}
1220
1221struct cap_msg_args {
1222	struct ceph_mds_session	*session;
1223	u64			ino, cid, follows;
1224	u64			flush_tid, oldest_flush_tid, size, max_size;
1225	u64			xattr_version;
1226	u64			change_attr;
1227	struct ceph_buffer	*xattr_buf;
1228	struct ceph_buffer	*old_xattr_buf;
1229	struct timespec64	atime, mtime, ctime, btime;
1230	int			op, caps, wanted, dirty;
1231	u32			seq, issue_seq, mseq, time_warp_seq;
1232	u32			flags;
1233	kuid_t			uid;
1234	kgid_t			gid;
1235	umode_t			mode;
1236	bool			inline_data;
1237	bool			wake;
1238	bool			encrypted;
1239	u32			fscrypt_auth_len;
1240	u8			fscrypt_auth[sizeof(struct ceph_fscrypt_auth)]; // for context
1241};
1242
1243/* Marshal up the cap msg to the MDS */
1244static void encode_cap_msg(struct ceph_msg *msg, struct cap_msg_args *arg)
1245{
1246	struct ceph_mds_caps *fc;
1247	void *p;
1248	struct ceph_mds_client *mdsc = arg->session->s_mdsc;
1249	struct ceph_osd_client *osdc = &mdsc->fsc->client->osdc;
1250
1251	doutc(mdsc->fsc->client,
1252	      "%s %llx %llx caps %s wanted %s dirty %s seq %u/%u"
1253	      " tid %llu/%llu mseq %u follows %lld size %llu/%llu"
1254	      " xattr_ver %llu xattr_len %d\n",
1255	      ceph_cap_op_name(arg->op), arg->cid, arg->ino,
1256	      ceph_cap_string(arg->caps), ceph_cap_string(arg->wanted),
1257	      ceph_cap_string(arg->dirty), arg->seq, arg->issue_seq,
1258	      arg->flush_tid, arg->oldest_flush_tid, arg->mseq, arg->follows,
1259	      arg->size, arg->max_size, arg->xattr_version,
1260	      arg->xattr_buf ? (int)arg->xattr_buf->vec.iov_len : 0);
1261
1262	msg->hdr.version = cpu_to_le16(12);
1263	msg->hdr.tid = cpu_to_le64(arg->flush_tid);
1264
1265	fc = msg->front.iov_base;
1266	memset(fc, 0, sizeof(*fc));
1267
1268	fc->cap_id = cpu_to_le64(arg->cid);
1269	fc->op = cpu_to_le32(arg->op);
1270	fc->seq = cpu_to_le32(arg->seq);
1271	fc->issue_seq = cpu_to_le32(arg->issue_seq);
1272	fc->migrate_seq = cpu_to_le32(arg->mseq);
1273	fc->caps = cpu_to_le32(arg->caps);
1274	fc->wanted = cpu_to_le32(arg->wanted);
1275	fc->dirty = cpu_to_le32(arg->dirty);
1276	fc->ino = cpu_to_le64(arg->ino);
1277	fc->snap_follows = cpu_to_le64(arg->follows);
1278
1279#if IS_ENABLED(CONFIG_FS_ENCRYPTION)
1280	if (arg->encrypted)
1281		fc->size = cpu_to_le64(round_up(arg->size,
1282						CEPH_FSCRYPT_BLOCK_SIZE));
1283	else
1284#endif
1285		fc->size = cpu_to_le64(arg->size);
1286	fc->max_size = cpu_to_le64(arg->max_size);
1287	ceph_encode_timespec64(&fc->mtime, &arg->mtime);
1288	ceph_encode_timespec64(&fc->atime, &arg->atime);
1289	ceph_encode_timespec64(&fc->ctime, &arg->ctime);
1290	fc->time_warp_seq = cpu_to_le32(arg->time_warp_seq);
1291
1292	fc->uid = cpu_to_le32(from_kuid(&init_user_ns, arg->uid));
1293	fc->gid = cpu_to_le32(from_kgid(&init_user_ns, arg->gid));
1294	fc->mode = cpu_to_le32(arg->mode);
1295
1296	fc->xattr_version = cpu_to_le64(arg->xattr_version);
1297	if (arg->xattr_buf) {
1298		msg->middle = ceph_buffer_get(arg->xattr_buf);
1299		fc->xattr_len = cpu_to_le32(arg->xattr_buf->vec.iov_len);
1300		msg->hdr.middle_len = cpu_to_le32(arg->xattr_buf->vec.iov_len);
1301	}
1302
1303	p = fc + 1;
1304	/* flock buffer size (version 2) */
1305	ceph_encode_32(&p, 0);
1306	/* inline version (version 4) */
1307	ceph_encode_64(&p, arg->inline_data ? 0 : CEPH_INLINE_NONE);
1308	/* inline data size */
1309	ceph_encode_32(&p, 0);
1310	/*
1311	 * osd_epoch_barrier (version 5)
1312	 * The epoch_barrier is protected osdc->lock, so READ_ONCE here in
1313	 * case it was recently changed
1314	 */
1315	ceph_encode_32(&p, READ_ONCE(osdc->epoch_barrier));
1316	/* oldest_flush_tid (version 6) */
1317	ceph_encode_64(&p, arg->oldest_flush_tid);
1318
1319	/*
1320	 * caller_uid/caller_gid (version 7)
1321	 *
1322	 * Currently, we don't properly track which caller dirtied the caps
1323	 * last, and force a flush of them when there is a conflict. For now,
1324	 * just set this to 0:0, to emulate how the MDS has worked up to now.
1325	 */
1326	ceph_encode_32(&p, 0);
1327	ceph_encode_32(&p, 0);
1328
1329	/* pool namespace (version 8) (mds always ignores this) */
1330	ceph_encode_32(&p, 0);
1331
1332	/* btime and change_attr (version 9) */
1333	ceph_encode_timespec64(p, &arg->btime);
1334	p += sizeof(struct ceph_timespec);
1335	ceph_encode_64(&p, arg->change_attr);
1336
1337	/* Advisory flags (version 10) */
1338	ceph_encode_32(&p, arg->flags);
1339
1340	/* dirstats (version 11) - these are r/o on the client */
1341	ceph_encode_64(&p, 0);
1342	ceph_encode_64(&p, 0);
1343
1344#if IS_ENABLED(CONFIG_FS_ENCRYPTION)
1345	/*
1346	 * fscrypt_auth and fscrypt_file (version 12)
1347	 *
1348	 * fscrypt_auth holds the crypto context (if any). fscrypt_file
1349	 * tracks the real i_size as an __le64 field (and we use a rounded-up
1350	 * i_size in the traditional size field).
1351	 */
1352	ceph_encode_32(&p, arg->fscrypt_auth_len);
1353	ceph_encode_copy(&p, arg->fscrypt_auth, arg->fscrypt_auth_len);
1354	ceph_encode_32(&p, sizeof(__le64));
1355	ceph_encode_64(&p, arg->size);
1356#else /* CONFIG_FS_ENCRYPTION */
1357	ceph_encode_32(&p, 0);
1358	ceph_encode_32(&p, 0);
1359#endif /* CONFIG_FS_ENCRYPTION */
1360}
1361
1362/*
1363 * Queue cap releases when an inode is dropped from our cache.
1364 */
1365void __ceph_remove_caps(struct ceph_inode_info *ci)
1366{
1367	struct inode *inode = &ci->netfs.inode;
1368	struct ceph_mds_client *mdsc = ceph_inode_to_fs_client(inode)->mdsc;
1369	struct rb_node *p;
1370
1371	/* lock i_ceph_lock, because ceph_d_revalidate(..., LOOKUP_RCU)
1372	 * may call __ceph_caps_issued_mask() on a freeing inode. */
1373	spin_lock(&ci->i_ceph_lock);
1374	p = rb_first(&ci->i_caps);
1375	while (p) {
1376		struct ceph_cap *cap = rb_entry(p, struct ceph_cap, ci_node);
1377		p = rb_next(p);
1378		ceph_remove_cap(mdsc, cap, true);
1379	}
1380	spin_unlock(&ci->i_ceph_lock);
1381}
1382
1383/*
1384 * Prepare to send a cap message to an MDS. Update the cap state, and populate
1385 * the arg struct with the parameters that will need to be sent. This should
1386 * be done under the i_ceph_lock to guard against changes to cap state.
1387 *
1388 * Make note of max_size reported/requested from mds, revoked caps
1389 * that have now been implemented.
1390 */
1391static void __prep_cap(struct cap_msg_args *arg, struct ceph_cap *cap,
1392		       int op, int flags, int used, int want, int retain,
1393		       int flushing, u64 flush_tid, u64 oldest_flush_tid)
1394{
1395	struct ceph_inode_info *ci = cap->ci;
1396	struct inode *inode = &ci->netfs.inode;
1397	struct ceph_client *cl = ceph_inode_to_client(inode);
1398	int held, revoking;
1399
1400	lockdep_assert_held(&ci->i_ceph_lock);
1401
1402	held = cap->issued | cap->implemented;
1403	revoking = cap->implemented & ~cap->issued;
1404	retain &= ~revoking;
1405
1406	doutc(cl, "%p %llx.%llx cap %p session %p %s -> %s (revoking %s)\n",
1407	      inode, ceph_vinop(inode), cap, cap->session,
1408	      ceph_cap_string(held), ceph_cap_string(held & retain),
1409	      ceph_cap_string(revoking));
1410	BUG_ON((retain & CEPH_CAP_PIN) == 0);
1411
1412	ci->i_ceph_flags &= ~CEPH_I_FLUSH;
1413
1414	cap->issued &= retain;  /* drop bits we don't want */
1415	/*
1416	 * Wake up any waiters on wanted -> needed transition. This is due to
1417	 * the weird transition from buffered to sync IO... we need to flush
1418	 * dirty pages _before_ allowing sync writes to avoid reordering.
1419	 */
1420	arg->wake = cap->implemented & ~cap->issued;
1421	cap->implemented &= cap->issued | used;
1422	cap->mds_wanted = want;
1423
1424	arg->session = cap->session;
1425	arg->ino = ceph_vino(inode).ino;
1426	arg->cid = cap->cap_id;
1427	arg->follows = flushing ? ci->i_head_snapc->seq : 0;
1428	arg->flush_tid = flush_tid;
1429	arg->oldest_flush_tid = oldest_flush_tid;
1430	arg->size = i_size_read(inode);
1431	ci->i_reported_size = arg->size;
1432	arg->max_size = ci->i_wanted_max_size;
1433	if (cap == ci->i_auth_cap) {
1434		if (want & CEPH_CAP_ANY_FILE_WR)
1435			ci->i_requested_max_size = arg->max_size;
1436		else
1437			ci->i_requested_max_size = 0;
1438	}
1439
1440	if (flushing & CEPH_CAP_XATTR_EXCL) {
1441		arg->old_xattr_buf = __ceph_build_xattrs_blob(ci);
1442		arg->xattr_version = ci->i_xattrs.version;
1443		arg->xattr_buf = ceph_buffer_get(ci->i_xattrs.blob);
1444	} else {
1445		arg->xattr_buf = NULL;
1446		arg->old_xattr_buf = NULL;
1447	}
1448
1449	arg->mtime = inode_get_mtime(inode);
1450	arg->atime = inode_get_atime(inode);
1451	arg->ctime = inode_get_ctime(inode);
1452	arg->btime = ci->i_btime;
1453	arg->change_attr = inode_peek_iversion_raw(inode);
1454
1455	arg->op = op;
1456	arg->caps = cap->implemented;
1457	arg->wanted = want;
1458	arg->dirty = flushing;
1459
1460	arg->seq = cap->seq;
1461	arg->issue_seq = cap->issue_seq;
1462	arg->mseq = cap->mseq;
1463	arg->time_warp_seq = ci->i_time_warp_seq;
1464
1465	arg->uid = inode->i_uid;
1466	arg->gid = inode->i_gid;
1467	arg->mode = inode->i_mode;
1468
1469	arg->inline_data = ci->i_inline_version != CEPH_INLINE_NONE;
1470	if (!(flags & CEPH_CLIENT_CAPS_PENDING_CAPSNAP) &&
1471	    !list_empty(&ci->i_cap_snaps)) {
1472		struct ceph_cap_snap *capsnap;
1473		list_for_each_entry_reverse(capsnap, &ci->i_cap_snaps, ci_item) {
1474			if (capsnap->cap_flush.tid)
1475				break;
1476			if (capsnap->need_flush) {
1477				flags |= CEPH_CLIENT_CAPS_PENDING_CAPSNAP;
1478				break;
1479			}
1480		}
1481	}
1482	arg->flags = flags;
1483	arg->encrypted = IS_ENCRYPTED(inode);
1484#if IS_ENABLED(CONFIG_FS_ENCRYPTION)
1485	if (ci->fscrypt_auth_len &&
1486	    WARN_ON_ONCE(ci->fscrypt_auth_len > sizeof(struct ceph_fscrypt_auth))) {
1487		/* Don't set this if it's too big */
1488		arg->fscrypt_auth_len = 0;
1489	} else {
1490		arg->fscrypt_auth_len = ci->fscrypt_auth_len;
1491		memcpy(arg->fscrypt_auth, ci->fscrypt_auth,
1492		       min_t(size_t, ci->fscrypt_auth_len,
1493			     sizeof(arg->fscrypt_auth)));
1494	}
1495#endif /* CONFIG_FS_ENCRYPTION */
1496}
1497
1498#if IS_ENABLED(CONFIG_FS_ENCRYPTION)
1499#define CAP_MSG_FIXED_FIELDS (sizeof(struct ceph_mds_caps) + \
1500		      4 + 8 + 4 + 4 + 8 + 4 + 4 + 4 + 8 + 8 + 4 + 8 + 8 + 4 + 4 + 8)
1501
1502static inline int cap_msg_size(struct cap_msg_args *arg)
1503{
1504	return CAP_MSG_FIXED_FIELDS + arg->fscrypt_auth_len;
1505}
1506#else
1507#define CAP_MSG_FIXED_FIELDS (sizeof(struct ceph_mds_caps) + \
1508		      4 + 8 + 4 + 4 + 8 + 4 + 4 + 4 + 8 + 8 + 4 + 8 + 8 + 4 + 4)
1509
1510static inline int cap_msg_size(struct cap_msg_args *arg)
1511{
1512	return CAP_MSG_FIXED_FIELDS;
1513}
1514#endif /* CONFIG_FS_ENCRYPTION */
1515
1516/*
1517 * Send a cap msg on the given inode.
1518 *
1519 * Caller should hold snap_rwsem (read), s_mutex.
1520 */
1521static void __send_cap(struct cap_msg_args *arg, struct ceph_inode_info *ci)
1522{
1523	struct ceph_msg *msg;
1524	struct inode *inode = &ci->netfs.inode;
1525	struct ceph_client *cl = ceph_inode_to_client(inode);
1526
1527	msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPS, cap_msg_size(arg), GFP_NOFS,
1528			   false);
1529	if (!msg) {
1530		pr_err_client(cl,
1531			      "error allocating cap msg: ino (%llx.%llx)"
1532			      " flushing %s tid %llu, requeuing cap.\n",
1533			      ceph_vinop(inode), ceph_cap_string(arg->dirty),
1534			      arg->flush_tid);
1535		spin_lock(&ci->i_ceph_lock);
1536		__cap_delay_requeue(arg->session->s_mdsc, ci);
1537		spin_unlock(&ci->i_ceph_lock);
1538		return;
1539	}
1540
1541	encode_cap_msg(msg, arg);
1542	ceph_con_send(&arg->session->s_con, msg);
1543	ceph_buffer_put(arg->old_xattr_buf);
1544	ceph_buffer_put(arg->xattr_buf);
1545	if (arg->wake)
1546		wake_up_all(&ci->i_cap_wq);
1547}
1548
1549static inline int __send_flush_snap(struct inode *inode,
1550				    struct ceph_mds_session *session,
1551				    struct ceph_cap_snap *capsnap,
1552				    u32 mseq, u64 oldest_flush_tid)
1553{
1554	struct cap_msg_args	arg;
1555	struct ceph_msg		*msg;
1556
1557	arg.session = session;
1558	arg.ino = ceph_vino(inode).ino;
1559	arg.cid = 0;
1560	arg.follows = capsnap->follows;
1561	arg.flush_tid = capsnap->cap_flush.tid;
1562	arg.oldest_flush_tid = oldest_flush_tid;
1563
1564	arg.size = capsnap->size;
1565	arg.max_size = 0;
1566	arg.xattr_version = capsnap->xattr_version;
1567	arg.xattr_buf = capsnap->xattr_blob;
1568	arg.old_xattr_buf = NULL;
1569
1570	arg.atime = capsnap->atime;
1571	arg.mtime = capsnap->mtime;
1572	arg.ctime = capsnap->ctime;
1573	arg.btime = capsnap->btime;
1574	arg.change_attr = capsnap->change_attr;
1575
1576	arg.op = CEPH_CAP_OP_FLUSHSNAP;
1577	arg.caps = capsnap->issued;
1578	arg.wanted = 0;
1579	arg.dirty = capsnap->dirty;
1580
1581	arg.seq = 0;
1582	arg.issue_seq = 0;
1583	arg.mseq = mseq;
1584	arg.time_warp_seq = capsnap->time_warp_seq;
1585
1586	arg.uid = capsnap->uid;
1587	arg.gid = capsnap->gid;
1588	arg.mode = capsnap->mode;
1589
1590	arg.inline_data = capsnap->inline_data;
1591	arg.flags = 0;
1592	arg.wake = false;
1593	arg.encrypted = IS_ENCRYPTED(inode);
1594
1595	/* No fscrypt_auth changes from a capsnap.*/
1596	arg.fscrypt_auth_len = 0;
1597
1598	msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPS, cap_msg_size(&arg),
1599			   GFP_NOFS, false);
1600	if (!msg)
1601		return -ENOMEM;
1602
1603	encode_cap_msg(msg, &arg);
1604	ceph_con_send(&arg.session->s_con, msg);
1605	return 0;
1606}
1607
1608/*
1609 * When a snapshot is taken, clients accumulate dirty metadata on
1610 * inodes with capabilities in ceph_cap_snaps to describe the file
1611 * state at the time the snapshot was taken.  This must be flushed
1612 * asynchronously back to the MDS once sync writes complete and dirty
1613 * data is written out.
1614 *
1615 * Called under i_ceph_lock.
1616 */
1617static void __ceph_flush_snaps(struct ceph_inode_info *ci,
1618			       struct ceph_mds_session *session)
1619		__releases(ci->i_ceph_lock)
1620		__acquires(ci->i_ceph_lock)
1621{
1622	struct inode *inode = &ci->netfs.inode;
1623	struct ceph_mds_client *mdsc = session->s_mdsc;
1624	struct ceph_client *cl = mdsc->fsc->client;
1625	struct ceph_cap_snap *capsnap;
1626	u64 oldest_flush_tid = 0;
1627	u64 first_tid = 1, last_tid = 0;
1628
1629	doutc(cl, "%p %llx.%llx session %p\n", inode, ceph_vinop(inode),
1630	      session);
1631
1632	list_for_each_entry(capsnap, &ci->i_cap_snaps, ci_item) {
1633		/*
1634		 * we need to wait for sync writes to complete and for dirty
1635		 * pages to be written out.
1636		 */
1637		if (capsnap->dirty_pages || capsnap->writing)
1638			break;
1639
1640		/* should be removed by ceph_try_drop_cap_snap() */
1641		BUG_ON(!capsnap->need_flush);
1642
1643		/* only flush each capsnap once */
1644		if (capsnap->cap_flush.tid > 0) {
1645			doutc(cl, "already flushed %p, skipping\n", capsnap);
1646			continue;
1647		}
1648
1649		spin_lock(&mdsc->cap_dirty_lock);
1650		capsnap->cap_flush.tid = ++mdsc->last_cap_flush_tid;
1651		list_add_tail(&capsnap->cap_flush.g_list,
1652			      &mdsc->cap_flush_list);
1653		if (oldest_flush_tid == 0)
1654			oldest_flush_tid = __get_oldest_flush_tid(mdsc);
1655		if (list_empty(&ci->i_flushing_item)) {
1656			list_add_tail(&ci->i_flushing_item,
1657				      &session->s_cap_flushing);
1658		}
1659		spin_unlock(&mdsc->cap_dirty_lock);
1660
1661		list_add_tail(&capsnap->cap_flush.i_list,
1662			      &ci->i_cap_flush_list);
1663
1664		if (first_tid == 1)
1665			first_tid = capsnap->cap_flush.tid;
1666		last_tid = capsnap->cap_flush.tid;
1667	}
1668
1669	ci->i_ceph_flags &= ~CEPH_I_FLUSH_SNAPS;
1670
1671	while (first_tid <= last_tid) {
1672		struct ceph_cap *cap = ci->i_auth_cap;
1673		struct ceph_cap_flush *cf = NULL, *iter;
1674		int ret;
1675
1676		if (!(cap && cap->session == session)) {
1677			doutc(cl, "%p %llx.%llx auth cap %p not mds%d, stop\n",
1678			      inode, ceph_vinop(inode), cap, session->s_mds);
1679			break;
1680		}
1681
1682		ret = -ENOENT;
1683		list_for_each_entry(iter, &ci->i_cap_flush_list, i_list) {
1684			if (iter->tid >= first_tid) {
1685				cf = iter;
1686				ret = 0;
1687				break;
1688			}
1689		}
1690		if (ret < 0)
1691			break;
1692
1693		first_tid = cf->tid + 1;
1694
1695		capsnap = container_of(cf, struct ceph_cap_snap, cap_flush);
1696		refcount_inc(&capsnap->nref);
1697		spin_unlock(&ci->i_ceph_lock);
1698
1699		doutc(cl, "%p %llx.%llx capsnap %p tid %llu %s\n", inode,
1700		      ceph_vinop(inode), capsnap, cf->tid,
1701		      ceph_cap_string(capsnap->dirty));
1702
1703		ret = __send_flush_snap(inode, session, capsnap, cap->mseq,
1704					oldest_flush_tid);
1705		if (ret < 0) {
1706			pr_err_client(cl, "error sending cap flushsnap, "
1707				      "ino (%llx.%llx) tid %llu follows %llu\n",
1708				      ceph_vinop(inode), cf->tid,
1709				      capsnap->follows);
1710		}
1711
1712		ceph_put_cap_snap(capsnap);
1713		spin_lock(&ci->i_ceph_lock);
1714	}
1715}
1716
1717void ceph_flush_snaps(struct ceph_inode_info *ci,
1718		      struct ceph_mds_session **psession)
1719{
1720	struct inode *inode = &ci->netfs.inode;
1721	struct ceph_mds_client *mdsc = ceph_inode_to_fs_client(inode)->mdsc;
1722	struct ceph_client *cl = ceph_inode_to_client(inode);
1723	struct ceph_mds_session *session = NULL;
1724	bool need_put = false;
1725	int mds;
1726
1727	doutc(cl, "%p %llx.%llx\n", inode, ceph_vinop(inode));
1728	if (psession)
1729		session = *psession;
1730retry:
1731	spin_lock(&ci->i_ceph_lock);
1732	if (!(ci->i_ceph_flags & CEPH_I_FLUSH_SNAPS)) {
1733		doutc(cl, " no capsnap needs flush, doing nothing\n");
1734		goto out;
1735	}
1736	if (!ci->i_auth_cap) {
1737		doutc(cl, " no auth cap (migrating?), doing nothing\n");
1738		goto out;
1739	}
1740
1741	mds = ci->i_auth_cap->session->s_mds;
1742	if (session && session->s_mds != mds) {
1743		doutc(cl, " oops, wrong session %p mutex\n", session);
1744		ceph_put_mds_session(session);
1745		session = NULL;
1746	}
1747	if (!session) {
1748		spin_unlock(&ci->i_ceph_lock);
1749		mutex_lock(&mdsc->mutex);
1750		session = __ceph_lookup_mds_session(mdsc, mds);
1751		mutex_unlock(&mdsc->mutex);
1752		goto retry;
1753	}
1754
1755	// make sure flushsnap messages are sent in proper order.
1756	if (ci->i_ceph_flags & CEPH_I_KICK_FLUSH)
1757		__kick_flushing_caps(mdsc, session, ci, 0);
1758
1759	__ceph_flush_snaps(ci, session);
1760out:
1761	spin_unlock(&ci->i_ceph_lock);
1762
1763	if (psession)
1764		*psession = session;
1765	else
1766		ceph_put_mds_session(session);
1767	/* we flushed them all; remove this inode from the queue */
1768	spin_lock(&mdsc->snap_flush_lock);
1769	if (!list_empty(&ci->i_snap_flush_item))
1770		need_put = true;
1771	list_del_init(&ci->i_snap_flush_item);
1772	spin_unlock(&mdsc->snap_flush_lock);
1773
1774	if (need_put)
1775		iput(inode);
1776}
1777
1778/*
1779 * Mark caps dirty.  If inode is newly dirty, return the dirty flags.
1780 * Caller is then responsible for calling __mark_inode_dirty with the
1781 * returned flags value.
1782 */
1783int __ceph_mark_dirty_caps(struct ceph_inode_info *ci, int mask,
1784			   struct ceph_cap_flush **pcf)
1785{
1786	struct ceph_mds_client *mdsc =
1787		ceph_sb_to_fs_client(ci->netfs.inode.i_sb)->mdsc;
1788	struct inode *inode = &ci->netfs.inode;
1789	struct ceph_client *cl = ceph_inode_to_client(inode);
1790	int was = ci->i_dirty_caps;
1791	int dirty = 0;
1792
1793	lockdep_assert_held(&ci->i_ceph_lock);
1794
1795	if (!ci->i_auth_cap) {
1796		pr_warn_client(cl, "%p %llx.%llx mask %s, "
1797			       "but no auth cap (session was closed?)\n",
1798				inode, ceph_vinop(inode),
1799				ceph_cap_string(mask));
1800		return 0;
1801	}
1802
1803	doutc(cl, "%p %llx.%llx %s dirty %s -> %s\n", inode,
1804	      ceph_vinop(inode), ceph_cap_string(mask),
1805	      ceph_cap_string(was), ceph_cap_string(was | mask));
1806	ci->i_dirty_caps |= mask;
1807	if (was == 0) {
1808		struct ceph_mds_session *session = ci->i_auth_cap->session;
1809
1810		WARN_ON_ONCE(ci->i_prealloc_cap_flush);
1811		swap(ci->i_prealloc_cap_flush, *pcf);
1812
1813		if (!ci->i_head_snapc) {
1814			WARN_ON_ONCE(!rwsem_is_locked(&mdsc->snap_rwsem));
1815			ci->i_head_snapc = ceph_get_snap_context(
1816				ci->i_snap_realm->cached_context);
1817		}
1818		doutc(cl, "%p %llx.%llx now dirty snapc %p auth cap %p\n",
1819		      inode, ceph_vinop(inode), ci->i_head_snapc,
1820		      ci->i_auth_cap);
1821		BUG_ON(!list_empty(&ci->i_dirty_item));
1822		spin_lock(&mdsc->cap_dirty_lock);
1823		list_add(&ci->i_dirty_item, &session->s_cap_dirty);
1824		spin_unlock(&mdsc->cap_dirty_lock);
1825		if (ci->i_flushing_caps == 0) {
1826			ihold(inode);
1827			dirty |= I_DIRTY_SYNC;
1828		}
1829	} else {
1830		WARN_ON_ONCE(!ci->i_prealloc_cap_flush);
1831	}
1832	BUG_ON(list_empty(&ci->i_dirty_item));
1833	if (((was | ci->i_flushing_caps) & CEPH_CAP_FILE_BUFFER) &&
1834	    (mask & CEPH_CAP_FILE_BUFFER))
1835		dirty |= I_DIRTY_DATASYNC;
1836	__cap_delay_requeue(mdsc, ci);
1837	return dirty;
1838}
1839
1840struct ceph_cap_flush *ceph_alloc_cap_flush(void)
1841{
1842	struct ceph_cap_flush *cf;
1843
1844	cf = kmem_cache_alloc(ceph_cap_flush_cachep, GFP_KERNEL);
1845	if (!cf)
1846		return NULL;
1847
1848	cf->is_capsnap = false;
1849	return cf;
1850}
1851
1852void ceph_free_cap_flush(struct ceph_cap_flush *cf)
1853{
1854	if (cf)
1855		kmem_cache_free(ceph_cap_flush_cachep, cf);
1856}
1857
1858static u64 __get_oldest_flush_tid(struct ceph_mds_client *mdsc)
1859{
1860	if (!list_empty(&mdsc->cap_flush_list)) {
1861		struct ceph_cap_flush *cf =
1862			list_first_entry(&mdsc->cap_flush_list,
1863					 struct ceph_cap_flush, g_list);
1864		return cf->tid;
1865	}
1866	return 0;
1867}
1868
1869/*
1870 * Remove cap_flush from the mdsc's or inode's flushing cap list.
1871 * Return true if caller needs to wake up flush waiters.
1872 */
1873static bool __detach_cap_flush_from_mdsc(struct ceph_mds_client *mdsc,
1874					 struct ceph_cap_flush *cf)
1875{
1876	struct ceph_cap_flush *prev;
1877	bool wake = cf->wake;
1878
1879	if (wake && cf->g_list.prev != &mdsc->cap_flush_list) {
1880		prev = list_prev_entry(cf, g_list);
1881		prev->wake = true;
1882		wake = false;
1883	}
1884	list_del_init(&cf->g_list);
1885	return wake;
1886}
1887
1888static bool __detach_cap_flush_from_ci(struct ceph_inode_info *ci,
1889				       struct ceph_cap_flush *cf)
1890{
1891	struct ceph_cap_flush *prev;
1892	bool wake = cf->wake;
1893
1894	if (wake && cf->i_list.prev != &ci->i_cap_flush_list) {
1895		prev = list_prev_entry(cf, i_list);
1896		prev->wake = true;
1897		wake = false;
1898	}
1899	list_del_init(&cf->i_list);
1900	return wake;
1901}
1902
1903/*
1904 * Add dirty inode to the flushing list.  Assigned a seq number so we
1905 * can wait for caps to flush without starving.
1906 *
1907 * Called under i_ceph_lock. Returns the flush tid.
1908 */
1909static u64 __mark_caps_flushing(struct inode *inode,
1910				struct ceph_mds_session *session, bool wake,
1911				u64 *oldest_flush_tid)
1912{
1913	struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(inode->i_sb)->mdsc;
1914	struct ceph_client *cl = ceph_inode_to_client(inode);
1915	struct ceph_inode_info *ci = ceph_inode(inode);
1916	struct ceph_cap_flush *cf = NULL;
1917	int flushing;
1918
1919	lockdep_assert_held(&ci->i_ceph_lock);
1920	BUG_ON(ci->i_dirty_caps == 0);
1921	BUG_ON(list_empty(&ci->i_dirty_item));
1922	BUG_ON(!ci->i_prealloc_cap_flush);
1923
1924	flushing = ci->i_dirty_caps;
1925	doutc(cl, "flushing %s, flushing_caps %s -> %s\n",
1926	      ceph_cap_string(flushing),
1927	      ceph_cap_string(ci->i_flushing_caps),
1928	      ceph_cap_string(ci->i_flushing_caps | flushing));
1929	ci->i_flushing_caps |= flushing;
1930	ci->i_dirty_caps = 0;
1931	doutc(cl, "%p %llx.%llx now !dirty\n", inode, ceph_vinop(inode));
1932
1933	swap(cf, ci->i_prealloc_cap_flush);
1934	cf->caps = flushing;
1935	cf->wake = wake;
1936
1937	spin_lock(&mdsc->cap_dirty_lock);
1938	list_del_init(&ci->i_dirty_item);
1939
1940	cf->tid = ++mdsc->last_cap_flush_tid;
1941	list_add_tail(&cf->g_list, &mdsc->cap_flush_list);
1942	*oldest_flush_tid = __get_oldest_flush_tid(mdsc);
1943
1944	if (list_empty(&ci->i_flushing_item)) {
1945		list_add_tail(&ci->i_flushing_item, &session->s_cap_flushing);
1946		mdsc->num_cap_flushing++;
1947	}
1948	spin_unlock(&mdsc->cap_dirty_lock);
1949
1950	list_add_tail(&cf->i_list, &ci->i_cap_flush_list);
1951
1952	return cf->tid;
1953}
1954
1955/*
1956 * try to invalidate mapping pages without blocking.
1957 */
1958static int try_nonblocking_invalidate(struct inode *inode)
1959	__releases(ci->i_ceph_lock)
1960	__acquires(ci->i_ceph_lock)
1961{
1962	struct ceph_client *cl = ceph_inode_to_client(inode);
1963	struct ceph_inode_info *ci = ceph_inode(inode);
1964	u32 invalidating_gen = ci->i_rdcache_gen;
1965
1966	spin_unlock(&ci->i_ceph_lock);
1967	ceph_fscache_invalidate(inode, false);
1968	invalidate_mapping_pages(&inode->i_data, 0, -1);
1969	spin_lock(&ci->i_ceph_lock);
1970
1971	if (inode->i_data.nrpages == 0 &&
1972	    invalidating_gen == ci->i_rdcache_gen) {
1973		/* success. */
1974		doutc(cl, "%p %llx.%llx success\n", inode,
1975		      ceph_vinop(inode));
1976		/* save any racing async invalidate some trouble */
1977		ci->i_rdcache_revoking = ci->i_rdcache_gen - 1;
1978		return 0;
1979	}
1980	doutc(cl, "%p %llx.%llx failed\n", inode, ceph_vinop(inode));
1981	return -1;
1982}
1983
1984bool __ceph_should_report_size(struct ceph_inode_info *ci)
1985{
1986	loff_t size = i_size_read(&ci->netfs.inode);
1987	/* mds will adjust max size according to the reported size */
1988	if (ci->i_flushing_caps & CEPH_CAP_FILE_WR)
1989		return false;
1990	if (size >= ci->i_max_size)
1991		return true;
1992	/* half of previous max_size increment has been used */
1993	if (ci->i_max_size > ci->i_reported_size &&
1994	    (size << 1) >= ci->i_max_size + ci->i_reported_size)
1995		return true;
1996	return false;
1997}
1998
1999/*
2000 * Swiss army knife function to examine currently used and wanted
2001 * versus held caps.  Release, flush, ack revoked caps to mds as
2002 * appropriate.
2003 *
2004 *  CHECK_CAPS_AUTHONLY - we should only check the auth cap
2005 *  CHECK_CAPS_FLUSH - we should flush any dirty caps immediately, without
2006 *    further delay.
2007 *  CHECK_CAPS_FLUSH_FORCE - we should flush any caps immediately, without
2008 *    further delay.
2009 */
2010void ceph_check_caps(struct ceph_inode_info *ci, int flags)
2011{
2012	struct inode *inode = &ci->netfs.inode;
2013	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
2014	struct ceph_client *cl = ceph_inode_to_client(inode);
2015	struct ceph_cap *cap;
2016	u64 flush_tid, oldest_flush_tid;
2017	int file_wanted, used, cap_used;
2018	int issued, implemented, want, retain, revoking, flushing = 0;
2019	int mds = -1;   /* keep track of how far we've gone through i_caps list
2020			   to avoid an infinite loop on retry */
2021	struct rb_node *p;
2022	bool queue_invalidate = false;
2023	bool tried_invalidate = false;
2024	bool queue_writeback = false;
2025	struct ceph_mds_session *session = NULL;
2026
2027	spin_lock(&ci->i_ceph_lock);
2028	if (ci->i_ceph_flags & CEPH_I_ASYNC_CREATE) {
2029		ci->i_ceph_flags |= CEPH_I_ASYNC_CHECK_CAPS;
2030
2031		/* Don't send messages until we get async create reply */
2032		spin_unlock(&ci->i_ceph_lock);
2033		return;
2034	}
2035
2036	if (ci->i_ceph_flags & CEPH_I_FLUSH)
2037		flags |= CHECK_CAPS_FLUSH;
2038retry:
2039	/* Caps wanted by virtue of active open files. */
2040	file_wanted = __ceph_caps_file_wanted(ci);
2041
2042	/* Caps which have active references against them */
2043	used = __ceph_caps_used(ci);
2044
2045	/*
2046	 * "issued" represents the current caps that the MDS wants us to have.
2047	 * "implemented" is the set that we have been granted, and includes the
2048	 * ones that have not yet been returned to the MDS (the "revoking" set,
2049	 * usually because they have outstanding references).
2050	 */
2051	issued = __ceph_caps_issued(ci, &implemented);
2052	revoking = implemented & ~issued;
2053
2054	want = file_wanted;
2055
2056	/* The ones we currently want to retain (may be adjusted below) */
2057	retain = file_wanted | used | CEPH_CAP_PIN;
2058	if (!mdsc->stopping && inode->i_nlink > 0) {
2059		if (file_wanted) {
2060			retain |= CEPH_CAP_ANY;       /* be greedy */
2061		} else if (S_ISDIR(inode->i_mode) &&
2062			   (issued & CEPH_CAP_FILE_SHARED) &&
2063			   __ceph_dir_is_complete(ci)) {
2064			/*
2065			 * If a directory is complete, we want to keep
2066			 * the exclusive cap. So that MDS does not end up
2067			 * revoking the shared cap on every create/unlink
2068			 * operation.
2069			 */
2070			if (IS_RDONLY(inode)) {
2071				want = CEPH_CAP_ANY_SHARED;
2072			} else {
2073				want |= CEPH_CAP_ANY_SHARED | CEPH_CAP_FILE_EXCL;
2074			}
2075			retain |= want;
2076		} else {
2077
2078			retain |= CEPH_CAP_ANY_SHARED;
2079			/*
2080			 * keep RD only if we didn't have the file open RW,
2081			 * because then the mds would revoke it anyway to
2082			 * journal max_size=0.
2083			 */
2084			if (ci->i_max_size == 0)
2085				retain |= CEPH_CAP_ANY_RD;
2086		}
2087	}
2088
2089	doutc(cl, "%p %llx.%llx file_want %s used %s dirty %s "
2090	      "flushing %s issued %s revoking %s retain %s %s%s%s%s\n",
2091	     inode, ceph_vinop(inode), ceph_cap_string(file_wanted),
2092	     ceph_cap_string(used), ceph_cap_string(ci->i_dirty_caps),
2093	     ceph_cap_string(ci->i_flushing_caps),
2094	     ceph_cap_string(issued), ceph_cap_string(revoking),
2095	     ceph_cap_string(retain),
2096	     (flags & CHECK_CAPS_AUTHONLY) ? " AUTHONLY" : "",
2097	     (flags & CHECK_CAPS_FLUSH) ? " FLUSH" : "",
2098	     (flags & CHECK_CAPS_NOINVAL) ? " NOINVAL" : "",
2099	     (flags & CHECK_CAPS_FLUSH_FORCE) ? " FLUSH_FORCE" : "");
2100
2101	/*
2102	 * If we no longer need to hold onto old our caps, and we may
2103	 * have cached pages, but don't want them, then try to invalidate.
2104	 * If we fail, it's because pages are locked.... try again later.
2105	 */
2106	if ((!(flags & CHECK_CAPS_NOINVAL) || mdsc->stopping) &&
2107	    S_ISREG(inode->i_mode) &&
2108	    !(ci->i_wb_ref || ci->i_wrbuffer_ref) &&   /* no dirty pages... */
2109	    inode->i_data.nrpages &&		/* have cached pages */
2110	    (revoking & (CEPH_CAP_FILE_CACHE|
2111			 CEPH_CAP_FILE_LAZYIO)) && /*  or revoking cache */
2112	    !tried_invalidate) {
2113		doutc(cl, "trying to invalidate on %p %llx.%llx\n",
2114		      inode, ceph_vinop(inode));
2115		if (try_nonblocking_invalidate(inode) < 0) {
2116			doutc(cl, "queuing invalidate\n");
2117			queue_invalidate = true;
2118			ci->i_rdcache_revoking = ci->i_rdcache_gen;
2119		}
2120		tried_invalidate = true;
2121		goto retry;
2122	}
2123
2124	for (p = rb_first(&ci->i_caps); p; p = rb_next(p)) {
2125		int mflags = 0;
2126		struct cap_msg_args arg;
2127
2128		cap = rb_entry(p, struct ceph_cap, ci_node);
2129
2130		/* avoid looping forever */
2131		if (mds >= cap->mds ||
2132		    ((flags & CHECK_CAPS_AUTHONLY) && cap != ci->i_auth_cap))
2133			continue;
2134
2135		/*
2136		 * If we have an auth cap, we don't need to consider any
2137		 * overlapping caps as used.
2138		 */
2139		cap_used = used;
2140		if (ci->i_auth_cap && cap != ci->i_auth_cap)
2141			cap_used &= ~ci->i_auth_cap->issued;
2142
2143		revoking = cap->implemented & ~cap->issued;
2144		doutc(cl, " mds%d cap %p used %s issued %s implemented %s revoking %s\n",
2145		      cap->mds, cap, ceph_cap_string(cap_used),
2146		      ceph_cap_string(cap->issued),
2147		      ceph_cap_string(cap->implemented),
2148		      ceph_cap_string(revoking));
2149
2150		/* completed revocation? going down and there are no caps? */
2151		if (revoking) {
2152			if ((revoking & cap_used) == 0) {
2153				doutc(cl, "completed revocation of %s\n",
2154				      ceph_cap_string(cap->implemented & ~cap->issued));
2155				goto ack;
2156			}
2157
2158			/*
2159			 * If the "i_wrbuffer_ref" was increased by mmap or generic
2160			 * cache write just before the ceph_check_caps() is called,
2161			 * the Fb capability revoking will fail this time. Then we
2162			 * must wait for the BDI's delayed work to flush the dirty
2163			 * pages and to release the "i_wrbuffer_ref", which will cost
2164			 * at most 5 seconds. That means the MDS needs to wait at
2165			 * most 5 seconds to finished the Fb capability's revocation.
2166			 *
2167			 * Let's queue a writeback for it.
2168			 */
2169			if (S_ISREG(inode->i_mode) && ci->i_wrbuffer_ref &&
2170			    (revoking & CEPH_CAP_FILE_BUFFER))
2171				queue_writeback = true;
2172		}
2173
2174		if (flags & CHECK_CAPS_FLUSH_FORCE) {
2175			doutc(cl, "force to flush caps\n");
2176			goto ack;
2177		}
2178
2179		if (cap == ci->i_auth_cap &&
2180		    (cap->issued & CEPH_CAP_FILE_WR)) {
2181			/* request larger max_size from MDS? */
2182			if (ci->i_wanted_max_size > ci->i_max_size &&
2183			    ci->i_wanted_max_size > ci->i_requested_max_size) {
2184				doutc(cl, "requesting new max_size\n");
2185				goto ack;
2186			}
2187
2188			/* approaching file_max? */
2189			if (__ceph_should_report_size(ci)) {
2190				doutc(cl, "i_size approaching max_size\n");
2191				goto ack;
2192			}
2193		}
2194		/* flush anything dirty? */
2195		if (cap == ci->i_auth_cap) {
2196			if ((flags & CHECK_CAPS_FLUSH) && ci->i_dirty_caps) {
2197				doutc(cl, "flushing dirty caps\n");
2198				goto ack;
2199			}
2200			if (ci->i_ceph_flags & CEPH_I_FLUSH_SNAPS) {
2201				doutc(cl, "flushing snap caps\n");
2202				goto ack;
2203			}
2204		}
2205
2206		/* want more caps from mds? */
2207		if (want & ~cap->mds_wanted) {
2208			if (want & ~(cap->mds_wanted | cap->issued))
2209				goto ack;
2210			if (!__cap_is_valid(cap))
2211				goto ack;
2212		}
2213
2214		/* things we might delay */
2215		if ((cap->issued & ~retain) == 0)
2216			continue;     /* nope, all good */
2217
2218ack:
2219		ceph_put_mds_session(session);
2220		session = ceph_get_mds_session(cap->session);
2221
2222		/* kick flushing and flush snaps before sending normal
2223		 * cap message */
2224		if (cap == ci->i_auth_cap &&
2225		    (ci->i_ceph_flags &
2226		     (CEPH_I_KICK_FLUSH | CEPH_I_FLUSH_SNAPS))) {
2227			if (ci->i_ceph_flags & CEPH_I_KICK_FLUSH)
2228				__kick_flushing_caps(mdsc, session, ci, 0);
2229			if (ci->i_ceph_flags & CEPH_I_FLUSH_SNAPS)
2230				__ceph_flush_snaps(ci, session);
2231
2232			goto retry;
2233		}
2234
2235		if (cap == ci->i_auth_cap && ci->i_dirty_caps) {
2236			flushing = ci->i_dirty_caps;
2237			flush_tid = __mark_caps_flushing(inode, session, false,
2238							 &oldest_flush_tid);
2239			if (flags & CHECK_CAPS_FLUSH &&
2240			    list_empty(&session->s_cap_dirty))
2241				mflags |= CEPH_CLIENT_CAPS_SYNC;
2242		} else {
2243			flushing = 0;
2244			flush_tid = 0;
2245			spin_lock(&mdsc->cap_dirty_lock);
2246			oldest_flush_tid = __get_oldest_flush_tid(mdsc);
2247			spin_unlock(&mdsc->cap_dirty_lock);
2248		}
2249
2250		mds = cap->mds;  /* remember mds, so we don't repeat */
2251
2252		__prep_cap(&arg, cap, CEPH_CAP_OP_UPDATE, mflags, cap_used,
2253			   want, retain, flushing, flush_tid, oldest_flush_tid);
2254
2255		spin_unlock(&ci->i_ceph_lock);
2256		__send_cap(&arg, ci);
2257		spin_lock(&ci->i_ceph_lock);
2258
2259		goto retry; /* retake i_ceph_lock and restart our cap scan. */
2260	}
2261
2262	/* periodically re-calculate caps wanted by open files */
2263	if (__ceph_is_any_real_caps(ci) &&
2264	    list_empty(&ci->i_cap_delay_list) &&
2265	    (file_wanted & ~CEPH_CAP_PIN) &&
2266	    !(used & (CEPH_CAP_FILE_RD | CEPH_CAP_ANY_FILE_WR))) {
2267		__cap_delay_requeue(mdsc, ci);
2268	}
2269
2270	spin_unlock(&ci->i_ceph_lock);
2271
2272	ceph_put_mds_session(session);
2273	if (queue_writeback)
2274		ceph_queue_writeback(inode);
2275	if (queue_invalidate)
2276		ceph_queue_invalidate(inode);
2277}
2278
2279/*
2280 * Try to flush dirty caps back to the auth mds.
2281 */
2282static int try_flush_caps(struct inode *inode, u64 *ptid)
2283{
2284	struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(inode->i_sb)->mdsc;
2285	struct ceph_inode_info *ci = ceph_inode(inode);
2286	int flushing = 0;
2287	u64 flush_tid = 0, oldest_flush_tid = 0;
2288
2289	spin_lock(&ci->i_ceph_lock);
2290retry_locked:
2291	if (ci->i_dirty_caps && ci->i_auth_cap) {
2292		struct ceph_cap *cap = ci->i_auth_cap;
2293		struct cap_msg_args arg;
2294		struct ceph_mds_session *session = cap->session;
2295
2296		if (session->s_state < CEPH_MDS_SESSION_OPEN) {
2297			spin_unlock(&ci->i_ceph_lock);
2298			goto out;
2299		}
2300
2301		if (ci->i_ceph_flags &
2302		    (CEPH_I_KICK_FLUSH | CEPH_I_FLUSH_SNAPS)) {
2303			if (ci->i_ceph_flags & CEPH_I_KICK_FLUSH)
2304				__kick_flushing_caps(mdsc, session, ci, 0);
2305			if (ci->i_ceph_flags & CEPH_I_FLUSH_SNAPS)
2306				__ceph_flush_snaps(ci, session);
2307			goto retry_locked;
2308		}
2309
2310		flushing = ci->i_dirty_caps;
2311		flush_tid = __mark_caps_flushing(inode, session, true,
2312						 &oldest_flush_tid);
2313
2314		__prep_cap(&arg, cap, CEPH_CAP_OP_FLUSH, CEPH_CLIENT_CAPS_SYNC,
2315			   __ceph_caps_used(ci), __ceph_caps_wanted(ci),
2316			   (cap->issued | cap->implemented),
2317			   flushing, flush_tid, oldest_flush_tid);
2318		spin_unlock(&ci->i_ceph_lock);
2319
2320		__send_cap(&arg, ci);
2321	} else {
2322		if (!list_empty(&ci->i_cap_flush_list)) {
2323			struct ceph_cap_flush *cf =
2324				list_last_entry(&ci->i_cap_flush_list,
2325						struct ceph_cap_flush, i_list);
2326			cf->wake = true;
2327			flush_tid = cf->tid;
2328		}
2329		flushing = ci->i_flushing_caps;
2330		spin_unlock(&ci->i_ceph_lock);
2331	}
2332out:
2333	*ptid = flush_tid;
2334	return flushing;
2335}
2336
2337/*
2338 * Return true if we've flushed caps through the given flush_tid.
2339 */
2340static int caps_are_flushed(struct inode *inode, u64 flush_tid)
2341{
2342	struct ceph_inode_info *ci = ceph_inode(inode);
2343	int ret = 1;
2344
2345	spin_lock(&ci->i_ceph_lock);
2346	if (!list_empty(&ci->i_cap_flush_list)) {
2347		struct ceph_cap_flush * cf =
2348			list_first_entry(&ci->i_cap_flush_list,
2349					 struct ceph_cap_flush, i_list);
2350		if (cf->tid <= flush_tid)
2351			ret = 0;
2352	}
2353	spin_unlock(&ci->i_ceph_lock);
2354	return ret;
2355}
2356
2357/*
2358 * flush the mdlog and wait for any unsafe requests to complete.
2359 */
2360static int flush_mdlog_and_wait_inode_unsafe_requests(struct inode *inode)
2361{
2362	struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(inode->i_sb)->mdsc;
2363	struct ceph_client *cl = ceph_inode_to_client(inode);
2364	struct ceph_inode_info *ci = ceph_inode(inode);
2365	struct ceph_mds_request *req1 = NULL, *req2 = NULL;
2366	int ret, err = 0;
2367
2368	spin_lock(&ci->i_unsafe_lock);
2369	if (S_ISDIR(inode->i_mode) && !list_empty(&ci->i_unsafe_dirops)) {
2370		req1 = list_last_entry(&ci->i_unsafe_dirops,
2371					struct ceph_mds_request,
2372					r_unsafe_dir_item);
2373		ceph_mdsc_get_request(req1);
2374	}
2375	if (!list_empty(&ci->i_unsafe_iops)) {
2376		req2 = list_last_entry(&ci->i_unsafe_iops,
2377					struct ceph_mds_request,
2378					r_unsafe_target_item);
2379		ceph_mdsc_get_request(req2);
2380	}
2381	spin_unlock(&ci->i_unsafe_lock);
2382
2383	/*
2384	 * Trigger to flush the journal logs in all the relevant MDSes
2385	 * manually, or in the worst case we must wait at most 5 seconds
2386	 * to wait the journal logs to be flushed by the MDSes periodically.
2387	 */
2388	if (req1 || req2) {
2389		struct ceph_mds_request *req;
2390		struct ceph_mds_session **sessions;
2391		struct ceph_mds_session *s;
2392		unsigned int max_sessions;
2393		int i;
2394
2395		mutex_lock(&mdsc->mutex);
2396		max_sessions = mdsc->max_sessions;
2397
2398		sessions = kcalloc(max_sessions, sizeof(s), GFP_KERNEL);
2399		if (!sessions) {
2400			mutex_unlock(&mdsc->mutex);
2401			err = -ENOMEM;
2402			goto out;
2403		}
2404
2405		spin_lock(&ci->i_unsafe_lock);
2406		if (req1) {
2407			list_for_each_entry(req, &ci->i_unsafe_dirops,
2408					    r_unsafe_dir_item) {
2409				s = req->r_session;
2410				if (!s)
2411					continue;
2412				if (!sessions[s->s_mds]) {
2413					s = ceph_get_mds_session(s);
2414					sessions[s->s_mds] = s;
2415				}
2416			}
2417		}
2418		if (req2) {
2419			list_for_each_entry(req, &ci->i_unsafe_iops,
2420					    r_unsafe_target_item) {
2421				s = req->r_session;
2422				if (!s)
2423					continue;
2424				if (!sessions[s->s_mds]) {
2425					s = ceph_get_mds_session(s);
2426					sessions[s->s_mds] = s;
2427				}
2428			}
2429		}
2430		spin_unlock(&ci->i_unsafe_lock);
2431
2432		/* the auth MDS */
2433		spin_lock(&ci->i_ceph_lock);
2434		if (ci->i_auth_cap) {
2435			s = ci->i_auth_cap->session;
2436			if (!sessions[s->s_mds])
2437				sessions[s->s_mds] = ceph_get_mds_session(s);
2438		}
2439		spin_unlock(&ci->i_ceph_lock);
2440		mutex_unlock(&mdsc->mutex);
2441
2442		/* send flush mdlog request to MDSes */
2443		for (i = 0; i < max_sessions; i++) {
2444			s = sessions[i];
2445			if (s) {
2446				send_flush_mdlog(s);
2447				ceph_put_mds_session(s);
2448			}
2449		}
2450		kfree(sessions);
2451	}
2452
2453	doutc(cl, "%p %llx.%llx wait on tid %llu %llu\n", inode,
2454	      ceph_vinop(inode), req1 ? req1->r_tid : 0ULL,
2455	      req2 ? req2->r_tid : 0ULL);
2456	if (req1) {
2457		ret = !wait_for_completion_timeout(&req1->r_safe_completion,
2458					ceph_timeout_jiffies(req1->r_timeout));
2459		if (ret)
2460			err = -EIO;
2461	}
2462	if (req2) {
2463		ret = !wait_for_completion_timeout(&req2->r_safe_completion,
2464					ceph_timeout_jiffies(req2->r_timeout));
2465		if (ret)
2466			err = -EIO;
2467	}
2468
2469out:
2470	if (req1)
2471		ceph_mdsc_put_request(req1);
2472	if (req2)
2473		ceph_mdsc_put_request(req2);
2474	return err;
2475}
2476
2477int ceph_fsync(struct file *file, loff_t start, loff_t end, int datasync)
2478{
2479	struct inode *inode = file->f_mapping->host;
2480	struct ceph_inode_info *ci = ceph_inode(inode);
2481	struct ceph_client *cl = ceph_inode_to_client(inode);
2482	u64 flush_tid;
2483	int ret, err;
2484	int dirty;
2485
2486	doutc(cl, "%p %llx.%llx%s\n", inode, ceph_vinop(inode),
2487	      datasync ? " datasync" : "");
2488
2489	ret = file_write_and_wait_range(file, start, end);
2490	if (datasync)
2491		goto out;
2492
2493	ret = ceph_wait_on_async_create(inode);
2494	if (ret)
2495		goto out;
2496
2497	dirty = try_flush_caps(inode, &flush_tid);
2498	doutc(cl, "dirty caps are %s\n", ceph_cap_string(dirty));
2499
2500	err = flush_mdlog_and_wait_inode_unsafe_requests(inode);
2501
2502	/*
2503	 * only wait on non-file metadata writeback (the mds
2504	 * can recover size and mtime, so we don't need to
2505	 * wait for that)
2506	 */
2507	if (!err && (dirty & ~CEPH_CAP_ANY_FILE_WR)) {
2508		err = wait_event_interruptible(ci->i_cap_wq,
2509					caps_are_flushed(inode, flush_tid));
2510	}
2511
2512	if (err < 0)
2513		ret = err;
2514
2515	err = file_check_and_advance_wb_err(file);
2516	if (err < 0)
2517		ret = err;
2518out:
2519	doutc(cl, "%p %llx.%llx%s result=%d\n", inode, ceph_vinop(inode),
2520	      datasync ? " datasync" : "", ret);
2521	return ret;
2522}
2523
2524/*
2525 * Flush any dirty caps back to the mds.  If we aren't asked to wait,
2526 * queue inode for flush but don't do so immediately, because we can
2527 * get by with fewer MDS messages if we wait for data writeback to
2528 * complete first.
2529 */
2530int ceph_write_inode(struct inode *inode, struct writeback_control *wbc)
2531{
2532	struct ceph_inode_info *ci = ceph_inode(inode);
2533	struct ceph_client *cl = ceph_inode_to_client(inode);
2534	u64 flush_tid;
2535	int err = 0;
2536	int dirty;
2537	int wait = (wbc->sync_mode == WB_SYNC_ALL && !wbc->for_sync);
2538
2539	doutc(cl, "%p %llx.%llx wait=%d\n", inode, ceph_vinop(inode), wait);
2540	ceph_fscache_unpin_writeback(inode, wbc);
2541	if (wait) {
2542		err = ceph_wait_on_async_create(inode);
2543		if (err)
2544			return err;
2545		dirty = try_flush_caps(inode, &flush_tid);
2546		if (dirty)
2547			err = wait_event_interruptible(ci->i_cap_wq,
2548				       caps_are_flushed(inode, flush_tid));
2549	} else {
2550		struct ceph_mds_client *mdsc =
2551			ceph_sb_to_fs_client(inode->i_sb)->mdsc;
2552
2553		spin_lock(&ci->i_ceph_lock);
2554		if (__ceph_caps_dirty(ci))
2555			__cap_delay_requeue_front(mdsc, ci);
2556		spin_unlock(&ci->i_ceph_lock);
2557	}
2558	return err;
2559}
2560
2561static void __kick_flushing_caps(struct ceph_mds_client *mdsc,
2562				 struct ceph_mds_session *session,
2563				 struct ceph_inode_info *ci,
2564				 u64 oldest_flush_tid)
2565	__releases(ci->i_ceph_lock)
2566	__acquires(ci->i_ceph_lock)
2567{
2568	struct inode *inode = &ci->netfs.inode;
2569	struct ceph_client *cl = mdsc->fsc->client;
2570	struct ceph_cap *cap;
2571	struct ceph_cap_flush *cf;
2572	int ret;
2573	u64 first_tid = 0;
2574	u64 last_snap_flush = 0;
2575
2576	/* Don't do anything until create reply comes in */
2577	if (ci->i_ceph_flags & CEPH_I_ASYNC_CREATE)
2578		return;
2579
2580	ci->i_ceph_flags &= ~CEPH_I_KICK_FLUSH;
2581
2582	list_for_each_entry_reverse(cf, &ci->i_cap_flush_list, i_list) {
2583		if (cf->is_capsnap) {
2584			last_snap_flush = cf->tid;
2585			break;
2586		}
2587	}
2588
2589	list_for_each_entry(cf, &ci->i_cap_flush_list, i_list) {
2590		if (cf->tid < first_tid)
2591			continue;
2592
2593		cap = ci->i_auth_cap;
2594		if (!(cap && cap->session == session)) {
2595			pr_err_client(cl, "%p auth cap %p not mds%d ???\n",
2596				      inode, cap, session->s_mds);
2597			break;
2598		}
2599
2600		first_tid = cf->tid + 1;
2601
2602		if (!cf->is_capsnap) {
2603			struct cap_msg_args arg;
2604
2605			doutc(cl, "%p %llx.%llx cap %p tid %llu %s\n",
2606			      inode, ceph_vinop(inode), cap, cf->tid,
2607			      ceph_cap_string(cf->caps));
2608			__prep_cap(&arg, cap, CEPH_CAP_OP_FLUSH,
2609					 (cf->tid < last_snap_flush ?
2610					  CEPH_CLIENT_CAPS_PENDING_CAPSNAP : 0),
2611					  __ceph_caps_used(ci),
2612					  __ceph_caps_wanted(ci),
2613					  (cap->issued | cap->implemented),
2614					  cf->caps, cf->tid, oldest_flush_tid);
2615			spin_unlock(&ci->i_ceph_lock);
2616			__send_cap(&arg, ci);
2617		} else {
2618			struct ceph_cap_snap *capsnap =
2619					container_of(cf, struct ceph_cap_snap,
2620						    cap_flush);
2621			doutc(cl, "%p %llx.%llx capsnap %p tid %llu %s\n",
2622			      inode, ceph_vinop(inode), capsnap, cf->tid,
2623			      ceph_cap_string(capsnap->dirty));
2624
2625			refcount_inc(&capsnap->nref);
2626			spin_unlock(&ci->i_ceph_lock);
2627
2628			ret = __send_flush_snap(inode, session, capsnap, cap->mseq,
2629						oldest_flush_tid);
2630			if (ret < 0) {
2631				pr_err_client(cl, "error sending cap flushsnap,"
2632					      " %p %llx.%llx tid %llu follows %llu\n",
2633					      inode, ceph_vinop(inode), cf->tid,
2634					      capsnap->follows);
2635			}
2636
2637			ceph_put_cap_snap(capsnap);
2638		}
2639
2640		spin_lock(&ci->i_ceph_lock);
2641	}
2642}
2643
2644void ceph_early_kick_flushing_caps(struct ceph_mds_client *mdsc,
2645				   struct ceph_mds_session *session)
2646{
2647	struct ceph_client *cl = mdsc->fsc->client;
2648	struct ceph_inode_info *ci;
2649	struct ceph_cap *cap;
2650	u64 oldest_flush_tid;
2651
2652	doutc(cl, "mds%d\n", session->s_mds);
2653
2654	spin_lock(&mdsc->cap_dirty_lock);
2655	oldest_flush_tid = __get_oldest_flush_tid(mdsc);
2656	spin_unlock(&mdsc->cap_dirty_lock);
2657
2658	list_for_each_entry(ci, &session->s_cap_flushing, i_flushing_item) {
2659		struct inode *inode = &ci->netfs.inode;
2660
2661		spin_lock(&ci->i_ceph_lock);
2662		cap = ci->i_auth_cap;
2663		if (!(cap && cap->session == session)) {
2664			pr_err_client(cl, "%p %llx.%llx auth cap %p not mds%d ???\n",
2665				      inode, ceph_vinop(inode), cap,
2666				      session->s_mds);
2667			spin_unlock(&ci->i_ceph_lock);
2668			continue;
2669		}
2670
2671
2672		/*
2673		 * if flushing caps were revoked, we re-send the cap flush
2674		 * in client reconnect stage. This guarantees MDS * processes
2675		 * the cap flush message before issuing the flushing caps to
2676		 * other client.
2677		 */
2678		if ((cap->issued & ci->i_flushing_caps) !=
2679		    ci->i_flushing_caps) {
2680			/* encode_caps_cb() also will reset these sequence
2681			 * numbers. make sure sequence numbers in cap flush
2682			 * message match later reconnect message */
2683			cap->seq = 0;
2684			cap->issue_seq = 0;
2685			cap->mseq = 0;
2686			__kick_flushing_caps(mdsc, session, ci,
2687					     oldest_flush_tid);
2688		} else {
2689			ci->i_ceph_flags |= CEPH_I_KICK_FLUSH;
2690		}
2691
2692		spin_unlock(&ci->i_ceph_lock);
2693	}
2694}
2695
2696void ceph_kick_flushing_caps(struct ceph_mds_client *mdsc,
2697			     struct ceph_mds_session *session)
2698{
2699	struct ceph_client *cl = mdsc->fsc->client;
2700	struct ceph_inode_info *ci;
2701	struct ceph_cap *cap;
2702	u64 oldest_flush_tid;
2703
2704	lockdep_assert_held(&session->s_mutex);
2705
2706	doutc(cl, "mds%d\n", session->s_mds);
2707
2708	spin_lock(&mdsc->cap_dirty_lock);
2709	oldest_flush_tid = __get_oldest_flush_tid(mdsc);
2710	spin_unlock(&mdsc->cap_dirty_lock);
2711
2712	list_for_each_entry(ci, &session->s_cap_flushing, i_flushing_item) {
2713		struct inode *inode = &ci->netfs.inode;
2714
2715		spin_lock(&ci->i_ceph_lock);
2716		cap = ci->i_auth_cap;
2717		if (!(cap && cap->session == session)) {
2718			pr_err_client(cl, "%p %llx.%llx auth cap %p not mds%d ???\n",
2719				      inode, ceph_vinop(inode), cap,
2720				      session->s_mds);
2721			spin_unlock(&ci->i_ceph_lock);
2722			continue;
2723		}
2724		if (ci->i_ceph_flags & CEPH_I_KICK_FLUSH) {
2725			__kick_flushing_caps(mdsc, session, ci,
2726					     oldest_flush_tid);
2727		}
2728		spin_unlock(&ci->i_ceph_lock);
2729	}
2730}
2731
2732void ceph_kick_flushing_inode_caps(struct ceph_mds_session *session,
2733				   struct ceph_inode_info *ci)
2734{
2735	struct ceph_mds_client *mdsc = session->s_mdsc;
2736	struct ceph_cap *cap = ci->i_auth_cap;
2737	struct inode *inode = &ci->netfs.inode;
2738
2739	lockdep_assert_held(&ci->i_ceph_lock);
2740
2741	doutc(mdsc->fsc->client, "%p %llx.%llx flushing %s\n",
2742	      inode, ceph_vinop(inode),
2743	      ceph_cap_string(ci->i_flushing_caps));
2744
2745	if (!list_empty(&ci->i_cap_flush_list)) {
2746		u64 oldest_flush_tid;
2747		spin_lock(&mdsc->cap_dirty_lock);
2748		list_move_tail(&ci->i_flushing_item,
2749			       &cap->session->s_cap_flushing);
2750		oldest_flush_tid = __get_oldest_flush_tid(mdsc);
2751		spin_unlock(&mdsc->cap_dirty_lock);
2752
2753		__kick_flushing_caps(mdsc, session, ci, oldest_flush_tid);
2754	}
2755}
2756
2757
2758/*
2759 * Take references to capabilities we hold, so that we don't release
2760 * them to the MDS prematurely.
2761 */
2762void ceph_take_cap_refs(struct ceph_inode_info *ci, int got,
2763			    bool snap_rwsem_locked)
2764{
2765	struct inode *inode = &ci->netfs.inode;
2766	struct ceph_client *cl = ceph_inode_to_client(inode);
2767
2768	lockdep_assert_held(&ci->i_ceph_lock);
2769
2770	if (got & CEPH_CAP_PIN)
2771		ci->i_pin_ref++;
2772	if (got & CEPH_CAP_FILE_RD)
2773		ci->i_rd_ref++;
2774	if (got & CEPH_CAP_FILE_CACHE)
2775		ci->i_rdcache_ref++;
2776	if (got & CEPH_CAP_FILE_EXCL)
2777		ci->i_fx_ref++;
2778	if (got & CEPH_CAP_FILE_WR) {
2779		if (ci->i_wr_ref == 0 && !ci->i_head_snapc) {
2780			BUG_ON(!snap_rwsem_locked);
2781			ci->i_head_snapc = ceph_get_snap_context(
2782					ci->i_snap_realm->cached_context);
2783		}
2784		ci->i_wr_ref++;
2785	}
2786	if (got & CEPH_CAP_FILE_BUFFER) {
2787		if (ci->i_wb_ref == 0)
2788			ihold(inode);
2789		ci->i_wb_ref++;
2790		doutc(cl, "%p %llx.%llx wb %d -> %d (?)\n", inode,
2791		      ceph_vinop(inode), ci->i_wb_ref-1, ci->i_wb_ref);
2792	}
2793}
2794
2795/*
2796 * Try to grab cap references.  Specify those refs we @want, and the
2797 * minimal set we @need.  Also include the larger offset we are writing
2798 * to (when applicable), and check against max_size here as well.
2799 * Note that caller is responsible for ensuring max_size increases are
2800 * requested from the MDS.
2801 *
2802 * Returns 0 if caps were not able to be acquired (yet), 1 if succeed,
2803 * or a negative error code. There are 3 special error codes:
2804 *  -EAGAIN:  need to sleep but non-blocking is specified
2805 *  -EFBIG:   ask caller to call check_max_size() and try again.
2806 *  -EUCLEAN: ask caller to call ceph_renew_caps() and try again.
2807 */
2808enum {
2809	/* first 8 bits are reserved for CEPH_FILE_MODE_FOO */
2810	NON_BLOCKING	= (1 << 8),
2811	CHECK_FILELOCK	= (1 << 9),
2812};
2813
2814static int try_get_cap_refs(struct inode *inode, int need, int want,
2815			    loff_t endoff, int flags, int *got)
2816{
2817	struct ceph_inode_info *ci = ceph_inode(inode);
2818	struct ceph_mds_client *mdsc = ceph_inode_to_fs_client(inode)->mdsc;
2819	struct ceph_client *cl = ceph_inode_to_client(inode);
2820	int ret = 0;
2821	int have, implemented;
2822	bool snap_rwsem_locked = false;
2823
2824	doutc(cl, "%p %llx.%llx need %s want %s\n", inode,
2825	      ceph_vinop(inode), ceph_cap_string(need),
2826	      ceph_cap_string(want));
2827
2828again:
2829	spin_lock(&ci->i_ceph_lock);
2830
2831	if ((flags & CHECK_FILELOCK) &&
2832	    (ci->i_ceph_flags & CEPH_I_ERROR_FILELOCK)) {
2833		doutc(cl, "%p %llx.%llx error filelock\n", inode,
2834		      ceph_vinop(inode));
2835		ret = -EIO;
2836		goto out_unlock;
2837	}
2838
2839	/* finish pending truncate */
2840	while (ci->i_truncate_pending) {
2841		spin_unlock(&ci->i_ceph_lock);
2842		if (snap_rwsem_locked) {
2843			up_read(&mdsc->snap_rwsem);
2844			snap_rwsem_locked = false;
2845		}
2846		__ceph_do_pending_vmtruncate(inode);
2847		spin_lock(&ci->i_ceph_lock);
2848	}
2849
2850	have = __ceph_caps_issued(ci, &implemented);
2851
2852	if (have & need & CEPH_CAP_FILE_WR) {
2853		if (endoff >= 0 && endoff > (loff_t)ci->i_max_size) {
2854			doutc(cl, "%p %llx.%llx endoff %llu > maxsize %llu\n",
2855			      inode, ceph_vinop(inode), endoff, ci->i_max_size);
2856			if (endoff > ci->i_requested_max_size)
2857				ret = ci->i_auth_cap ? -EFBIG : -EUCLEAN;
2858			goto out_unlock;
2859		}
2860		/*
2861		 * If a sync write is in progress, we must wait, so that we
2862		 * can get a final snapshot value for size+mtime.
2863		 */
2864		if (__ceph_have_pending_cap_snap(ci)) {
2865			doutc(cl, "%p %llx.%llx cap_snap_pending\n", inode,
2866			      ceph_vinop(inode));
2867			goto out_unlock;
2868		}
2869	}
2870
2871	if ((have & need) == need) {
2872		/*
2873		 * Look at (implemented & ~have & not) so that we keep waiting
2874		 * on transition from wanted -> needed caps.  This is needed
2875		 * for WRBUFFER|WR -> WR to avoid a new WR sync write from
2876		 * going before a prior buffered writeback happens.
2877		 *
2878		 * For RDCACHE|RD -> RD, there is not need to wait and we can
2879		 * just exclude the revoking caps and force to sync read.
2880		 */
2881		int not = want & ~(have & need);
2882		int revoking = implemented & ~have;
2883		int exclude = revoking & not;
2884		doutc(cl, "%p %llx.%llx have %s but not %s (revoking %s)\n",
2885		      inode, ceph_vinop(inode), ceph_cap_string(have),
2886		      ceph_cap_string(not), ceph_cap_string(revoking));
2887		if (!exclude || !(exclude & CEPH_CAP_FILE_BUFFER)) {
2888			if (!snap_rwsem_locked &&
2889			    !ci->i_head_snapc &&
2890			    (need & CEPH_CAP_FILE_WR)) {
2891				if (!down_read_trylock(&mdsc->snap_rwsem)) {
2892					/*
2893					 * we can not call down_read() when
2894					 * task isn't in TASK_RUNNING state
2895					 */
2896					if (flags & NON_BLOCKING) {
2897						ret = -EAGAIN;
2898						goto out_unlock;
2899					}
2900
2901					spin_unlock(&ci->i_ceph_lock);
2902					down_read(&mdsc->snap_rwsem);
2903					snap_rwsem_locked = true;
2904					goto again;
2905				}
2906				snap_rwsem_locked = true;
2907			}
2908			if ((have & want) == want)
2909				*got = need | (want & ~exclude);
2910			else
2911				*got = need;
2912			ceph_take_cap_refs(ci, *got, true);
2913			ret = 1;
2914		}
2915	} else {
2916		int session_readonly = false;
2917		int mds_wanted;
2918		if (ci->i_auth_cap &&
2919		    (need & (CEPH_CAP_FILE_WR | CEPH_CAP_FILE_EXCL))) {
2920			struct ceph_mds_session *s = ci->i_auth_cap->session;
2921			spin_lock(&s->s_cap_lock);
2922			session_readonly = s->s_readonly;
2923			spin_unlock(&s->s_cap_lock);
2924		}
2925		if (session_readonly) {
2926			doutc(cl, "%p %llx.%llx need %s but mds%d readonly\n",
2927			      inode, ceph_vinop(inode), ceph_cap_string(need),
2928			      ci->i_auth_cap->mds);
2929			ret = -EROFS;
2930			goto out_unlock;
2931		}
2932
2933		if (ceph_inode_is_shutdown(inode)) {
2934			doutc(cl, "%p %llx.%llx inode is shutdown\n",
2935			      inode, ceph_vinop(inode));
2936			ret = -ESTALE;
2937			goto out_unlock;
2938		}
2939		mds_wanted = __ceph_caps_mds_wanted(ci, false);
2940		if (need & ~mds_wanted) {
2941			doutc(cl, "%p %llx.%llx need %s > mds_wanted %s\n",
2942			      inode, ceph_vinop(inode), ceph_cap_string(need),
2943			      ceph_cap_string(mds_wanted));
2944			ret = -EUCLEAN;
2945			goto out_unlock;
2946		}
2947
2948		doutc(cl, "%p %llx.%llx have %s need %s\n", inode,
2949		      ceph_vinop(inode), ceph_cap_string(have),
2950		      ceph_cap_string(need));
2951	}
2952out_unlock:
2953
2954	__ceph_touch_fmode(ci, mdsc, flags);
2955
2956	spin_unlock(&ci->i_ceph_lock);
2957	if (snap_rwsem_locked)
2958		up_read(&mdsc->snap_rwsem);
2959
2960	if (!ret)
2961		ceph_update_cap_mis(&mdsc->metric);
2962	else if (ret == 1)
2963		ceph_update_cap_hit(&mdsc->metric);
2964
2965	doutc(cl, "%p %llx.%llx ret %d got %s\n", inode,
2966	      ceph_vinop(inode), ret, ceph_cap_string(*got));
2967	return ret;
2968}
2969
2970/*
2971 * Check the offset we are writing up to against our current
2972 * max_size.  If necessary, tell the MDS we want to write to
2973 * a larger offset.
2974 */
2975static void check_max_size(struct inode *inode, loff_t endoff)
2976{
2977	struct ceph_inode_info *ci = ceph_inode(inode);
2978	struct ceph_client *cl = ceph_inode_to_client(inode);
2979	int check = 0;
2980
2981	/* do we need to explicitly request a larger max_size? */
2982	spin_lock(&ci->i_ceph_lock);
2983	if (endoff >= ci->i_max_size && endoff > ci->i_wanted_max_size) {
2984		doutc(cl, "write %p %llx.%llx at large endoff %llu, req max_size\n",
2985		      inode, ceph_vinop(inode), endoff);
2986		ci->i_wanted_max_size = endoff;
2987	}
2988	/* duplicate ceph_check_caps()'s logic */
2989	if (ci->i_auth_cap &&
2990	    (ci->i_auth_cap->issued & CEPH_CAP_FILE_WR) &&
2991	    ci->i_wanted_max_size > ci->i_max_size &&
2992	    ci->i_wanted_max_size > ci->i_requested_max_size)
2993		check = 1;
2994	spin_unlock(&ci->i_ceph_lock);
2995	if (check)
2996		ceph_check_caps(ci, CHECK_CAPS_AUTHONLY);
2997}
2998
2999static inline int get_used_fmode(int caps)
3000{
3001	int fmode = 0;
3002	if (caps & CEPH_CAP_FILE_RD)
3003		fmode |= CEPH_FILE_MODE_RD;
3004	if (caps & CEPH_CAP_FILE_WR)
3005		fmode |= CEPH_FILE_MODE_WR;
3006	return fmode;
3007}
3008
3009int ceph_try_get_caps(struct inode *inode, int need, int want,
3010		      bool nonblock, int *got)
3011{
3012	int ret, flags;
3013
3014	BUG_ON(need & ~CEPH_CAP_FILE_RD);
3015	BUG_ON(want & ~(CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO |
3016			CEPH_CAP_FILE_SHARED | CEPH_CAP_FILE_EXCL |
3017			CEPH_CAP_ANY_DIR_OPS));
3018	if (need) {
3019		ret = ceph_pool_perm_check(inode, need);
3020		if (ret < 0)
3021			return ret;
3022	}
3023
3024	flags = get_used_fmode(need | want);
3025	if (nonblock)
3026		flags |= NON_BLOCKING;
3027
3028	ret = try_get_cap_refs(inode, need, want, 0, flags, got);
3029	/* three special error codes */
3030	if (ret == -EAGAIN || ret == -EFBIG || ret == -EUCLEAN)
3031		ret = 0;
3032	return ret;
3033}
3034
3035/*
3036 * Wait for caps, and take cap references.  If we can't get a WR cap
3037 * due to a small max_size, make sure we check_max_size (and possibly
3038 * ask the mds) so we don't get hung up indefinitely.
3039 */
3040int __ceph_get_caps(struct inode *inode, struct ceph_file_info *fi, int need,
3041		    int want, loff_t endoff, int *got)
3042{
3043	struct ceph_inode_info *ci = ceph_inode(inode);
3044	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
3045	int ret, _got, flags;
3046
3047	ret = ceph_pool_perm_check(inode, need);
3048	if (ret < 0)
3049		return ret;
3050
3051	if (fi && (fi->fmode & CEPH_FILE_MODE_WR) &&
3052	    fi->filp_gen != READ_ONCE(fsc->filp_gen))
3053		return -EBADF;
3054
3055	flags = get_used_fmode(need | want);
3056
3057	while (true) {
3058		flags &= CEPH_FILE_MODE_MASK;
3059		if (vfs_inode_has_locks(inode))
3060			flags |= CHECK_FILELOCK;
3061		_got = 0;
3062		ret = try_get_cap_refs(inode, need, want, endoff,
3063				       flags, &_got);
3064		WARN_ON_ONCE(ret == -EAGAIN);
3065		if (!ret) {
3066#ifdef CONFIG_DEBUG_FS
3067			struct ceph_mds_client *mdsc = fsc->mdsc;
3068			struct cap_wait cw;
3069#endif
3070			DEFINE_WAIT_FUNC(wait, woken_wake_function);
3071
3072#ifdef CONFIG_DEBUG_FS
3073			cw.ino = ceph_ino(inode);
3074			cw.tgid = current->tgid;
3075			cw.need = need;
3076			cw.want = want;
3077
3078			spin_lock(&mdsc->caps_list_lock);
3079			list_add(&cw.list, &mdsc->cap_wait_list);
3080			spin_unlock(&mdsc->caps_list_lock);
3081#endif
3082
3083			/* make sure used fmode not timeout */
3084			ceph_get_fmode(ci, flags, FMODE_WAIT_BIAS);
3085			add_wait_queue(&ci->i_cap_wq, &wait);
3086
3087			flags |= NON_BLOCKING;
3088			while (!(ret = try_get_cap_refs(inode, need, want,
3089							endoff, flags, &_got))) {
3090				if (signal_pending(current)) {
3091					ret = -ERESTARTSYS;
3092					break;
3093				}
3094				wait_woken(&wait, TASK_INTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT);
3095			}
3096
3097			remove_wait_queue(&ci->i_cap_wq, &wait);
3098			ceph_put_fmode(ci, flags, FMODE_WAIT_BIAS);
3099
3100#ifdef CONFIG_DEBUG_FS
3101			spin_lock(&mdsc->caps_list_lock);
3102			list_del(&cw.list);
3103			spin_unlock(&mdsc->caps_list_lock);
3104#endif
3105
3106			if (ret == -EAGAIN)
3107				continue;
3108		}
3109
3110		if (fi && (fi->fmode & CEPH_FILE_MODE_WR) &&
3111		    fi->filp_gen != READ_ONCE(fsc->filp_gen)) {
3112			if (ret >= 0 && _got)
3113				ceph_put_cap_refs(ci, _got);
3114			return -EBADF;
3115		}
3116
3117		if (ret < 0) {
3118			if (ret == -EFBIG || ret == -EUCLEAN) {
3119				int ret2 = ceph_wait_on_async_create(inode);
3120				if (ret2 < 0)
3121					return ret2;
3122			}
3123			if (ret == -EFBIG) {
3124				check_max_size(inode, endoff);
3125				continue;
3126			}
3127			if (ret == -EUCLEAN) {
3128				/* session was killed, try renew caps */
3129				ret = ceph_renew_caps(inode, flags);
3130				if (ret == 0)
3131					continue;
3132			}
3133			return ret;
3134		}
3135
3136		if (S_ISREG(ci->netfs.inode.i_mode) &&
3137		    ceph_has_inline_data(ci) &&
3138		    (_got & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) &&
3139		    i_size_read(inode) > 0) {
3140			struct page *page =
3141				find_get_page(inode->i_mapping, 0);
3142			if (page) {
3143				bool uptodate = PageUptodate(page);
3144
3145				put_page(page);
3146				if (uptodate)
3147					break;
3148			}
3149			/*
3150			 * drop cap refs first because getattr while
3151			 * holding * caps refs can cause deadlock.
3152			 */
3153			ceph_put_cap_refs(ci, _got);
3154			_got = 0;
3155
3156			/*
3157			 * getattr request will bring inline data into
3158			 * page cache
3159			 */
3160			ret = __ceph_do_getattr(inode, NULL,
3161						CEPH_STAT_CAP_INLINE_DATA,
3162						true);
3163			if (ret < 0)
3164				return ret;
3165			continue;
3166		}
3167		break;
3168	}
3169	*got = _got;
3170	return 0;
3171}
3172
3173int ceph_get_caps(struct file *filp, int need, int want, loff_t endoff,
3174		  int *got)
3175{
3176	struct ceph_file_info *fi = filp->private_data;
3177	struct inode *inode = file_inode(filp);
3178
3179	return __ceph_get_caps(inode, fi, need, want, endoff, got);
3180}
3181
3182/*
3183 * Take cap refs.  Caller must already know we hold at least one ref
3184 * on the caps in question or we don't know this is safe.
3185 */
3186void ceph_get_cap_refs(struct ceph_inode_info *ci, int caps)
3187{
3188	spin_lock(&ci->i_ceph_lock);
3189	ceph_take_cap_refs(ci, caps, false);
3190	spin_unlock(&ci->i_ceph_lock);
3191}
3192
3193
3194/*
3195 * drop cap_snap that is not associated with any snapshot.
3196 * we don't need to send FLUSHSNAP message for it.
3197 */
3198static int ceph_try_drop_cap_snap(struct ceph_inode_info *ci,
3199				  struct ceph_cap_snap *capsnap)
3200{
3201	struct inode *inode = &ci->netfs.inode;
3202	struct ceph_client *cl = ceph_inode_to_client(inode);
3203
3204	if (!capsnap->need_flush &&
3205	    !capsnap->writing && !capsnap->dirty_pages) {
3206		doutc(cl, "%p follows %llu\n", capsnap, capsnap->follows);
3207		BUG_ON(capsnap->cap_flush.tid > 0);
3208		ceph_put_snap_context(capsnap->context);
3209		if (!list_is_last(&capsnap->ci_item, &ci->i_cap_snaps))
3210			ci->i_ceph_flags |= CEPH_I_FLUSH_SNAPS;
3211
3212		list_del(&capsnap->ci_item);
3213		ceph_put_cap_snap(capsnap);
3214		return 1;
3215	}
3216	return 0;
3217}
3218
3219enum put_cap_refs_mode {
3220	PUT_CAP_REFS_SYNC = 0,
3221	PUT_CAP_REFS_ASYNC,
3222};
3223
3224/*
3225 * Release cap refs.
3226 *
3227 * If we released the last ref on any given cap, call ceph_check_caps
3228 * to release (or schedule a release).
3229 *
3230 * If we are releasing a WR cap (from a sync write), finalize any affected
3231 * cap_snap, and wake up any waiters.
3232 */
3233static void __ceph_put_cap_refs(struct ceph_inode_info *ci, int had,
3234				enum put_cap_refs_mode mode)
3235{
3236	struct inode *inode = &ci->netfs.inode;
3237	struct ceph_client *cl = ceph_inode_to_client(inode);
3238	int last = 0, put = 0, flushsnaps = 0, wake = 0;
3239	bool check_flushsnaps = false;
3240
3241	spin_lock(&ci->i_ceph_lock);
3242	if (had & CEPH_CAP_PIN)
3243		--ci->i_pin_ref;
3244	if (had & CEPH_CAP_FILE_RD)
3245		if (--ci->i_rd_ref == 0)
3246			last++;
3247	if (had & CEPH_CAP_FILE_CACHE)
3248		if (--ci->i_rdcache_ref == 0)
3249			last++;
3250	if (had & CEPH_CAP_FILE_EXCL)
3251		if (--ci->i_fx_ref == 0)
3252			last++;
3253	if (had & CEPH_CAP_FILE_BUFFER) {
3254		if (--ci->i_wb_ref == 0) {
3255			last++;
3256			/* put the ref held by ceph_take_cap_refs() */
3257			put++;
3258			check_flushsnaps = true;
3259		}
3260		doutc(cl, "%p %llx.%llx wb %d -> %d (?)\n", inode,
3261		      ceph_vinop(inode), ci->i_wb_ref+1, ci->i_wb_ref);
3262	}
3263	if (had & CEPH_CAP_FILE_WR) {
3264		if (--ci->i_wr_ref == 0) {
3265			/*
3266			 * The Fb caps will always be took and released
3267			 * together with the Fw caps.
3268			 */
3269			WARN_ON_ONCE(ci->i_wb_ref);
3270
3271			last++;
3272			check_flushsnaps = true;
3273			if (ci->i_wrbuffer_ref_head == 0 &&
3274			    ci->i_dirty_caps == 0 &&
3275			    ci->i_flushing_caps == 0) {
3276				BUG_ON(!ci->i_head_snapc);
3277				ceph_put_snap_context(ci->i_head_snapc);
3278				ci->i_head_snapc = NULL;
3279			}
3280			/* see comment in __ceph_remove_cap() */
3281			if (!__ceph_is_any_real_caps(ci) && ci->i_snap_realm)
3282				ceph_change_snap_realm(inode, NULL);
3283		}
3284	}
3285	if (check_flushsnaps && __ceph_have_pending_cap_snap(ci)) {
3286		struct ceph_cap_snap *capsnap =
3287			list_last_entry(&ci->i_cap_snaps,
3288					struct ceph_cap_snap,
3289					ci_item);
3290
3291		capsnap->writing = 0;
3292		if (ceph_try_drop_cap_snap(ci, capsnap))
3293			/* put the ref held by ceph_queue_cap_snap() */
3294			put++;
3295		else if (__ceph_finish_cap_snap(ci, capsnap))
3296			flushsnaps = 1;
3297		wake = 1;
3298	}
3299	spin_unlock(&ci->i_ceph_lock);
3300
3301	doutc(cl, "%p %llx.%llx had %s%s%s\n", inode, ceph_vinop(inode),
3302	      ceph_cap_string(had), last ? " last" : "", put ? " put" : "");
3303
3304	switch (mode) {
3305	case PUT_CAP_REFS_SYNC:
3306		if (last)
3307			ceph_check_caps(ci, 0);
3308		else if (flushsnaps)
3309			ceph_flush_snaps(ci, NULL);
3310		break;
3311	case PUT_CAP_REFS_ASYNC:
3312		if (last)
3313			ceph_queue_check_caps(inode);
3314		else if (flushsnaps)
3315			ceph_queue_flush_snaps(inode);
3316		break;
3317	default:
3318		break;
3319	}
3320	if (wake)
3321		wake_up_all(&ci->i_cap_wq);
3322	while (put-- > 0)
3323		iput(inode);
3324}
3325
3326void ceph_put_cap_refs(struct ceph_inode_info *ci, int had)
3327{
3328	__ceph_put_cap_refs(ci, had, PUT_CAP_REFS_SYNC);
3329}
3330
3331void ceph_put_cap_refs_async(struct ceph_inode_info *ci, int had)
3332{
3333	__ceph_put_cap_refs(ci, had, PUT_CAP_REFS_ASYNC);
3334}
3335
3336/*
3337 * Release @nr WRBUFFER refs on dirty pages for the given @snapc snap
3338 * context.  Adjust per-snap dirty page accounting as appropriate.
3339 * Once all dirty data for a cap_snap is flushed, flush snapped file
3340 * metadata back to the MDS.  If we dropped the last ref, call
3341 * ceph_check_caps.
3342 */
3343void ceph_put_wrbuffer_cap_refs(struct ceph_inode_info *ci, int nr,
3344				struct ceph_snap_context *snapc)
3345{
3346	struct inode *inode = &ci->netfs.inode;
3347	struct ceph_client *cl = ceph_inode_to_client(inode);
3348	struct ceph_cap_snap *capsnap = NULL, *iter;
3349	int put = 0;
3350	bool last = false;
3351	bool flush_snaps = false;
3352	bool complete_capsnap = false;
3353
3354	spin_lock(&ci->i_ceph_lock);
3355	ci->i_wrbuffer_ref -= nr;
3356	if (ci->i_wrbuffer_ref == 0) {
3357		last = true;
3358		put++;
3359	}
3360
3361	if (ci->i_head_snapc == snapc) {
3362		ci->i_wrbuffer_ref_head -= nr;
3363		if (ci->i_wrbuffer_ref_head == 0 &&
3364		    ci->i_wr_ref == 0 &&
3365		    ci->i_dirty_caps == 0 &&
3366		    ci->i_flushing_caps == 0) {
3367			BUG_ON(!ci->i_head_snapc);
3368			ceph_put_snap_context(ci->i_head_snapc);
3369			ci->i_head_snapc = NULL;
3370		}
3371		doutc(cl, "on %p %llx.%llx head %d/%d -> %d/%d %s\n",
3372		      inode, ceph_vinop(inode), ci->i_wrbuffer_ref+nr,
3373		      ci->i_wrbuffer_ref_head+nr, ci->i_wrbuffer_ref,
3374		      ci->i_wrbuffer_ref_head, last ? " LAST" : "");
3375	} else {
3376		list_for_each_entry(iter, &ci->i_cap_snaps, ci_item) {
3377			if (iter->context == snapc) {
3378				capsnap = iter;
3379				break;
3380			}
3381		}
3382
3383		if (!capsnap) {
3384			/*
3385			 * The capsnap should already be removed when removing
3386			 * auth cap in the case of a forced unmount.
3387			 */
3388			WARN_ON_ONCE(ci->i_auth_cap);
3389			goto unlock;
3390		}
3391
3392		capsnap->dirty_pages -= nr;
3393		if (capsnap->dirty_pages == 0) {
3394			complete_capsnap = true;
3395			if (!capsnap->writing) {
3396				if (ceph_try_drop_cap_snap(ci, capsnap)) {
3397					put++;
3398				} else {
3399					ci->i_ceph_flags |= CEPH_I_FLUSH_SNAPS;
3400					flush_snaps = true;
3401				}
3402			}
3403		}
3404		doutc(cl, "%p %llx.%llx cap_snap %p snap %lld %d/%d -> %d/%d %s%s\n",
3405		      inode, ceph_vinop(inode), capsnap, capsnap->context->seq,
3406		      ci->i_wrbuffer_ref+nr, capsnap->dirty_pages + nr,
3407		      ci->i_wrbuffer_ref, capsnap->dirty_pages,
3408		      last ? " (wrbuffer last)" : "",
3409		      complete_capsnap ? " (complete capsnap)" : "");
3410	}
3411
3412unlock:
3413	spin_unlock(&ci->i_ceph_lock);
3414
3415	if (last) {
3416		ceph_check_caps(ci, 0);
3417	} else if (flush_snaps) {
3418		ceph_flush_snaps(ci, NULL);
3419	}
3420	if (complete_capsnap)
3421		wake_up_all(&ci->i_cap_wq);
3422	while (put-- > 0) {
3423		iput(inode);
3424	}
3425}
3426
3427/*
3428 * Invalidate unlinked inode's aliases, so we can drop the inode ASAP.
3429 */
3430static void invalidate_aliases(struct inode *inode)
3431{
3432	struct ceph_client *cl = ceph_inode_to_client(inode);
3433	struct dentry *dn, *prev = NULL;
3434
3435	doutc(cl, "%p %llx.%llx\n", inode, ceph_vinop(inode));
3436	d_prune_aliases(inode);
3437	/*
3438	 * For non-directory inode, d_find_alias() only returns
3439	 * hashed dentry. After calling d_invalidate(), the
3440	 * dentry becomes unhashed.
3441	 *
3442	 * For directory inode, d_find_alias() can return
3443	 * unhashed dentry. But directory inode should have
3444	 * one alias at most.
3445	 */
3446	while ((dn = d_find_alias(inode))) {
3447		if (dn == prev) {
3448			dput(dn);
3449			break;
3450		}
3451		d_invalidate(dn);
3452		if (prev)
3453			dput(prev);
3454		prev = dn;
3455	}
3456	if (prev)
3457		dput(prev);
3458}
3459
3460struct cap_extra_info {
3461	struct ceph_string *pool_ns;
3462	/* inline data */
3463	u64 inline_version;
3464	void *inline_data;
3465	u32 inline_len;
3466	/* dirstat */
3467	bool dirstat_valid;
3468	u64 nfiles;
3469	u64 nsubdirs;
3470	u64 change_attr;
3471	/* currently issued */
3472	int issued;
3473	struct timespec64 btime;
3474	u8 *fscrypt_auth;
3475	u32 fscrypt_auth_len;
3476	u64 fscrypt_file_size;
3477};
3478
3479/*
3480 * Handle a cap GRANT message from the MDS.  (Note that a GRANT may
3481 * actually be a revocation if it specifies a smaller cap set.)
3482 *
3483 * caller holds s_mutex and i_ceph_lock, we drop both.
3484 */
3485static void handle_cap_grant(struct inode *inode,
3486			     struct ceph_mds_session *session,
3487			     struct ceph_cap *cap,
3488			     struct ceph_mds_caps *grant,
3489			     struct ceph_buffer *xattr_buf,
3490			     struct cap_extra_info *extra_info)
3491	__releases(ci->i_ceph_lock)
3492	__releases(session->s_mdsc->snap_rwsem)
3493{
3494	struct ceph_client *cl = ceph_inode_to_client(inode);
3495	struct ceph_inode_info *ci = ceph_inode(inode);
3496	int seq = le32_to_cpu(grant->seq);
3497	int newcaps = le32_to_cpu(grant->caps);
3498	int used, wanted, dirty;
3499	u64 size = le64_to_cpu(grant->size);
3500	u64 max_size = le64_to_cpu(grant->max_size);
3501	unsigned char check_caps = 0;
3502	bool was_stale = cap->cap_gen < atomic_read(&session->s_cap_gen);
3503	bool wake = false;
3504	bool writeback = false;
3505	bool queue_trunc = false;
3506	bool queue_invalidate = false;
3507	bool deleted_inode = false;
3508	bool fill_inline = false;
3509	bool revoke_wait = false;
3510	int flags = 0;
3511
3512	/*
3513	 * If there is at least one crypto block then we'll trust
3514	 * fscrypt_file_size. If the real length of the file is 0, then
3515	 * ignore it (it has probably been truncated down to 0 by the MDS).
3516	 */
3517	if (IS_ENCRYPTED(inode) && size)
3518		size = extra_info->fscrypt_file_size;
3519
3520	doutc(cl, "%p %llx.%llx cap %p mds%d seq %d %s\n", inode,
3521	      ceph_vinop(inode), cap, session->s_mds, seq,
3522	      ceph_cap_string(newcaps));
3523	doutc(cl, " size %llu max_size %llu, i_size %llu\n", size,
3524	      max_size, i_size_read(inode));
3525
3526
3527	/*
3528	 * If CACHE is being revoked, and we have no dirty buffers,
3529	 * try to invalidate (once).  (If there are dirty buffers, we
3530	 * will invalidate _after_ writeback.)
3531	 */
3532	if (S_ISREG(inode->i_mode) && /* don't invalidate readdir cache */
3533	    ((cap->issued & ~newcaps) & CEPH_CAP_FILE_CACHE) &&
3534	    (newcaps & CEPH_CAP_FILE_LAZYIO) == 0 &&
3535	    !(ci->i_wrbuffer_ref || ci->i_wb_ref)) {
3536		if (try_nonblocking_invalidate(inode)) {
3537			/* there were locked pages.. invalidate later
3538			   in a separate thread. */
3539			if (ci->i_rdcache_revoking != ci->i_rdcache_gen) {
3540				queue_invalidate = true;
3541				ci->i_rdcache_revoking = ci->i_rdcache_gen;
3542			}
3543		}
3544	}
3545
3546	if (was_stale)
3547		cap->issued = cap->implemented = CEPH_CAP_PIN;
3548
3549	/*
3550	 * auth mds of the inode changed. we received the cap export message,
3551	 * but still haven't received the cap import message. handle_cap_export
3552	 * updated the new auth MDS' cap.
3553	 *
3554	 * "ceph_seq_cmp(seq, cap->seq) <= 0" means we are processing a message
3555	 * that was sent before the cap import message. So don't remove caps.
3556	 */
3557	if (ceph_seq_cmp(seq, cap->seq) <= 0) {
3558		WARN_ON(cap != ci->i_auth_cap);
3559		WARN_ON(cap->cap_id != le64_to_cpu(grant->cap_id));
3560		seq = cap->seq;
3561		newcaps |= cap->issued;
3562	}
3563
3564	/* side effects now are allowed */
3565	cap->cap_gen = atomic_read(&session->s_cap_gen);
3566	cap->seq = seq;
3567
3568	__check_cap_issue(ci, cap, newcaps);
3569
3570	inode_set_max_iversion_raw(inode, extra_info->change_attr);
3571
3572	if ((newcaps & CEPH_CAP_AUTH_SHARED) &&
3573	    (extra_info->issued & CEPH_CAP_AUTH_EXCL) == 0) {
3574		umode_t mode = le32_to_cpu(grant->mode);
3575
3576		if (inode_wrong_type(inode, mode))
3577			pr_warn_once("inode type changed! (ino %llx.%llx is 0%o, mds says 0%o)\n",
3578				     ceph_vinop(inode), inode->i_mode, mode);
3579		else
3580			inode->i_mode = mode;
3581		inode->i_uid = make_kuid(&init_user_ns, le32_to_cpu(grant->uid));
3582		inode->i_gid = make_kgid(&init_user_ns, le32_to_cpu(grant->gid));
3583		ci->i_btime = extra_info->btime;
3584		doutc(cl, "%p %llx.%llx mode 0%o uid.gid %d.%d\n", inode,
3585		      ceph_vinop(inode), inode->i_mode,
3586		      from_kuid(&init_user_ns, inode->i_uid),
3587		      from_kgid(&init_user_ns, inode->i_gid));
3588#if IS_ENABLED(CONFIG_FS_ENCRYPTION)
3589		if (ci->fscrypt_auth_len != extra_info->fscrypt_auth_len ||
3590		    memcmp(ci->fscrypt_auth, extra_info->fscrypt_auth,
3591			   ci->fscrypt_auth_len))
3592			pr_warn_ratelimited_client(cl,
3593				"cap grant attempt to change fscrypt_auth on non-I_NEW inode (old len %d new len %d)\n",
3594				ci->fscrypt_auth_len,
3595				extra_info->fscrypt_auth_len);
3596#endif
3597	}
3598
3599	if ((newcaps & CEPH_CAP_LINK_SHARED) &&
3600	    (extra_info->issued & CEPH_CAP_LINK_EXCL) == 0) {
3601		set_nlink(inode, le32_to_cpu(grant->nlink));
3602		if (inode->i_nlink == 0)
3603			deleted_inode = true;
3604	}
3605
3606	if ((extra_info->issued & CEPH_CAP_XATTR_EXCL) == 0 &&
3607	    grant->xattr_len) {
3608		int len = le32_to_cpu(grant->xattr_len);
3609		u64 version = le64_to_cpu(grant->xattr_version);
3610
3611		if (version > ci->i_xattrs.version) {
3612			doutc(cl, " got new xattrs v%llu on %p %llx.%llx len %d\n",
3613			      version, inode, ceph_vinop(inode), len);
3614			if (ci->i_xattrs.blob)
3615				ceph_buffer_put(ci->i_xattrs.blob);
3616			ci->i_xattrs.blob = ceph_buffer_get(xattr_buf);
3617			ci->i_xattrs.version = version;
3618			ceph_forget_all_cached_acls(inode);
3619			ceph_security_invalidate_secctx(inode);
3620		}
3621	}
3622
3623	if (newcaps & CEPH_CAP_ANY_RD) {
3624		struct timespec64 mtime, atime, ctime;
3625		/* ctime/mtime/atime? */
3626		ceph_decode_timespec64(&mtime, &grant->mtime);
3627		ceph_decode_timespec64(&atime, &grant->atime);
3628		ceph_decode_timespec64(&ctime, &grant->ctime);
3629		ceph_fill_file_time(inode, extra_info->issued,
3630				    le32_to_cpu(grant->time_warp_seq),
3631				    &ctime, &mtime, &atime);
3632	}
3633
3634	if ((newcaps & CEPH_CAP_FILE_SHARED) && extra_info->dirstat_valid) {
3635		ci->i_files = extra_info->nfiles;
3636		ci->i_subdirs = extra_info->nsubdirs;
3637	}
3638
3639	if (newcaps & (CEPH_CAP_ANY_FILE_RD | CEPH_CAP_ANY_FILE_WR)) {
3640		/* file layout may have changed */
3641		s64 old_pool = ci->i_layout.pool_id;
3642		struct ceph_string *old_ns;
3643
3644		ceph_file_layout_from_legacy(&ci->i_layout, &grant->layout);
3645		old_ns = rcu_dereference_protected(ci->i_layout.pool_ns,
3646					lockdep_is_held(&ci->i_ceph_lock));
3647		rcu_assign_pointer(ci->i_layout.pool_ns, extra_info->pool_ns);
3648
3649		if (ci->i_layout.pool_id != old_pool ||
3650		    extra_info->pool_ns != old_ns)
3651			ci->i_ceph_flags &= ~CEPH_I_POOL_PERM;
3652
3653		extra_info->pool_ns = old_ns;
3654
3655		/* size/truncate_seq? */
3656		queue_trunc = ceph_fill_file_size(inode, extra_info->issued,
3657					le32_to_cpu(grant->truncate_seq),
3658					le64_to_cpu(grant->truncate_size),
3659					size);
3660	}
3661
3662	if (ci->i_auth_cap == cap && (newcaps & CEPH_CAP_ANY_FILE_WR)) {
3663		if (max_size != ci->i_max_size) {
3664			doutc(cl, "max_size %lld -> %llu\n", ci->i_max_size,
3665			      max_size);
3666			ci->i_max_size = max_size;
3667			if (max_size >= ci->i_wanted_max_size) {
3668				ci->i_wanted_max_size = 0;  /* reset */
3669				ci->i_requested_max_size = 0;
3670			}
3671			wake = true;
3672		}
3673	}
3674
3675	/* check cap bits */
3676	wanted = __ceph_caps_wanted(ci);
3677	used = __ceph_caps_used(ci);
3678	dirty = __ceph_caps_dirty(ci);
3679	doutc(cl, " my wanted = %s, used = %s, dirty %s\n",
3680	      ceph_cap_string(wanted), ceph_cap_string(used),
3681	      ceph_cap_string(dirty));
3682
3683	if ((was_stale || le32_to_cpu(grant->op) == CEPH_CAP_OP_IMPORT) &&
3684	    (wanted & ~(cap->mds_wanted | newcaps))) {
3685		/*
3686		 * If mds is importing cap, prior cap messages that update
3687		 * 'wanted' may get dropped by mds (migrate seq mismatch).
3688		 *
3689		 * We don't send cap message to update 'wanted' if what we
3690		 * want are already issued. If mds revokes caps, cap message
3691		 * that releases caps also tells mds what we want. But if
3692		 * caps got revoked by mds forcedly (session stale). We may
3693		 * haven't told mds what we want.
3694		 */
3695		check_caps = 1;
3696	}
3697
3698	/* revocation, grant, or no-op? */
3699	if (cap->issued & ~newcaps) {
3700		int revoking = cap->issued & ~newcaps;
3701
3702		doutc(cl, "revocation: %s -> %s (revoking %s)\n",
3703		      ceph_cap_string(cap->issued), ceph_cap_string(newcaps),
3704		      ceph_cap_string(revoking));
3705		if (S_ISREG(inode->i_mode) &&
3706		    (revoking & used & CEPH_CAP_FILE_BUFFER)) {
3707			writeback = true;  /* initiate writeback; will delay ack */
3708			revoke_wait = true;
3709		} else if (queue_invalidate &&
3710			 revoking == CEPH_CAP_FILE_CACHE &&
3711			 (newcaps & CEPH_CAP_FILE_LAZYIO) == 0) {
3712			revoke_wait = true; /* do nothing yet, invalidation will be queued */
3713		} else if (cap == ci->i_auth_cap) {
3714			check_caps = 1; /* check auth cap only */
3715		} else {
3716			check_caps = 2; /* check all caps */
3717		}
3718		/* If there is new caps, try to wake up the waiters */
3719		if (~cap->issued & newcaps)
3720			wake = true;
3721		cap->issued = newcaps;
3722		cap->implemented |= newcaps;
3723	} else if (cap->issued == newcaps) {
3724		doutc(cl, "caps unchanged: %s -> %s\n",
3725		      ceph_cap_string(cap->issued),
3726		      ceph_cap_string(newcaps));
3727	} else {
3728		doutc(cl, "grant: %s -> %s\n", ceph_cap_string(cap->issued),
3729		      ceph_cap_string(newcaps));
3730		/* non-auth MDS is revoking the newly grant caps ? */
3731		if (cap == ci->i_auth_cap &&
3732		    __ceph_caps_revoking_other(ci, cap, newcaps))
3733		    check_caps = 2;
3734
3735		cap->issued = newcaps;
3736		cap->implemented |= newcaps; /* add bits only, to
3737					      * avoid stepping on a
3738					      * pending revocation */
3739		wake = true;
3740	}
3741	BUG_ON(cap->issued & ~cap->implemented);
3742
3743	/* don't let check_caps skip sending a response to MDS for revoke msgs */
3744	if (!revoke_wait && le32_to_cpu(grant->op) == CEPH_CAP_OP_REVOKE) {
3745		cap->mds_wanted = 0;
3746		flags |= CHECK_CAPS_FLUSH_FORCE;
3747		if (cap == ci->i_auth_cap)
3748			check_caps = 1; /* check auth cap only */
3749		else
3750			check_caps = 2; /* check all caps */
3751	}
3752
3753	if (extra_info->inline_version > 0 &&
3754	    extra_info->inline_version >= ci->i_inline_version) {
3755		ci->i_inline_version = extra_info->inline_version;
3756		if (ci->i_inline_version != CEPH_INLINE_NONE &&
3757		    (newcaps & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)))
3758			fill_inline = true;
3759	}
3760
3761	if (le32_to_cpu(grant->op) == CEPH_CAP_OP_IMPORT) {
3762		if (ci->i_auth_cap == cap) {
3763			if (newcaps & ~extra_info->issued)
3764				wake = true;
3765
3766			if (ci->i_requested_max_size > max_size ||
3767			    !(le32_to_cpu(grant->wanted) & CEPH_CAP_ANY_FILE_WR)) {
3768				/* re-request max_size if necessary */
3769				ci->i_requested_max_size = 0;
3770				wake = true;
3771			}
3772
3773			ceph_kick_flushing_inode_caps(session, ci);
3774		}
3775		up_read(&session->s_mdsc->snap_rwsem);
3776	}
3777	spin_unlock(&ci->i_ceph_lock);
3778
3779	if (fill_inline)
3780		ceph_fill_inline_data(inode, NULL, extra_info->inline_data,
3781				      extra_info->inline_len);
3782
3783	if (queue_trunc)
3784		ceph_queue_vmtruncate(inode);
3785
3786	if (writeback)
3787		/*
3788		 * queue inode for writeback: we can't actually call
3789		 * filemap_write_and_wait, etc. from message handler
3790		 * context.
3791		 */
3792		ceph_queue_writeback(inode);
3793	if (queue_invalidate)
3794		ceph_queue_invalidate(inode);
3795	if (deleted_inode)
3796		invalidate_aliases(inode);
3797	if (wake)
3798		wake_up_all(&ci->i_cap_wq);
3799
3800	mutex_unlock(&session->s_mutex);
3801	if (check_caps == 1)
3802		ceph_check_caps(ci, flags | CHECK_CAPS_AUTHONLY | CHECK_CAPS_NOINVAL);
3803	else if (check_caps == 2)
3804		ceph_check_caps(ci, flags | CHECK_CAPS_NOINVAL);
3805}
3806
3807/*
3808 * Handle FLUSH_ACK from MDS, indicating that metadata we sent to the
3809 * MDS has been safely committed.
3810 */
3811static void handle_cap_flush_ack(struct inode *inode, u64 flush_tid,
3812				 struct ceph_mds_caps *m,
3813				 struct ceph_mds_session *session,
3814				 struct ceph_cap *cap)
3815	__releases(ci->i_ceph_lock)
3816{
3817	struct ceph_inode_info *ci = ceph_inode(inode);
3818	struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(inode->i_sb)->mdsc;
3819	struct ceph_client *cl = mdsc->fsc->client;
3820	struct ceph_cap_flush *cf, *tmp_cf;
3821	LIST_HEAD(to_remove);
3822	unsigned seq = le32_to_cpu(m->seq);
3823	int dirty = le32_to_cpu(m->dirty);
3824	int cleaned = 0;
3825	bool drop = false;
3826	bool wake_ci = false;
3827	bool wake_mdsc = false;
3828
3829	list_for_each_entry_safe(cf, tmp_cf, &ci->i_cap_flush_list, i_list) {
3830		/* Is this the one that was flushed? */
3831		if (cf->tid == flush_tid)
3832			cleaned = cf->caps;
3833
3834		/* Is this a capsnap? */
3835		if (cf->is_capsnap)
3836			continue;
3837
3838		if (cf->tid <= flush_tid) {
3839			/*
3840			 * An earlier or current tid. The FLUSH_ACK should
3841			 * represent a superset of this flush's caps.
3842			 */
3843			wake_ci |= __detach_cap_flush_from_ci(ci, cf);
3844			list_add_tail(&cf->i_list, &to_remove);
3845		} else {
3846			/*
3847			 * This is a later one. Any caps in it are still dirty
3848			 * so don't count them as cleaned.
3849			 */
3850			cleaned &= ~cf->caps;
3851			if (!cleaned)
3852				break;
3853		}
3854	}
3855
3856	doutc(cl, "%p %llx.%llx mds%d seq %d on %s cleaned %s, flushing %s -> %s\n",
3857	      inode, ceph_vinop(inode), session->s_mds, seq,
3858	      ceph_cap_string(dirty), ceph_cap_string(cleaned),
3859	      ceph_cap_string(ci->i_flushing_caps),
3860	      ceph_cap_string(ci->i_flushing_caps & ~cleaned));
3861
3862	if (list_empty(&to_remove) && !cleaned)
3863		goto out;
3864
3865	ci->i_flushing_caps &= ~cleaned;
3866
3867	spin_lock(&mdsc->cap_dirty_lock);
3868
3869	list_for_each_entry(cf, &to_remove, i_list)
3870		wake_mdsc |= __detach_cap_flush_from_mdsc(mdsc, cf);
3871
3872	if (ci->i_flushing_caps == 0) {
3873		if (list_empty(&ci->i_cap_flush_list)) {
3874			list_del_init(&ci->i_flushing_item);
3875			if (!list_empty(&session->s_cap_flushing)) {
3876				struct inode *inode =
3877					    &list_first_entry(&session->s_cap_flushing,
3878							      struct ceph_inode_info,
3879							      i_flushing_item)->netfs.inode;
3880				doutc(cl, " mds%d still flushing cap on %p %llx.%llx\n",
3881				      session->s_mds, inode, ceph_vinop(inode));
3882			}
3883		}
3884		mdsc->num_cap_flushing--;
3885		doutc(cl, " %p %llx.%llx now !flushing\n", inode,
3886		      ceph_vinop(inode));
3887
3888		if (ci->i_dirty_caps == 0) {
3889			doutc(cl, " %p %llx.%llx now clean\n", inode,
3890			      ceph_vinop(inode));
3891			BUG_ON(!list_empty(&ci->i_dirty_item));
3892			drop = true;
3893			if (ci->i_wr_ref == 0 &&
3894			    ci->i_wrbuffer_ref_head == 0) {
3895				BUG_ON(!ci->i_head_snapc);
3896				ceph_put_snap_context(ci->i_head_snapc);
3897				ci->i_head_snapc = NULL;
3898			}
3899		} else {
3900			BUG_ON(list_empty(&ci->i_dirty_item));
3901		}
3902	}
3903	spin_unlock(&mdsc->cap_dirty_lock);
3904
3905out:
3906	spin_unlock(&ci->i_ceph_lock);
3907
3908	while (!list_empty(&to_remove)) {
3909		cf = list_first_entry(&to_remove,
3910				      struct ceph_cap_flush, i_list);
3911		list_del_init(&cf->i_list);
3912		if (!cf->is_capsnap)
3913			ceph_free_cap_flush(cf);
3914	}
3915
3916	if (wake_ci)
3917		wake_up_all(&ci->i_cap_wq);
3918	if (wake_mdsc)
3919		wake_up_all(&mdsc->cap_flushing_wq);
3920	if (drop)
3921		iput(inode);
3922}
3923
3924void __ceph_remove_capsnap(struct inode *inode, struct ceph_cap_snap *capsnap,
3925			   bool *wake_ci, bool *wake_mdsc)
3926{
3927	struct ceph_inode_info *ci = ceph_inode(inode);
3928	struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(inode->i_sb)->mdsc;
3929	struct ceph_client *cl = mdsc->fsc->client;
3930	bool ret;
3931
3932	lockdep_assert_held(&ci->i_ceph_lock);
3933
3934	doutc(cl, "removing capsnap %p, %p %llx.%llx ci %p\n", capsnap,
3935	      inode, ceph_vinop(inode), ci);
3936
3937	list_del_init(&capsnap->ci_item);
3938	ret = __detach_cap_flush_from_ci(ci, &capsnap->cap_flush);
3939	if (wake_ci)
3940		*wake_ci = ret;
3941
3942	spin_lock(&mdsc->cap_dirty_lock);
3943	if (list_empty(&ci->i_cap_flush_list))
3944		list_del_init(&ci->i_flushing_item);
3945
3946	ret = __detach_cap_flush_from_mdsc(mdsc, &capsnap->cap_flush);
3947	if (wake_mdsc)
3948		*wake_mdsc = ret;
3949	spin_unlock(&mdsc->cap_dirty_lock);
3950}
3951
3952void ceph_remove_capsnap(struct inode *inode, struct ceph_cap_snap *capsnap,
3953			 bool *wake_ci, bool *wake_mdsc)
3954{
3955	struct ceph_inode_info *ci = ceph_inode(inode);
3956
3957	lockdep_assert_held(&ci->i_ceph_lock);
3958
3959	WARN_ON_ONCE(capsnap->dirty_pages || capsnap->writing);
3960	__ceph_remove_capsnap(inode, capsnap, wake_ci, wake_mdsc);
3961}
3962
3963/*
3964 * Handle FLUSHSNAP_ACK.  MDS has flushed snap data to disk and we can
3965 * throw away our cap_snap.
3966 *
3967 * Caller hold s_mutex.
3968 */
3969static void handle_cap_flushsnap_ack(struct inode *inode, u64 flush_tid,
3970				     struct ceph_mds_caps *m,
3971				     struct ceph_mds_session *session)
3972{
3973	struct ceph_inode_info *ci = ceph_inode(inode);
3974	struct ceph_mds_client *mdsc = ceph_sb_to_fs_client(inode->i_sb)->mdsc;
3975	struct ceph_client *cl = mdsc->fsc->client;
3976	u64 follows = le64_to_cpu(m->snap_follows);
3977	struct ceph_cap_snap *capsnap = NULL, *iter;
3978	bool wake_ci = false;
3979	bool wake_mdsc = false;
3980
3981	doutc(cl, "%p %llx.%llx ci %p mds%d follows %lld\n", inode,
3982	      ceph_vinop(inode), ci, session->s_mds, follows);
3983
3984	spin_lock(&ci->i_ceph_lock);
3985	list_for_each_entry(iter, &ci->i_cap_snaps, ci_item) {
3986		if (iter->follows == follows) {
3987			if (iter->cap_flush.tid != flush_tid) {
3988				doutc(cl, " cap_snap %p follows %lld "
3989				      "tid %lld != %lld\n", iter,
3990				      follows, flush_tid,
3991				      iter->cap_flush.tid);
3992				break;
3993			}
3994			capsnap = iter;
3995			break;
3996		} else {
3997			doutc(cl, " skipping cap_snap %p follows %lld\n",
3998			      iter, iter->follows);
3999		}
4000	}
4001	if (capsnap)
4002		ceph_remove_capsnap(inode, capsnap, &wake_ci, &wake_mdsc);
4003	spin_unlock(&ci->i_ceph_lock);
4004
4005	if (capsnap) {
4006		ceph_put_snap_context(capsnap->context);
4007		ceph_put_cap_snap(capsnap);
4008		if (wake_ci)
4009			wake_up_all(&ci->i_cap_wq);
4010		if (wake_mdsc)
4011			wake_up_all(&mdsc->cap_flushing_wq);
4012		iput(inode);
4013	}
4014}
4015
4016/*
4017 * Handle TRUNC from MDS, indicating file truncation.
4018 *
4019 * caller hold s_mutex.
4020 */
4021static bool handle_cap_trunc(struct inode *inode,
4022			     struct ceph_mds_caps *trunc,
4023			     struct ceph_mds_session *session,
4024			     struct cap_extra_info *extra_info)
4025{
4026	struct ceph_inode_info *ci = ceph_inode(inode);
4027	struct ceph_client *cl = ceph_inode_to_client(inode);
4028	int mds = session->s_mds;
4029	int seq = le32_to_cpu(trunc->seq);
4030	u32 truncate_seq = le32_to_cpu(trunc->truncate_seq);
4031	u64 truncate_size = le64_to_cpu(trunc->truncate_size);
4032	u64 size = le64_to_cpu(trunc->size);
4033	int implemented = 0;
4034	int dirty = __ceph_caps_dirty(ci);
4035	int issued = __ceph_caps_issued(ceph_inode(inode), &implemented);
4036	bool queue_trunc = false;
4037
4038	lockdep_assert_held(&ci->i_ceph_lock);
4039
4040	issued |= implemented | dirty;
4041
4042	/*
4043	 * If there is at least one crypto block then we'll trust
4044	 * fscrypt_file_size. If the real length of the file is 0, then
4045	 * ignore it (it has probably been truncated down to 0 by the MDS).
4046	 */
4047	if (IS_ENCRYPTED(inode) && size)
4048		size = extra_info->fscrypt_file_size;
4049
4050	doutc(cl, "%p %llx.%llx mds%d seq %d to %lld truncate seq %d\n",
4051	      inode, ceph_vinop(inode), mds, seq, truncate_size, truncate_seq);
4052	queue_trunc = ceph_fill_file_size(inode, issued,
4053					  truncate_seq, truncate_size, size);
4054	return queue_trunc;
4055}
4056
4057/*
4058 * Handle EXPORT from MDS.  Cap is being migrated _from_ this mds to a
4059 * different one.  If we are the most recent migration we've seen (as
4060 * indicated by mseq), make note of the migrating cap bits for the
4061 * duration (until we see the corresponding IMPORT).
4062 *
4063 * caller holds s_mutex
4064 */
4065static void handle_cap_export(struct inode *inode, struct ceph_mds_caps *ex,
4066			      struct ceph_mds_cap_peer *ph,
4067			      struct ceph_mds_session *session)
4068{
4069	struct ceph_mds_client *mdsc = ceph_inode_to_fs_client(inode)->mdsc;
4070	struct ceph_client *cl = mdsc->fsc->client;
4071	struct ceph_mds_session *tsession = NULL;
4072	struct ceph_cap *cap, *tcap, *new_cap = NULL;
4073	struct ceph_inode_info *ci = ceph_inode(inode);
4074	u64 t_cap_id;
4075	u32 t_issue_seq, t_mseq;
4076	int target, issued;
4077	int mds = session->s_mds;
4078
4079	if (ph) {
4080		t_cap_id = le64_to_cpu(ph->cap_id);
4081		t_issue_seq = le32_to_cpu(ph->issue_seq);
4082		t_mseq = le32_to_cpu(ph->mseq);
4083		target = le32_to_cpu(ph->mds);
4084	} else {
4085		t_cap_id = t_issue_seq = t_mseq = 0;
4086		target = -1;
4087	}
4088
4089	doutc(cl, " cap %llx.%llx export to peer %d piseq %u pmseq %u\n",
4090	      ceph_vinop(inode), target, t_issue_seq, t_mseq);
4091retry:
4092	down_read(&mdsc->snap_rwsem);
4093	spin_lock(&ci->i_ceph_lock);
4094	cap = __get_cap_for_mds(ci, mds);
4095	if (!cap || cap->cap_id != le64_to_cpu(ex->cap_id))
4096		goto out_unlock;
4097
4098	if (target < 0) {
4099		ceph_remove_cap(mdsc, cap, false);
4100		goto out_unlock;
4101	}
4102
4103	/*
4104	 * now we know we haven't received the cap import message yet
4105	 * because the exported cap still exist.
4106	 */
4107
4108	issued = cap->issued;
4109	if (issued != cap->implemented)
4110		pr_err_ratelimited_client(cl, "issued != implemented: "
4111					  "%p %llx.%llx mds%d seq %d mseq %d"
4112					  " issued %s implemented %s\n",
4113					  inode, ceph_vinop(inode), mds,
4114					  cap->seq, cap->mseq,
4115					  ceph_cap_string(issued),
4116					  ceph_cap_string(cap->implemented));
4117
4118
4119	tcap = __get_cap_for_mds(ci, target);
4120	if (tcap) {
4121		/* already have caps from the target */
4122		if (tcap->cap_id == t_cap_id &&
4123		    ceph_seq_cmp(tcap->seq, t_issue_seq) < 0) {
4124			doutc(cl, " updating import cap %p mds%d\n", tcap,
4125			      target);
4126			tcap->cap_id = t_cap_id;
4127			tcap->seq = t_issue_seq - 1;
4128			tcap->issue_seq = t_issue_seq - 1;
4129			tcap->issued |= issued;
4130			tcap->implemented |= issued;
4131			if (cap == ci->i_auth_cap) {
4132				ci->i_auth_cap = tcap;
4133				change_auth_cap_ses(ci, tcap->session);
4134			}
4135		}
4136		ceph_remove_cap(mdsc, cap, false);
4137		goto out_unlock;
4138	} else if (tsession) {
4139		/* add placeholder for the export target */
4140		int flag = (cap == ci->i_auth_cap) ? CEPH_CAP_FLAG_AUTH : 0;
4141		tcap = new_cap;
4142		ceph_add_cap(inode, tsession, t_cap_id, issued, 0,
4143			     t_issue_seq - 1, t_mseq, (u64)-1, flag, &new_cap);
4144
4145		if (!list_empty(&ci->i_cap_flush_list) &&
4146		    ci->i_auth_cap == tcap) {
4147			spin_lock(&mdsc->cap_dirty_lock);
4148			list_move_tail(&ci->i_flushing_item,
4149				       &tcap->session->s_cap_flushing);
4150			spin_unlock(&mdsc->cap_dirty_lock);
4151		}
4152
4153		ceph_remove_cap(mdsc, cap, false);
4154		goto out_unlock;
4155	}
4156
4157	spin_unlock(&ci->i_ceph_lock);
4158	up_read(&mdsc->snap_rwsem);
4159	mutex_unlock(&session->s_mutex);
4160
4161	/* open target session */
4162	tsession = ceph_mdsc_open_export_target_session(mdsc, target);
4163	if (!IS_ERR(tsession)) {
4164		if (mds > target) {
4165			mutex_lock(&session->s_mutex);
4166			mutex_lock_nested(&tsession->s_mutex,
4167					  SINGLE_DEPTH_NESTING);
4168		} else {
4169			mutex_lock(&tsession->s_mutex);
4170			mutex_lock_nested(&session->s_mutex,
4171					  SINGLE_DEPTH_NESTING);
4172		}
4173		new_cap = ceph_get_cap(mdsc, NULL);
4174	} else {
4175		WARN_ON(1);
4176		tsession = NULL;
4177		target = -1;
4178		mutex_lock(&session->s_mutex);
4179	}
4180	goto retry;
4181
4182out_unlock:
4183	spin_unlock(&ci->i_ceph_lock);
4184	up_read(&mdsc->snap_rwsem);
4185	mutex_unlock(&session->s_mutex);
4186	if (tsession) {
4187		mutex_unlock(&tsession->s_mutex);
4188		ceph_put_mds_session(tsession);
4189	}
4190	if (new_cap)
4191		ceph_put_cap(mdsc, new_cap);
4192}
4193
4194/*
4195 * Handle cap IMPORT.
4196 *
4197 * caller holds s_mutex. acquires i_ceph_lock
4198 */
4199static void handle_cap_import(struct ceph_mds_client *mdsc,
4200			      struct inode *inode, struct ceph_mds_caps *im,
4201			      struct ceph_mds_cap_peer *ph,
4202			      struct ceph_mds_session *session,
4203			      struct ceph_cap **target_cap, int *old_issued)
4204{
4205	struct ceph_inode_info *ci = ceph_inode(inode);
4206	struct ceph_client *cl = mdsc->fsc->client;
4207	struct ceph_cap *cap, *ocap, *new_cap = NULL;
4208	int mds = session->s_mds;
4209	int issued;
4210	unsigned caps = le32_to_cpu(im->caps);
4211	unsigned wanted = le32_to_cpu(im->wanted);
4212	unsigned seq = le32_to_cpu(im->seq);
4213	unsigned mseq = le32_to_cpu(im->migrate_seq);
4214	u64 realmino = le64_to_cpu(im->realm);
4215	u64 cap_id = le64_to_cpu(im->cap_id);
4216	u64 p_cap_id;
4217	u32 piseq = 0;
4218	u32 pmseq = 0;
4219	int peer;
4220
4221	if (ph) {
4222		p_cap_id = le64_to_cpu(ph->cap_id);
4223		peer = le32_to_cpu(ph->mds);
4224		piseq = le32_to_cpu(ph->issue_seq);
4225		pmseq = le32_to_cpu(ph->mseq);
4226	} else {
4227		p_cap_id = 0;
4228		peer = -1;
4229	}
4230
4231	doutc(cl, " cap %llx.%llx import from peer %d piseq %u pmseq %u\n",
4232	      ceph_vinop(inode), peer, piseq, pmseq);
4233retry:
4234	cap = __get_cap_for_mds(ci, mds);
4235	if (!cap) {
4236		if (!new_cap) {
4237			spin_unlock(&ci->i_ceph_lock);
4238			new_cap = ceph_get_cap(mdsc, NULL);
4239			spin_lock(&ci->i_ceph_lock);
4240			goto retry;
4241		}
4242		cap = new_cap;
4243	} else {
4244		if (new_cap) {
4245			ceph_put_cap(mdsc, new_cap);
4246			new_cap = NULL;
4247		}
4248	}
4249
4250	__ceph_caps_issued(ci, &issued);
4251	issued |= __ceph_caps_dirty(ci);
4252
4253	ceph_add_cap(inode, session, cap_id, caps, wanted, seq, mseq,
4254		     realmino, CEPH_CAP_FLAG_AUTH, &new_cap);
4255
4256	ocap = peer >= 0 ? __get_cap_for_mds(ci, peer) : NULL;
4257	if (ocap && ocap->cap_id == p_cap_id) {
4258		doutc(cl, " remove export cap %p mds%d flags %d\n",
4259		      ocap, peer, ph->flags);
4260		if ((ph->flags & CEPH_CAP_FLAG_AUTH) &&
4261		    (ocap->seq != piseq ||
4262		     ocap->mseq != pmseq)) {
4263			pr_err_ratelimited_client(cl, "mismatched seq/mseq: "
4264					"%p %llx.%llx mds%d seq %d mseq %d"
4265					" importer mds%d has peer seq %d mseq %d\n",
4266					inode, ceph_vinop(inode), peer,
4267					ocap->seq, ocap->mseq, mds, piseq, pmseq);
4268		}
4269		ceph_remove_cap(mdsc, ocap, (ph->flags & CEPH_CAP_FLAG_RELEASE));
4270	}
4271
4272	*old_issued = issued;
4273	*target_cap = cap;
4274}
4275
4276#ifdef CONFIG_FS_ENCRYPTION
4277static int parse_fscrypt_fields(void **p, void *end,
4278				struct cap_extra_info *extra)
4279{
4280	u32 len;
4281
4282	ceph_decode_32_safe(p, end, extra->fscrypt_auth_len, bad);
4283	if (extra->fscrypt_auth_len) {
4284		ceph_decode_need(p, end, extra->fscrypt_auth_len, bad);
4285		extra->fscrypt_auth = kmalloc(extra->fscrypt_auth_len,
4286					      GFP_KERNEL);
4287		if (!extra->fscrypt_auth)
4288			return -ENOMEM;
4289		ceph_decode_copy_safe(p, end, extra->fscrypt_auth,
4290					extra->fscrypt_auth_len, bad);
4291	}
4292
4293	ceph_decode_32_safe(p, end, len, bad);
4294	if (len >= sizeof(u64)) {
4295		ceph_decode_64_safe(p, end, extra->fscrypt_file_size, bad);
4296		len -= sizeof(u64);
4297	}
4298	ceph_decode_skip_n(p, end, len, bad);
4299	return 0;
4300bad:
4301	return -EIO;
4302}
4303#else
4304static int parse_fscrypt_fields(void **p, void *end,
4305				struct cap_extra_info *extra)
4306{
4307	u32 len;
4308
4309	/* Don't care about these fields unless we're encryption-capable */
4310	ceph_decode_32_safe(p, end, len, bad);
4311	if (len)
4312		ceph_decode_skip_n(p, end, len, bad);
4313	ceph_decode_32_safe(p, end, len, bad);
4314	if (len)
4315		ceph_decode_skip_n(p, end, len, bad);
4316	return 0;
4317bad:
4318	return -EIO;
4319}
4320#endif
4321
4322/*
4323 * Handle a caps message from the MDS.
4324 *
4325 * Identify the appropriate session, inode, and call the right handler
4326 * based on the cap op.
4327 */
4328void ceph_handle_caps(struct ceph_mds_session *session,
4329		      struct ceph_msg *msg)
4330{
4331	struct ceph_mds_client *mdsc = session->s_mdsc;
4332	struct ceph_client *cl = mdsc->fsc->client;
4333	struct inode *inode;
4334	struct ceph_inode_info *ci;
4335	struct ceph_cap *cap;
4336	struct ceph_mds_caps *h;
4337	struct ceph_mds_cap_peer *peer = NULL;
4338	struct ceph_snap_realm *realm = NULL;
4339	int op;
4340	int msg_version = le16_to_cpu(msg->hdr.version);
4341	u32 seq, mseq, issue_seq;
4342	struct ceph_vino vino;
4343	void *snaptrace;
4344	size_t snaptrace_len;
4345	void *p, *end;
4346	struct cap_extra_info extra_info = {};
4347	bool queue_trunc;
4348	bool close_sessions = false;
4349	bool do_cap_release = false;
4350
4351	if (!ceph_inc_mds_stopping_blocker(mdsc, session))
4352		return;
4353
4354	/* decode */
4355	end = msg->front.iov_base + msg->front.iov_len;
4356	if (msg->front.iov_len < sizeof(*h))
4357		goto bad;
4358	h = msg->front.iov_base;
4359	op = le32_to_cpu(h->op);
4360	vino.ino = le64_to_cpu(h->ino);
4361	vino.snap = CEPH_NOSNAP;
4362	seq = le32_to_cpu(h->seq);
4363	mseq = le32_to_cpu(h->migrate_seq);
4364	issue_seq = le32_to_cpu(h->issue_seq);
4365
4366	snaptrace = h + 1;
4367	snaptrace_len = le32_to_cpu(h->snap_trace_len);
4368	p = snaptrace + snaptrace_len;
4369
4370	if (msg_version >= 2) {
4371		u32 flock_len;
4372		ceph_decode_32_safe(&p, end, flock_len, bad);
4373		if (p + flock_len > end)
4374			goto bad;
4375		p += flock_len;
4376	}
4377
4378	if (msg_version >= 3) {
4379		if (op == CEPH_CAP_OP_IMPORT) {
4380			if (p + sizeof(*peer) > end)
4381				goto bad;
4382			peer = p;
4383			p += sizeof(*peer);
4384		} else if (op == CEPH_CAP_OP_EXPORT) {
4385			/* recorded in unused fields */
4386			peer = (void *)&h->size;
4387		}
4388	}
4389
4390	if (msg_version >= 4) {
4391		ceph_decode_64_safe(&p, end, extra_info.inline_version, bad);
4392		ceph_decode_32_safe(&p, end, extra_info.inline_len, bad);
4393		if (p + extra_info.inline_len > end)
4394			goto bad;
4395		extra_info.inline_data = p;
4396		p += extra_info.inline_len;
4397	}
4398
4399	if (msg_version >= 5) {
4400		struct ceph_osd_client	*osdc = &mdsc->fsc->client->osdc;
4401		u32			epoch_barrier;
4402
4403		ceph_decode_32_safe(&p, end, epoch_barrier, bad);
4404		ceph_osdc_update_epoch_barrier(osdc, epoch_barrier);
4405	}
4406
4407	if (msg_version >= 8) {
4408		u32 pool_ns_len;
4409
4410		/* version >= 6 */
4411		ceph_decode_skip_64(&p, end, bad);	// flush_tid
4412		/* version >= 7 */
4413		ceph_decode_skip_32(&p, end, bad);	// caller_uid
4414		ceph_decode_skip_32(&p, end, bad);	// caller_gid
4415		/* version >= 8 */
4416		ceph_decode_32_safe(&p, end, pool_ns_len, bad);
4417		if (pool_ns_len > 0) {
4418			ceph_decode_need(&p, end, pool_ns_len, bad);
4419			extra_info.pool_ns =
4420				ceph_find_or_create_string(p, pool_ns_len);
4421			p += pool_ns_len;
4422		}
4423	}
4424
4425	if (msg_version >= 9) {
4426		struct ceph_timespec *btime;
4427
4428		if (p + sizeof(*btime) > end)
4429			goto bad;
4430		btime = p;
4431		ceph_decode_timespec64(&extra_info.btime, btime);
4432		p += sizeof(*btime);
4433		ceph_decode_64_safe(&p, end, extra_info.change_attr, bad);
4434	}
4435
4436	if (msg_version >= 11) {
4437		/* version >= 10 */
4438		ceph_decode_skip_32(&p, end, bad); // flags
4439		/* version >= 11 */
4440		extra_info.dirstat_valid = true;
4441		ceph_decode_64_safe(&p, end, extra_info.nfiles, bad);
4442		ceph_decode_64_safe(&p, end, extra_info.nsubdirs, bad);
4443	}
4444
4445	if (msg_version >= 12) {
4446		if (parse_fscrypt_fields(&p, end, &extra_info))
4447			goto bad;
4448	}
4449
4450	/* lookup ino */
4451	inode = ceph_find_inode(mdsc->fsc->sb, vino);
4452	doutc(cl, " caps mds%d op %s ino %llx.%llx inode %p seq %u iseq %u mseq %u\n",
4453	      session->s_mds, ceph_cap_op_name(op), vino.ino, vino.snap, inode,
4454	      seq, issue_seq, mseq);
4455
4456	trace_ceph_handle_caps(mdsc, session, op, &vino, ceph_inode(inode),
4457			       seq, issue_seq, mseq);
4458
4459	mutex_lock(&session->s_mutex);
4460
4461	if (!inode) {
4462		doutc(cl, " i don't have ino %llx\n", vino.ino);
4463
4464		switch (op) {
4465		case CEPH_CAP_OP_IMPORT:
4466		case CEPH_CAP_OP_REVOKE:
4467		case CEPH_CAP_OP_GRANT:
4468			do_cap_release = true;
4469			break;
4470		default:
4471			break;
4472		}
4473		goto flush_cap_releases;
4474	}
4475	ci = ceph_inode(inode);
4476
4477	/* these will work even if we don't have a cap yet */
4478	switch (op) {
4479	case CEPH_CAP_OP_FLUSHSNAP_ACK:
4480		handle_cap_flushsnap_ack(inode, le64_to_cpu(msg->hdr.tid),
4481					 h, session);
4482		goto done;
4483
4484	case CEPH_CAP_OP_EXPORT:
4485		handle_cap_export(inode, h, peer, session);
4486		goto done_unlocked;
4487
4488	case CEPH_CAP_OP_IMPORT:
4489		realm = NULL;
4490		if (snaptrace_len) {
4491			down_write(&mdsc->snap_rwsem);
4492			if (ceph_update_snap_trace(mdsc, snaptrace,
4493						   snaptrace + snaptrace_len,
4494						   false, &realm)) {
4495				up_write(&mdsc->snap_rwsem);
4496				close_sessions = true;
4497				goto done;
4498			}
4499			downgrade_write(&mdsc->snap_rwsem);
4500		} else {
4501			down_read(&mdsc->snap_rwsem);
4502		}
4503		spin_lock(&ci->i_ceph_lock);
4504		handle_cap_import(mdsc, inode, h, peer, session,
4505				  &cap, &extra_info.issued);
4506		handle_cap_grant(inode, session, cap,
4507				 h, msg->middle, &extra_info);
4508		if (realm)
4509			ceph_put_snap_realm(mdsc, realm);
4510		goto done_unlocked;
4511	}
4512
4513	/* the rest require a cap */
4514	spin_lock(&ci->i_ceph_lock);
4515	cap = __get_cap_for_mds(ceph_inode(inode), session->s_mds);
4516	if (!cap) {
4517		doutc(cl, " no cap on %p ino %llx.%llx from mds%d\n",
4518		      inode, ceph_ino(inode), ceph_snap(inode),
4519		      session->s_mds);
4520		spin_unlock(&ci->i_ceph_lock);
4521		switch (op) {
4522		case CEPH_CAP_OP_REVOKE:
4523		case CEPH_CAP_OP_GRANT:
4524			do_cap_release = true;
4525			break;
4526		default:
4527			break;
4528		}
4529		goto flush_cap_releases;
4530	}
4531
4532	/* note that each of these drops i_ceph_lock for us */
4533	switch (op) {
4534	case CEPH_CAP_OP_REVOKE:
4535	case CEPH_CAP_OP_GRANT:
4536		__ceph_caps_issued(ci, &extra_info.issued);
4537		extra_info.issued |= __ceph_caps_dirty(ci);
4538		handle_cap_grant(inode, session, cap,
4539				 h, msg->middle, &extra_info);
4540		goto done_unlocked;
4541
4542	case CEPH_CAP_OP_FLUSH_ACK:
4543		handle_cap_flush_ack(inode, le64_to_cpu(msg->hdr.tid),
4544				     h, session, cap);
4545		break;
4546
4547	case CEPH_CAP_OP_TRUNC:
4548		queue_trunc = handle_cap_trunc(inode, h, session,
4549						&extra_info);
4550		spin_unlock(&ci->i_ceph_lock);
4551		if (queue_trunc)
4552			ceph_queue_vmtruncate(inode);
4553		break;
4554
4555	default:
4556		spin_unlock(&ci->i_ceph_lock);
4557		pr_err_client(cl, "unknown cap op %d %s\n", op,
4558			      ceph_cap_op_name(op));
4559	}
4560
4561done:
4562	mutex_unlock(&session->s_mutex);
4563done_unlocked:
4564	iput(inode);
4565out:
4566	ceph_dec_mds_stopping_blocker(mdsc);
4567
4568	ceph_put_string(extra_info.pool_ns);
4569
4570	/* Defer closing the sessions after s_mutex lock being released */
4571	if (close_sessions)
4572		ceph_mdsc_close_sessions(mdsc);
4573
4574	kfree(extra_info.fscrypt_auth);
4575	return;
4576
4577flush_cap_releases:
4578	/*
4579	 * send any cap release message to try to move things
4580	 * along for the mds (who clearly thinks we still have this
4581	 * cap).
4582	 */
4583	if (do_cap_release) {
4584		cap = ceph_get_cap(mdsc, NULL);
4585		cap->cap_ino = vino.ino;
4586		cap->queue_release = 1;
4587		cap->cap_id = le64_to_cpu(h->cap_id);
4588		cap->mseq = mseq;
4589		cap->seq = seq;
4590		cap->issue_seq = seq;
4591		spin_lock(&session->s_cap_lock);
4592		__ceph_queue_cap_release(session, cap);
4593		spin_unlock(&session->s_cap_lock);
4594	}
4595	ceph_flush_session_cap_releases(mdsc, session);
4596	goto done;
4597
4598bad:
4599	pr_err_client(cl, "corrupt message\n");
4600	ceph_msg_dump(msg);
4601	goto out;
4602}
4603
4604/*
4605 * Delayed work handler to process end of delayed cap release LRU list.
4606 *
4607 * If new caps are added to the list while processing it, these won't get
4608 * processed in this run.  In this case, the ci->i_hold_caps_max will be
4609 * returned so that the work can be scheduled accordingly.
4610 */
4611unsigned long ceph_check_delayed_caps(struct ceph_mds_client *mdsc)
4612{
4613	struct ceph_client *cl = mdsc->fsc->client;
4614	struct inode *inode;
4615	struct ceph_inode_info *ci;
4616	struct ceph_mount_options *opt = mdsc->fsc->mount_options;
4617	unsigned long delay_max = opt->caps_wanted_delay_max * HZ;
4618	unsigned long loop_start = jiffies;
4619	unsigned long delay = 0;
4620
4621	doutc(cl, "begin\n");
4622	spin_lock(&mdsc->cap_delay_lock);
4623	while (!list_empty(&mdsc->cap_delay_list)) {
4624		ci = list_first_entry(&mdsc->cap_delay_list,
4625				      struct ceph_inode_info,
4626				      i_cap_delay_list);
4627		if (time_before(loop_start, ci->i_hold_caps_max - delay_max)) {
4628			doutc(cl, "caps added recently.  Exiting loop");
4629			delay = ci->i_hold_caps_max;
4630			break;
4631		}
4632		if ((ci->i_ceph_flags & CEPH_I_FLUSH) == 0 &&
4633		    time_before(jiffies, ci->i_hold_caps_max))
4634			break;
4635		list_del_init(&ci->i_cap_delay_list);
4636
4637		inode = igrab(&ci->netfs.inode);
4638		if (inode) {
4639			spin_unlock(&mdsc->cap_delay_lock);
4640			doutc(cl, "on %p %llx.%llx\n", inode,
4641			      ceph_vinop(inode));
4642			ceph_check_caps(ci, 0);
4643			iput(inode);
4644			spin_lock(&mdsc->cap_delay_lock);
4645		}
4646
4647		/*
4648		 * Make sure too many dirty caps or general
4649		 * slowness doesn't block mdsc delayed work,
4650		 * preventing send_renew_caps() from running.
4651		 */
4652		if (time_after_eq(jiffies, loop_start + 5 * HZ))
4653			break;
4654	}
4655	spin_unlock(&mdsc->cap_delay_lock);
4656	doutc(cl, "done\n");
4657
4658	return delay;
4659}
4660
4661/*
4662 * Flush all dirty caps to the mds
4663 */
4664static void flush_dirty_session_caps(struct ceph_mds_session *s)
4665{
4666	struct ceph_mds_client *mdsc = s->s_mdsc;
4667	struct ceph_client *cl = mdsc->fsc->client;
4668	struct ceph_inode_info *ci;
4669	struct inode *inode;
4670
4671	doutc(cl, "begin\n");
4672	spin_lock(&mdsc->cap_dirty_lock);
4673	while (!list_empty(&s->s_cap_dirty)) {
4674		ci = list_first_entry(&s->s_cap_dirty, struct ceph_inode_info,
4675				      i_dirty_item);
4676		inode = &ci->netfs.inode;
4677		ihold(inode);
4678		doutc(cl, "%p %llx.%llx\n", inode, ceph_vinop(inode));
4679		spin_unlock(&mdsc->cap_dirty_lock);
4680		ceph_wait_on_async_create(inode);
4681		ceph_check_caps(ci, CHECK_CAPS_FLUSH);
4682		iput(inode);
4683		spin_lock(&mdsc->cap_dirty_lock);
4684	}
4685	spin_unlock(&mdsc->cap_dirty_lock);
4686	doutc(cl, "done\n");
4687}
4688
4689void ceph_flush_dirty_caps(struct ceph_mds_client *mdsc)
4690{
4691	ceph_mdsc_iterate_sessions(mdsc, flush_dirty_session_caps, true);
4692}
4693
4694/*
4695 * Flush all cap releases to the mds
4696 */
4697static void flush_cap_releases(struct ceph_mds_session *s)
4698{
4699	struct ceph_mds_client *mdsc = s->s_mdsc;
4700	struct ceph_client *cl = mdsc->fsc->client;
4701
4702	doutc(cl, "begin\n");
4703	spin_lock(&s->s_cap_lock);
4704	if (s->s_num_cap_releases)
4705		ceph_flush_session_cap_releases(mdsc, s);
4706	spin_unlock(&s->s_cap_lock);
4707	doutc(cl, "done\n");
4708
4709}
4710
4711void ceph_flush_cap_releases(struct ceph_mds_client *mdsc)
4712{
4713	ceph_mdsc_iterate_sessions(mdsc, flush_cap_releases, true);
4714}
4715
4716void __ceph_touch_fmode(struct ceph_inode_info *ci,
4717			struct ceph_mds_client *mdsc, int fmode)
4718{
4719	unsigned long now = jiffies;
4720	if (fmode & CEPH_FILE_MODE_RD)
4721		ci->i_last_rd = now;
4722	if (fmode & CEPH_FILE_MODE_WR)
4723		ci->i_last_wr = now;
4724	/* queue periodic check */
4725	if (fmode &&
4726	    __ceph_is_any_real_caps(ci) &&
4727	    list_empty(&ci->i_cap_delay_list))
4728		__cap_delay_requeue(mdsc, ci);
4729}
4730
4731void ceph_get_fmode(struct ceph_inode_info *ci, int fmode, int count)
4732{
4733	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(ci->netfs.inode.i_sb);
4734	int bits = (fmode << 1) | 1;
4735	bool already_opened = false;
4736	int i;
4737
4738	if (count == 1)
4739		atomic64_inc(&mdsc->metric.opened_files);
4740
4741	spin_lock(&ci->i_ceph_lock);
4742	for (i = 0; i < CEPH_FILE_MODE_BITS; i++) {
4743		/*
4744		 * If any of the mode ref is larger than 0,
4745		 * that means it has been already opened by
4746		 * others. Just skip checking the PIN ref.
4747		 */
4748		if (i && ci->i_nr_by_mode[i])
4749			already_opened = true;
4750
4751		if (bits & (1 << i))
4752			ci->i_nr_by_mode[i] += count;
4753	}
4754
4755	if (!already_opened)
4756		percpu_counter_inc(&mdsc->metric.opened_inodes);
4757	spin_unlock(&ci->i_ceph_lock);
4758}
4759
4760/*
4761 * Drop open file reference.  If we were the last open file,
4762 * we may need to release capabilities to the MDS (or schedule
4763 * their delayed release).
4764 */
4765void ceph_put_fmode(struct ceph_inode_info *ci, int fmode, int count)
4766{
4767	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(ci->netfs.inode.i_sb);
4768	int bits = (fmode << 1) | 1;
4769	bool is_closed = true;
4770	int i;
4771
4772	if (count == 1)
4773		atomic64_dec(&mdsc->metric.opened_files);
4774
4775	spin_lock(&ci->i_ceph_lock);
4776	for (i = 0; i < CEPH_FILE_MODE_BITS; i++) {
4777		if (bits & (1 << i)) {
4778			BUG_ON(ci->i_nr_by_mode[i] < count);
4779			ci->i_nr_by_mode[i] -= count;
4780		}
4781
4782		/*
4783		 * If any of the mode ref is not 0 after
4784		 * decreased, that means it is still opened
4785		 * by others. Just skip checking the PIN ref.
4786		 */
4787		if (i && ci->i_nr_by_mode[i])
4788			is_closed = false;
4789	}
4790
4791	if (is_closed)
4792		percpu_counter_dec(&mdsc->metric.opened_inodes);
4793	spin_unlock(&ci->i_ceph_lock);
4794}
4795
4796/*
4797 * For a soon-to-be unlinked file, drop the LINK caps. If it
4798 * looks like the link count will hit 0, drop any other caps (other
4799 * than PIN) we don't specifically want (due to the file still being
4800 * open).
4801 */
4802int ceph_drop_caps_for_unlink(struct inode *inode)
4803{
4804	struct ceph_inode_info *ci = ceph_inode(inode);
4805	int drop = CEPH_CAP_LINK_SHARED | CEPH_CAP_LINK_EXCL;
4806
4807	spin_lock(&ci->i_ceph_lock);
4808	if (inode->i_nlink == 1) {
4809		drop |= ~(__ceph_caps_wanted(ci) | CEPH_CAP_PIN);
4810
4811		if (__ceph_caps_dirty(ci)) {
4812			struct ceph_mds_client *mdsc =
4813				ceph_inode_to_fs_client(inode)->mdsc;
4814
4815			doutc(mdsc->fsc->client, "%p %llx.%llx\n", inode,
4816			      ceph_vinop(inode));
4817			spin_lock(&mdsc->cap_delay_lock);
4818			ci->i_ceph_flags |= CEPH_I_FLUSH;
4819			if (!list_empty(&ci->i_cap_delay_list))
4820				list_del_init(&ci->i_cap_delay_list);
4821			list_add_tail(&ci->i_cap_delay_list,
4822				      &mdsc->cap_unlink_delay_list);
4823			spin_unlock(&mdsc->cap_delay_lock);
4824
4825			/*
4826			 * Fire the work immediately, because the MDS maybe
4827			 * waiting for caps release.
4828			 */
4829			ceph_queue_cap_unlink_work(mdsc);
4830		}
4831	}
4832	spin_unlock(&ci->i_ceph_lock);
4833	return drop;
4834}
4835
4836/*
4837 * Helpers for embedding cap and dentry lease releases into mds
4838 * requests.
4839 *
4840 * @force is used by dentry_release (below) to force inclusion of a
4841 * record for the directory inode, even when there aren't any caps to
4842 * drop.
4843 */
4844int ceph_encode_inode_release(void **p, struct inode *inode,
4845			      int mds, int drop, int unless, int force)
4846{
4847	struct ceph_inode_info *ci = ceph_inode(inode);
4848	struct ceph_client *cl = ceph_inode_to_client(inode);
4849	struct ceph_cap *cap;
4850	struct ceph_mds_request_release *rel = *p;
4851	int used, dirty;
4852	int ret = 0;
4853
4854	spin_lock(&ci->i_ceph_lock);
4855	used = __ceph_caps_used(ci);
4856	dirty = __ceph_caps_dirty(ci);
4857
4858	doutc(cl, "%p %llx.%llx mds%d used|dirty %s drop %s unless %s\n",
4859	      inode, ceph_vinop(inode), mds, ceph_cap_string(used|dirty),
4860	      ceph_cap_string(drop), ceph_cap_string(unless));
4861
4862	/* only drop unused, clean caps */
4863	drop &= ~(used | dirty);
4864
4865	cap = __get_cap_for_mds(ci, mds);
4866	if (cap && __cap_is_valid(cap)) {
4867		unless &= cap->issued;
4868		if (unless) {
4869			if (unless & CEPH_CAP_AUTH_EXCL)
4870				drop &= ~CEPH_CAP_AUTH_SHARED;
4871			if (unless & CEPH_CAP_LINK_EXCL)
4872				drop &= ~CEPH_CAP_LINK_SHARED;
4873			if (unless & CEPH_CAP_XATTR_EXCL)
4874				drop &= ~CEPH_CAP_XATTR_SHARED;
4875			if (unless & CEPH_CAP_FILE_EXCL)
4876				drop &= ~CEPH_CAP_FILE_SHARED;
4877		}
4878
4879		if (force || (cap->issued & drop)) {
4880			if (cap->issued & drop) {
4881				int wanted = __ceph_caps_wanted(ci);
4882				doutc(cl, "%p %llx.%llx cap %p %s -> %s, "
4883				      "wanted %s -> %s\n", inode,
4884				      ceph_vinop(inode), cap,
4885				      ceph_cap_string(cap->issued),
4886				      ceph_cap_string(cap->issued & ~drop),
4887				      ceph_cap_string(cap->mds_wanted),
4888				      ceph_cap_string(wanted));
4889
4890				cap->issued &= ~drop;
4891				cap->implemented &= ~drop;
4892				cap->mds_wanted = wanted;
4893				if (cap == ci->i_auth_cap &&
4894				    !(wanted & CEPH_CAP_ANY_FILE_WR))
4895					ci->i_requested_max_size = 0;
4896			} else {
4897				doutc(cl, "%p %llx.%llx cap %p %s (force)\n",
4898				      inode, ceph_vinop(inode), cap,
4899				      ceph_cap_string(cap->issued));
4900			}
4901
4902			rel->ino = cpu_to_le64(ceph_ino(inode));
4903			rel->cap_id = cpu_to_le64(cap->cap_id);
4904			rel->seq = cpu_to_le32(cap->seq);
4905			rel->issue_seq = cpu_to_le32(cap->issue_seq);
4906			rel->mseq = cpu_to_le32(cap->mseq);
4907			rel->caps = cpu_to_le32(cap->implemented);
4908			rel->wanted = cpu_to_le32(cap->mds_wanted);
4909			rel->dname_len = 0;
4910			rel->dname_seq = 0;
4911			*p += sizeof(*rel);
4912			ret = 1;
4913		} else {
4914			doutc(cl, "%p %llx.%llx cap %p %s (noop)\n",
4915			      inode, ceph_vinop(inode), cap,
4916			      ceph_cap_string(cap->issued));
4917		}
4918	}
4919	spin_unlock(&ci->i_ceph_lock);
4920	return ret;
4921}
4922
4923/**
4924 * ceph_encode_dentry_release - encode a dentry release into an outgoing request
4925 * @p: outgoing request buffer
4926 * @dentry: dentry to release
4927 * @dir: dir to release it from
4928 * @mds: mds that we're speaking to
4929 * @drop: caps being dropped
4930 * @unless: unless we have these caps
4931 *
4932 * Encode a dentry release into an outgoing request buffer. Returns 1 if the
4933 * thing was released, or a negative error code otherwise.
4934 */
4935int ceph_encode_dentry_release(void **p, struct dentry *dentry,
4936			       struct inode *dir,
4937			       int mds, int drop, int unless)
4938{
4939	struct ceph_mds_request_release *rel = *p;
4940	struct ceph_dentry_info *di = ceph_dentry(dentry);
4941	struct ceph_client *cl;
4942	int force = 0;
4943	int ret;
4944
4945	/* This shouldn't happen */
4946	BUG_ON(!dir);
4947
4948	/*
4949	 * force an record for the directory caps if we have a dentry lease.
4950	 * this is racy (can't take i_ceph_lock and d_lock together), but it
4951	 * doesn't have to be perfect; the mds will revoke anything we don't
4952	 * release.
4953	 */
4954	spin_lock(&dentry->d_lock);
4955	if (di->lease_session && di->lease_session->s_mds == mds)
4956		force = 1;
4957	spin_unlock(&dentry->d_lock);
4958
4959	ret = ceph_encode_inode_release(p, dir, mds, drop, unless, force);
4960
4961	cl = ceph_inode_to_client(dir);
4962	spin_lock(&dentry->d_lock);
4963	if (ret && di->lease_session && di->lease_session->s_mds == mds) {
4964		int len = dentry->d_name.len;
4965		doutc(cl, "%p mds%d seq %d\n",  dentry, mds,
4966		      (int)di->lease_seq);
4967		rel->dname_seq = cpu_to_le32(di->lease_seq);
4968		__ceph_mdsc_drop_dentry_lease(dentry);
4969		memcpy(*p, dentry->d_name.name, len);
4970		spin_unlock(&dentry->d_lock);
4971		if (IS_ENCRYPTED(dir) && fscrypt_has_encryption_key(dir)) {
4972			len = ceph_encode_encrypted_dname(dir, *p, len);
4973			if (len < 0)
4974				return len;
4975		}
4976		rel->dname_len = cpu_to_le32(len);
4977		*p += len;
4978	} else {
4979		spin_unlock(&dentry->d_lock);
4980	}
4981	return ret;
4982}
4983
4984static int remove_capsnaps(struct ceph_mds_client *mdsc, struct inode *inode)
4985{
4986	struct ceph_inode_info *ci = ceph_inode(inode);
4987	struct ceph_client *cl = mdsc->fsc->client;
4988	struct ceph_cap_snap *capsnap;
4989	int capsnap_release = 0;
4990
4991	lockdep_assert_held(&ci->i_ceph_lock);
4992
4993	doutc(cl, "removing capsnaps, ci is %p, %p %llx.%llx\n",
4994	      ci, inode, ceph_vinop(inode));
4995
4996	while (!list_empty(&ci->i_cap_snaps)) {
4997		capsnap = list_first_entry(&ci->i_cap_snaps,
4998					   struct ceph_cap_snap, ci_item);
4999		__ceph_remove_capsnap(inode, capsnap, NULL, NULL);
5000		ceph_put_snap_context(capsnap->context);
5001		ceph_put_cap_snap(capsnap);
5002		capsnap_release++;
5003	}
5004	wake_up_all(&ci->i_cap_wq);
5005	wake_up_all(&mdsc->cap_flushing_wq);
5006	return capsnap_release;
5007}
5008
5009int ceph_purge_inode_cap(struct inode *inode, struct ceph_cap *cap, bool *invalidate)
5010{
5011	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
5012	struct ceph_mds_client *mdsc = fsc->mdsc;
5013	struct ceph_client *cl = fsc->client;
5014	struct ceph_inode_info *ci = ceph_inode(inode);
5015	bool is_auth;
5016	bool dirty_dropped = false;
5017	int iputs = 0;
5018
5019	lockdep_assert_held(&ci->i_ceph_lock);
5020
5021	doutc(cl, "removing cap %p, ci is %p, %p %llx.%llx\n",
5022	      cap, ci, inode, ceph_vinop(inode));
5023
5024	is_auth = (cap == ci->i_auth_cap);
5025	__ceph_remove_cap(cap, false);
5026	if (is_auth) {
5027		struct ceph_cap_flush *cf;
5028
5029		if (ceph_inode_is_shutdown(inode)) {
5030			if (inode->i_data.nrpages > 0)
5031				*invalidate = true;
5032			if (ci->i_wrbuffer_ref > 0)
5033				mapping_set_error(&inode->i_data, -EIO);
5034		}
5035
5036		spin_lock(&mdsc->cap_dirty_lock);
5037
5038		/* trash all of the cap flushes for this inode */
5039		while (!list_empty(&ci->i_cap_flush_list)) {
5040			cf = list_first_entry(&ci->i_cap_flush_list,
5041					      struct ceph_cap_flush, i_list);
5042			list_del_init(&cf->g_list);
5043			list_del_init(&cf->i_list);
5044			if (!cf->is_capsnap)
5045				ceph_free_cap_flush(cf);
5046		}
5047
5048		if (!list_empty(&ci->i_dirty_item)) {
5049			pr_warn_ratelimited_client(cl,
5050				" dropping dirty %s state for %p %llx.%llx\n",
5051				ceph_cap_string(ci->i_dirty_caps),
5052				inode, ceph_vinop(inode));
5053			ci->i_dirty_caps = 0;
5054			list_del_init(&ci->i_dirty_item);
5055			dirty_dropped = true;
5056		}
5057		if (!list_empty(&ci->i_flushing_item)) {
5058			pr_warn_ratelimited_client(cl,
5059				" dropping dirty+flushing %s state for %p %llx.%llx\n",
5060				ceph_cap_string(ci->i_flushing_caps),
5061				inode, ceph_vinop(inode));
5062			ci->i_flushing_caps = 0;
5063			list_del_init(&ci->i_flushing_item);
5064			mdsc->num_cap_flushing--;
5065			dirty_dropped = true;
5066		}
5067		spin_unlock(&mdsc->cap_dirty_lock);
5068
5069		if (dirty_dropped) {
5070			mapping_set_error(inode->i_mapping, -EIO);
5071
5072			if (ci->i_wrbuffer_ref_head == 0 &&
5073			    ci->i_wr_ref == 0 &&
5074			    ci->i_dirty_caps == 0 &&
5075			    ci->i_flushing_caps == 0) {
5076				ceph_put_snap_context(ci->i_head_snapc);
5077				ci->i_head_snapc = NULL;
5078			}
5079		}
5080
5081		if (atomic_read(&ci->i_filelock_ref) > 0) {
5082			/* make further file lock syscall return -EIO */
5083			ci->i_ceph_flags |= CEPH_I_ERROR_FILELOCK;
5084			pr_warn_ratelimited_client(cl,
5085				" dropping file locks for %p %llx.%llx\n",
5086				inode, ceph_vinop(inode));
5087		}
5088
5089		if (!ci->i_dirty_caps && ci->i_prealloc_cap_flush) {
5090			cf = ci->i_prealloc_cap_flush;
5091			ci->i_prealloc_cap_flush = NULL;
5092			if (!cf->is_capsnap)
5093				ceph_free_cap_flush(cf);
5094		}
5095
5096		if (!list_empty(&ci->i_cap_snaps))
5097			iputs = remove_capsnaps(mdsc, inode);
5098	}
5099	if (dirty_dropped)
5100		++iputs;
5101	return iputs;
5102}