Whamcloud - gitweb
- CROW-related fixes from b_hd_mdref
[fs/lustre-release.git] / lustre / mds / mds_open.c
index 3a8fd77..e76ff94 100644 (file)
@@ -111,7 +111,7 @@ static void mds_mfd_destroy(struct mds_file_data *mfd)
         mds_mfd_put(mfd);
 }
 
-
+#ifdef IFILTERDATA_ACTUALLY_USED
 /* Caller must hold mds->mds_epoch_sem */
 static int mds_alloc_filterdata(struct inode *inode)
 {
@@ -131,6 +131,7 @@ static void mds_free_filterdata(struct inode *inode)
         inode->i_filterdata = NULL;
         iput(inode);
 }
+#endif /*IFILTERDATA_ACTUALLY_USED*/
 
 /* Write access to a file: executors cause a negative count,
  * writers a positive count.  The semaphore is needed to perform
@@ -155,7 +156,7 @@ static int mds_get_write_access(struct mds_obd *mds, struct inode *inode,
                 RETURN(-ETXTBSY);
         }
 
-
+#ifdef IFILTERDATA_ACTUALLY_USED
         if (MDS_FILTERDATA(inode) && MDS_FILTERDATA(inode)->io_epoch != 0) {
                 CDEBUG(D_INODE, "continuing MDS epoch "LPU64" for ino %lu/%u\n",
                        MDS_FILTERDATA(inode)->io_epoch, inode->i_ino,
@@ -169,14 +170,17 @@ static int mds_get_write_access(struct mds_obd *mds, struct inode *inode,
                 rc = -ENOMEM;
                 goto out;
         }
+#endif /*IFILTERDATA_ACTUALLY_USED*/
         if (epoch > mds->mds_io_epoch)
                 mds->mds_io_epoch = epoch;
         else
                 mds->mds_io_epoch++;
+#ifdef IFILTERDATA_ACTUALLY_USED
         MDS_FILTERDATA(inode)->io_epoch = mds->mds_io_epoch;
         CDEBUG(D_INODE, "starting MDS epoch "LPU64" for ino %lu/%u\n",
                mds->mds_io_epoch, inode->i_ino, inode->i_generation);
  out:
+#endif /*IFILTERDATA_ACTUALLY_USED*/
         if (rc == 0)
                 atomic_inc(&inode->i_writecount);
         up(&mds->mds_epoch_sem);
