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