fs/ceph/inode.c

   1 // SPDX-License-Identifier: GPL-2.0
   2 #include <linux/ceph/ceph_debug.h>
   3
   4 #include <linux/module.h>
   5 #include <linux/fs.h>
   6 #include <linux/slab.h>
   7 #include <linux/string.h>
   8 #include <linux/uaccess.h>
   9 #include <linux/kernel.h>
  10 #include <linux/writeback.h>
  11 #include <linux/vmalloc.h>
  12 #include <linux/xattr.h>
  13 #include <linux/posix_acl.h>
  14 #include <linux/random.h>
  15 #include <linux/sort.h>
  16 #include <linux/iversion.h>
  17
  18 #include "super.h"
  19 #include "mds_client.h"
  20 #include "cache.h"
  21 #include <linux/ceph/decode.h>
  22
  23 /*
  24  * Ceph inode operations
  25  *
  26  * Implement basic inode helpers (get, alloc) and inode ops (getattr,
  27  * setattr, etc.), xattr helpers, and helpers for assimilating
  28  * metadata returned by the MDS into our cache.
  29  *
  30  * Also define helpers for doing asynchronous writeback, invalidation,
  31  * and truncation for the benefit of those who can't afford to block
  32  * (typically because they are in the message handler path).
  33  */
  34
  35 static const struct inode_operations ceph_symlink_iops;
  36
  37 static void ceph_inode_work(struct work_struct *work);
  38
  39 /*
  40  * find or create an inode, given the ceph ino number
  41  */
  42 static int ceph_set_ino_cb(struct inode *inode, void *data)
  43 {
  44         struct ceph_inode_info *ci = ceph_inode(inode);
  45         struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
  46
  47         ci->i_vino = *(struct ceph_vino *)data;
  48         inode->i_ino = ceph_vino_to_ino_t(ci->i_vino);
  49         inode_set_iversion_raw(inode, 0);
  50         percpu_counter_inc(&mdsc->metric.total_inodes);
  51
  52         return 0;
  53 }
  54
  55 struct inode *ceph_get_inode(struct super_block *sb, struct ceph_vino vino)
  56 {
  57         struct inode *inode;
  58
  59         inode = iget5_locked(sb, (unsigned long)vino.ino, ceph_ino_compare,
  60                              ceph_set_ino_cb, &vino);
  61         if (!inode)
  62                 return ERR_PTR(-ENOMEM);
  63
  64         dout("get_inode on %llu=%llx.%llx got %p new %d\n", ceph_present_inode(inode),
  65              ceph_vinop(inode), inode, !!(inode->i_state & I_NEW));
  66         return inode;
  67 }
  68
  69 /*
  70  * get/constuct snapdir inode for a given directory
  71  */
  72 struct inode *ceph_get_snapdir(struct inode *parent)
  73 {
  74         struct ceph_vino vino = {
  75                 .ino = ceph_ino(parent),
  76                 .snap = CEPH_SNAPDIR,
  77         };
  78         struct inode *inode = ceph_get_inode(parent->i_sb, vino);
  79         struct ceph_inode_info *ci = ceph_inode(inode);
  80
  81         BUG_ON(!S_ISDIR(parent->i_mode));
  82         if (IS_ERR(inode))
  83                 return inode;
  84         inode->i_mode = parent->i_mode;
  85         inode->i_uid = parent->i_uid;
  86         inode->i_gid = parent->i_gid;
  87         inode->i_mtime = parent->i_mtime;
  88         inode->i_ctime = parent->i_ctime;
  89         inode->i_atime = parent->i_atime;
  90         inode->i_op = &ceph_snapdir_iops;
  91         inode->i_fop = &ceph_snapdir_fops;
  92         ci->i_snap_caps = CEPH_CAP_PIN; /* so we can open */
  93         ci->i_rbytes = 0;
  94         ci->i_btime = ceph_inode(parent)->i_btime;
  95
  96         if (inode->i_state & I_NEW)
  97                 unlock_new_inode(inode);
  98
  99         return inode;
 100 }
 101
 102 const struct inode_operations ceph_file_iops = {
 103         .permission = ceph_permission,
 104         .setattr = ceph_setattr,
 105         .getattr = ceph_getattr,
 106         .listxattr = ceph_listxattr,
 107         .get_acl = ceph_get_acl,
 108         .set_acl = ceph_set_acl,
 109 };
 110
 111
 112 /*
 113  * We use a 'frag tree' to keep track of the MDS's directory fragments
 114  * for a given inode (usually there is just a single fragment).  We
 115  * need to know when a child frag is delegated to a new MDS, or when
 116  * it is flagged as replicated, so we can direct our requests
 117  * accordingly.
 118  */
 119
 120 /*
 121  * find/create a frag in the tree
 122  */
 123 static struct ceph_inode_frag *__get_or_create_frag(struct ceph_inode_info *ci,
 124                                                     u32 f)
 125 {
 126         struct rb_node **p;
 127         struct rb_node *parent = NULL;
 128         struct ceph_inode_frag *frag;
 129         int c;
 130
 131         p = &ci->i_fragtree.rb_node;
 132         while (*p) {
 133                 parent = *p;
 134                 frag = rb_entry(parent, struct ceph_inode_frag, node);
 135                 c = ceph_frag_compare(f, frag->frag);
 136                 if (c < 0)
 137                         p = &(*p)->rb_left;
 138                 else if (c > 0)
 139                         p = &(*p)->rb_right;
 140                 else
 141                         return frag;
 142         }
 143
 144         frag = kmalloc(sizeof(*frag), GFP_NOFS);
 145         if (!frag)
 146                 return ERR_PTR(-ENOMEM);
 147
 148         frag->frag = f;
 149         frag->split_by = 0;
 150         frag->mds = -1;
 151         frag->ndist = 0;
 152
 153         rb_link_node(&frag->node, parent, p);
 154         rb_insert_color(&frag->node, &ci->i_fragtree);
 155
 156         dout("get_or_create_frag added %llx.%llx frag %x\n",
 157              ceph_vinop(&ci->vfs_inode), f);
 158         return frag;
 159 }
 160
 161 /*
 162  * find a specific frag @f
 163  */
 164 struct ceph_inode_frag *__ceph_find_frag(struct ceph_inode_info *ci, u32 f)
 165 {
 166         struct rb_node *n = ci->i_fragtree.rb_node;
 167
 168         while (n) {
 169                 struct ceph_inode_frag *frag =
 170                         rb_entry(n, struct ceph_inode_frag, node);
 171                 int c = ceph_frag_compare(f, frag->frag);
 172                 if (c < 0)
 173                         n = n->rb_left;
 174                 else if (c > 0)
 175                         n = n->rb_right;
 176                 else
 177                         return frag;
 178         }
 179         return NULL;
 180 }
 181
 182 /*
 183  * Choose frag containing the given value @v.  If @pfrag is
 184  * specified, copy the frag delegation info to the caller if
 185  * it is present.
 186  */
 187 static u32 __ceph_choose_frag(struct ceph_inode_info *ci, u32 v,
 188                               struct ceph_inode_frag *pfrag, int *found)
 189 {
 190         u32 t = ceph_frag_make(0, 0);
 191         struct ceph_inode_frag *frag;
 192         unsigned nway, i;
 193         u32 n;
 194
 195         if (found)
 196                 *found = 0;
 197
 198         while (1) {
 199                 WARN_ON(!ceph_frag_contains_value(t, v));
 200                 frag = __ceph_find_frag(ci, t);
 201                 if (!frag)
 202                         break; /* t is a leaf */
 203                 if (frag->split_by == 0) {
 204                         if (pfrag)
 205                                 memcpy(pfrag, frag, sizeof(*pfrag));
 206                         if (found)
 207                                 *found = 1;
 208                         break;
 209                 }
 210
 211                 /* choose child */
 212                 nway = 1 << frag->split_by;
 213                 dout("choose_frag(%x) %x splits by %d (%d ways)\n", v, t,
 214                      frag->split_by, nway);
 215                 for (i = 0; i < nway; i++) {
 216                         n = ceph_frag_make_child(t, frag->split_by, i);
 217                         if (ceph_frag_contains_value(n, v)) {
 218                                 t = n;
 219                                 break;
 220                         }
 221                 }
 222                 BUG_ON(i == nway);
 223         }
 224         dout("choose_frag(%x) = %x\n", v, t);
 225
 226         return t;
 227 }
 228
 229 u32 ceph_choose_frag(struct ceph_inode_info *ci, u32 v,
 230                      struct ceph_inode_frag *pfrag, int *found)
 231 {
 232         u32 ret;
 233         mutex_lock(&ci->i_fragtree_mutex);
 234         ret = __ceph_choose_frag(ci, v, pfrag, found);
 235         mutex_unlock(&ci->i_fragtree_mutex);
 236         return ret;
 237 }
 238
 239 /*
 240  * Process dirfrag (delegation) info from the mds.  Include leaf
 241  * fragment in tree ONLY if ndist > 0.  Otherwise, only
 242  * branches/splits are included in i_fragtree)
 243  */
 244 static int ceph_fill_dirfrag(struct inode *inode,
 245                              struct ceph_mds_reply_dirfrag *dirinfo)
 246 {
 247         struct ceph_inode_info *ci = ceph_inode(inode);
 248         struct ceph_inode_frag *frag;
 249         u32 id = le32_to_cpu(dirinfo->frag);
 250         int mds = le32_to_cpu(dirinfo->auth);
 251         int ndist = le32_to_cpu(dirinfo->ndist);
 252         int diri_auth = -1;
 253         int i;
 254         int err = 0;
 255
 256         spin_lock(&ci->i_ceph_lock);
 257         if (ci->i_auth_cap)
 258                 diri_auth = ci->i_auth_cap->mds;
 259         spin_unlock(&ci->i_ceph_lock);
 260
 261         if (mds == -1) /* CDIR_AUTH_PARENT */
 262                 mds = diri_auth;
 263
 264         mutex_lock(&ci->i_fragtree_mutex);
 265         if (ndist == 0 && mds == diri_auth) {
 266                 /* no delegation info needed. */
 267                 frag = __ceph_find_frag(ci, id);
 268                 if (!frag)
 269                         goto out;
 270                 if (frag->split_by == 0) {
 271                         /* tree leaf, remove */
 272                         dout("fill_dirfrag removed %llx.%llx frag %x"
 273                              " (no ref)\n", ceph_vinop(inode), id);
 274                         rb_erase(&frag->node, &ci->i_fragtree);
 275                         kfree(frag);
 276                 } else {
 277                         /* tree branch, keep and clear */
 278                         dout("fill_dirfrag cleared %llx.%llx frag %x"
 279                              " referral\n", ceph_vinop(inode), id);
 280                         frag->mds = -1;
 281                         frag->ndist = 0;
 282                 }
 283                 goto out;
 284         }
 285
 286
 287         /* find/add this frag to store mds delegation info */
 288         frag = __get_or_create_frag(ci, id);
 289         if (IS_ERR(frag)) {
 290                 /* this is not the end of the world; we can continue
 291                    with bad/inaccurate delegation info */
 292                 pr_err("fill_dirfrag ENOMEM on mds ref %llx.%llx fg %x\n",
 293                        ceph_vinop(inode), le32_to_cpu(dirinfo->frag));
 294                 err = -ENOMEM;
 295                 goto out;
 296         }
 297
 298         frag->mds = mds;
 299         frag->ndist = min_t(u32, ndist, CEPH_MAX_DIRFRAG_REP);
 300         for (i = 0; i < frag->ndist; i++)
 301                 frag->dist[i] = le32_to_cpu(dirinfo->dist[i]);
 302         dout("fill_dirfrag %llx.%llx frag %x ndist=%d\n",
 303              ceph_vinop(inode), frag->frag, frag->ndist);
 304
 305 out:
 306         mutex_unlock(&ci->i_fragtree_mutex);
 307         return err;
 308 }
 309
 310 static int frag_tree_split_cmp(const void *l, const void *r)
 311 {
 312         struct ceph_frag_tree_split *ls = (struct ceph_frag_tree_split*)l;
 313         struct ceph_frag_tree_split *rs = (struct ceph_frag_tree_split*)r;
 314         return ceph_frag_compare(le32_to_cpu(ls->frag),
 315                                  le32_to_cpu(rs->frag));
 316 }
 317
 318 static bool is_frag_child(u32 f, struct ceph_inode_frag *frag)
 319 {
 320         if (!frag)
 321                 return f == ceph_frag_make(0, 0);
 322         if (ceph_frag_bits(f) != ceph_frag_bits(frag->frag) + frag->split_by)
 323                 return false;
 324         return ceph_frag_contains_value(frag->frag, ceph_frag_value(f));
 325 }
 326
 327 static int ceph_fill_fragtree(struct inode *inode,
 328                               struct ceph_frag_tree_head *fragtree,
 329                               struct ceph_mds_reply_dirfrag *dirinfo)
 330 {
 331         struct ceph_inode_info *ci = ceph_inode(inode);
 332         struct ceph_inode_frag *frag, *prev_frag = NULL;
 333         struct rb_node *rb_node;
 334         unsigned i, split_by, nsplits;
 335         u32 id;
 336         bool update = false;
 337
 338         mutex_lock(&ci->i_fragtree_mutex);
 339         nsplits = le32_to_cpu(fragtree->nsplits);
 340         if (nsplits != ci->i_fragtree_nsplits) {
 341                 update = true;
 342         } else if (nsplits) {
 343                 i = prandom_u32() % nsplits;
 344                 id = le32_to_cpu(fragtree->splits[i].frag);
 345                 if (!__ceph_find_frag(ci, id))
 346                         update = true;
 347         } else if (!RB_EMPTY_ROOT(&ci->i_fragtree)) {
 348                 rb_node = rb_first(&ci->i_fragtree);
 349                 frag = rb_entry(rb_node, struct ceph_inode_frag, node);
 350                 if (frag->frag != ceph_frag_make(0, 0) || rb_next(rb_node))
 351                         update = true;
 352         }
 353         if (!update && dirinfo) {
 354                 id = le32_to_cpu(dirinfo->frag);
 355                 if (id != __ceph_choose_frag(ci, id, NULL, NULL))
 356                         update = true;
 357         }
 358         if (!update)
 359                 goto out_unlock;
 360
 361         if (nsplits > 1) {
 362                 sort(fragtree->splits, nsplits, sizeof(fragtree->splits[0]),
 363                      frag_tree_split_cmp, NULL);
 364         }
 365
 366         dout("fill_fragtree %llx.%llx\n", ceph_vinop(inode));
 367         rb_node = rb_first(&ci->i_fragtree);
 368         for (i = 0; i < nsplits; i++) {
 369                 id = le32_to_cpu(fragtree->splits[i].frag);
 370                 split_by = le32_to_cpu(fragtree->splits[i].by);
 371                 if (split_by == 0 || ceph_frag_bits(id) + split_by > 24) {
 372                         pr_err("fill_fragtree %llx.%llx invalid split %d/%u, "
 373                                "frag %x split by %d\n", ceph_vinop(inode),
 374                                i, nsplits, id, split_by);
 375                         continue;
 376                 }
 377                 frag = NULL;
 378                 while (rb_node) {
 379                         frag = rb_entry(rb_node, struct ceph_inode_frag, node);
 380                         if (ceph_frag_compare(frag->frag, id) >= 0) {
 381                                 if (frag->frag != id)
 382                                         frag = NULL;
 383                                 else
 384                                         rb_node = rb_next(rb_node);
 385                                 break;
 386                         }
 387                         rb_node = rb_next(rb_node);
 388                         /* delete stale split/leaf node */
 389                         if (frag->split_by > 0 ||
 390                             !is_frag_child(frag->frag, prev_frag)) {
 391                                 rb_erase(&frag->node, &ci->i_fragtree);
 392                                 if (frag->split_by > 0)
 393                                         ci->i_fragtree_nsplits--;
 394                                 kfree(frag);
 395                         }
 396                         frag = NULL;
 397                 }
 398                 if (!frag) {
 399                         frag = __get_or_create_frag(ci, id);
 400                         if (IS_ERR(frag))
 401                                 continue;
 402                 }
 403                 if (frag->split_by == 0)
 404                         ci->i_fragtree_nsplits++;
 405                 frag->split_by = split_by;
 406                 dout(" frag %x split by %d\n", frag->frag, frag->split_by);
 407                 prev_frag = frag;
 408         }
 409         while (rb_node) {
 410                 frag = rb_entry(rb_node, struct ceph_inode_frag, node);
 411                 rb_node = rb_next(rb_node);
 412                 /* delete stale split/leaf node */
 413                 if (frag->split_by > 0 ||
 414                     !is_frag_child(frag->frag, prev_frag)) {
 415                         rb_erase(&frag->node, &ci->i_fragtree);
 416                         if (frag->split_by > 0)
 417                                 ci->i_fragtree_nsplits--;
 418                         kfree(frag);
 419                 }
 420         }
 421 out_unlock:
 422         mutex_unlock(&ci->i_fragtree_mutex);
 423         return 0;
 424 }
 425
 426 /*
 427  * initialize a newly allocated inode.
 428  */
 429 struct inode *ceph_alloc_inode(struct super_block *sb)
 430 {
 431         struct ceph_inode_info *ci;
 432         int i;
 433
 434         ci = kmem_cache_alloc(ceph_inode_cachep, GFP_NOFS);
 435         if (!ci)
 436                 return NULL;
 437
 438         dout("alloc_inode %p\n", &ci->vfs_inode);
 439
 440         spin_lock_init(&ci->i_ceph_lock);
 441
 442         ci->i_version = 0;
 443         ci->i_inline_version = 0;
 444         ci->i_time_warp_seq = 0;
 445         ci->i_ceph_flags = 0;
 446         atomic64_set(&ci->i_ordered_count, 1);
 447         atomic64_set(&ci->i_release_count, 1);
 448         atomic64_set(&ci->i_complete_seq[0], 0);
 449         atomic64_set(&ci->i_complete_seq[1], 0);
 450         ci->i_symlink = NULL;
 451
 452         ci->i_max_bytes = 0;
 453         ci->i_max_files = 0;
 454
 455         memset(&ci->i_dir_layout, 0, sizeof(ci->i_dir_layout));
 456         memset(&ci->i_cached_layout, 0, sizeof(ci->i_cached_layout));
 457         RCU_INIT_POINTER(ci->i_layout.pool_ns, NULL);
 458
 459         ci->i_fragtree = RB_ROOT;
 460         mutex_init(&ci->i_fragtree_mutex);
 461
 462         ci->i_xattrs.blob = NULL;
 463         ci->i_xattrs.prealloc_blob = NULL;
 464         ci->i_xattrs.dirty = false;
 465         ci->i_xattrs.index = RB_ROOT;
 466         ci->i_xattrs.count = 0;
 467         ci->i_xattrs.names_size = 0;
 468         ci->i_xattrs.vals_size = 0;
 469         ci->i_xattrs.version = 0;
 470         ci->i_xattrs.index_version = 0;
 471
 472         ci->i_caps = RB_ROOT;
 473         ci->i_auth_cap = NULL;
 474         ci->i_dirty_caps = 0;
 475         ci->i_flushing_caps = 0;
 476         INIT_LIST_HEAD(&ci->i_dirty_item);
 477         INIT_LIST_HEAD(&ci->i_flushing_item);
 478         ci->i_prealloc_cap_flush = NULL;
 479         INIT_LIST_HEAD(&ci->i_cap_flush_list);
 480         init_waitqueue_head(&ci->i_cap_wq);
 481         ci->i_hold_caps_max = 0;
 482         INIT_LIST_HEAD(&ci->i_cap_delay_list);
 483         INIT_LIST_HEAD(&ci->i_cap_snaps);
 484         ci->i_head_snapc = NULL;
 485         ci->i_snap_caps = 0;
 486
 487         ci->i_last_rd = ci->i_last_wr = jiffies - 3600 * HZ;
 488         for (i = 0; i < CEPH_FILE_MODE_BITS; i++)
 489                 ci->i_nr_by_mode[i] = 0;
 490
 491         mutex_init(&ci->i_truncate_mutex);
 492         ci->i_truncate_seq = 0;
 493         ci->i_truncate_size = 0;
 494         ci->i_truncate_pending = 0;
 495
 496         ci->i_max_size = 0;
 497         ci->i_reported_size = 0;
 498         ci->i_wanted_max_size = 0;
 499         ci->i_requested_max_size = 0;
 500
 501         ci->i_pin_ref = 0;
 502         ci->i_rd_ref = 0;
 503         ci->i_rdcache_ref = 0;
 504         ci->i_wr_ref = 0;
 505         ci->i_wb_ref = 0;
 506         ci->i_fx_ref = 0;
 507         ci->i_wrbuffer_ref = 0;
 508         ci->i_wrbuffer_ref_head = 0;
 509         atomic_set(&ci->i_filelock_ref, 0);
 510         atomic_set(&ci->i_shared_gen, 1);
 511         ci->i_rdcache_gen = 0;
 512         ci->i_rdcache_revoking = 0;
 513
 514         INIT_LIST_HEAD(&ci->i_unsafe_dirops);
 515         INIT_LIST_HEAD(&ci->i_unsafe_iops);
 516         spin_lock_init(&ci->i_unsafe_lock);
 517
 518         ci->i_snap_realm = NULL;
 519         INIT_LIST_HEAD(&ci->i_snap_realm_item);
 520         INIT_LIST_HEAD(&ci->i_snap_flush_item);
 521
 522         INIT_WORK(&ci->i_work, ceph_inode_work);
 523         ci->i_work_mask = 0;
 524         memset(&ci->i_btime, '\0', sizeof(ci->i_btime));
 525
 526         ceph_fscache_inode_init(ci);
 527
 528         ci->i_meta_err = 0;
 529
 530         return &ci->vfs_inode;
 531 }
 532
 533 void ceph_free_inode(struct inode *inode)
 534 {
 535         struct ceph_inode_info *ci = ceph_inode(inode);
 536
 537         kfree(ci->i_symlink);
 538         kmem_cache_free(ceph_inode_cachep, ci);
 539 }
 540
 541 void ceph_evict_inode(struct inode *inode)
 542 {
 543         struct ceph_inode_info *ci = ceph_inode(inode);
 544         struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
 545         struct ceph_inode_frag *frag;
 546         struct rb_node *n;
 547
 548         dout("evict_inode %p ino %llx.%llx\n", inode, ceph_vinop(inode));
 549
 550         percpu_counter_dec(&mdsc->metric.total_inodes);
 551
 552         truncate_inode_pages_final(&inode->i_data);
 553         clear_inode(inode);
 554
 555         ceph_fscache_unregister_inode_cookie(ci);
 556
 557         __ceph_remove_caps(ci);
 558
 559         if (__ceph_has_any_quota(ci))
 560                 ceph_adjust_quota_realms_count(inode, false);
 561
 562         /*
 563          * we may still have a snap_realm reference if there are stray
 564          * caps in i_snap_caps.
 565          */
 566         if (ci->i_snap_realm) {
 567                 if (ceph_snap(inode) == CEPH_NOSNAP) {
 568                         struct ceph_snap_realm *realm = ci->i_snap_realm;
 569                         dout(" dropping residual ref to snap realm %p\n",
 570                              realm);
 571                         spin_lock(&realm->inodes_with_caps_lock);
 572                         list_del_init(&ci->i_snap_realm_item);
 573                         ci->i_snap_realm = NULL;
 574                         if (realm->ino == ci->i_vino.ino)
 575                                 realm->inode = NULL;
 576                         spin_unlock(&realm->inodes_with_caps_lock);
 577                         ceph_put_snap_realm(mdsc, realm);
 578                 } else {
 579                         ceph_put_snapid_map(mdsc, ci->i_snapid_map);
 580                         ci->i_snap_realm = NULL;
 581                 }
 582         }
 583
 584         while ((n = rb_first(&ci->i_fragtree)) != NULL) {
 585                 frag = rb_entry(n, struct ceph_inode_frag, node);
 586                 rb_erase(n, &ci->i_fragtree);
 587                 kfree(frag);
 588         }
 589         ci->i_fragtree_nsplits = 0;
 590
 591         __ceph_destroy_xattrs(ci);
 592         if (ci->i_xattrs.blob)
 593                 ceph_buffer_put(ci->i_xattrs.blob);
 594         if (ci->i_xattrs.prealloc_blob)
 595                 ceph_buffer_put(ci->i_xattrs.prealloc_blob);
 596
 597         ceph_put_string(rcu_dereference_raw(ci->i_layout.pool_ns));
 598         ceph_put_string(rcu_dereference_raw(ci->i_cached_layout.pool_ns));
 599 }
 600
 601 static inline blkcnt_t calc_inode_blocks(u64 size)
 602 {
 603         return (size + (1<<9) - 1) >> 9;
 604 }
 605
 606 /*
 607  * Helpers to fill in size, ctime, mtime, and atime.  We have to be
 608  * careful because either the client or MDS may have more up to date
 609  * info, depending on which capabilities are held, and whether
 610  * time_warp_seq or truncate_seq have increased.  (Ordinarily, mtime
 611  * and size are monotonically increasing, except when utimes() or
 612  * truncate() increments the corresponding _seq values.)
 613  */
 614 int ceph_fill_file_size(struct inode *inode, int issued,
 615                         u32 truncate_seq, u64 truncate_size, u64 size)
 616 {
 617         struct ceph_inode_info *ci = ceph_inode(inode);
 618         int queue_trunc = 0;
 619
 620         if (ceph_seq_cmp(truncate_seq, ci->i_truncate_seq) > 0 ||
 621             (truncate_seq == ci->i_truncate_seq && size > inode->i_size)) {
 622                 dout("size %lld -> %llu\n", inode->i_size, size);
 623                 if (size > 0 && S_ISDIR(inode->i_mode)) {
 624                         pr_err("fill_file_size non-zero size for directory\n");
 625                         size = 0;
 626                 }
 627                 i_size_write(inode, size);
 628                 inode->i_blocks = calc_inode_blocks(size);
 629                 ci->i_reported_size = size;
 630                 if (truncate_seq != ci->i_truncate_seq) {
 631                         dout("truncate_seq %u -> %u\n",
 632                              ci->i_truncate_seq, truncate_seq);
 633                         ci->i_truncate_seq = truncate_seq;
 634
 635                         /* the MDS should have revoked these caps */
 636                         WARN_ON_ONCE(issued & (CEPH_CAP_FILE_EXCL |
 637                                                CEPH_CAP_FILE_RD |
 638                                                CEPH_CAP_FILE_WR |
 639                                                CEPH_CAP_FILE_LAZYIO));
 640                         /*
 641                          * If we hold relevant caps, or in the case where we're
 642                          * not the only client referencing this file and we
 643                          * don't hold those caps, then we need to check whether
 644                          * the file is either opened or mmaped
 645                          */
 646                         if ((issued & (CEPH_CAP_FILE_CACHE|
 647                                        CEPH_CAP_FILE_BUFFER)) ||
 648                             mapping_mapped(inode->i_mapping) ||
 649                             __ceph_is_file_opened(ci)) {
 650                                 ci->i_truncate_pending++;
 651                                 queue_trunc = 1;
 652                         }
 653                 }
 654         }
 655         if (ceph_seq_cmp(truncate_seq, ci->i_truncate_seq) >= 0 &&
 656             ci->i_truncate_size != truncate_size) {
 657                 dout("truncate_size %lld -> %llu\n", ci->i_truncate_size,
 658                      truncate_size);
 659                 ci->i_truncate_size = truncate_size;
 660         }
 661
 662         if (queue_trunc)
 663                 ceph_fscache_invalidate(inode);
 664
 665         return queue_trunc;
 666 }
 667
 668 void ceph_fill_file_time(struct inode *inode, int issued,
 669                          u64 time_warp_seq, struct timespec64 *ctime,
 670                          struct timespec64 *mtime, struct timespec64 *atime)
 671 {
 672         struct ceph_inode_info *ci = ceph_inode(inode);
 673         int warn = 0;
 674
 675         if (issued & (CEPH_CAP_FILE_EXCL|
 676                       CEPH_CAP_FILE_WR|
 677                       CEPH_CAP_FILE_BUFFER|
 678                       CEPH_CAP_AUTH_EXCL|
 679                       CEPH_CAP_XATTR_EXCL)) {
 680                 if (ci->i_version == 0 ||
 681                     timespec64_compare(ctime, &inode->i_ctime) > 0) {
 682                         dout("ctime %lld.%09ld -> %lld.%09ld inc w/ cap\n",
 683                              inode->i_ctime.tv_sec, inode->i_ctime.tv_nsec,
 684                              ctime->tv_sec, ctime->tv_nsec);
 685                         inode->i_ctime = *ctime;
 686                 }
 687                 if (ci->i_version == 0 ||
 688                     ceph_seq_cmp(time_warp_seq, ci->i_time_warp_seq) > 0) {
 689                         /* the MDS did a utimes() */
 690                         dout("mtime %lld.%09ld -> %lld.%09ld "
 691                              "tw %d -> %d\n",
 692                              inode->i_mtime.tv_sec, inode->i_mtime.tv_nsec,
 693                              mtime->tv_sec, mtime->tv_nsec,
 694                              ci->i_time_warp_seq, (int)time_warp_seq);
 695
 696                         inode->i_mtime = *mtime;
 697                         inode->i_atime = *atime;
 698                         ci->i_time_warp_seq = time_warp_seq;
 699                 } else if (time_warp_seq == ci->i_time_warp_seq) {
 700                         /* nobody did utimes(); take the max */
 701                         if (timespec64_compare(mtime, &inode->i_mtime) > 0) {
 702                                 dout("mtime %lld.%09ld -> %lld.%09ld inc\n",
 703                                      inode->i_mtime.tv_sec,
 704                                      inode->i_mtime.tv_nsec,
 705                                      mtime->tv_sec, mtime->tv_nsec);
 706                                 inode->i_mtime = *mtime;
 707                         }
 708                         if (timespec64_compare(atime, &inode->i_atime) > 0) {
 709                                 dout("atime %lld.%09ld -> %lld.%09ld inc\n",
 710                                      inode->i_atime.tv_sec,
 711                                      inode->i_atime.tv_nsec,
 712                                      atime->tv_sec, atime->tv_nsec);
 713                                 inode->i_atime = *atime;
 714                         }
 715                 } else if (issued & CEPH_CAP_FILE_EXCL) {
 716                         /* we did a utimes(); ignore mds values */
 717                 } else {
 718                         warn = 1;
 719                 }
 720         } else {
 721                 /* we have no write|excl caps; whatever the MDS says is true */
 722                 if (ceph_seq_cmp(time_warp_seq, ci->i_time_warp_seq) >= 0) {
 723                         inode->i_ctime = *ctime;
 724                         inode->i_mtime = *mtime;
 725                         inode->i_atime = *atime;
 726                         ci->i_time_warp_seq = time_warp_seq;
 727                 } else {
 728                         warn = 1;
 729                 }
 730         }
 731         if (warn) /* time_warp_seq shouldn't go backwards */
 732                 dout("%p mds time_warp_seq %llu < %u\n",
 733                      inode, time_warp_seq, ci->i_time_warp_seq);
 734 }
 735
 736 /*
 737  * Populate an inode based on info from mds.  May be called on new or
 738  * existing inodes.
 739  */
 740 int ceph_fill_inode(struct inode *inode, struct page *locked_page,
 741                     struct ceph_mds_reply_info_in *iinfo,
 742                     struct ceph_mds_reply_dirfrag *dirinfo,
 743                     struct ceph_mds_session *session, int cap_fmode,
 744                     struct ceph_cap_reservation *caps_reservation)
 745 {
 746         struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
 747         struct ceph_mds_reply_inode *info = iinfo->in;
 748         struct ceph_inode_info *ci = ceph_inode(inode);
 749         int issued, new_issued, info_caps;
 750         struct timespec64 mtime, atime, ctime;
 751         struct ceph_buffer *xattr_blob = NULL;
 752         struct ceph_buffer *old_blob = NULL;
 753         struct ceph_string *pool_ns = NULL;
 754         struct ceph_cap *new_cap = NULL;
 755         int err = 0;
 756         bool wake = false;
 757         bool queue_trunc = false;
 758         bool new_version = false;
 759         bool fill_inline = false;
 760
 761         dout("%s %p ino %llx.%llx v %llu had %llu\n", __func__,
 762              inode, ceph_vinop(inode), le64_to_cpu(info->version),
 763              ci->i_version);
 764
 765         info_caps = le32_to_cpu(info->cap.caps);
 766
 767         /* prealloc new cap struct */
 768         if (info_caps && ceph_snap(inode) == CEPH_NOSNAP) {
 769                 new_cap = ceph_get_cap(mdsc, caps_reservation);
 770                 if (!new_cap)
 771                         return -ENOMEM;
 772         }
 773
 774         /*
 775          * prealloc xattr data, if it looks like we'll need it.  only
 776          * if len > 4 (meaning there are actually xattrs; the first 4
 777          * bytes are the xattr count).
 778          */
 779         if (iinfo->xattr_len > 4) {
 780                 xattr_blob = ceph_buffer_new(iinfo->xattr_len, GFP_NOFS);
 781                 if (!xattr_blob)
 782                         pr_err("%s ENOMEM xattr blob %d bytes\n", __func__,
 783                                iinfo->xattr_len);
 784         }
 785
 786         if (iinfo->pool_ns_len > 0)
 787                 pool_ns = ceph_find_or_create_string(iinfo->pool_ns_data,
 788                                                      iinfo->pool_ns_len);
 789
 790         if (ceph_snap(inode) != CEPH_NOSNAP && !ci->i_snapid_map)
 791                 ci->i_snapid_map = ceph_get_snapid_map(mdsc, ceph_snap(inode));
 792
 793         spin_lock(&ci->i_ceph_lock);
 794
 795         /*
 796          * provided version will be odd if inode value is projected,
 797          * even if stable.  skip the update if we have newer stable
 798          * info (ours>=theirs, e.g. due to racing mds replies), unless
 799          * we are getting projected (unstable) info (in which case the
 800          * version is odd, and we want ours>theirs).
 801          *   us   them
 802          *   2    2     skip
 803          *   3    2     skip
 804          *   3    3     update
 805          */
 806         if (ci->i_version == 0 ||
 807             ((info->cap.flags & CEPH_CAP_FLAG_AUTH) &&
 808              le64_to_cpu(info->version) > (ci->i_version & ~1)))
 809                 new_version = true;
 810
 811         /* Update change_attribute */
 812         inode_set_max_iversion_raw(inode, iinfo->change_attr);
 813
 814         __ceph_caps_issued(ci, &issued);
 815         issued |= __ceph_caps_dirty(ci);
 816         new_issued = ~issued & info_caps;
 817
 818         /* update inode */
 819         inode->i_rdev = le32_to_cpu(info->rdev);
 820         /* directories have fl_stripe_unit set to zero */
 821         if (le32_to_cpu(info->layout.fl_stripe_unit))
 822                 inode->i_blkbits =
 823                         fls(le32_to_cpu(info->layout.fl_stripe_unit)) - 1;
 824         else
 825                 inode->i_blkbits = CEPH_BLOCK_SHIFT;
 826
 827         __ceph_update_quota(ci, iinfo->max_bytes, iinfo->max_files);
 828
 829         if ((new_version || (new_issued & CEPH_CAP_AUTH_SHARED)) &&
 830             (issued & CEPH_CAP_AUTH_EXCL) == 0) {
 831                 inode->i_mode = le32_to_cpu(info->mode);
 832                 inode->i_uid = make_kuid(&init_user_ns, le32_to_cpu(info->uid));
 833                 inode->i_gid = make_kgid(&init_user_ns, le32_to_cpu(info->gid));
 834                 dout("%p mode 0%o uid.gid %d.%d\n", inode, inode->i_mode,
 835                      from_kuid(&init_user_ns, inode->i_uid),
 836                      from_kgid(&init_user_ns, inode->i_gid));
 837                 ceph_decode_timespec64(&ci->i_btime, &iinfo->btime);
 838                 ceph_decode_timespec64(&ci->i_snap_btime, &iinfo->snap_btime);
 839         }
 840
 841         if ((new_version || (new_issued & CEPH_CAP_LINK_SHARED)) &&
 842             (issued & CEPH_CAP_LINK_EXCL) == 0)
 843                 set_nlink(inode, le32_to_cpu(info->nlink));
 844
 845         if (new_version || (new_issued & CEPH_CAP_ANY_RD)) {
 846                 /* be careful with mtime, atime, size */
 847                 ceph_decode_timespec64(&atime, &info->atime);
 848                 ceph_decode_timespec64(&mtime, &info->mtime);
 849                 ceph_decode_timespec64(&ctime, &info->ctime);
 850                 ceph_fill_file_time(inode, issued,
 851                                 le32_to_cpu(info->time_warp_seq),
 852                                 &ctime, &mtime, &atime);
 853         }
 854
 855         if (new_version || (info_caps & CEPH_CAP_FILE_SHARED)) {
 856                 ci->i_files = le64_to_cpu(info->files);
 857                 ci->i_subdirs = le64_to_cpu(info->subdirs);
 858         }
 859
 860         if (new_version ||
 861             (new_issued & (CEPH_CAP_ANY_FILE_RD | CEPH_CAP_ANY_FILE_WR))) {
 862                 s64 old_pool = ci->i_layout.pool_id;
 863                 struct ceph_string *old_ns;
 864
 865                 ceph_file_layout_from_legacy(&ci->i_layout, &info->layout);
 866                 old_ns = rcu_dereference_protected(ci->i_layout.pool_ns,
 867                                         lockdep_is_held(&ci->i_ceph_lock));
 868                 rcu_assign_pointer(ci->i_layout.pool_ns, pool_ns);
 869
 870                 if (ci->i_layout.pool_id != old_pool || pool_ns != old_ns)
 871                         ci->i_ceph_flags &= ~CEPH_I_POOL_PERM;
 872
 873                 pool_ns = old_ns;
 874
 875                 queue_trunc = ceph_fill_file_size(inode, issued,
 876                                         le32_to_cpu(info->truncate_seq),
 877                                         le64_to_cpu(info->truncate_size),
 878                                         le64_to_cpu(info->size));
 879                 /* only update max_size on auth cap */
 880                 if ((info->cap.flags & CEPH_CAP_FLAG_AUTH) &&
 881                     ci->i_max_size != le64_to_cpu(info->max_size)) {
 882                         dout("max_size %lld -> %llu\n", ci->i_max_size,
 883                                         le64_to_cpu(info->max_size));
 884                         ci->i_max_size = le64_to_cpu(info->max_size);
 885                 }
 886         }
 887
 888         /* layout and rstat are not tracked by capability, update them if
 889          * the inode info is from auth mds */
 890         if (new_version || (info->cap.flags & CEPH_CAP_FLAG_AUTH)) {
 891                 if (S_ISDIR(inode->i_mode)) {
 892                         ci->i_dir_layout = iinfo->dir_layout;
 893                         ci->i_rbytes = le64_to_cpu(info->rbytes);
 894                         ci->i_rfiles = le64_to_cpu(info->rfiles);
 895                         ci->i_rsubdirs = le64_to_cpu(info->rsubdirs);
 896                         ci->i_dir_pin = iinfo->dir_pin;
 897                         ceph_decode_timespec64(&ci->i_rctime, &info->rctime);
 898                 }
 899         }
 900
 901         /* xattrs */
 902         /* note that if i_xattrs.len <= 4, i_xattrs.data will still be NULL. */
 903         if ((ci->i_xattrs.version == 0 || !(issued & CEPH_CAP_XATTR_EXCL))  &&
 904             le64_to_cpu(info->xattr_version) > ci->i_xattrs.version) {
 905                 if (ci->i_xattrs.blob)
 906                         old_blob = ci->i_xattrs.blob;
 907                 ci->i_xattrs.blob = xattr_blob;
 908                 if (xattr_blob)
 909                         memcpy(ci->i_xattrs.blob->vec.iov_base,
 910                                iinfo->xattr_data, iinfo->xattr_len);
 911                 ci->i_xattrs.version = le64_to_cpu(info->xattr_version);
 912                 ceph_forget_all_cached_acls(inode);
 913                 ceph_security_invalidate_secctx(inode);
 914                 xattr_blob = NULL;
 915         }
 916
 917         /* finally update i_version */
 918         if (le64_to_cpu(info->version) > ci->i_version)
 919                 ci->i_version = le64_to_cpu(info->version);
 920
 921         inode->i_mapping->a_ops = &ceph_aops;
 922
 923         switch (inode->i_mode & S_IFMT) {
 924         case S_IFIFO:
 925         case S_IFBLK:
 926         case S_IFCHR:
 927         case S_IFSOCK:
 928                 inode->i_blkbits = PAGE_SHIFT;
 929                 init_special_inode(inode, inode->i_mode, inode->i_rdev);
 930                 inode->i_op = &ceph_file_iops;
 931                 break;
 932         case S_IFREG:
 933                 inode->i_op = &ceph_file_iops;
 934                 inode->i_fop = &ceph_file_fops;
 935                 break;
 936         case S_IFLNK:
 937                 inode->i_op = &ceph_symlink_iops;
 938                 if (!ci->i_symlink) {
 939                         u32 symlen = iinfo->symlink_len;
 940                         char *sym;
 941
 942                         spin_unlock(&ci->i_ceph_lock);
 943
 944                         if (symlen != i_size_read(inode)) {
 945                                 pr_err("%s %llx.%llx BAD symlink "
 946                                         "size %lld\n", __func__,
 947                                         ceph_vinop(inode),
 948                                         i_size_read(inode));
 949                                 i_size_write(inode, symlen);
 950                                 inode->i_blocks = calc_inode_blocks(symlen);
 951                         }
 952
 953                         err = -ENOMEM;
 954                         sym = kstrndup(iinfo->symlink, symlen, GFP_NOFS);
 955                         if (!sym)
 956                                 goto out;
 957
 958                         spin_lock(&ci->i_ceph_lock);
 959                         if (!ci->i_symlink)
 960                                 ci->i_symlink = sym;
 961                         else
 962                                 kfree(sym); /* lost a race */
 963                 }
 964                 inode->i_link = ci->i_symlink;
 965                 break;
 966         case S_IFDIR:
 967                 inode->i_op = &ceph_dir_iops;
 968                 inode->i_fop = &ceph_dir_fops;
 969                 break;
 970         default:
 971                 pr_err("%s %llx.%llx BAD mode 0%o\n", __func__,
 972                        ceph_vinop(inode), inode->i_mode);
 973         }
 974
 975         /* were we issued a capability? */
 976         if (info_caps) {
 977                 if (ceph_snap(inode) == CEPH_NOSNAP) {
 978                         ceph_add_cap(inode, session,
 979                                      le64_to_cpu(info->cap.cap_id),
 980                                      info_caps,
 981                                      le32_to_cpu(info->cap.wanted),
 982                                      le32_to_cpu(info->cap.seq),
 983                                      le32_to_cpu(info->cap.mseq),
 984                                      le64_to_cpu(info->cap.realm),
 985                                      info->cap.flags, &new_cap);
 986
 987                         /* set dir completion flag? */
 988                         if (S_ISDIR(inode->i_mode) &&
 989                             ci->i_files == 0 && ci->i_subdirs == 0 &&
 990                             (info_caps & CEPH_CAP_FILE_SHARED) &&
 991                             (issued & CEPH_CAP_FILE_EXCL) == 0 &&
 992                             !__ceph_dir_is_complete(ci)) {
 993                                 dout(" marking %p complete (empty)\n", inode);
 994                                 i_size_write(inode, 0);
 995                                 __ceph_dir_set_complete(ci,
 996                                         atomic64_read(&ci->i_release_count),
 997                                         atomic64_read(&ci->i_ordered_count));
 998                         }
 999
1000                         wake = true;
1001                 } else {
1002                         dout(" %p got snap_caps %s\n", inode,
1003                              ceph_cap_string(info_caps));
1004                         ci->i_snap_caps |= info_caps;
1005                 }
1006         }
1007
1008         if (iinfo->inline_version > 0 &&
1009             iinfo->inline_version >= ci->i_inline_version) {
1010                 int cache_caps = CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO;
1011                 ci->i_inline_version = iinfo->inline_version;
1012                 if (ci->i_inline_version != CEPH_INLINE_NONE &&
1013                     (locked_page || (info_caps & cache_caps)))
1014                         fill_inline = true;
1015         }
1016
1017         if (cap_fmode >= 0) {
1018                 if (!info_caps)
1019                         pr_warn("mds issued no caps on %llx.%llx\n",
1020                                 ceph_vinop(inode));
1021                 __ceph_touch_fmode(ci, mdsc, cap_fmode);
1022         }
1023
1024         spin_unlock(&ci->i_ceph_lock);
1025
1026         if (fill_inline)
1027                 ceph_fill_inline_data(inode, locked_page,
1028                                       iinfo->inline_data, iinfo->inline_len);
1029
1030         if (wake)
1031                 wake_up_all(&ci->i_cap_wq);
1032
1033         /* queue truncate if we saw i_size decrease */
1034         if (queue_trunc)
1035                 ceph_queue_vmtruncate(inode);
1036
1037         /* populate frag tree */
1038         if (S_ISDIR(inode->i_mode))
1039                 ceph_fill_fragtree(inode, &info->fragtree, dirinfo);
1040
1041         /* update delegation info? */
1042         if (dirinfo)
1043                 ceph_fill_dirfrag(inode, dirinfo);
1044
1045         err = 0;
1046 out:
1047         if (new_cap)
1048                 ceph_put_cap(mdsc, new_cap);
1049         ceph_buffer_put(old_blob);
1050         ceph_buffer_put(xattr_blob);
1051         ceph_put_string(pool_ns);
1052         return err;
1053 }
1054
1055 /*
1056  * caller should hold session s_mutex and dentry->d_lock.
1057  */
1058 static void __update_dentry_lease(struct inode *dir, struct dentry *dentry,
1059                                   struct ceph_mds_reply_lease *lease,
1060                                   struct ceph_mds_session *session,
1061                                   unsigned long from_time,
1062                                   struct ceph_mds_session **old_lease_session)
1063 {
1064         struct ceph_dentry_info *di = ceph_dentry(dentry);
1065         unsigned mask = le16_to_cpu(lease->mask);
1066         long unsigned duration = le32_to_cpu(lease->duration_ms);
1067         long unsigned ttl = from_time + (duration * HZ) / 1000;
1068         long unsigned half_ttl = from_time + (duration * HZ / 2) / 1000;
1069
1070         dout("update_dentry_lease %p duration %lu ms ttl %lu\n",
1071              dentry, duration, ttl);
1072
1073         /* only track leases on regular dentries */
1074         if (ceph_snap(dir) != CEPH_NOSNAP)
1075                 return;
1076
1077         if (mask & CEPH_LEASE_PRIMARY_LINK)
1078                 di->flags |= CEPH_DENTRY_PRIMARY_LINK;
1079         else
1080                 di->flags &= ~CEPH_DENTRY_PRIMARY_LINK;
1081
1082         di->lease_shared_gen = atomic_read(&ceph_inode(dir)->i_shared_gen);
1083         if (!(mask & CEPH_LEASE_VALID)) {
1084                 __ceph_dentry_dir_lease_touch(di);
1085                 return;
1086         }
1087
1088         if (di->lease_gen == session->s_cap_gen &&
1089             time_before(ttl, di->time))
1090                 return;  /* we already have a newer lease. */
1091
1092         if (di->lease_session && di->lease_session != session) {
1093                 *old_lease_session = di->lease_session;
1094                 di->lease_session = NULL;
1095         }
1096
1097         if (!di->lease_session)
1098                 di->lease_session = ceph_get_mds_session(session);
1099         di->lease_gen = session->s_cap_gen;
1100         di->lease_seq = le32_to_cpu(lease->seq);
1101         di->lease_renew_after = half_ttl;
1102         di->lease_renew_from = 0;
1103         di->time = ttl;
1104
1105         __ceph_dentry_lease_touch(di);
1106 }
1107
1108 static inline void update_dentry_lease(struct inode *dir, struct dentry *dentry,
1109                                         struct ceph_mds_reply_lease *lease,
1110                                         struct ceph_mds_session *session,
1111                                         unsigned long from_time)
1112 {
1113         struct ceph_mds_session *old_lease_session = NULL;
1114         spin_lock(&dentry->d_lock);
1115         __update_dentry_lease(dir, dentry, lease, session, from_time,
1116                               &old_lease_session);
1117         spin_unlock(&dentry->d_lock);
1118         if (old_lease_session)
1119                 ceph_put_mds_session(old_lease_session);
1120 }
1121
1122 /*
1123  * update dentry lease without having parent inode locked
1124  */
1125 static void update_dentry_lease_careful(struct dentry *dentry,
1126                                         struct ceph_mds_reply_lease *lease,
1127                                         struct ceph_mds_session *session,
1128                                         unsigned long from_time,
1129                                         char *dname, u32 dname_len,
1130                                         struct ceph_vino *pdvino,
1131                                         struct ceph_vino *ptvino)
1132
1133 {
1134         struct inode *dir;
1135         struct ceph_mds_session *old_lease_session = NULL;
1136
1137         spin_lock(&dentry->d_lock);
1138         /* make sure dentry's name matches target */
1139         if (dentry->d_name.len != dname_len ||
1140             memcmp(dentry->d_name.name, dname, dname_len))
1141                 goto out_unlock;
1142
1143         dir = d_inode(dentry->d_parent);
1144         /* make sure parent matches dvino */
1145         if (!ceph_ino_compare(dir, pdvino))
1146                 goto out_unlock;
1147
1148         /* make sure dentry's inode matches target. NULL ptvino means that
1149          * we expect a negative dentry */
1150         if (ptvino) {
1151                 if (d_really_is_negative(dentry))
1152                         goto out_unlock;
1153                 if (!ceph_ino_compare(d_inode(dentry), ptvino))
1154                         goto out_unlock;
1155         } else {
1156                 if (d_really_is_positive(dentry))
1157                         goto out_unlock;
1158         }
1159
1160         __update_dentry_lease(dir, dentry, lease, session,
1161                               from_time, &old_lease_session);
1162 out_unlock:
1163         spin_unlock(&dentry->d_lock);
1164         if (old_lease_session)
1165                 ceph_put_mds_session(old_lease_session);
1166 }
1167
1168 /*
1169  * splice a dentry to an inode.
1170  * caller must hold directory i_mutex for this to be safe.
1171  */
1172 static int splice_dentry(struct dentry **pdn, struct inode *in)
1173 {
1174         struct dentry *dn = *pdn;
1175         struct dentry *realdn;
1176
1177         BUG_ON(d_inode(dn));
1178
1179         if (S_ISDIR(in->i_mode)) {
1180                 /* If inode is directory, d_splice_alias() below will remove
1181                  * 'realdn' from its origin parent. We need to ensure that
1182                  * origin parent's readdir cache will not reference 'realdn'
1183                  */
1184                 realdn = d_find_any_alias(in);
1185                 if (realdn) {
1186                         struct ceph_dentry_info *di = ceph_dentry(realdn);
1187                         spin_lock(&realdn->d_lock);
1188
1189                         realdn->d_op->d_prune(realdn);
1190
1191                         di->time = jiffies;
1192                         di->lease_shared_gen = 0;
1193                         di->offset = 0;
1194
1195                         spin_unlock(&realdn->d_lock);
1196                         dput(realdn);
1197                 }
1198         }
1199
1200         /* dn must be unhashed */
1201         if (!d_unhashed(dn))
1202                 d_drop(dn);
1203         realdn = d_splice_alias(in, dn);
1204         if (IS_ERR(realdn)) {
1205                 pr_err("splice_dentry error %ld %p inode %p ino %llx.%llx\n",
1206                        PTR_ERR(realdn), dn, in, ceph_vinop(in));
1207                 return PTR_ERR(realdn);
1208         }
1209
1210         if (realdn) {
1211                 dout("dn %p (%d) spliced with %p (%d) "
1212                      "inode %p ino %llx.%llx\n",
1213                      dn, d_count(dn),
1214                      realdn, d_count(realdn),
1215                      d_inode(realdn), ceph_vinop(d_inode(realdn)));
1216                 dput(dn);
1217                 *pdn = realdn;
1218         } else {
1219                 BUG_ON(!ceph_dentry(dn));
1220                 dout("dn %p attached to %p ino %llx.%llx\n",
1221                      dn, d_inode(dn), ceph_vinop(d_inode(dn)));
1222         }
1223         return 0;
1224 }
1225
1226 /*
1227  * Incorporate results into the local cache.  This is either just
1228  * one inode, or a directory, dentry, and possibly linked-to inode (e.g.,
1229  * after a lookup).
1230  *
1231  * A reply may contain
1232  *         a directory inode along with a dentry.
1233  *  and/or a target inode
1234  *
1235  * Called with snap_rwsem (read).
1236  */
1237 int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
1238 {
1239         struct ceph_mds_session *session = req->r_session;
1240         struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;
1241         struct inode *in = NULL;
1242         struct ceph_vino tvino, dvino;
1243         struct ceph_fs_client *fsc = ceph_sb_to_client(sb);
1244         int err = 0;
1245
1246         dout("fill_trace %p is_dentry %d is_target %d\n", req,
1247              rinfo->head->is_dentry, rinfo->head->is_target);
1248
1249         if (!rinfo->head->is_target && !rinfo->head->is_dentry) {
1250                 dout("fill_trace reply is empty!\n");
1251                 if (rinfo->head->result == 0 && req->r_parent)
1252                         ceph_invalidate_dir_request(req);
1253                 return 0;
1254         }
1255
1256         if (rinfo->head->is_dentry) {
1257                 struct inode *dir = req->r_parent;
1258
1259                 if (dir) {
1260                         err = ceph_fill_inode(dir, NULL, &rinfo->diri,
1261                                               rinfo->dirfrag, session, -1,
1262                                               &req->r_caps_reservation);
1263                         if (err < 0)
1264                                 goto done;
1265                 } else {
1266                         WARN_ON_ONCE(1);
1267                 }
1268
1269                 if (dir && req->r_op == CEPH_MDS_OP_LOOKUPNAME &&
1270                     test_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags) &&
1271                     !test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
1272                         struct qstr dname;
1273                         struct dentry *dn, *parent;
1274
1275                         BUG_ON(!rinfo->head->is_target);
1276                         BUG_ON(req->r_dentry);
1277
1278                         parent = d_find_any_alias(dir);
1279                         BUG_ON(!parent);
1280
1281                         dname.name = rinfo->dname;
1282                         dname.len = rinfo->dname_len;
1283                         dname.hash = full_name_hash(parent, dname.name, dname.len);
1284                         tvino.ino = le64_to_cpu(rinfo->targeti.in->ino);
1285                         tvino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
1286 retry_lookup:
1287                         dn = d_lookup(parent, &dname);
1288                         dout("d_lookup on parent=%p name=%.*s got %p\n",
1289                              parent, dname.len, dname.name, dn);
1290
1291                         if (!dn) {
1292                                 dn = d_alloc(parent, &dname);
1293                                 dout("d_alloc %p '%.*s' = %p\n", parent,
1294                                      dname.len, dname.name, dn);
1295                                 if (!dn) {
1296                                         dput(parent);
1297                                         err = -ENOMEM;
1298                                         goto done;
1299                                 }
1300                                 err = 0;
1301                         } else if (d_really_is_positive(dn) &&
1302                                    (ceph_ino(d_inode(dn)) != tvino.ino ||
1303                                     ceph_snap(d_inode(dn)) != tvino.snap)) {
1304                                 dout(" dn %p points to wrong inode %p\n",
1305                                      dn, d_inode(dn));
1306                                 ceph_dir_clear_ordered(dir);
1307                                 d_delete(dn);
1308                                 dput(dn);
1309                                 goto retry_lookup;
1310                         }
1311
1312                         req->r_dentry = dn;
1313                         dput(parent);
1314                 }
1315         }
1316
1317         if (rinfo->head->is_target) {
1318                 /* Should be filled in by handle_reply */
1319                 BUG_ON(!req->r_target_inode);
1320
1321                 in = req->r_target_inode;
1322                 err = ceph_fill_inode(in, req->r_locked_page, &rinfo->targeti,
1323                                 NULL, session,
1324                                 (!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags) &&
1325                                  !test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags) &&
1326                                  rinfo->head->result == 0) ?  req->r_fmode : -1,
1327                                 &req->r_caps_reservation);
1328                 if (err < 0) {
1329                         pr_err("ceph_fill_inode badness %p %llx.%llx\n",
1330                                 in, ceph_vinop(in));
1331                         req->r_target_inode = NULL;
1332                         if (in->i_state & I_NEW)
1333                                 discard_new_inode(in);
1334                         else
1335                                 iput(in);
1336                         goto done;
1337                 }
1338                 if (in->i_state & I_NEW)
1339                         unlock_new_inode(in);
1340         }
1341
1342         /*
1343          * ignore null lease/binding on snapdir ENOENT, or else we
1344          * will have trouble splicing in the virtual snapdir later
1345          */
1346         if (rinfo->head->is_dentry &&
1347             !test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags) &&
1348             test_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags) &&
1349             (rinfo->head->is_target || strncmp(req->r_dentry->d_name.name,
1350                                                fsc->mount_options->snapdir_name,
1351                                                req->r_dentry->d_name.len))) {
1352                 /*
1353                  * lookup link rename   : null -> possibly existing inode
1354                  * mknod symlink mkdir  : null -> new inode
1355                  * unlink               : linked -> null
1356                  */
1357                 struct inode *dir = req->r_parent;
1358                 struct dentry *dn = req->r_dentry;
1359                 bool have_dir_cap, have_lease;
1360
1361                 BUG_ON(!dn);
1362                 BUG_ON(!dir);
1363                 BUG_ON(d_inode(dn->d_parent) != dir);
1364
1365                 dvino.ino = le64_to_cpu(rinfo->diri.in->ino);
1366                 dvino.snap = le64_to_cpu(rinfo->diri.in->snapid);
1367
1368                 BUG_ON(ceph_ino(dir) != dvino.ino);
1369                 BUG_ON(ceph_snap(dir) != dvino.snap);
1370
1371                 /* do we have a lease on the whole dir? */
1372                 have_dir_cap =
1373                         (le32_to_cpu(rinfo->diri.in->cap.caps) &
1374                          CEPH_CAP_FILE_SHARED);
1375
1376                 /* do we have a dn lease? */
1377                 have_lease = have_dir_cap ||
1378                         le32_to_cpu(rinfo->dlease->duration_ms);
1379                 if (!have_lease)
1380                         dout("fill_trace  no dentry lease or dir cap\n");
1381
1382                 /* rename? */
1383                 if (req->r_old_dentry && req->r_op == CEPH_MDS_OP_RENAME) {
1384                         struct inode *olddir = req->r_old_dentry_dir;
1385                         BUG_ON(!olddir);
1386
1387                         dout(" src %p '%pd' dst %p '%pd'\n",
1388                              req->r_old_dentry,
1389                              req->r_old_dentry,
1390                              dn, dn);
1391                         dout("fill_trace doing d_move %p -> %p\n",
1392                              req->r_old_dentry, dn);
1393
1394                         /* d_move screws up sibling dentries' offsets */
1395                         ceph_dir_clear_ordered(dir);
1396                         ceph_dir_clear_ordered(olddir);
1397
1398                         d_move(req->r_old_dentry, dn);
1399                         dout(" src %p '%pd' dst %p '%pd'\n",
1400                              req->r_old_dentry,
1401                              req->r_old_dentry,
1402                              dn, dn);
1403
1404                         /* ensure target dentry is invalidated, despite
1405                            rehashing bug in vfs_rename_dir */
1406                         ceph_invalidate_dentry_lease(dn);
1407
1408                         dout("dn %p gets new offset %lld\n", req->r_old_dentry,
1409                              ceph_dentry(req->r_old_dentry)->offset);
1410
1411                         /* swap r_dentry and r_old_dentry in case that
1412                          * splice_dentry() gets called later. This is safe
1413                          * because no other place will use them */
1414                         req->r_dentry = req->r_old_dentry;
1415                         req->r_old_dentry = dn;
1416                         dn = req->r_dentry;
1417                 }
1418
1419                 /* null dentry? */
1420                 if (!rinfo->head->is_target) {
1421                         dout("fill_trace null dentry\n");
1422                         if (d_really_is_positive(dn)) {
1423                                 dout("d_delete %p\n", dn);
1424                                 ceph_dir_clear_ordered(dir);
1425                                 d_delete(dn);
1426                         } else if (have_lease) {
1427                                 if (d_unhashed(dn))
1428                                         d_add(dn, NULL);
1429                                 update_dentry_lease(dir, dn,
1430                                                     rinfo->dlease, session,
1431                                                     req->r_request_started);
1432                         }
1433                         goto done;
1434                 }
1435
1436                 /* attach proper inode */
1437                 if (d_really_is_negative(dn)) {
1438                         ceph_dir_clear_ordered(dir);
1439                         ihold(in);
1440                         err = splice_dentry(&req->r_dentry, in);
1441                         if (err < 0)
1442                                 goto done;
1443                         dn = req->r_dentry;  /* may have spliced */
1444                 } else if (d_really_is_positive(dn) && d_inode(dn) != in) {
1445                         dout(" %p links to %p %llx.%llx, not %llx.%llx\n",
1446                              dn, d_inode(dn), ceph_vinop(d_inode(dn)),
1447                              ceph_vinop(in));
1448                         d_invalidate(dn);
1449                         have_lease = false;
1450                 }
1451
1452                 if (have_lease) {
1453                         update_dentry_lease(dir, dn,
1454                                             rinfo->dlease, session,
1455                                             req->r_request_started);
1456                 }
1457                 dout(" final dn %p\n", dn);
1458         } else if ((req->r_op == CEPH_MDS_OP_LOOKUPSNAP ||
1459                     req->r_op == CEPH_MDS_OP_MKSNAP) &&
1460                    test_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags) &&
1461                    !test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
1462                 struct inode *dir = req->r_parent;
1463
1464                 /* fill out a snapdir LOOKUPSNAP dentry */
1465                 BUG_ON(!dir);
1466                 BUG_ON(ceph_snap(dir) != CEPH_SNAPDIR);
1467                 BUG_ON(!req->r_dentry);
1468                 dout(" linking snapped dir %p to dn %p\n", in, req->r_dentry);
1469                 ceph_dir_clear_ordered(dir);
1470                 ihold(in);
1471                 err = splice_dentry(&req->r_dentry, in);
1472                 if (err < 0)
1473                         goto done;
1474         } else if (rinfo->head->is_dentry && req->r_dentry) {
1475                 /* parent inode is not locked, be carefull */
1476                 struct ceph_vino *ptvino = NULL;
1477                 dvino.ino = le64_to_cpu(rinfo->diri.in->ino);
1478                 dvino.snap = le64_to_cpu(rinfo->diri.in->snapid);
1479                 if (rinfo->head->is_target) {
1480                         tvino.ino = le64_to_cpu(rinfo->targeti.in->ino);
1481                         tvino.snap = le64_to_cpu(rinfo->targeti.in->snapid);
1482                         ptvino = &tvino;
1483                 }
1484                 update_dentry_lease_careful(req->r_dentry, rinfo->dlease,
1485                                             session, req->r_request_started,
1486                                             rinfo->dname, rinfo->dname_len,
1487                                             &dvino, ptvino);
1488         }
1489 done:
1490         dout("fill_trace done err=%d\n", err);
1491         return err;
1492 }
1493
1494 /*
1495  * Prepopulate our cache with readdir results, leases, etc.
1496  */
1497 static int readdir_prepopulate_inodes_only(struct ceph_mds_request *req,
1498                                            struct ceph_mds_session *session)
1499 {
1500         struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;
1501         int i, err = 0;
1502
1503         for (i = 0; i < rinfo->dir_nr; i++) {
1504                 struct ceph_mds_reply_dir_entry *rde = rinfo->dir_entries + i;
1505                 struct ceph_vino vino;
1506                 struct inode *in;
1507                 int rc;
1508
1509                 vino.ino = le64_to_cpu(rde->inode.in->ino);
1510                 vino.snap = le64_to_cpu(rde->inode.in->snapid);
1511
1512                 in = ceph_get_inode(req->r_dentry->d_sb, vino);
1513                 if (IS_ERR(in)) {
1514                         err = PTR_ERR(in);
1515                         dout("new_inode badness got %d\n", err);
1516                         continue;
1517                 }
1518                 rc = ceph_fill_inode(in, NULL, &rde->inode, NULL, session,
1519                                      -1, &req->r_caps_reservation);
1520                 if (rc < 0) {
1521                         pr_err("ceph_fill_inode badness on %p got %d\n",
1522                                in, rc);
1523                         err = rc;
1524                         if (in->i_state & I_NEW) {
1525                                 ihold(in);
1526                                 discard_new_inode(in);
1527                         }
1528                 } else if (in->i_state & I_NEW) {
1529                         unlock_new_inode(in);
1530                 }
1531
1532                 /* avoid calling iput_final() in mds dispatch threads */
1533                 ceph_async_iput(in);
1534         }
1535
1536         return err;
1537 }
1538
1539 void ceph_readdir_cache_release(struct ceph_readdir_cache_control *ctl)
1540 {
1541         if (ctl->page) {
1542                 kunmap(ctl->page);
1543                 put_page(ctl->page);
1544                 ctl->page = NULL;
1545         }
1546 }
1547
1548 static int fill_readdir_cache(struct inode *dir, struct dentry *dn,
1549                               struct ceph_readdir_cache_control *ctl,
1550                               struct ceph_mds_request *req)
1551 {
1552         struct ceph_inode_info *ci = ceph_inode(dir);
1553         unsigned nsize = PAGE_SIZE / sizeof(struct dentry*);
1554         unsigned idx = ctl->index % nsize;
1555         pgoff_t pgoff = ctl->index / nsize;
1556
1557         if (!ctl->page || pgoff != page_index(ctl->page)) {
1558                 ceph_readdir_cache_release(ctl);
1559                 if (idx == 0)
1560                         ctl->page = grab_cache_page(&dir->i_data, pgoff);
1561                 else
1562                         ctl->page = find_lock_page(&dir->i_data, pgoff);
1563                 if (!ctl->page) {
1564                         ctl->index = -1;
1565                         return idx == 0 ? -ENOMEM : 0;
1566                 }
1567                 /* reading/filling the cache are serialized by
1568                  * i_mutex, no need to use page lock */
1569                 unlock_page(ctl->page);
1570                 ctl->dentries = kmap(ctl->page);
1571                 if (idx == 0)
1572                         memset(ctl->dentries, 0, PAGE_SIZE);
1573         }
1574
1575         if (req->r_dir_release_cnt == atomic64_read(&ci->i_release_count) &&
1576             req->r_dir_ordered_cnt == atomic64_read(&ci->i_ordered_count)) {
1577                 dout("readdir cache dn %p idx %d\n", dn, ctl->index);
1578                 ctl->dentries[idx] = dn;
1579                 ctl->index++;
1580         } else {
1581                 dout("disable readdir cache\n");
1582                 ctl->index = -1;
1583         }
1584         return 0;
1585 }
1586
1587 int ceph_readdir_prepopulate(struct ceph_mds_request *req,
1588                              struct ceph_mds_session *session)
1589 {
1590         struct dentry *parent = req->r_dentry;
1591         struct ceph_inode_info *ci = ceph_inode(d_inode(parent));
1592         struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;
1593         struct qstr dname;
1594         struct dentry *dn;
1595         struct inode *in;
1596         int err = 0, skipped = 0, ret, i;
1597         u32 frag = le32_to_cpu(req->r_args.readdir.frag);
1598         u32 last_hash = 0;
1599         u32 fpos_offset;
1600         struct ceph_readdir_cache_control cache_ctl = {};
1601
1602         if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
1603                 return readdir_prepopulate_inodes_only(req, session);
1604
1605         if (rinfo->hash_order) {
1606                 if (req->r_path2) {
1607                         last_hash = ceph_str_hash(ci->i_dir_layout.dl_dir_hash,
1608                                                   req->r_path2,
1609                                                   strlen(req->r_path2));
1610                         last_hash = ceph_frag_value(last_hash);
1611                 } else if (rinfo->offset_hash) {
1612                         /* mds understands offset_hash */
1613                         WARN_ON_ONCE(req->r_readdir_offset != 2);
1614                         last_hash = le32_to_cpu(req->r_args.readdir.offset_hash);
1615                 }
1616         }
1617
1618         if (rinfo->dir_dir &&
1619             le32_to_cpu(rinfo->dir_dir->frag) != frag) {
1620                 dout("readdir_prepopulate got new frag %x -> %x\n",
1621                      frag, le32_to_cpu(rinfo->dir_dir->frag));
1622                 frag = le32_to_cpu(rinfo->dir_dir->frag);
1623                 if (!rinfo->hash_order)
1624                         req->r_readdir_offset = 2;
1625         }
1626
1627         if (le32_to_cpu(rinfo->head->op) == CEPH_MDS_OP_LSSNAP) {
1628                 dout("readdir_prepopulate %d items under SNAPDIR dn %p\n",
1629                      rinfo->dir_nr, parent);
1630         } else {
1631                 dout("readdir_prepopulate %d items under dn %p\n",
1632                      rinfo->dir_nr, parent);
1633                 if (rinfo->dir_dir)
1634                         ceph_fill_dirfrag(d_inode(parent), rinfo->dir_dir);
1635
1636                 if (ceph_frag_is_leftmost(frag) &&
1637                     req->r_readdir_offset == 2 &&
1638                     !(rinfo->hash_order && last_hash)) {
1639                         /* note dir version at start of readdir so we can
1640                          * tell if any dentries get dropped */
1641                         req->r_dir_release_cnt =
1642                                 atomic64_read(&ci->i_release_count);
1643                         req->r_dir_ordered_cnt =
1644                                 atomic64_read(&ci->i_ordered_count);
1645                         req->r_readdir_cache_idx = 0;
1646                 }
1647         }
1648
1649         cache_ctl.index = req->r_readdir_cache_idx;
1650         fpos_offset = req->r_readdir_offset;
1651
1652         /* FIXME: release caps/leases if error occurs */
1653         for (i = 0; i < rinfo->dir_nr; i++) {
1654                 struct ceph_mds_reply_dir_entry *rde = rinfo->dir_entries + i;
1655                 struct ceph_vino tvino;
1656
1657                 dname.name = rde->name;
1658                 dname.len = rde->name_len;
1659                 dname.hash = full_name_hash(parent, dname.name, dname.len);
1660
1661                 tvino.ino = le64_to_cpu(rde->inode.in->ino);
1662                 tvino.snap = le64_to_cpu(rde->inode.in->snapid);
1663
1664                 if (rinfo->hash_order) {
1665                         u32 hash = ceph_str_hash(ci->i_dir_layout.dl_dir_hash,
1666                                                  rde->name, rde->name_len);
1667                         hash = ceph_frag_value(hash);
1668                         if (hash != last_hash)
1669                                 fpos_offset = 2;
1670                         last_hash = hash;
1671                         rde->offset = ceph_make_fpos(hash, fpos_offset++, true);
1672                 } else {
1673                         rde->offset = ceph_make_fpos(frag, fpos_offset++, false);
1674                 }
1675
1676 retry_lookup:
1677                 dn = d_lookup(parent, &dname);
1678                 dout("d_lookup on parent=%p name=%.*s got %p\n",
1679                      parent, dname.len, dname.name, dn);
1680
1681                 if (!dn) {
1682                         dn = d_alloc(parent, &dname);
1683                         dout("d_alloc %p '%.*s' = %p\n", parent,
1684                              dname.len, dname.name, dn);
1685                         if (!dn) {
1686                                 dout("d_alloc badness\n");
1687                                 err = -ENOMEM;
1688                                 goto out;
1689                         }
1690                 } else if (d_really_is_positive(dn) &&
1691                            (ceph_ino(d_inode(dn)) != tvino.ino ||
1692                             ceph_snap(d_inode(dn)) != tvino.snap)) {
1693                         struct ceph_dentry_info *di = ceph_dentry(dn);
1694                         dout(" dn %p points to wrong inode %p\n",
1695                              dn, d_inode(dn));
1696
1697                         spin_lock(&dn->d_lock);
1698                         if (di->offset > 0 &&
1699                             di->lease_shared_gen ==
1700                             atomic_read(&ci->i_shared_gen)) {
1701                                 __ceph_dir_clear_ordered(ci);
1702                                 di->offset = 0;
1703                         }
1704                         spin_unlock(&dn->d_lock);
1705
1706                         d_delete(dn);
1707                         dput(dn);
1708                         goto retry_lookup;
1709                 }
1710
1711                 /* inode */
1712                 if (d_really_is_positive(dn)) {
1713                         in = d_inode(dn);
1714                 } else {
1715                         in = ceph_get_inode(parent->d_sb, tvino);
1716                         if (IS_ERR(in)) {
1717                                 dout("new_inode badness\n");
1718                                 d_drop(dn);
1719                                 dput(dn);
1720                                 err = PTR_ERR(in);
1721                                 goto out;
1722                         }
1723                 }
1724
1725                 ret = ceph_fill_inode(in, NULL, &rde->inode, NULL, session,
1726                                       -1, &req->r_caps_reservation);
1727                 if (ret < 0) {
1728                         pr_err("ceph_fill_inode badness on %p\n", in);
1729                         if (d_really_is_negative(dn)) {
1730                                 /* avoid calling iput_final() in mds
1731                                  * dispatch threads */
1732                                 if (in->i_state & I_NEW) {
1733                                         ihold(in);
1734                                         discard_new_inode(in);
1735                                 }
1736                                 ceph_async_iput(in);
1737                         }
1738                         d_drop(dn);
1739                         err = ret;
1740                         goto next_item;
1741                 }
1742                 if (in->i_state & I_NEW)
1743                         unlock_new_inode(in);
1744
1745                 if (d_really_is_negative(dn)) {
1746                         if (ceph_security_xattr_deadlock(in)) {
1747                                 dout(" skip splicing dn %p to inode %p"
1748                                      " (security xattr deadlock)\n", dn, in);
1749                                 ceph_async_iput(in);
1750                                 skipped++;
1751                                 goto next_item;
1752                         }
1753
1754                         err = splice_dentry(&dn, in);
1755                         if (err < 0)
1756                                 goto next_item;
1757                 }
1758
1759                 ceph_dentry(dn)->offset = rde->offset;
1760
1761                 update_dentry_lease(d_inode(parent), dn,
1762                                     rde->lease, req->r_session,
1763                                     req->r_request_started);
1764
1765                 if (err == 0 && skipped == 0 && cache_ctl.index >= 0) {
1766                         ret = fill_readdir_cache(d_inode(parent), dn,
1767                                                  &cache_ctl, req);
1768                         if (ret < 0)
1769                                 err = ret;
1770                 }
1771 next_item:
1772                 dput(dn);
1773         }
1774 out:
1775         if (err == 0 && skipped == 0) {
1776                 set_bit(CEPH_MDS_R_DID_PREPOPULATE, &req->r_req_flags);
1777                 req->r_readdir_cache_idx = cache_ctl.index;
1778         }
1779         ceph_readdir_cache_release(&cache_ctl);
1780         dout("readdir_prepopulate done\n");
1781         return err;
1782 }
1783
1784 bool ceph_inode_set_size(struct inode *inode, loff_t size)
1785 {
1786         struct ceph_inode_info *ci = ceph_inode(inode);
1787         bool ret;
1788
1789         spin_lock(&ci->i_ceph_lock);
1790         dout("set_size %p %llu -> %llu\n", inode, inode->i_size, size);
1791         i_size_write(inode, size);
1792         inode->i_blocks = calc_inode_blocks(size);
1793
1794         ret = __ceph_should_report_size(ci);
1795
1796         spin_unlock(&ci->i_ceph_lock);
1797         return ret;
1798 }
1799
1800 /*
1801  * Put reference to inode, but avoid calling iput_final() in current thread.
1802  * iput_final() may wait for reahahead pages. The wait can cause deadlock in
1803  * some contexts.
1804  */
1805 void ceph_async_iput(struct inode *inode)
1806 {
1807         if (!inode)
1808                 return;
1809         for (;;) {
1810                 if (atomic_add_unless(&inode->i_count, -1, 1))
1811                         break;
1812                 if (queue_work(ceph_inode_to_client(inode)->inode_wq,
1813                                &ceph_inode(inode)->i_work))
1814                         break;
1815                 /* queue work failed, i_count must be at least 2 */
1816         }
1817 }
1818
1819 /*
1820  * Write back inode data in a worker thread.  (This can't be done
1821  * in the message handler context.)
1822  */
1823 void ceph_queue_writeback(struct inode *inode)
1824 {
1825         struct ceph_inode_info *ci = ceph_inode(inode);
1826         set_bit(CEPH_I_WORK_WRITEBACK, &ci->i_work_mask);
1827
1828         ihold(inode);
1829         if (queue_work(ceph_inode_to_client(inode)->inode_wq,
1830                        &ci->i_work)) {
1831                 dout("ceph_queue_writeback %p\n", inode);
1832         } else {
1833                 dout("ceph_queue_writeback %p already queued, mask=%lx\n",
1834                      inode, ci->i_work_mask);
1835                 iput(inode);
1836         }
1837 }
1838
1839 /*
1840  * queue an async invalidation
1841  */
1842 void ceph_queue_invalidate(struct inode *inode)
1843 {
1844         struct ceph_inode_info *ci = ceph_inode(inode);
1845         set_bit(CEPH_I_WORK_INVALIDATE_PAGES, &ci->i_work_mask);
1846
1847         ihold(inode);
1848         if (queue_work(ceph_inode_to_client(inode)->inode_wq,
1849                        &ceph_inode(inode)->i_work)) {
1850                 dout("ceph_queue_invalidate %p\n", inode);
1851         } else {
1852                 dout("ceph_queue_invalidate %p already queued, mask=%lx\n",
1853                      inode, ci->i_work_mask);
1854                 iput(inode);
1855         }
1856 }
1857
1858 /*
1859  * Queue an async vmtruncate.  If we fail to queue work, we will handle
1860  * the truncation the next time we call __ceph_do_pending_vmtruncate.
1861  */
1862 void ceph_queue_vmtruncate(struct inode *inode)
1863 {
1864         struct ceph_inode_info *ci = ceph_inode(inode);
1865         set_bit(CEPH_I_WORK_VMTRUNCATE, &ci->i_work_mask);
1866
1867         ihold(inode);
1868         if (queue_work(ceph_inode_to_client(inode)->inode_wq,
1869                        &ci->i_work)) {
1870                 dout("ceph_queue_vmtruncate %p\n", inode);
1871         } else {
1872                 dout("ceph_queue_vmtruncate %p already queued, mask=%lx\n",
1873                      inode, ci->i_work_mask);
1874                 iput(inode);
1875         }
1876 }
1877
1878 static void ceph_do_invalidate_pages(struct inode *inode)
1879 {
1880         struct ceph_inode_info *ci = ceph_inode(inode);
1881         struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
1882         u32 orig_gen;
1883         int check = 0;
1884
1885         mutex_lock(&ci->i_truncate_mutex);
1886
1887         if (READ_ONCE(fsc->mount_state) >= CEPH_MOUNT_SHUTDOWN) {
1888                 pr_warn_ratelimited("invalidate_pages %p %lld forced umount\n",
1889                                     inode, ceph_ino(inode));
1890                 mapping_set_error(inode->i_mapping, -EIO);
1891                 truncate_pagecache(inode, 0);
1892                 mutex_unlock(&ci->i_truncate_mutex);
1893                 goto out;
1894         }
1895
1896         spin_lock(&ci->i_ceph_lock);
1897         dout("invalidate_pages %p gen %d revoking %d\n", inode,
1898              ci->i_rdcache_gen, ci->i_rdcache_revoking);
1899         if (ci->i_rdcache_revoking != ci->i_rdcache_gen) {
1900                 if (__ceph_caps_revoking_other(ci, NULL, CEPH_CAP_FILE_CACHE))
1901                         check = 1;
1902                 spin_unlock(&ci->i_ceph_lock);
1903                 mutex_unlock(&ci->i_truncate_mutex);
1904                 goto out;
1905         }
1906         orig_gen = ci->i_rdcache_gen;
1907         spin_unlock(&ci->i_ceph_lock);
1908
1909         if (invalidate_inode_pages2(inode->i_mapping) < 0) {
1910                 pr_err("invalidate_pages %p fails\n", inode);
1911         }
1912
1913         spin_lock(&ci->i_ceph_lock);
1914         if (orig_gen == ci->i_rdcache_gen &&
1915             orig_gen == ci->i_rdcache_revoking) {
1916                 dout("invalidate_pages %p gen %d successful\n", inode,
1917                      ci->i_rdcache_gen);
1918                 ci->i_rdcache_revoking--;
1919                 check = 1;
1920         } else {
1921                 dout("invalidate_pages %p gen %d raced, now %d revoking %d\n",
1922                      inode, orig_gen, ci->i_rdcache_gen,
1923                      ci->i_rdcache_revoking);
1924                 if (__ceph_caps_revoking_other(ci, NULL, CEPH_CAP_FILE_CACHE))
1925                         check = 1;
1926         }
1927         spin_unlock(&ci->i_ceph_lock);
1928         mutex_unlock(&ci->i_truncate_mutex);
1929 out:
1930         if (check)
1931                 ceph_check_caps(ci, 0, NULL);
1932 }
1933
1934 /*
1935  * Make sure any pending truncation is applied before doing anything
1936  * that may depend on it.
1937  */
1938 void __ceph_do_pending_vmtruncate(struct inode *inode)
1939 {
1940         struct ceph_inode_info *ci = ceph_inode(inode);
1941         u64 to;
1942         int wrbuffer_refs, finish = 0;
1943
1944         mutex_lock(&ci->i_truncate_mutex);
1945 retry:
1946         spin_lock(&ci->i_ceph_lock);
1947         if (ci->i_truncate_pending == 0) {
1948                 dout("__do_pending_vmtruncate %p none pending\n", inode);
1949                 spin_unlock(&ci->i_ceph_lock);
1950                 mutex_unlock(&ci->i_truncate_mutex);
1951                 return;
1952         }
1953
1954         /*
1955          * make sure any dirty snapped pages are flushed before we
1956          * possibly truncate them.. so write AND block!
1957          */
1958         if (ci->i_wrbuffer_ref_head < ci->i_wrbuffer_ref) {
1959                 spin_unlock(&ci->i_ceph_lock);
1960                 dout("__do_pending_vmtruncate %p flushing snaps first\n",
1961                      inode);
1962                 filemap_write_and_wait_range(&inode->i_data, 0,
1963                                              inode->i_sb->s_maxbytes);
1964                 goto retry;
1965         }
1966
1967         /* there should be no reader or writer */
1968         WARN_ON_ONCE(ci->i_rd_ref || ci->i_wr_ref);
1969
1970         to = ci->i_truncate_size;
1971         wrbuffer_refs = ci->i_wrbuffer_ref;
1972         dout("__do_pending_vmtruncate %p (%d) to %lld\n", inode,
1973              ci->i_truncate_pending, to);
1974         spin_unlock(&ci->i_ceph_lock);
1975
1976         truncate_pagecache(inode, to);
1977
1978         spin_lock(&ci->i_ceph_lock);
1979         if (to == ci->i_truncate_size) {
1980                 ci->i_truncate_pending = 0;
1981                 finish = 1;
1982         }
1983         spin_unlock(&ci->i_ceph_lock);
1984         if (!finish)
1985                 goto retry;
1986
1987         mutex_unlock(&ci->i_truncate_mutex);
1988
1989         if (wrbuffer_refs == 0)
1990                 ceph_check_caps(ci, 0, NULL);
1991
1992         wake_up_all(&ci->i_cap_wq);
1993 }
1994
1995 static void ceph_inode_work(struct work_struct *work)
1996 {
1997         struct ceph_inode_info *ci = container_of(work, struct ceph_inode_info,
1998                                                  i_work);
1999         struct inode *inode = &ci->vfs_inode;
2000
2001         if (test_and_clear_bit(CEPH_I_WORK_WRITEBACK, &ci->i_work_mask)) {
2002                 dout("writeback %p\n", inode);
2003                 filemap_fdatawrite(&inode->i_data);
2004         }
2005         if (test_and_clear_bit(CEPH_I_WORK_INVALIDATE_PAGES, &ci->i_work_mask))
2006                 ceph_do_invalidate_pages(inode);
2007
2008         if (test_and_clear_bit(CEPH_I_WORK_VMTRUNCATE, &ci->i_work_mask))
2009                 __ceph_do_pending_vmtruncate(inode);
2010
2011         iput(inode);
2012 }
2013
2014 /*
2015  * symlinks
2016  */
2017 static const struct inode_operations ceph_symlink_iops = {
2018         .get_link = simple_get_link,
2019         .setattr = ceph_setattr,
2020         .getattr = ceph_getattr,
2021         .listxattr = ceph_listxattr,
2022 };
2023
2024 int __ceph_setattr(struct inode *inode, struct iattr *attr)
2025 {
2026         struct ceph_inode_info *ci = ceph_inode(inode);
2027         unsigned int ia_valid = attr->ia_valid;
2028         struct ceph_mds_request *req;
2029         struct ceph_mds_client *mdsc = ceph_sb_to_client(inode->i_sb)->mdsc;
2030         struct ceph_cap_flush *prealloc_cf;
2031         int issued;
2032         int release = 0, dirtied = 0;
2033         int mask = 0;
2034         int err = 0;
2035         int inode_dirty_flags = 0;
2036         bool lock_snap_rwsem = false;
2037
2038         prealloc_cf = ceph_alloc_cap_flush();
2039         if (!prealloc_cf)
2040                 return -ENOMEM;
2041
2042         req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_SETATTR,
2043                                        USE_AUTH_MDS);
2044         if (IS_ERR(req)) {
2045                 ceph_free_cap_flush(prealloc_cf);
2046                 return PTR_ERR(req);
2047         }
2048
2049         spin_lock(&ci->i_ceph_lock);
2050         issued = __ceph_caps_issued(ci, NULL);
2051
2052         if (!ci->i_head_snapc &&
2053             (issued & (CEPH_CAP_ANY_EXCL | CEPH_CAP_FILE_WR))) {
2054                 lock_snap_rwsem = true;
2055                 if (!down_read_trylock(&mdsc->snap_rwsem)) {
2056                         spin_unlock(&ci->i_ceph_lock);
2057                         down_read(&mdsc->snap_rwsem);
2058                         spin_lock(&ci->i_ceph_lock);
2059                         issued = __ceph_caps_issued(ci, NULL);
2060                 }
2061         }
2062
2063         dout("setattr %p issued %s\n", inode, ceph_cap_string(issued));
2064
2065         if (ia_valid & ATTR_UID) {
2066                 dout("setattr %p uid %d -> %d\n", inode,
2067                      from_kuid(&init_user_ns, inode->i_uid),
2068                      from_kuid(&init_user_ns, attr->ia_uid));
2069                 if (issued & CEPH_CAP_AUTH_EXCL) {
2070                         inode->i_uid = attr->ia_uid;
2071                         dirtied |= CEPH_CAP_AUTH_EXCL;
2072                 } else if ((issued & CEPH_CAP_AUTH_SHARED) == 0 ||
2073                            !uid_eq(attr->ia_uid, inode->i_uid)) {
2074                         req->r_args.setattr.uid = cpu_to_le32(
2075                                 from_kuid(&init_user_ns, attr->ia_uid));
2076                         mask |= CEPH_SETATTR_UID;
2077                         release |= CEPH_CAP_AUTH_SHARED;
2078                 }
2079         }
2080         if (ia_valid & ATTR_GID) {
2081                 dout("setattr %p gid %d -> %d\n", inode,
2082                      from_kgid(&init_user_ns, inode->i_gid),
2083                      from_kgid(&init_user_ns, attr->ia_gid));
2084                 if (issued & CEPH_CAP_AUTH_EXCL) {
2085                         inode->i_gid = attr->ia_gid;
2086                         dirtied |= CEPH_CAP_AUTH_EXCL;
2087                 } else if ((issued & CEPH_CAP_AUTH_SHARED) == 0 ||
2088                            !gid_eq(attr->ia_gid, inode->i_gid)) {
2089                         req->r_args.setattr.gid = cpu_to_le32(
2090                                 from_kgid(&init_user_ns, attr->ia_gid));
2091                         mask |= CEPH_SETATTR_GID;
2092                         release |= CEPH_CAP_AUTH_SHARED;
2093                 }
2094         }
2095         if (ia_valid & ATTR_MODE) {
2096                 dout("setattr %p mode 0%o -> 0%o\n", inode, inode->i_mode,
2097                      attr->ia_mode);
2098                 if (issued & CEPH_CAP_AUTH_EXCL) {
2099                         inode->i_mode = attr->ia_mode;
2100                         dirtied |= CEPH_CAP_AUTH_EXCL;
2101                 } else if ((issued & CEPH_CAP_AUTH_SHARED) == 0 ||
2102                            attr->ia_mode != inode->i_mode) {
2103                         inode->i_mode = attr->ia_mode;
2104                         req->r_args.setattr.mode = cpu_to_le32(attr->ia_mode);
2105                         mask |= CEPH_SETATTR_MODE;
2106                         release |= CEPH_CAP_AUTH_SHARED;
2107                 }
2108         }
2109
2110         if (ia_valid & ATTR_ATIME) {
2111                 dout("setattr %p atime %lld.%ld -> %lld.%ld\n", inode,
2112                      inode->i_atime.tv_sec, inode->i_atime.tv_nsec,
2113                      attr->ia_atime.tv_sec, attr->ia_atime.tv_nsec);
2114                 if (issued & CEPH_CAP_FILE_EXCL) {
2115                         ci->i_time_warp_seq++;
2116                         inode->i_atime = attr->ia_atime;
2117                         dirtied |= CEPH_CAP_FILE_EXCL;
2118                 } else if ((issued & CEPH_CAP_FILE_WR) &&
2119                            timespec64_compare(&inode->i_atime,
2120                                             &attr->ia_atime) < 0) {
2121                         inode->i_atime = attr->ia_atime;
2122                         dirtied |= CEPH_CAP_FILE_WR;
2123                 } else if ((issued & CEPH_CAP_FILE_SHARED) == 0 ||
2124                            !timespec64_equal(&inode->i_atime, &attr->ia_atime)) {
2125                         ceph_encode_timespec64(&req->r_args.setattr.atime,
2126                                                &attr->ia_atime);
2127                         mask |= CEPH_SETATTR_ATIME;
2128                         release |= CEPH_CAP_FILE_SHARED |
2129                                    CEPH_CAP_FILE_RD | CEPH_CAP_FILE_WR;
2130                 }
2131         }
2132         if (ia_valid & ATTR_SIZE) {
2133                 dout("setattr %p size %lld -> %lld\n", inode,
2134                      inode->i_size, attr->ia_size);
2135                 if ((issued & CEPH_CAP_FILE_EXCL) &&
2136                     attr->ia_size > inode->i_size) {
2137                         i_size_write(inode, attr->ia_size);
2138                         inode->i_blocks = calc_inode_blocks(attr->ia_size);
2139                         ci->i_reported_size = attr->ia_size;
2140                         dirtied |= CEPH_CAP_FILE_EXCL;
2141                         ia_valid |= ATTR_MTIME;
2142                 } else if ((issued & CEPH_CAP_FILE_SHARED) == 0 ||
2143                            attr->ia_size != inode->i_size) {
2144                         req->r_args.setattr.size = cpu_to_le64(attr->ia_size);
2145                         req->r_args.setattr.old_size =
2146                                 cpu_to_le64(inode->i_size);
2147                         mask |= CEPH_SETATTR_SIZE;
2148                         release |= CEPH_CAP_FILE_SHARED | CEPH_CAP_FILE_EXCL |
2149                                    CEPH_CAP_FILE_RD | CEPH_CAP_FILE_WR;
2150                 }
2151         }
2152         if (ia_valid & ATTR_MTIME) {
2153                 dout("setattr %p mtime %lld.%ld -> %lld.%ld\n", inode,
2154                      inode->i_mtime.tv_sec, inode->i_mtime.tv_nsec,
2155                      attr->ia_mtime.tv_sec, attr->ia_mtime.tv_nsec);
2156                 if (issued & CEPH_CAP_FILE_EXCL) {
2157                         ci->i_time_warp_seq++;
2158                         inode->i_mtime = attr->ia_mtime;
2159                         dirtied |= CEPH_CAP_FILE_EXCL;
2160                 } else if ((issued & CEPH_CAP_FILE_WR) &&
2161                            timespec64_compare(&inode->i_mtime,
2162                                             &attr->ia_mtime) < 0) {
2163                         inode->i_mtime = attr->ia_mtime;
2164                         dirtied |= CEPH_CAP_FILE_WR;
2165                 } else if ((issued & CEPH_CAP_FILE_SHARED) == 0 ||
2166                            !timespec64_equal(&inode->i_mtime, &attr->ia_mtime)) {
2167                         ceph_encode_timespec64(&req->r_args.setattr.mtime,
2168                                                &attr->ia_mtime);
2169                         mask |= CEPH_SETATTR_MTIME;
2170                         release |= CEPH_CAP_FILE_SHARED |
2171                                    CEPH_CAP_FILE_RD | CEPH_CAP_FILE_WR;
2172                 }
2173         }
2174
2175         /* these do nothing */
2176         if (ia_valid & ATTR_CTIME) {
2177                 bool only = (ia_valid & (ATTR_SIZE|ATTR_MTIME|ATTR_ATIME|
2178                                          ATTR_MODE|ATTR_UID|ATTR_GID)) == 0;
2179                 dout("setattr %p ctime %lld.%ld -> %lld.%ld (%s)\n", inode,
2180                      inode->i_ctime.tv_sec, inode->i_ctime.tv_nsec,
2181                      attr->ia_ctime.tv_sec, attr->ia_ctime.tv_nsec,
2182                      only ? "ctime only" : "ignored");
2183                 if (only) {
2184                         /*
2185                          * if kernel wants to dirty ctime but nothing else,
2186                          * we need to choose a cap to dirty under, or do
2187                          * a almost-no-op setattr
2188                          */
2189                         if (issued & CEPH_CAP_AUTH_EXCL)
2190                                 dirtied |= CEPH_CAP_AUTH_EXCL;
2191                         else if (issued & CEPH_CAP_FILE_EXCL)
2192                                 dirtied |= CEPH_CAP_FILE_EXCL;
2193                         else if (issued & CEPH_CAP_XATTR_EXCL)
2194                                 dirtied |= CEPH_CAP_XATTR_EXCL;
2195                         else
2196                                 mask |= CEPH_SETATTR_CTIME;
2197                 }
2198         }
2199         if (ia_valid & ATTR_FILE)
2200                 dout("setattr %p ATTR_FILE ... hrm!\n", inode);
2201
2202         if (dirtied) {
2203                 inode_dirty_flags = __ceph_mark_dirty_caps(ci, dirtied,
2204                                                            &prealloc_cf);
2205                 inode->i_ctime = attr->ia_ctime;
2206         }
2207
2208         release &= issued;
2209         spin_unlock(&ci->i_ceph_lock);
2210         if (lock_snap_rwsem)
2211                 up_read(&mdsc->snap_rwsem);
2212
2213         if (inode_dirty_flags)
2214                 __mark_inode_dirty(inode, inode_dirty_flags);
2215
2216
2217         if (mask) {
2218                 req->r_inode = inode;
2219                 ihold(inode);
2220                 req->r_inode_drop = release;
2221                 req->r_args.setattr.mask = cpu_to_le32(mask);
2222                 req->r_num_caps = 1;
2223                 req->r_stamp = attr->ia_ctime;
2224                 err = ceph_mdsc_do_request(mdsc, NULL, req);
2225         }
2226         dout("setattr %p result=%d (%s locally, %d remote)\n", inode, err,
2227              ceph_cap_string(dirtied), mask);
2228
2229         ceph_mdsc_put_request(req);
2230         ceph_free_cap_flush(prealloc_cf);
2231
2232         if (err >= 0 && (mask & CEPH_SETATTR_SIZE))
2233                 __ceph_do_pending_vmtruncate(inode);
2234
2235         return err;
2236 }
2237
2238 /*
2239  * setattr
2240  */
2241 int ceph_setattr(struct dentry *dentry, struct iattr *attr)
2242 {
2243         struct inode *inode = d_inode(dentry);
2244         struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
2245         int err;
2246
2247         if (ceph_snap(inode) != CEPH_NOSNAP)
2248                 return -EROFS;
2249
2250         err = setattr_prepare(dentry, attr);
2251         if (err != 0)
2252                 return err;
2253
2254         if ((attr->ia_valid & ATTR_SIZE) &&
2255             attr->ia_size > max(inode->i_size, fsc->max_file_size))
2256                 return -EFBIG;
2257
2258         if ((attr->ia_valid & ATTR_SIZE) &&
2259             ceph_quota_is_max_bytes_exceeded(inode, attr->ia_size))
2260                 return -EDQUOT;
2261
2262         err = __ceph_setattr(inode, attr);
2263
2264         if (err >= 0 && (attr->ia_valid & ATTR_MODE))
2265                 err = posix_acl_chmod(inode, attr->ia_mode);
2266
2267         return err;
2268 }
2269
2270 /*
2271  * Verify that we have a lease on the given mask.  If not,
2272  * do a getattr against an mds.
2273  */
2274 int __ceph_do_getattr(struct inode *inode, struct page *locked_page,
2275                       int mask, bool force)
2276 {
2277         struct ceph_fs_client *fsc = ceph_sb_to_client(inode->i_sb);
2278         struct ceph_mds_client *mdsc = fsc->mdsc;
2279         struct ceph_mds_request *req;
2280         int mode;
2281         int err;
2282
2283         if (ceph_snap(inode) == CEPH_SNAPDIR) {
2284                 dout("do_getattr inode %p SNAPDIR\n", inode);
2285                 return 0;
2286         }
2287
2288         dout("do_getattr inode %p mask %s mode 0%o\n",
2289              inode, ceph_cap_string(mask), inode->i_mode);
2290         if (!force && ceph_caps_issued_mask_metric(ceph_inode(inode), mask, 1))
2291                         return 0;
2292
2293         mode = (mask & CEPH_STAT_RSTAT) ? USE_AUTH_MDS : USE_ANY_MDS;
2294         req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_GETATTR, mode);
2295         if (IS_ERR(req))
2296                 return PTR_ERR(req);
2297         req->r_inode = inode;
2298         ihold(inode);
2299         req->r_num_caps = 1;
2300         req->r_args.getattr.mask = cpu_to_le32(mask);
2301         req->r_locked_page = locked_page;
2302         err = ceph_mdsc_do_request(mdsc, NULL, req);
2303         if (locked_page && err == 0) {
2304                 u64 inline_version = req->r_reply_info.targeti.inline_version;
2305                 if (inline_version == 0) {
2306                         /* the reply is supposed to contain inline data */
2307                         err = -EINVAL;
2308                 } else if (inline_version == CEPH_INLINE_NONE) {
2309                         err = -ENODATA;
2310                 } else {
2311                         err = req->r_reply_info.targeti.inline_len;
2312                 }
2313         }
2314         ceph_mdsc_put_request(req);
2315         dout("do_getattr result=%d\n", err);
2316         return err;
2317 }
2318
2319
2320 /*
2321  * Check inode permissions.  We verify we have a valid value for
2322  * the AUTH cap, then call the generic handler.
2323  */
2324 int ceph_permission(struct inode *inode, int mask)
2325 {
2326         int err;
2327
2328         if (mask & MAY_NOT_BLOCK)
2329                 return -ECHILD;
2330
2331         err = ceph_do_getattr(inode, CEPH_CAP_AUTH_SHARED, false);
2332
2333         if (!err)
2334                 err = generic_permission(inode, mask);
2335         return err;
2336 }
2337
2338 /* Craft a mask of needed caps given a set of requested statx attrs. */
2339 static int statx_to_caps(u32 want, umode_t mode)
2340 {
2341         int mask = 0;
2342
2343         if (want & (STATX_MODE|STATX_UID|STATX_GID|STATX_CTIME|STATX_BTIME))
2344                 mask |= CEPH_CAP_AUTH_SHARED;
2345
2346         if (want & (STATX_NLINK|STATX_CTIME)) {
2347                 /*
2348                  * The link count for directories depends on inode->i_subdirs,
2349                  * and that is only updated when Fs caps are held.
2350                  */
2351                 if (S_ISDIR(mode))
2352                         mask |= CEPH_CAP_FILE_SHARED;
2353                 else
2354                         mask |= CEPH_CAP_LINK_SHARED;
2355         }
2356
2357         if (want & (STATX_ATIME|STATX_MTIME|STATX_CTIME|STATX_SIZE|
2358                     STATX_BLOCKS))
2359                 mask |= CEPH_CAP_FILE_SHARED;
2360
2361         if (want & (STATX_CTIME))
2362                 mask |= CEPH_CAP_XATTR_SHARED;
2363
2364         return mask;
2365 }
2366
2367 /*
2368  * Get all the attributes. If we have sufficient caps for the requested attrs,
2369  * then we can avoid talking to the MDS at all.
2370  */
2371 int ceph_getattr(const struct path *path, struct kstat *stat,
2372                  u32 request_mask, unsigned int flags)
2373 {
2374         struct inode *inode = d_inode(path->dentry);
2375         struct ceph_inode_info *ci = ceph_inode(inode);
2376         u32 valid_mask = STATX_BASIC_STATS;
2377         int err = 0;
2378
2379         /* Skip the getattr altogether if we're asked not to sync */
2380         if (!(flags & AT_STATX_DONT_SYNC)) {
2381                 err = ceph_do_getattr(inode,
2382                                 statx_to_caps(request_mask, inode->i_mode),
2383                                 flags & AT_STATX_FORCE_SYNC);
2384                 if (err)
2385                         return err;
2386         }
2387
2388         generic_fillattr(inode, stat);
2389         stat->ino = ceph_present_inode(inode);
2390
2391         /*
2392          * btime on newly-allocated inodes is 0, so if this is still set to
2393          * that, then assume that it's not valid.
2394          */
2395         if (ci->i_btime.tv_sec || ci->i_btime.tv_nsec) {
2396                 stat->btime = ci->i_btime;
2397                 valid_mask |= STATX_BTIME;
2398         }
2399
2400         if (ceph_snap(inode) == CEPH_NOSNAP)
2401                 stat->dev = inode->i_sb->s_dev;
2402         else
2403                 stat->dev = ci->i_snapid_map ? ci->i_snapid_map->dev : 0;
2404
2405         if (S_ISDIR(inode->i_mode)) {
2406                 if (ceph_test_mount_opt(ceph_sb_to_client(inode->i_sb),
2407                                         RBYTES))
2408                         stat->size = ci->i_rbytes;
2409                 else
2410                         stat->size = ci->i_files + ci->i_subdirs;
2411                 stat->blocks = 0;
2412                 stat->blksize = 65536;
2413                 /*
2414                  * Some applications rely on the number of st_nlink
2415                  * value on directories to be either 0 (if unlinked)
2416                  * or 2 + number of subdirectories.
2417                  */
2418                 if (stat->nlink == 1)
2419                         /* '.' + '..' + subdirs */
2420                         stat->nlink = 1 + 1 + ci->i_subdirs;
2421         }
2422
2423         stat->result_mask = request_mask & valid_mask;
2424         return err;
2425 }