@@ -201,7 +205,9 @@ static int mds_put_write_access(struct mds_obd *mds, struct inode *inode,
         if (!unlinking && !(body->valid & OBD_MD_FLSIZE))
                 GOTO(out, rc = EAGAIN);
 #endif
+#ifdef IFILTERDATA_ACTUALLY_USED
         mds_free_filterdata(inode);
+#endif
  out:
         up(&mds->mds_epoch_sem);
         return rc;
@@ -241,33 +247,34 @@ static struct mds_file_data *mds_dentry_open(struct dentry *dentry,
         struct mds_obd *mds = mds_req2mds(req);
         struct mds_file_data *mfd;
         struct mds_body *body;
-        int error;
+        int rc = 0;
         ENTRY;
 
         mfd = mds_mfd_new();
         if (mfd == NULL) {
                 CERROR("mds: out of memory\n");
-                GOTO(cleanup_dentry, error = -ENOMEM);
+                GOTO(cleanup_dentry, rc = -ENOMEM);
         }
 
         body = lustre_msg_buf(req->rq_repmsg, 1, sizeof (*body));
 
         if (flags & FMODE_WRITE) {
                 /* FIXME: in recovery, need to pass old epoch here */
-                error = mds_get_write_access(mds, dentry->d_inode, 0);
-                if (error)
-                        GOTO(cleanup_mfd, error);
+                rc = mds_get_write_access(mds, dentry->d_inode, 0);
+                if (rc)
+                        GOTO(cleanup_mfd, rc);
+#ifdef IFILTERDATA_ACTUALLY_USED
                 body->io_epoch = MDS_FILTERDATA(dentry->d_inode)->io_epoch;
+#endif
         } else if (flags & FMODE_EXEC) {
-                error = mds_deny_write_access(mds, dentry->d_inode);
-                if (error)
-                        GOTO(cleanup_mfd, error);
+                rc = mds_deny_write_access(mds, dentry->d_inode);
+                if (rc)
+                        GOTO(cleanup_mfd, rc);
         }
 
         dget(dentry);
 
-        /* Mark the file as open to handle open-unlink. */
-
+        /* mark the file as open to handle open-unlink. */
         DOWN_WRITE_I_ALLOC_SEM(dentry->d_inode);
         mds_orphan_open_inc(dentry->d_inode);
         UP_WRITE_I_ALLOC_SEM(dentry->d_inode);
@@ -282,41 +289,42 @@ static struct mds_file_data *mds_dentry_open(struct dentry *dentry,
         mds_mfd_put(mfd);
 
         body->handle.cookie = mfd->mfd_handle.h_cookie;
-
         RETURN(mfd);
-
 cleanup_mfd:
         mds_mfd_put(mfd);
         mds_mfd_destroy(mfd);
 cleanup_dentry:
-        return ERR_PTR(error);
+        return ERR_PTR(rc);
 }
 
-static void mds_objids_from_lmm(obd_id *ids, struct lov_mds_md *lmm,
-                                struct lov_desc *desc)
+static inline void
+mds_objids_from_lmm(obd_id *ids, struct lov_mds_md *lmm,
+                    struct lov_desc *desc)
 {
         int i;
+        
         for (i = 0; i < le32_to_cpu(lmm->lmm_stripe_count); i++) {
                 ids[le32_to_cpu(lmm->lmm_objects[i].l_ost_idx)] =
                         le64_to_cpu(lmm->lmm_objects[i].l_object_id);
         }
 }
 
-/* Must be called with i_sem held */
-static int mds_create_objects(struct ptlrpc_request *req, int offset,
-                              struct mds_update_record *rec,
-                              struct mds_obd *mds, struct obd_device *obd,
-                              struct dentry *dchild, void **handle, 
-                              obd_id **ids)
+/* must be called with i_sem held */
+int
+mds_create_objects(struct obd_device *obd, struct ptlrpc_request *req,
+                   int offset, struct mds_update_record *rec,
+                   struct dentry *dchild, void **handle,
+                   obd_id **ids)
 {
-        struct obdo *oa = NULL;
+        struct inode *inode = dchild->d_inode;
+        struct mds_obd *mds = &obd->u.mds;
         struct obd_trans_info oti = { 0 };
-        struct mds_body *body;
         struct lov_stripe_md *lsm = NULL;
         struct lov_mds_md *lmm = NULL;
-        struct inode *inode = dchild->d_inode;
-        void *lmm_buf;
         int rc, lmm_bufsize, lmm_size;
+        struct obdo *oa = NULL;
+        struct mds_body *body;
+        void *lmm_buf;
         ENTRY;
 
         if (rec->ur_flags & MDS_OPEN_DELAY_CREATE ||
@@ -334,7 +342,7 @@ static int mds_create_objects(struct ptlrpc_request *req, int offset,
         if (*ids == NULL)
                 RETURN(-ENOMEM);
         oti.oti_objid = *ids;
-
+                
         /* replay case */
         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) {
                 LASSERT(id_ino(rec->ur_id2));
@@ -344,44 +352,46 @@ static int mds_create_objects(struct ptlrpc_request *req, int offset,
                 LASSERT(lmm);
 
                 if (*handle == NULL)
-                        *handle = fsfilt_start(obd,inode,FSFILT_OP_CREATE,NULL);
+                        *handle = fsfilt_start(obd, inode, FSFILT_OP_CREATE, NULL);
                 if (IS_ERR(*handle)) {
                         rc = PTR_ERR(*handle);
                         *handle = NULL;
-                        GOTO(out_oa, rc);
+                        RETURN(rc);
                 }
 
                 mds_objids_from_lmm(*ids, lmm, &mds->mds_dt_desc);
 
                 lmm_buf = lustre_msg_buf(req->rq_repmsg, offset, 0);
                 lmm_bufsize = req->rq_repmsg->buflens[offset];
-                LASSERT(lmm_buf);
+                LASSERT(lmm_buf != NULL);
                 LASSERT(lmm_bufsize >= lmm_size);
+
                 memcpy(lmm_buf, lmm, lmm_size);
-                rc = fsfilt_set_md(obd, inode, *handle, lmm, lmm_size);
+                rc = fsfilt_set_md(obd, inode, *handle, lmm,
+                                   lmm_size, EA_LOV);
                 if (rc)
                         CERROR("open replay failed to set md:%d\n", rc);
                 RETURN(0);
         }
 
         if (OBD_FAIL_CHECK_ONCE(OBD_FAIL_MDS_ALLOC_OBDO))
-                GOTO(out_ids, rc = -ENOMEM);
+                RETURN(-ENOMEM);
 
         oa = obdo_alloc();
         if (oa == NULL)
-                GOTO(out_ids, rc = -ENOMEM);
+                RETURN(-ENOMEM);
         oa->o_mode = S_IFREG | 0600;
         oa->o_id = inode->i_ino;
         oa->o_gr = FILTER_GROUP_FIRST_MDS + mds->mds_num;
         oa->o_generation = inode->i_generation;
         oa->o_uid = 0; /* must have 0 uid / gid on OST */
         oa->o_gid = 0;
-        oa->o_valid = OBD_MD_FLID | OBD_MD_FLGENER | OBD_MD_FLTYPE |
-                OBD_MD_FLMODE | OBD_MD_FLUID | OBD_MD_FLGID | OBD_MD_FLGROUP;
+        oa->o_valid = OBD_MD_FLID | OBD_MD_FLTYPE | OBD_MD_FLMODE |
+                        OBD_MD_FLUID | OBD_MD_FLGID | OBD_MD_FLGROUP;
         oa->o_size = 0;
 
-        obdo_from_inode(oa, inode, OBD_MD_FLTYPE|OBD_MD_FLATIME|OBD_MD_FLMTIME|
-                        OBD_MD_FLCTIME);
+        obdo_from_inode(oa, inode, OBD_MD_FLTYPE | OBD_MD_FLATIME |
+                        OBD_MD_FLMTIME | OBD_MD_FLCTIME);
 
         if (!(rec->ur_flags & MDS_OPEN_HAS_OBJS)) {
                 /* check if things like lfs setstripe are sending us the ea */
@@ -392,18 +402,33 @@ static int mds_create_objects(struct ptlrpc_request *req, int offset,
                         if (rc)
                                 GOTO(out_oa, rc);
                 } else {
-                        /* Per-directory striping default code removed, because
-                         * it uses the same unnamed EA storage as the directory
-                         * striping for CMD. -p */
-                } 
+                        OBD_ALLOC(lmm, mds->mds_max_mdsize);
+                        if (lmm == NULL)
+                                GOTO(out_oa, rc = -ENOMEM);
+
+                        lmm_size = mds->mds_max_mdsize;
+                        rc = mds_get_md(obd, dchild->d_parent->d_inode,
+                                        lmm, &lmm_size, 1, 0);
+                        if (rc > 0)
+                                rc = obd_iocontrol(OBD_IOC_LOV_SETSTRIPE,
+                                                   mds->mds_dt_exp,
+                                                   0, &lsm, lmm);
+                        OBD_FREE(lmm, mds->mds_max_mdsize);
+                        if (rc)
+                                GOTO(out_oa, rc);
+                }
+
                 LASSERT(oa->o_gr >= FILTER_GROUP_FIRST_MDS);
-                rc = obd_create(mds->mds_dt_exp, oa, &lsm, &oti);
+                oti.oti_flags |= OBD_MODE_CROW;
+                rc = obd_create(mds->mds_dt_exp, oa, NULL, 0, &lsm, &oti);
+
                 if (rc) {
                         int level = D_ERROR;
                         if (rc == -ENOSPC)
                                 level = D_INODE;
-                        CDEBUG(level, "error creating objects for "
-                                      "inode %lu: rc = %d\n",
+                        CDEBUG((rc == -ENOSPC ? D_INODE : D_ERROR),
+                               "error creating objects for "
+                               "inode %lu: rc = %d\n",
                                inode->i_ino, rc);
                         if (rc > 0) {
                                 CERROR("obd_create returned invalid "
@@ -415,16 +440,17 @@ static int mds_create_objects(struct ptlrpc_request *req, int offset,
         } else {
                 rc = obd_iocontrol(OBD_IOC_LOV_SETEA, mds->mds_dt_exp,
                                    0, &lsm, rec->ur_eadata);
-                if (rc) {
+                if (rc)
                         GOTO(out_oa, rc);
-                }
+
                 lsm->lsm_object_id = oa->o_id;
                 lsm->lsm_object_gr = oa->o_gr;
         }
         if (inode->i_size) {
                 oa->o_size = inode->i_size;
-                obdo_from_inode(oa, inode, OBD_MD_FLTYPE|OBD_MD_FLATIME|
-                                OBD_MD_FLMTIME| OBD_MD_FLCTIME| OBD_MD_FLSIZE);
+                obdo_from_inode(oa, inode, OBD_MD_FLTYPE | OBD_MD_FLATIME |
+                                OBD_MD_FLMTIME | OBD_MD_FLCTIME | OBD_MD_FLSIZE);
+                
                 rc = obd_setattr(mds->mds_dt_exp, oa, lsm, &oti);
                 if (rc) {
                         CERROR("error setting attrs for inode %lu: rc %d\n",
@@ -445,7 +471,11 @@ static int mds_create_objects(struct ptlrpc_request *req, int offset,
         rc = obd_packmd(mds->mds_dt_exp, &lmm, lsm);
         if (!id_ino(rec->ur_id2))
                 obd_free_memmd(mds->mds_dt_exp, &lsm);
-        LASSERT(rc >= 0);
+        if (rc < 0) {
+                CERROR("cannot pack lsm, err = %d\n", rc);
+                GOTO(out_oa, rc);
+        }
+
         lmm_size = rc;
         body->eadatasize = rc;
 
@@ -454,10 +484,12 @@ static int mds_create_objects(struct ptlrpc_request *req, int offset,
         if (IS_ERR(*handle)) {
                 rc = PTR_ERR(*handle);
                 *handle = NULL;
-                GOTO(out_ids, rc);
+                GOTO(out_oa, rc);
         }
 
-        rc = fsfilt_set_md(obd, inode, *handle, lmm, lmm_size);
+        rc = fsfilt_set_md(obd, inode, *handle, lmm,
+                           lmm_size, EA_LOV);
+        
         lmm_buf = lustre_msg_buf(req->rq_repmsg, offset, 0);
         lmm_bufsize = req->rq_repmsg->buflens[offset];
         LASSERT(lmm_buf);
@@ -465,17 +497,65 @@ static int mds_create_objects(struct ptlrpc_request *req, int offset,
 
         memcpy(lmm_buf, lmm, lmm_size);
         obd_free_diskmd(mds->mds_dt_exp, &lmm);
- out_oa:
+out_oa:
         oti_free_cookies(&oti);
         obdo_free(oa);
- out_ids:
-        if (rc) {
-                OBD_FREE(*ids, mds->mds_dt_desc.ld_tgt_count * sizeof(**ids));
-                *ids = NULL;
-        }
+
+        if (lsm)
+                obd_free_memmd(mds->mds_dt_exp, &lsm);
         RETURN(rc);
 }
 
+int
+mds_destroy_object(struct obd_device *obd,
+                   struct inode *inode, int async)
+{
+        struct mds_obd *mds = &obd->u.mds;
+        struct lov_mds_md *lmm = NULL;
+        int rc, lmm_size;
+        ENTRY;
+
+        LASSERT(inode != NULL);
+
+        if (inode->i_nlink != 0) {
+                CDEBUG(D_INODE, "attempt to destroy OSS object when "
+                       "i_nlink == %d\n", (int)inode->i_nlink);
+                RETURN(0);
+        }
+        
+        OBD_ALLOC(lmm, mds->mds_max_mdsize);
+        if (lmm == NULL)
+                RETURN(-ENOMEM);
+
+        lmm_size = mds->mds_max_mdsize;
+        rc = mds_get_md(obd, inode, lmm, &lmm_size, 1, 0);
+        if (rc < 0) {
+                CERROR("no stripe info for %lu/%lu inode\n",
+                       (unsigned long)inode->i_ino,
+                       (unsigned long)inode->i_generation);
+                GOTO(out_free_lmm, rc);
+        }
+
+        if (rc > 0) {
+                /* asynchronously unlink objecect on OSS */
+                rc = mds_unlink_object(mds, inode, lmm, lmm_size,
+                                       NULL, 0, async);
+                if (rc) {
+                        CERROR("error unlinking object on OSS, "
+                               "err %d\n", rc);
+                        GOTO(out_free_lmm, rc);
+                }
+        } else {
+                CDEBUG(D_INODE, "no stripping info found for inode "
+                     "%lu/%lu\n", (unsigned long)inode->i_ino, 
+                     (unsigned long)inode->i_generation);
+        }
+        EXIT;
+out_free_lmm:
+        OBD_FREE(lmm, mds->mds_max_mdsize);
+        return rc;
+}
+
 static void reconstruct_open(struct mds_update_record *rec, int offset,
                              struct ptlrpc_request *req,
                              struct lustre_handle *child_lockh)
@@ -485,7 +565,7 @@ static void reconstruct_open(struct mds_update_record *rec, int offset,
         struct mds_obd *mds = mds_req2mds(req);
         struct mds_file_data *mfd;
         struct obd_device *obd = req->rq_export->exp_obd;
-        struct dentry *parent, *dchild;
+        struct dentry *parent = NULL, *dchild;
         struct ldlm_reply *rep;
         struct mds_body *body;
         struct list_head *t;
@@ -507,14 +587,19 @@ static void reconstruct_open(struct mds_update_record *rec, int offset,
                 return; /* error looking up parent or child */
         }
 
-        parent = mds_id2dentry(obd, rec->ur_id1, NULL);
-        LASSERTF(!IS_ERR(parent), "lid "DLID4" rc %ld\n", 
-                 OLID4(rec->ur_id1), PTR_ERR(parent));
-
-        dchild = ll_lookup_one_len(rec->ur_name, parent, 
-                                   rec->ur_namelen - 1);
-        LASSERTF(!IS_ERR(dchild), "parent "DLID4" child %s rc %ld\n",
-                 OLID4(rec->ur_id1), rec->ur_name, PTR_ERR(dchild));
+        if (rec->ur_namelen == 1) {
+                CDEBUG(D_HA, "OPEN by fid "DLID4" (RESENT)\n",
+                       OLID4(rec->ur_id1));
+                dchild = mds_id2dentry(obd, rec->ur_id1, NULL);
+        } else {
+                parent = mds_id2dentry(obd, rec->ur_id1, NULL);
+                LASSERTF(!IS_ERR(parent), "lid "DLID4" rc %ld\n", 
+                                OLID4(rec->ur_id1), PTR_ERR(parent));
+                dchild = ll_lookup_one_len(rec->ur_name, parent, 
+                                rec->ur_namelen - 1);
+                LASSERTF(!IS_ERR(dchild), "parent "DLID4" child %s rc %ld\n",
+                                OLID4(rec->ur_id1), rec->ur_name, PTR_ERR(dchild));
+        }
 
         if (!dchild->d_inode)
                 GOTO(out_dput, 0); /* child not present to open */
@@ -530,7 +615,7 @@ static void reconstruct_open(struct mds_update_record *rec, int offset,
         mds_pack_inode2body(obd, body, dchild->d_inode, 1);
         if (S_ISREG(dchild->d_inode->i_mode)) {
                 rc = mds_pack_md(obd, req->rq_repmsg, 2, body,
-                                 dchild->d_inode, 1);
+                                 dchild->d_inode, 1, 0);
 
                 if (rc)
                         LASSERT(rc == req->rq_status);
@@ -593,7 +678,8 @@ static void reconstruct_open(struct mds_update_record *rec, int offset,
  out_dput:
         if (put_child)
                 l_dput(dchild);
-        l_dput(parent);
+        if (parent)
+                l_dput(parent);
         EXIT;
 }
 
@@ -614,12 +700,12 @@ static int accmode(int flags)
 /* Handles object creation, actual opening, and I/O epoch */
 static int mds_finish_open(struct ptlrpc_request *req, struct dentry *dchild,
                            struct mds_body *body, int flags, void **handle,
-                           struct mds_update_record *rec,struct ldlm_reply *rep)
+                           struct mds_update_record *rec, struct ldlm_reply *rep)
 {
-        struct mds_obd *mds = mds_req2mds(req);
         struct obd_device *obd = req->rq_export->exp_obd;
+        struct mds_obd *mds = mds_req2mds(req);
         struct mds_file_data *mfd = NULL;
-        obd_id *ids = NULL; /* object IDs created */
+        obd_id *ids = NULL;
         unsigned mode;
         int rc = 0;
         ENTRY;
@@ -627,51 +713,63 @@ static int mds_finish_open(struct ptlrpc_request *req, struct dentry *dchild,
         /* atomically create objects if necessary */
         down(&dchild->d_inode->i_sem);
         mode = dchild->d_inode->i_mode;
+
         if ((S_ISREG(mode) && !(body->valid & OBD_MD_FLEASIZE)) || 
             (S_ISDIR(mode) && !(body->valid & OBD_MD_FLDIREA))) {
                 rc = mds_pack_md(obd, req->rq_repmsg, 2, body,
-                                 dchild->d_inode, 0);
+                                 dchild->d_inode, 0, 0);
                 if (rc) {
                         up(&dchild->d_inode->i_sem);
                         RETURN(rc);
                 }
         }
+
         if (rec != NULL) {
-                /* no EA: create objects */
                 if ((body->valid & OBD_MD_FLEASIZE) &&
                     (rec->ur_flags & MDS_OPEN_HAS_EA)) {
                         up(&dchild->d_inode->i_sem);
                         RETURN(-EEXIST);
                 }
+                
                 if (!(body->valid & OBD_MD_FLEASIZE)) {
                         /* no EA: create objects */
-                        rc = mds_create_objects(req, 2, rec, mds, obd,
+                        rc = mds_create_objects(obd, req, 2, rec,
                                                 dchild, handle, &ids);
                         if (rc) {
-                                CERROR("mds_create_objects: rc = %d\n", rc);
+                                CERROR("mds_create_object: rc = %d\n", rc);
                                 up(&dchild->d_inode->i_sem);
                                 RETURN(rc);
                         }
                 }
+                
                 if (S_ISREG(dchild->d_inode->i_mode) &&
                     (body->valid & OBD_MD_FLEASIZE)) {
                         rc = mds_revalidate_lov_ea(obd, dchild->d_inode,
                                                    req->rq_repmsg, 2);
                         if (!rc)
                                 rc = mds_pack_md(obd, req->rq_repmsg, 2, body,
-                                                 dchild->d_inode, 0);
+                                                 dchild->d_inode, 0, 0);
                         if (rc) {
                                 up(&dchild->d_inode->i_sem);
                                 RETURN(rc);
                         }
                 }
         }
+        
+        rc = mds_pack_acl(obd, req->rq_repmsg, 3, body, dchild->d_inode);
+        if (rc < 0) {
+                CERROR("mds_pack_acl: rc = %d\n", rc);
+                up(&dchild->d_inode->i_sem);
+                RETURN(rc);
+        }
+
         /* If the inode has no EA data, then MDSs hold size, mtime */
         if (S_ISREG(dchild->d_inode->i_mode) &&
             !(body->valid & OBD_MD_FLEASIZE)) {
                 body->valid |= (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
                                 OBD_MD_FLATIME | OBD_MD_FLMTIME);
         }
+
         up(&dchild->d_inode->i_sem);
 
         intent_set_disposition(rep, DISP_OPEN_OPEN);
@@ -681,12 +779,12 @@ static int mds_finish_open(struct ptlrpc_request *req, struct dentry *dchild,
 
         CDEBUG(D_INODE, "mfd %p, cookie "LPX64"\n", mfd,
                mfd->mfd_handle.h_cookie);
+
         if (ids != NULL) {
                 mds_dt_update_objids(obd, ids);
                 OBD_FREE(ids, sizeof(*ids) * mds->mds_dt_desc.ld_tgt_count);
         }
-        //if (rc)
-        //        mds_mfd_destroy(mfd);
+
         RETURN(rc);
 }
 
@@ -705,9 +803,7 @@ static int mds_open_by_id(struct ptlrpc_request *req,
         ENTRY;
 
         down(&pending_dir->i_sem);
-        
         idlen = ll_id2str(idname, id_ino(id), id_gen(id));
-        
         dchild = lookup_one_len(idname, mds->mds_pending_dir,
                                 idlen);
         if (IS_ERR(dchild)) {
@@ -718,8 +814,8 @@ static int mds_open_by_id(struct ptlrpc_request *req,
                 RETURN(rc);
         }
 
+        up(&pending_dir->i_sem);
         if (dchild->d_inode != NULL) {
-                up(&pending_dir->i_sem);
                 mds_inode_set_orphan(dchild->d_inode);
                 mds_pack_inode2body(req2obd(req), body,
                                     dchild->d_inode, 1);
@@ -730,8 +826,7 @@ static int mds_open_by_id(struct ptlrpc_request *req,
                        idname);
                 goto open;
         }
-        dput(dchild);
-        up(&pending_dir->i_sem);
+        l_dput(dchild);
 
         /*
          * we didn't find it in PENDING so it isn't an orphan.  See if it was a
@@ -752,7 +847,6 @@ static int mds_open_by_id(struct ptlrpc_request *req,
         rc = mds_finish_transno(mds, dchild ? dchild->d_inode : NULL, handle,
                                 req, rc, rep ? rep->lock_policy_res1 : 0);
         /* XXX what do we do here if mds_finish_transno itself failed? */
-
         l_dput(dchild);
         RETURN(rc);
 }
@@ -789,23 +883,11 @@ int mds_pin(struct ptlrpc_request *req, int offset)
 int mds_lock_new_child(struct obd_device *obd, struct inode *inode,
                        struct lustre_handle *child_lockh)
 {
-        struct ldlm_res_id child_res_id = { .name = { 0, 0, 0, 0 } };
+        struct ldlm_res_id child_res_id = { .name = { inode->i_ino, 0, 1, 0 } };
         struct lustre_handle lockh;
-        struct lustre_id sid;
         int lock_flags = 0;
         int rc;
-
-        down(&inode->i_sem);
-        rc = mds_read_inode_sid(obd, inode, &sid);
-        up(&inode->i_sem);
-        if (rc) {
-                CERROR("Can't read inode self id, err = %d\n", rc);
-                RETURN(rc);
-        }
-
-        child_res_id.name[0] = id_fid(&sid);
-        child_res_id.name[1] = id_group(&sid);
-        child_res_id.name[2] = 1;
+        ENTRY;
 
         if (child_lockh == NULL)
                 child_lockh = &lockh;
@@ -822,31 +904,16 @@ int mds_lock_new_child(struct obd_device *obd, struct inode *inode,
         RETURN(rc);
 }
 
-static int is_mount_object(struct dentry *dparent)
-{
-        struct dentry *dchild;
-
-        if (!(dparent->d_inode->i_mode & S_ISUID))
-                return 0;
-
-        dchild = lookup_one_len(".mntinfo", dparent, strlen(".mntinfo"));
-        if (IS_ERR(dchild) || dchild == NULL)
-                return 0;
-
-        dput(dchild);
-        return 1;
-}
-
 int mds_open(struct mds_update_record *rec, int offset,
              struct ptlrpc_request *req, struct lustre_handle *child_lockh)
 {
         /* XXX ALLOCATE _something_ - 464 bytes on stack here */
         struct obd_device *obd = req->rq_export->exp_obd;
+        struct mds_export_data *med = &req->rq_export->u.eu_mds_data;
         struct mds_obd *mds = mds_req2mds(req);
         struct ldlm_reply *rep = NULL;
         struct mds_body *body = NULL;
         struct dentry *dchild = NULL, *dparent = NULL;
-        struct mds_export_data *med;
         struct lustre_handle parent_lockh[2] = {{0}, {0}};
         int rc = 0, cleanup_phase = 0, acc_mode, created = 0;
         int parent_mode = LCK_PR;
@@ -854,12 +921,21 @@ int mds_open(struct mds_update_record *rec, int offset,
         struct dentry_params dp;
         struct mea *mea = NULL;
         int mea_size, update_mode;
+        int child_mode = LCK_PR;
+        /* Always returning LOOKUP lock if open succesful to guard
+           dentry on client. */
+        ldlm_policy_data_t policy = {.l_inodebits={MDS_INODELOCK_LOOKUP}};
+        struct ldlm_res_id child_res_id = { .name = {0}};
+        int lock_flags = 0;
         ENTRY;
 
         DEBUG_REQ(D_INODE, req, "parent "DLID4" name %*s mode %o",
                   OLID4(rec->ur_id1), rec->ur_namelen - 1, rec->ur_name,
                   rec->ur_mode);
 
+        OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_PAUSE_OPEN | OBD_FAIL_ONCE,
+                         (obd_timeout + 1) / 4);
+
         if (offset == 3) { /* intent */
                 rep = lustre_msg_buf(req->rq_repmsg, 0, sizeof (*rep));
                 body = lustre_msg_buf(req->rq_repmsg, 1, sizeof (*body));
@@ -884,10 +960,12 @@ int mds_open(struct mds_update_record *rec, int offset,
 
                 LASSERT(id_ino(rec->ur_id2));
 
-                rc = mds_open_by_id(req, rec->ur_id2, body, rec->ur_flags,
-                                    rec, rep);
-                if (rc != -ENOENT)
+                rc = mds_open_by_id(req, rec->ur_id2, body,
+                                    rec->ur_flags, rec, rep);
+                if (rc != -ENOENT) {
+                        mds_body_do_reverse_map(med, body);
                         RETURN(rc);
+                }
 
                 /* We didn't find the correct inode on disk either, so we
                  * need to re-create it via a regular replay. */
@@ -898,7 +976,6 @@ int mds_open(struct mds_update_record *rec, int offset,
 
         LASSERT(offset == 3); /* If we got here, we must be called via intent */
 
-        med = &req->rq_export->exp_mds_data;
         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OPEN_PACK)) {
                 CERROR("test case OBD_FAIL_MDS_OPEN_PACK\n");
                 RETURN(-ENOMEM);
@@ -907,7 +984,7 @@ int mds_open(struct mds_update_record *rec, int offset,
         acc_mode = accmode(rec->ur_flags);
 
         /* Step 1: Find and lock the parent */
-        if (rec->ur_flags & O_CREAT) {
+        if (rec->ur_flags & MDS_OPEN_CREAT) {
                 /* XXX Well, in fact we only need this lock mode change if
                    in addition to O_CREAT, the file does not exist.
                    But we do not know if it exists or not yet */
@@ -925,15 +1002,22 @@ int mds_open(struct mds_update_record *rec, int offset,
                 }
                 goto got_child;
         }
-        
+       
+restart:
         dparent = mds_id2locked_dentry(obd, rec->ur_id1, NULL, parent_mode,
                                        parent_lockh, &update_mode, rec->ur_name,
                                        rec->ur_namelen - 1, MDS_INODELOCK_UPDATE);
         if (IS_ERR(dparent)) {
                 rc = PTR_ERR(dparent);
-                if (rc != -ENOENT)
+                if (rc != -ENOENT) {
                         CERROR("parent lookup for "DLID4" failed, error %d\n",
                                OLID4(rec->ur_id1), rc);
+                } else {
+                        /* Just cannot find parent - make it look like
+                           usual negative lookup to avoid extra MDS RPC */
+                        intent_set_disposition(rep, DISP_LOOKUP_EXECD);
+                        intent_set_disposition(rep, DISP_LOOKUP_NEG);
+                }
                 GOTO(cleanup, rc);
         }
         LASSERT(dparent->d_inode != NULL);
@@ -974,33 +1058,55 @@ int mds_open(struct mds_update_record *rec, int offset,
 got_child:
         cleanup_phase = 2; /* child dentry */
 
-        intent_set_disposition(rep, DISP_LOOKUP_EXECD);
-
         if (dchild->d_flags & DCACHE_CROSS_REF) {
-                struct ldlm_res_id res_id = { . name = {0} };
-                ldlm_policy_data_t policy;
-                int flags = 0;
-
-                mds_pack_dentry2body(obd, body, dchild, 1);
-                intent_set_disposition(rep, DISP_LOOKUP_POS);
-
-                CDEBUG(D_OTHER, "cross reference: "DLID4"\n",
-                       OLID4(&body->id1));
-                
-                res_id.name[0] = dchild->d_fid;
-                res_id.name[1] = dchild->d_mdsnum;
+                CDEBUG(D_OTHER, "cross reference: "DLID4"\n", OLID4(rec->ur_id1));
+                LASSERT(rec->ur_namelen > 1);
                 
-                policy.l_inodebits.bits = MDS_INODELOCK_LOOKUP;
-                rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace,
-                                      res_id, LDLM_IBITS, &policy,
-                                      LCK_PR, &flags, mds_blocking_ast,
-                                      ldlm_completion_ast, NULL, NULL,
-                                      NULL, 0, NULL, child_lockh);
+                /* we're gonna acquire LOOKUP lock on the child,
+                 * but we have already locked parent and our order
+                 * may conflict with enqueue_order_locks(). so,
+                 * drop parent lock and acquire both the locks in
+                 * common order. bug 6190 */
 #ifdef S_PDIROPS
                 if (parent_lockh[1].cookie != 0)
                         ldlm_lock_decref(parent_lockh + 1, update_mode);
 #endif
                 ldlm_lock_decref(parent_lockh, parent_mode);
+                l_dput(dchild);
+                l_dput(dparent);
+                cleanup_phase = 0;
+
+                rc = mds_get_parent_child_locked(obd, mds, rec->ur_id1,
+                                                 parent_lockh, &dparent,
+                                                 LCK_PR, MDS_INODELOCK_UPDATE,
+                                                 &update_mode, rec->ur_name,
+                                                 rec->ur_namelen, child_lockh,
+                                                 &dchild, LCK_PR,
+                                                 MDS_INODELOCK_LOOKUP);
+
+                if (rc)
+                        GOTO(cleanup, rc);
+
+#ifdef S_PDIROPS
+                if (parent_lockh[1].cookie != 0)
+                        ldlm_lock_decref(parent_lockh + 1, update_mode);
+#endif
+                ldlm_lock_decref(parent_lockh, LCK_PR);
+
+                if (dchild->d_inode || !(dchild->d_flags & DCACHE_CROSS_REF)) {
+                        CDEBUG(D_OTHER, "race: name changed (%p)\n",
+                               dchild->d_inode);
+                        if (dchild->d_inode)
+                                ldlm_lock_decref(child_lockh, LCK_PR);
+                        l_dput(dchild);
+                        l_dput(dparent);
+                        GOTO(restart, rc);
+                }
+
+                mds_pack_dentry2body(obd, body, dchild, 1);
+                intent_set_disposition(rep, DISP_LOOKUP_POS);
+                intent_set_disposition(rep, DISP_LOOKUP_EXECD);
+
                 if (mea)
                         OBD_FREE(mea, mea_size);
                 l_dput(dchild);
@@ -1008,6 +1114,8 @@ got_child:
                 RETURN(rc);
         }
         
+        intent_set_disposition(rep, DISP_LOOKUP_EXECD);
+
         if (dchild->d_inode)
                 intent_set_disposition(rep, DISP_LOOKUP_POS);
         else
@@ -1101,7 +1209,6 @@ got_child:
                         MD_COUNTER_INCREMENT(obd, create);
                 }
 
-                down(&dchild->d_inode->i_sem);
                 if (ino) {
                         rc = mds_update_inode_sid(obd, dchild->d_inode,
                                                   handle, rec->ur_id2);
@@ -1123,11 +1230,11 @@ got_child:
                                        "rc = %d\n", rc);
                         }
                 }
-                up(&dchild->d_inode->i_sem);
-                
+
                 if (!(rec->ur_flags & O_EXCL)) { /* bug 3313 */
                         rc = fsfilt_commit(obd, dchild->d_inode->i_sb,
-                                           dchild->d_inode, handle, 0);
+                                           dchild->d_inode, handle, 
+                                           req->rq_export->exp_sync);
                         handle = NULL;
                 }
 
@@ -1136,7 +1243,7 @@ got_child:
         mds_pack_inode2body(obd, body, dchild->d_inode, 1);
        
         LASSERTF(!mds_inode_is_orphan(dchild->d_inode),
-                 "dchild %*s (%p) inode %p\n", dchild->d_name.len,
+                 "dchild %.*s (%p) inode %p\n", dchild->d_name.len,
                  dchild->d_name.name, dchild, dchild->d_inode);
 
         if (S_ISREG(dchild->d_inode->i_mode)) {
@@ -1164,46 +1271,89 @@ got_child:
                 GOTO(cleanup, rc = -EEXIST); // returns a lock to the client
         }
 
-        /* if we are following a symlink, don't open */
-        if (S_ISLNK(dchild->d_inode->i_mode))
-                GOTO(cleanup, rc = 0);
-
-        if ((rec->ur_flags & MDS_OPEN_DIRECTORY) &&
-            !S_ISDIR(dchild->d_inode->i_mode))
-                GOTO(cleanup, rc = -ENOTDIR);
-
         if (S_ISDIR(dchild->d_inode->i_mode)) {
                 if (rec->ur_flags & MDS_OPEN_CREAT ||
                     rec->ur_flags & FMODE_WRITE) {
-                        /*we are trying to create or write a exist dir*/
+                        /* we are trying to create or write a exist dir */
                         GOTO(cleanup, rc = -EISDIR);
                 }
                 if (ll_permission(dchild->d_inode, acc_mode, NULL))
                         GOTO(cleanup, rc = -EACCES);
 
-                if (is_mount_object(dchild)) {
-                        CERROR("Found possible GNS mount object %*s; not "
-                               "opening.\n", dchild->d_name.len,
-                               dchild->d_name.name);
-                        GOTO(cleanup, rc = 0); // success, but don't really open
-                }
+                /* 
+                 * here was checking for possible GNS mount points to skip their
+                 * open. I removed it as its detection based only on SUID bit is
+                 * not reliable. Opening such a dirs should not cause any
+                 * problems, at least tests show that. --umka
+                 */
         }
 
+        /* if we are following a symlink, don't open */
+        if (S_ISLNK(dchild->d_inode->i_mode))
+                GOTO(cleanup_no_trans, rc = 0);
+
+        if ((rec->ur_flags & MDS_OPEN_DIRECTORY) &&
+            !S_ISDIR(dchild->d_inode->i_mode))
+                GOTO(cleanup, rc = -ENOTDIR);
+               
+       if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OPEN_CREATE)) {
+               obd_fail_loc = OBD_FAIL_LDLM_REPLY | OBD_FAIL_ONCE;
+               GOTO(cleanup, rc = -EAGAIN);
+       }
+
+#warning "disable opencache lock for CMD2"
+#if 0
+        /* Obtain OPEN lock as well */
+        policy.l_inodebits.bits |= MDS_INODELOCK_OPEN;
+
+        /* We cannot use acc_mode here, because it is zeroed in case of
+           creating a file, so we get wrong lockmode */
+        if (accmode(rec->ur_flags) & MAY_WRITE)
+               child_mode = LCK_CW;
+        else if (accmode(rec->ur_flags) & MAY_EXEC)
+                child_mode = LCK_PR;
+        else
+                child_mode = LCK_CR;
+
+        if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY)) {
+               /* In case of replay we do not get a lock assuming that the
+                   caller has it already */
+                child_res_id.name[0] = id_fid(&body->id1);
+                child_res_id.name[1] = id_group(&body->id1);
+
+                rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace,
+                                      child_res_id, LDLM_IBITS, &policy,
+                                      child_mode, &lock_flags,
+                                      mds_blocking_ast, ldlm_completion_ast,
+                                      NULL, NULL, NULL, 0, NULL, child_lockh);
+                if (rc != ELDLM_OK)
+                        GOTO(cleanup, rc);
+
+                cleanup_phase = 3;
+        }
+#endif
+
         /* Step 5: mds_open it */
         rc = mds_finish_open(req, dchild, body, rec->ur_flags, &handle,
                              rec, rep);
-        GOTO(cleanup, rc);
 
- cleanup:
+       EXIT;
+cleanup:
         rc = mds_finish_transno(mds, dchild ? dchild->d_inode : NULL, handle,
                                 req, rc, rep ? rep->lock_policy_res1 : 0);
 
+cleanup_no_trans:
         switch (cleanup_phase) {
+        case 3:
+                if (rc) {
+                        ldlm_lock_decref(child_lockh, child_mode);
+                        child_lockh->cookie = 0;
+                }
         case 2:
                 if (rc && created) {
                         int err = vfs_unlink(dparent->d_inode, dchild);
                         if (err) {
-                                CERROR("unlink(%*s) in error path: %d\n",
+                                CERROR("unlink(%.*s) in error path: %d\n",
                                        dchild->d_name.len, dchild->d_name.name,
                                        err);
                         }
@@ -1227,6 +1377,21 @@ got_child:
         }
         if (mea)
                 OBD_FREE(mea, mea_size);
+        if (rc == 0)
+                atomic_inc(&mds->mds_open_count);
+
+        mds_body_do_reverse_map(med, body);
+
+        /*
+         * If we have not taken the "open" lock, we may not return 0 here,
+         * because caller expects 0 to mean "lock is taken", and it needs
+         * nonzero return here for caller to return EDLM_LOCK_ABORTED to
+         * client. Later caller should rewrite the return value back to zero
+         * if it to be used any further.
+         */
+        if ((cleanup_phase != 3) && !rc)
+                rc = ENOLCK;
+
         RETURN(rc);
 }
 
@@ -1284,12 +1449,6 @@ int mds_mfd_close(struct ptlrpc_request *req, int offset,
                 int stripe_count = 0;
                 LASSERT(rc == 0); /* mds_put_write_access must have succeeded */
 
-                if (obd->obd_recovering) {
-                        CDEBUG(D_HA, "not remove orphan %s until recovery"
-                               " is over\n", idname);
-                        GOTO(out, rc);
-                }
-
                 CDEBUG(D_HA, "destroying orphan object %s\n", idname);
                 
                 if ((S_ISREG(inode->i_mode) && inode->i_nlink != 1) ||
@@ -1308,9 +1467,20 @@ int mds_mfd_close(struct ptlrpc_request *req, int offset,
                                                idlen);
                 if (IS_ERR(pending_child))
                         GOTO(cleanup, rc = PTR_ERR(pending_child));
-                LASSERT(pending_child->d_inode != NULL);
+                if (pending_child->d_inode == NULL) {
+                        CERROR("orphan %s has been removed\n", idname);
+                        GOTO(cleanup, rc = 0);
+                }
 
                 cleanup_phase = 2; /* dput(pending_child) when finished */
+                if (S_ISDIR(pending_child->d_inode->i_mode)) {
+                        rc = vfs_rmdir(pending_dir, pending_child);
+                        if (rc)
+                                CERROR("error unlinking orphan dir %s: rc %d\n",
+                                       idname, rc);
+                        goto out;
+                }
+
                 if (req != NULL && req->rq_repmsg != NULL) {
                         lmm = lustre_msg_buf(req->rq_repmsg, 1, 0);
                         stripe_count = le32_to_cpu(lmm->lmm_stripe_count);
@@ -1327,10 +1497,7 @@ int mds_mfd_close(struct ptlrpc_request *req, int offset,
                 pending_child->d_fsdata = (void *) &dp;
                 dp.p_inum = 0;
                 dp.p_ptr = req;
-                if (S_ISDIR(pending_child->d_inode->i_mode))
-                        rc = vfs_rmdir(pending_dir, pending_child);
-                else
-                        rc = vfs_unlink(pending_dir, pending_child);
+                rc = vfs_unlink(pending_dir, pending_child);
                 if (rc)
                         CERROR("error unlinking orphan %s: rc %d\n",
                                idname, rc);
@@ -1343,6 +1510,13 @@ int mds_mfd_close(struct ptlrpc_request *req, int offset,
                                       req->rq_repmsg->buflens[2], &lcl) > 0) {
                         reply_body->valid |= OBD_MD_FLCOOKIE;
                 }
+               
+               rc = mds_destroy_object(obd, inode, 1);
+               if (rc) {
+                       CERROR("cannot destroy OSS object on close, err %d\n",
+                              rc);
+                       rc = 0;
+               }
 
                 goto out; /* Don't bother updating attrs on unlinked inode */
         }
@@ -1408,10 +1582,17 @@ out:
         mds_mfd_destroy(mfd);
 
  cleanup:
+        atomic_dec(&mds->mds_open_count);
         if (req != NULL && reply_body != NULL) {
                 rc = mds_finish_transno(mds, pending_dir, handle, req, rc, 0);
         } else if (handle) {
-                int err = fsfilt_commit(obd, mds->mds_sb, pending_dir, handle, 0);
+                int err, force_sync = 0;
+
+                if (req && req->rq_export)
+                        force_sync = req->rq_export->exp_sync;
+
+                err = fsfilt_commit(obd, mds->mds_sb, pending_dir, handle, 
+                                    force_sync);
                 if (err) {
                         CERROR("error committing close: %d\n", err);
                         if (!rc)
@@ -1453,13 +1634,12 @@ int mds_close(struct ptlrpc_request *req, int offset)
         }
 
         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) {
-                DEBUG_REQ(D_HA, req, "close replay\n");
+                DEBUG_REQ(D_HA, req, "close replay");
                 memcpy(lustre_msg_buf(req->rq_repmsg, 2, 0),
                        lustre_msg_buf(req->rq_reqmsg, offset + 1, 0),
                        req->rq_repmsg->buflens[2]);
         }
 
-
         body = lustre_swab_reqbuf(req, offset, sizeof(*body),
                                   lustre_swab_mds_body);
         if (body == NULL) {
@@ -1474,8 +1654,7 @@ int mds_close(struct ptlrpc_request *req, int offset)
         mfd = mds_handle2mfd(&body->handle);
         if (mfd == NULL) {
                 DEBUG_REQ(D_ERROR, req, "no handle for file close ino "LPD64
-                          ": cookie "LPX64, body->id1.li_stc.u.e3s.l3s_ino,
-                          body->handle.cookie);
+                          ": cookie "LPX64, id_ino(&body->id1), body->handle.cookie);
                 req->rq_status = -ESTALE;
                 RETURN(-ESTALE);
         }
@@ -1494,7 +1673,7 @@ int mds_close(struct ptlrpc_request *req, int offset)
                                     (body->valid & OBD_MD_FID) ? 1 : 0);
                 
                 mds_pack_md(obd, req->rq_repmsg, 1, rep_body, 
-                           inode, MDS_PACK_MD_LOCK);
+                           inode, MDS_PACK_MD_LOCK, 0);
         }
         spin_lock(&med->med_open_lock);
         list_del(&mfd->mfd_list);