- Do an additional getattr in ll_lookup2 after we get the lock, to refresh the

inode attributes
- mds_blocking_ast didn't wait for all lock holders to finish with the lock
before cancelling it; fixed.
- adjusted more of the wildly inconsistent error code reporting, this time in
mds_getattr
- removed the bad array-walking code that I introduced yesterday to osc_enqueue

diff --git a/lustre/ldlm/ldlm_lock.c b/lustre/ldlm/ldlm_lock.c
index ed5a3c3..fea73d7 100644 (file)
--- a/lustre/ldlm/ldlm_lock.c
+++ b/lustre/ldlm/ldlm_lock.c
@@ -450,8 +450,8 @@ static int ldlm_lock_compat_list(struct ldlm_lock *lock, int send_cbs,
                 rc = 0;
 
                 if (send_cbs && child->l_blocking_ast != NULL) {
-                        CDEBUG(D_OTHER, "incompatible; sending blocking "
-                               "AST.\n");
+                        CDEBUG(D_OTHER, "lock %p incompatible; sending "
+                               "blocking AST.\n", child);
                         ldlm_add_ast_work_item(child, lock);
                 }
         }

diff --git a/lustre/llite/file.c b/lustre/llite/file.c
index 54c3e57..9ac392c 100644 (file)
--- a/lustre/llite/file.c
+++ b/lustre/llite/file.c
@@ -73,7 +73,7 @@ static int ll_file_open(struct inode *inode, struct file *file)
                         }
                         md = lli->lli_smd;
                 }
-                if (lli->lli_smd && lli->lli_smd->lmd_object_id == 0)
+                if (lli->lli_smd->lmd_object_id == 0)
                         LBUG();
                 up(&lli->lli_open_sem);
         }
@@ -94,8 +94,6 @@ static int ll_file_open(struct inode *inode, struct file *file)
                 CERROR("mdc_open didn't assign fd_mdshandle\n");
                 /* XXX handle this how, abort or is it non-fatal? */
         }
-        if (!fd->fd_mdshandle)
-                CERROR("mdc_open didn't assign fd_mdshandle\n");
 
         if (oa == NULL && (oa = obdo_alloc()) == NULL)
                 GOTO(out_mdc, rc = -EINVAL);
@@ -366,7 +364,7 @@ ll_file_write(struct file *file, const char *buf, size_t count, loff_t *ppos)
                                   ll_lock_callback, inode, sizeof(*inode),
                                   lockhs);
                 if (err != ELDLM_OK) {
-                        OBD_FREE(lockhs, md->lmd_stripe_count * sizeof(*lockhs));
+                        OBD_FREE(lockhs, md->lmd_stripe_count *sizeof(*lockhs));
                         CERROR("lock enqueue: err: %d\n", err);
                         RETURN(err);
                 }
