lustre/xattr: separate ACL and XATTR caches
authorAndrew Perepechko <andrew_perepechko@xyratex.com>
Sun, 9 Feb 2014 07:51:48 +0000 (02:51 -0500)
committerGreg Kroah-Hartman <gregkh@linuxfoundation.org>
Tue, 11 Feb 2014 20:09:58 +0000 (12:09 -0800)
This patch separates ACL and XATTR caches, so that
when updating an ACL only LOOKUP lock is needed and
when updating another XATTR only XATTR lock is needed.

This patch also reverts XATTR cache support for setxattr
because client performing REINT under even PR lock
will deadlock if an active server operation (like unlink)
attempts to cancel all locks, and setxattr has to wait
for it (MDC max-in-flight is 1).

This patch disables the r/o cache if the data is
unreasonably large (larger than maximum single EA
size).

Signed-off-by: Andrew Perepechko <andrew_perepechko@xyratex.com>
Signed-off-by: Nathaniel Clark <nathaniel.l.clark@intel.com>
Reviewed-on: http://review.whamcloud.com/7208
Intel-bug-id: https://jira.hpdd.intel.com/browse/LU-3669
Reviewed-by: Andreas Dilger <andreas.dilger@intel.com>
Reviewed-by: John L. Hammond <john.hammond@intel.com>
Signed-off-by: Oleg Drokin <oleg.drokin@intel.com>
Signed-off-by: Greg Kroah-Hartman <gregkh@linuxfoundation.org>
drivers/staging/lustre/lustre/include/lustre/lustre_idl.h
drivers/staging/lustre/lustre/ldlm/ldlm_lock.c
drivers/staging/lustre/lustre/llite/llite_internal.h
drivers/staging/lustre/lustre/llite/xattr.c
drivers/staging/lustre/lustre/llite/xattr_cache.c
drivers/staging/lustre/lustre/mdc/mdc_internal.h
drivers/staging/lustre/lustre/mdc/mdc_locks.c
drivers/staging/lustre/lustre/mdc/mdc_reint.c
drivers/staging/lustre/lustre/mdc/mdc_request.c
drivers/staging/lustre/lustre/ptlrpc/layout.c

index 05c77c0b2d75f613b8d1afbdfd7ab6e3c3490b0a..4183a359f1f5356bd36f0a9f5fb8fad3c7423215 100644 (file)
@@ -1747,7 +1747,6 @@ static inline __u32 lov_mds_md_size(__u16 stripes, __u32 lmm_magic)
                          OBD_MD_FLGID   | OBD_MD_FLFLAGS | OBD_MD_FLNLINK | \
                          OBD_MD_FLGENER | OBD_MD_FLRDEV  | OBD_MD_FLGROUP)
 
