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