@@ -380,7 +378,7 @@ ll_file_write(struct file *file, const char *buf, size_t count, loff_t *ppos)
         if (!(fd->fd_flags & LL_FILE_IGNORE_LOCK)) {
                 err = obd_cancel(&sbi->ll_osc_conn, md, LCK_PW, lockhs);
                 if (err != ELDLM_OK) {
-                        OBD_FREE(lockhs, md->lmd_stripe_count * sizeof(*lockhs));
+                        OBD_FREE(lockhs, md->lmd_stripe_count *sizeof(*lockhs));
                         CERROR("lock cancel: err: %d\n", err);
                         RETURN(err);
                 }

diff --git a/lustre/llite/namei.c b/lustre/llite/namei.c
index 334c49d..cc5412f 100644 (file)
--- a/lustre/llite/namei.c
+++ b/lustre/llite/namei.c
@@ -133,12 +133,13 @@ static struct dentry *ll_lookup2(struct inode * dir, struct dentry *dentry,
                                  struct lookup_intent *it)
 {
         struct ptlrpc_request *request = NULL;
+        struct ptlrpc_request *getattr_req = NULL;
         struct inode * inode = NULL;
         struct ll_sb_info *sbi = ll_i2sbi(dir);
         struct ll_inode_md md;
         struct lustre_handle lockh;
-        int err, offset;
         struct lookup_intent lookup_it = { IT_LOOKUP };
+        int err, offset, mode;
         obd_id ino = 0;
 
         ENTRY;
@@ -195,6 +196,27 @@ static struct dentry *ll_lookup2(struct inode * dir, struct dentry *dentry,
                         it->it_data = NULL;
                         GOTO(out_req, NULL);
                 }
+
+                /* Do a getattr now that we have the lock */
+                if ((it->it_op == IT_UNLINK || it->it_op == IT_RMDIR) &&
+                    it->it_status == 0)
+                        /* the unlink/rmdir succeeded, there's nothing to
+                         * lookup */
+                        goto iget;
+                md.body = lustre_msg_buf(request->rq_repmsg, offset);
+                ino = md.body->fid1.id;
+                mode = md.body->mode;
+                ptlrpc_free_req(request);
+                request = NULL;
+                err = mdc_getattr(&sbi->ll_mdc_conn, ino, mode,
+                                  OBD_MD_FLNOTOBD|OBD_MD_FLEASIZE, 0, &request);
+                if (err) {
+                        CERROR("failure %d inode %Ld\n", err, (long long)ino);
+                        ptlrpc_free_req(request);
+#warning FIXME: must release lock here
+                        RETURN(ERR_PTR(-abs(err)));
+                }
+                offset = 0;
         } else {
                 struct ll_inode_info *lli = ll_i2info(dir);
                 int type;
@@ -215,6 +237,7 @@ static struct dentry *ll_lookup2(struct inode * dir, struct dentry *dentry,
                 }
         }
 
+ iget:
         md.body = lustre_msg_buf(request->rq_repmsg, offset);
         if (S_ISREG(md.body->mode)) {
                 if (request->rq_repmsg->bufcount < offset + 1)
@@ -236,6 +259,7 @@ static struct dentry *ll_lookup2(struct inode * dir, struct dentry *dentry,
         EXIT;
  neg_req:
         ptlrpc_free_req(request);
+        ptlrpc_free_req(getattr_req);
  negative:
         dentry->d_op = &ll_d_ops;
         d_add(dentry, inode);

diff --git a/lustre/mds/handler.c b/lustre/mds/handler.c
index 546a0a2..7cb1fd0 100644 (file)
--- a/lustre/mds/handler.c
+++ b/lustre/mds/handler.c
@@ -421,12 +421,26 @@ static int mds_getlovinfo(struct ptlrpc_request *req)
 int mds_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
                      void *data, __u32 data_len)
 {
-        struct lustre_handle lockh;
+        int do_ast;
         ENTRY;
 
-        ldlm_lock2handle(lock, &lockh);
-        if (ldlm_cli_cancel(&lockh) < 0)
-                LBUG();
+        l_lock(&lock->l_resource->lr_namespace->ns_lock);
+        lock->l_flags |= LDLM_FL_CBPENDING;
+        do_ast = (!lock->l_readers && !lock->l_writers);
+        l_unlock(&lock->l_resource->lr_namespace->ns_lock);
+
+        if (do_ast) {
+                struct lustre_handle lockh;
+                int rc;
+
+                LDLM_DEBUG(lock, "already unused, calling ldlm_cli_cancel");
+                ldlm_lock2handle(lock, &lockh);
+                rc = ldlm_cli_cancel(&lockh);
+                if (rc < 0)
+                        CERROR("ldlm_cli_cancel: %d\n", rc);
+        } else
+                LDLM_DEBUG(lock, "Lock still has references, will be"
+                           "cancelled later");
         RETURN(0);
 }
 
@@ -565,14 +579,16 @@ static int mds_getattr(int offset, struct ptlrpc_request *req)
         struct dentry *de;
         struct inode *inode;
         struct mds_body *body;
-        int rc, size[2] = {sizeof(*body)}, bufcount = 1;
+        int rc = 0, size[2] = {sizeof(*body)}, bufcount = 1;
         ENTRY;
 
         body = lustre_msg_buf(req->rq_reqmsg, offset);
         push_ctxt(&saved, &mds->mds_ctxt);
         de = mds_fid2dentry(mds, &body->fid1, NULL);
-        if (IS_ERR(de))
+        if (IS_ERR(de)) {
+                req->rq_status = -ENOENT;
                 GOTO(out_pop, rc = -ENOENT);
+        }
 
         inode = de->d_inode;
         if (S_ISREG(body->fid1.f_type)) {
@@ -586,18 +602,18 @@ static int mds_getattr(int offset, struct ptlrpc_request *req)
         rc = lustre_pack_msg(bufcount, size, NULL, &req->rq_replen,
                              &req->rq_repmsg);
         if (rc || OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK)) {
-                CERROR("mds: out of memory\n");
-                GOTO(out, rc);
+                CERROR("out of memory or FAIL_MDS_GETATTR_PACK\n");
+                req->rq_status = rc;
+                GOTO(out, rc = 0);
         }
 
-        rc = mds_getattr_internal(mds, de, req, offset, 0);
+        req->rq_status = mds_getattr_internal(mds, de, req, offset, 0);
 
 out:
         l_dput(de);
 out_pop:
         pop_ctxt(&saved);
-        req->rq_status = rc;
-        RETURN(0);
+        RETURN(rc);
 }
 
 static int mds_statfs(struct ptlrpc_request *req)

diff --git a/lustre/mds/mds_reint.c b/lustre/mds/mds_reint.c
index f626289..e28066a 100644 (file)
--- a/lustre/mds/mds_reint.c
+++ b/lustre/mds/mds_reint.c
@@ -397,13 +397,10 @@ static int mds_reint_unlink(struct mds_update_record *rec, int offset,
         struct mds_obd *mds = mds_req2mds(req);
         struct obd_device *obd = req->rq_export->exp_obd;
         char *name;
-        int namelen;
         struct inode *dir, *inode;
-        int lock_mode;
         struct lustre_handle lockh, child_lockh;
         void *handle;
-        int rc = 0;
-        int err;
+        int namelen, lock_mode, err, rc = 0;
         ENTRY;
 
         /* a name was supplied by the client; fid1 is the directory */

diff --git a/lustre/osc/osc_request.c b/lustre/osc/osc_request.c
index 7f7d995..feb1894 100644 (file)
--- a/lustre/osc/osc_request.c
+++ b/lustre/osc/osc_request.c
@@ -593,12 +593,15 @@ static int osc_enqueue(struct lustre_handle *connh, struct lov_stripe_md *md,
                        struct lustre_handle *parent_lock, 
                        __u32 type, void *extentp, int extent_len, __u32 mode,
                        int *flags, void *callback, void *data, int datalen,
-                       struct lustre_handle *lockhs)
+                       struct lustre_handle *lockh)
 {
         __u64 res_id[RES_NAME_SIZE] = { md->lmd_object_id };
         struct obd_device *obddev = class_conn2obd(connh);
         struct ldlm_extent *extent = extentp;
-        int rc, i;
+        struct ldlm_lock *lock;
+        struct inode *inode = data;
+        struct ll_inode_info *lli = ll_i2info(inode);
+        int rc;
         __u32 mode2;
 
         /* Filesystem locks are given a bit of special treatment: first we
@@ -609,7 +612,7 @@ static int osc_enqueue(struct lustre_handle *connh, struct lov_stripe_md *md,
         /* Next, search for already existing extent locks that will cover us */
         //osc_con2dlmcl(conn, &cl, &connection, &rconn);
         rc = ldlm_lock_match(obddev->obd_namespace, res_id, type, extent,
-                             sizeof(extent), mode, lockhs);
+                             sizeof(extent), mode, lockh);
         if (rc == 1) {
                 /* We already have a lock, and it's referenced */
                 return 0;
@@ -625,19 +628,19 @@ static int osc_enqueue(struct lustre_handle *connh, struct lov_stripe_md *md,
                 mode2 = LCK_PW;
 
         rc = ldlm_lock_match(obddev->obd_namespace, res_id, type, extent,
-                             sizeof(extent), mode2, lockhs);
+                             sizeof(extent), mode2, lockh);
         if (rc == 1) {
                 int flags;
                 /* FIXME: This is not incredibly elegant, but it might
                  * be more elegant than adding another parameter to
                  * lock_match.  I want a second opinion. */
-                ldlm_lock_addref(lockhs, mode);
-                ldlm_lock_decref(lockhs, mode2);
+                ldlm_lock_addref(lockh, mode);
+                ldlm_lock_decref(lockh, mode2);
 
                 if (mode == LCK_PR)
                         return 0;
 
-                rc = ldlm_cli_convert(lockhs, mode, &flags);
+                rc = ldlm_cli_convert(lockh, mode, &flags);
                 if (rc)
                         LBUG();
 
@@ -647,28 +650,20 @@ static int osc_enqueue(struct lustre_handle *connh, struct lov_stripe_md *md,
         rc = ldlm_cli_enqueue(connh, NULL, obddev->obd_namespace,
                               parent_lock, res_id, type, extent,
                               sizeof(extent), mode, flags, ldlm_completion_ast,
-                              callback, data, datalen, lockhs);
+                              callback, data, datalen, lockh);
         if (rc)
                 return rc;
 
         /* This code must change if we ever stop passing an inode in as data */
         /* This is ldlm and llite code.  It makes me sad that it's in
          * osc_request.c --phil */
-        l_lock(&obddev->obd_namespace->ns_lock);
-        for (i = 0; i < md->lmd_stripe_count; i++) {
-                struct ldlm_lock *lock = ldlm_handle2lock(&(lockhs[i]));
-                struct inode *inode = data;
-                struct ll_inode_info *lli = ll_i2info(inode);
-
-                if (!lock) {
-                        CERROR("invalid lock in array\n");
-                        continue;
-                }
-
+        lock = ldlm_handle2lock(&lockh);
+        if (lock) {
                 /* Lock already has an extra ref from handle2lock */
+                l_lock(&obddev->obd_namespace->ns_lock);
                 list_add(&lock->l_inode_link, &lli->lli_osc_locks);
+                l_unlock(&obddev->obd_namespace->ns_lock);
         }
-        l_unlock(&obddev->obd_namespace->ns_lock);
 
         return rc;
 }