-#define OBD_MD_FLXATTRLOCKED OBD_MD_FLGETATTRLOCK
 #define OBD_MD_FLXATTRALL (OBD_MD_FLXATTR | OBD_MD_FLXATTRLS)
 
 /* don't forget obdo_fid which is way down at the bottom so it can
index 692623beee12f26f8848070379e88b9b20e1f03d..0548aca45e8e96223658557b493cf3db1dce49fe 100644 (file)
@@ -145,8 +145,6 @@ char *ldlm_it2str(int it)
                return "getxattr";
        case IT_LAYOUT:
                return "layout";
-       case IT_SETXATTR:
-               return "setxattr";
        default:
                CERROR("Unknown intent %d\n", it);
                return "UNKNOWN";
index 28669eafb3964656a901328cc8266ddc1b630208..bc17c295794380e69128beaea10891d3575c13a6 100644 (file)
@@ -296,13 +296,6 @@ int ll_xattr_cache_get(struct inode *inode,
                        size_t size,
                        __u64 valid);
 
-int ll_xattr_cache_update(struct inode *inode,
-                       const char *name,
-                       const char *newval,
-                       size_t size,
-                       __u64 valid,
-                       int flags);
-
 /*
  * Locking to guarantee consistency of non-atomic updates to long long i_size,
  * consistency between file size and KMS.
index af83580c6d48bd290487c692e229d2b833d153c4..b1ed4d9ea6beaf752c72136551f6c1ea1f067ad5 100644 (file)
@@ -183,17 +183,11 @@ int ll_setxattr_common(struct inode *inode, const char *name,
                valid |= rce_ops2valid(rce->rce_ops);
        }
 #endif
-       if (sbi->ll_xattr_cache_enabled &&
-           (rce == NULL || rce->rce_ops == RMT_LSETFACL)) {
-               rc = ll_xattr_cache_update(inode, name, pv, size, valid, flags);
-       } else {
                oc = ll_mdscapa_get(inode);
                rc = md_setxattr(sbi->ll_md_exp, ll_inode2fid(inode), oc,
                                valid, name, pv, size, 0, flags,
                                ll_i2suppgid(inode), &req);
                capa_put(oc);
-       }
-
 #ifdef CONFIG_FS_POSIX_ACL
        if (new_value != NULL)
                lustre_posix_acl_xattr_free(new_value, size);
@@ -292,6 +286,7 @@ int ll_getxattr_common(struct inode *inode, const char *name,
        void *xdata;
        struct obd_capa *oc;
        struct rmtacl_ctl_entry *rce = NULL;
+       struct ll_inode_info *lli = ll_i2info(inode);
 
        CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu/%u(%p)\n",
               inode->i_ino, inode->i_generation, inode);
@@ -339,7 +334,7 @@ int ll_getxattr_common(struct inode *inode, const char *name,
         */
        if (xattr_type == XATTR_ACL_ACCESS_T &&
            !(sbi->ll_flags & LL_SBI_RMT_CLIENT)) {
-               struct ll_inode_info *lli = ll_i2info(inode);
+
                struct posix_acl *acl;
 
                spin_lock(&lli->lli_lock);
@@ -358,13 +353,27 @@ int ll_getxattr_common(struct inode *inode, const char *name,
 #endif
 
 do_getxattr:
-       if (sbi->ll_xattr_cache_enabled && (rce == NULL ||
-                                           rce->rce_ops == RMT_LGETFACL ||
-                                           rce->rce_ops == RMT_LSETFACL)) {
+       if (sbi->ll_xattr_cache_enabled && xattr_type != XATTR_ACL_ACCESS_T) {
                rc = ll_xattr_cache_get(inode, name, buffer, size, valid);
+               if (rc == -EAGAIN)
+                       goto getxattr_nocache;
                if (rc < 0)
                        GOTO(out_xattr, rc);
+
+               /* Add "system.posix_acl_access" to the list */
+               if (lli->lli_posix_acl != NULL && valid & OBD_MD_FLXATTRLS) {
+                       if (size == 0) {
+                               rc += sizeof(XATTR_NAME_ACL_ACCESS);
+                       } else if (size - rc >= sizeof(XATTR_NAME_ACL_ACCESS)) {
+                               memcpy(buffer + rc, XATTR_NAME_ACL_ACCESS,
+                                      sizeof(XATTR_NAME_ACL_ACCESS));
+                               rc += sizeof(XATTR_NAME_ACL_ACCESS);
+                       } else {
+                               GOTO(out_xattr, rc = -ERANGE);
+                       }
+               }
        } else {
+getxattr_nocache:
                oc = ll_mdscapa_get(inode);
                rc = md_getxattr(sbi->ll_md_exp, ll_inode2fid(inode), oc,
                                valid | (rce ? rce_ops2valid(rce->rce_ops) : 0),
index 3e3be1f135022bbf0887ec7d3e925b51b29eec45..616d5bdf854dee06e931eb95802977a7f1931448 100644 (file)
@@ -98,13 +98,13 @@ static int ll_xattr_cache_find(struct list_head *cache,
 }
 
 /**
- * This adds or updates an xattr.
+ * This adds an xattr.
  *
  * Add @xattr_name attr with @xattr_val value and @xattr_val_len length,
- * if the attribute already exists, then update its value.
  *
  * \retval 0       success
  * \retval -ENOMEM if no memory could be allocated for the cached attr
+ * \retval -EPROTO if duplicate xattr is being added
  */
 static int ll_xattr_cache_add(struct list_head *cache,
                              const char *xattr_name,
@@ -116,27 +116,8 @@ static int ll_xattr_cache_add(struct list_head *cache,
 
 
        if (ll_xattr_cache_find(cache, xattr_name, &xattr) == 0) {
-               /* Found a cached EA, update it */
-
-               if (xattr_val_len != xattr->xe_vallen) {
-                       char *val;
-                       OBD_ALLOC(val, xattr_val_len);
-                       if (val == NULL) {
-                               CDEBUG(D_CACHE,
-                                      "failed to allocate %u bytes for xattr %s update\n",
-                                      xattr_val_len, xattr_name);
-                               return -ENOMEM;
-                       }
-                       OBD_FREE(xattr->xe_value, xattr->xe_vallen);
-                       xattr->xe_value = val;
-                       xattr->xe_vallen = xattr_val_len;
-               }
-               memcpy(xattr->xe_value, xattr_val, xattr_val_len);
-
-               CDEBUG(D_CACHE, "update: [%s]=%.*s\n", xattr_name,
-                       xattr_val_len, xattr_val);
-
-               return 0;
+               CDEBUG(D_CACHE, "duplicate xattr: [%s]\n", xattr_name);
+               return -EPROTO;
        }
 
        OBD_SLAB_ALLOC_PTR_GFP(xattr, xattr_kmem, __GFP_IO);
@@ -292,7 +273,7 @@ int ll_xattr_cache_destroy(struct inode *inode)
 }
 
 /**
- * Match or enqueue a PR or PW LDLM lock.
+ * Match or enqueue a PR lock.
  *
  * Find or request an LDLM lock with xattr data.
  * Since LDLM does not provide API for atomic match_or_enqueue,
@@ -322,9 +303,7 @@ static int ll_xattr_find_get_lock(struct inode *inode,
 
        mutex_lock(&lli->lli_xattrs_enq_lock);
        /* Try matching first. */
-       mode = ll_take_md_lock(inode, MDS_INODELOCK_XATTR, &lockh, 0,
-                              oit->it_op == IT_SETXATTR ? LCK_PW :
-                                                          (LCK_PR | LCK_PW));
+       mode = ll_take_md_lock(inode, MDS_INODELOCK_XATTR, &lockh, 0, LCK_PR);
        if (mode != 0) {
                /* fake oit in mdc_revalidate_lock() manner */
                oit->d.lustre.it_lock_handle = lockh.cookie;
@@ -340,13 +319,7 @@ static int ll_xattr_find_get_lock(struct inode *inode,
                return PTR_ERR(op_data);
        }
 
-       op_data->op_valid = OBD_MD_FLXATTR | OBD_MD_FLXATTRLS |
-                           OBD_MD_FLXATTRLOCKED;
-#ifdef CONFIG_FS_POSIX_ACL
-       /* If working with ACLs, we would like to cache local ACLs */
-       if (sbi->ll_flags & LL_SBI_RMT_CLIENT)
-               op_data->op_valid |= OBD_MD_FLRMTLGETFACL;
-#endif
+       op_data->op_valid = OBD_MD_FLXATTR | OBD_MD_FLXATTRLS;
 
        rc = md_enqueue(exp, &einfo, oit, op_data, &lockh, NULL, 0, NULL, 0);
        ll_finish_md_op_data(op_data);
@@ -409,7 +382,11 @@ static int ll_xattr_cache_refill(struct inode *inode, struct lookup_intent *oit)
        if (oit->d.lustre.it_status < 0) {
                CDEBUG(D_CACHE, "getxattr intent returned %d for fid "DFID"\n",
                       oit->d.lustre.it_status, PFID(ll_inode2fid(inode)));
-               GOTO(out_destroy, rc = oit->d.lustre.it_status);
+               rc = oit->d.lustre.it_status;
+               /* xattr data is so large that we don't want to cache it */
+               if (rc == -ERANGE)
+                       rc = -EAGAIN;
+               GOTO(out_destroy, rc);
        }
 
        body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
@@ -447,6 +424,11 @@ static int ll_xattr_cache_refill(struct inode *inode, struct lookup_intent *oit)
                        rc = -EPROTO;
                } else if (OBD_FAIL_CHECK(OBD_FAIL_LLITE_XATTR_ENOMEM)) {
                        rc = -ENOMEM;
+               } else if (!strcmp(xdata, XATTR_NAME_ACL_ACCESS)) {
+                       /* Filter out ACL ACCESS since it's cached separately */
+                       CDEBUG(D_CACHE, "not caching %s\n",
+                              XATTR_NAME_ACL_ACCESS);
+                       rc = 0;
                } else {
                        rc = ll_xattr_cache_add(&lli->lli_xattrs, xdata, xval,
                                                *xsizes);
@@ -467,8 +449,7 @@ static int ll_xattr_cache_refill(struct inode *inode, struct lookup_intent *oit)
 
        GOTO(out_maybe_drop, rc);
 out_maybe_drop:
-       /* drop lock on error or getxattr */
-       if (rc != 0 || oit->it_op != IT_SETXATTR)
+
                ll_intent_drop_lock(oit);
 
        if (rc != 0)
@@ -553,65 +534,3 @@ out:
 
        return rc;
 }
-
-
-/**
- * Set/update an xattr value or remove xattr using the write-through cache.
- *
- * Set/update the xattr value (if @valid has OBD_MD_FLXATTR) of @name to @newval
- * or
- * remove the xattr @name (@valid has OBD_MD_FLXATTRRM set) from @inode.
- * @flags is either XATTR_CREATE or XATTR_REPLACE as defined by setxattr(2)
- *
- * \retval 0        no error occured
- * \retval -EPROTO  network protocol error
- * \retval -ENOMEM  not enough memory for the cache
- * \retval -ERANGE  the buffer is not large enough
- * \retval -ENODATA no such attr (in the removal case)
- */
-int ll_xattr_cache_update(struct inode *inode,
-                       const char *name,
-                       const char *newval,
-                       size_t size,
-                       __u64 valid,
-                       int flags)
-{
-       struct lookup_intent oit = { .it_op = IT_SETXATTR };
-       struct ll_sb_info *sbi = ll_i2sbi(inode);
-       struct ptlrpc_request *req = NULL;
-       struct ll_inode_info *lli = ll_i2info(inode);
-       struct obd_capa *oc;
-       int rc;
-
-
-
-       LASSERT(!!(valid & OBD_MD_FLXATTR) ^ !!(valid & OBD_MD_FLXATTRRM));
-
-       rc = ll_xattr_cache_refill(inode, &oit);
-       if (rc)
-               return rc;
-
-       oc = ll_mdscapa_get(inode);
-       rc = md_setxattr(sbi->ll_md_exp, ll_inode2fid(inode), oc,
-                       valid | OBD_MD_FLXATTRLOCKED, name, newval,
-                       size, 0, flags, ll_i2suppgid(inode), &req);
-       capa_put(oc);
-
-       if (rc) {
-               ll_intent_drop_lock(&oit);
-               GOTO(out, rc);
-       }
-
-       if (valid & OBD_MD_FLXATTR)
-               rc = ll_xattr_cache_add(&lli->lli_xattrs, name, newval, size);
-       else if (valid & OBD_MD_FLXATTRRM)
-               rc = ll_xattr_cache_del(&lli->lli_xattrs, name);
-
-       ll_intent_drop_lock(&oit);
-       GOTO(out, rc);
-out:
-       up_write(&lli->lli_xattrs_list_rwsem);
-       ptlrpc_req_finished(req);
-
-       return rc;
-}
index 506982996c0e6fda4f252a69c487d9b74b331f66..fc21777b53c622dce5abff995d1fb40d11d4967e 100644 (file)
@@ -101,7 +101,7 @@ int mdc_enqueue(struct obd_export *exp, struct ldlm_enqueue_info *einfo,
                struct lustre_handle *lockh, void *lmm, int lmmsize,
                struct ptlrpc_request **req, __u64 extra_lock_flags);
 
-int mdc_resource_get_unused(struct obd_export *exp, struct lu_fid *fid,
+int mdc_resource_get_unused(struct obd_export *exp, const struct lu_fid *fid,
                            struct list_head *cancels, ldlm_mode_t mode,
                            __u64 bits);
 /* mdc/mdc_request.c */
index 8aa7c80c200218f6577095854c19229ed79a6410..288a41ec60c9153c1feadeaddf9aba7e9398692b 100644 (file)
@@ -378,13 +378,6 @@ mdc_intent_getxattr_pack(struct obd_export *exp,
 
        mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
 
-       if (it->it_op == IT_SETXATTR)
-               /* If we want to upgrade to LCK_PW, let's cancel LCK_PR
-                * locks now. This avoids unnecessary ASTs. */
-               count = mdc_resource_get_unused(exp, &op_data->op_fid1,
-                                               &cancels, LCK_PW,
-                                               MDS_INODELOCK_XATTR);
-
        rc = ldlm_prep_enqueue_req(exp, req, &cancels, count);
        if (rc) {
                ptlrpc_request_free(req);
@@ -842,7 +835,7 @@ resend:
                        return -EOPNOTSUPP;
                req = mdc_intent_layout_pack(exp, it, op_data);
                lvb_type = LVB_T_LAYOUT;
-       } else if (it->it_op & (IT_GETXATTR | IT_SETXATTR)) {
+       } else if (it->it_op & IT_GETXATTR) {
                req = mdc_intent_getxattr_pack(exp, it, op_data);
        } else {
                LBUG();
index 9f3a345f34e04579f316f52f905bc1c37d9acbc6..1aea154e122be7fc47a5ca731d6db9e5b632652b 100644 (file)
@@ -66,7 +66,7 @@ static int mdc_reint(struct ptlrpc_request *request,
 /* Find and cancel locally locks matched by inode @bits & @mode in the resource
  * found by @fid. Found locks are added into @cancel list. Returns the amount of
  * locks added to @cancels list. */
-int mdc_resource_get_unused(struct obd_export *exp, struct lu_fid *fid,
+int mdc_resource_get_unused(struct obd_export *exp, const struct lu_fid *fid,
                            struct list_head *cancels, ldlm_mode_t mode,
                            __u64 bits)
 {
index 83013927e13152d8b414dc715ec19df49f827dac..17c8e1486daa7ec36e702356c674fb73bed98f73 100644 (file)
@@ -355,10 +355,32 @@ static int mdc_xattr_common(struct obd_export *exp,const struct req_format *fmt,
                                     input_size);
        }
 
-       rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, opcode);
-       if (rc) {
-               ptlrpc_request_free(req);
-               return rc;
+       /* Flush local XATTR locks to get rid of a possible cancel RPC */
+       if (opcode == MDS_REINT && fid_is_sane(fid) &&
+           exp->exp_connect_data.ocd_ibits_known & MDS_INODELOCK_XATTR) {
+               LIST_HEAD(cancels);
+               int count;
+
+               /* Without that packing would fail */
+               if (input_size == 0)
+                       req_capsule_set_size(&req->rq_pill, &RMF_EADATA,
+                                            RCL_CLIENT, 0);
+
+               count = mdc_resource_get_unused(exp, fid,
+                                               &cancels, LCK_EX,
+                                               MDS_INODELOCK_XATTR);
+
+               rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
+               if (rc) {
+                       ptlrpc_request_free(req);
+                       return rc;
+               }
+       } else {
+               rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, opcode);
+               if (rc) {
+                       ptlrpc_request_free(req);
+                       return rc;
+               }
        }
 
        if (opcode == MDS_REINT) {
index 9b8f691cbeb78c838003ec9953855a60fb702268..41c12e00129f9cb1bb9035ea0e10be46171fad7a 100644 (file)
@@ -295,7 +295,8 @@ static const struct req_msg_field *mds_reint_setxattr_client[] = {
        &RMF_REC_REINT,
        &RMF_CAPA1,
        &RMF_NAME,
-       &RMF_EADATA
+       &RMF_EADATA,
+       &RMF_DLM_REQ
 };
 
 static const struct req_msg_field *mdt_swap_layouts[] = {