Branch b1_6

b=16929
i=johann
i=tom.wang (wangdi)
i=alexey.lyashkov (shadow)

Disable join file on MDS on production releases.

lustre/mds/mds_open.c

@@ -420,7 +420,7 @@ static int mds_create_objects(struct ptlrpc_request *req, int offset,
                         if (rc)
                                 GOTO(out_oa, rc);
                 }
-                rc = obd_create(mds->mds_osc_exp, oinfo.oi_oa, 
+                rc = obd_create(mds->mds_osc_exp, oinfo.oi_oa,
                                 &oinfo.oi_md, &oti);
                 if (rc) {
                         int level = D_ERROR;
@@ -543,7 +543,7 @@ static void reconstruct_open(struct mds_update_record *rec, int offset,
         parent = mds_fid2dentry(mds, rec->ur_fid1, NULL);
         if (IS_ERR(parent)) {
                 rc = PTR_ERR(parent);
-                LCONSOLE_WARN("Parent "LPU64"/%u lookup error %d." 
+                LCONSOLE_WARN("Parent "LPU64"/%u lookup error %d."
                               " Evicting client %s with export %s.\n",
                               rec->ur_fid1->id, rec->ur_fid1->generation, rc,
                               obd_uuid2str(&exp->exp_client_uuid),
@@ -556,7 +556,7 @@ static void reconstruct_open(struct mds_update_record *rec, int offset,
         dchild = ll_lookup_one_len(rec->ur_name, parent, rec->ur_namelen - 1);
         if (IS_ERR(dchild)) {
                 rc = PTR_ERR(dchild);
-                LCONSOLE_WARN("Child "LPU64"/%u lookup error %d." 
+                LCONSOLE_WARN("Child "LPU64"/%u lookup error %d."
                               " Evicting client %s with export %s.\n",
                               rec->ur_fid1->id, rec->ur_fid1->generation, rc,
                               obd_uuid2str(&exp->exp_client_uuid),
@@ -648,9 +648,16 @@ static void reconstruct_open(struct mds_update_record *rec, int offset,
          * to detect a re-open */
         if (mfd == NULL) {
                 if (rec->ur_flags & MDS_OPEN_JOIN_FILE) {
+#if LUSTRE_FIX >= 50
+                        /* Allow file join in beta builds to allow debugging */
                         rc = mds_join_file(rec, req, dchild, NULL);
                         if (rc)
                                 GOTO(out_dput, rc);
+#else
+                        CWARN("file join is not supported in this version of "
+                              "Lustre\n");
+                        GOTO(out_dput, req->rq_status = rc = -EOPNOTSUPP);
+#endif
                 }
                 mntget(mds->mds_vfsmnt);
                 CERROR("Re-opened file \n");
@@ -734,11 +741,18 @@ static int mds_finish_open(struct ptlrpc_request *req, struct dentry *dchild,
                         RETURN(-EEXIST);
                 }
                 if (rec->ur_flags & MDS_OPEN_JOIN_FILE) {
+#if LUSTRE_FIX >= 50
+                        /* Allow file join in beta builds to allow debugging */
                         UNLOCK_INODE_MUTEX(dchild->d_inode);
                         rc = mds_join_file(rec, req, dchild, lockh);
                         if (rc)
                                 RETURN(rc);
                         LOCK_INODE_MUTEX(dchild->d_inode);
+#else
+                        CWARN("file join is not supported in this version of "
+                              "Lustre\n");
+                        RETURN(-EOPNOTSUPP);
+#endif
                 }
                 if (!(body->valid & OBD_MD_FLEASIZE) &&
                     !(body->valid & OBD_MD_FLMODEASIZE)) {
@@ -1024,14 +1038,14 @@ int mds_open(struct mds_update_record *rec, int offset,
         }
 
         /* Step 2: Lookup the child */
-      
+
         if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) &&
             (rec->ur_flags & MDS_OPEN_LOCK) && (rec->ur_namelen == 1)) {
                 /* hack for nfsd with no_subtree_check, it will use anon
                  * dentry w/o filename to open the file. the anon dentry's
                  * parent was set to itself, so rec->ur_fid1 is the file.
                  * And in MDC it cannot derive the dentry's parent dentry,
-                 * hence the file's name, so we hack here in MDS, 
+                 * hence the file's name, so we hack here in MDS,
                  * refer to bug 13030. */
                 dchild = mds_fid2dentry(mds, rec->ur_fid1, NULL);
         } else {
@@ -1211,7 +1225,7 @@ found_child:
         else
                 child_mode = LCK_CR;
 
-        if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) && 
+        if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) &&
              (rec->ur_flags & MDS_OPEN_LOCK)) {
                 /* In case of replay we do not get a lock assuming that the
                    caller has it already */
@@ -1219,8 +1233,8 @@ found_child:
                 child_res_id.name[1] = dchild->d_inode->i_generation;
 
                 rc = ldlm_cli_enqueue_local(obd->obd_namespace, &child_res_id,
-                                            LDLM_IBITS, &policy, child_mode, 
-                                            &lock_flags, ldlm_blocking_ast, 
+                                            LDLM_IBITS, &policy, child_mode,
+                                            &lock_flags, ldlm_blocking_ast,
                                             ldlm_completion_ast, NULL, NULL,
                                             0, NULL, child_lockh);
                 if (rc != ELDLM_OK)
@@ -1342,7 +1356,7 @@ int mds_mfd_close(struct ptlrpc_request *req, int offset,
         } else if (mfd->mfd_mode & MDS_FMODE_EXEC) {
                 mds_allow_write_access(inode);
         }
-        /* here writecount change also needs protection from orphan write sem. 
+        /* here writecount change also needs protection from orphan write sem.
          * so drop orphan write sem after mds_put_write_access, bz 12888. */
         MDS_UP_WRITE_ORPHAN_SEM(inode);
 
@@ -1373,7 +1387,7 @@ int mds_mfd_close(struct ptlrpc_request *req, int offset,
 
                 cleanup_phase = 2; /* dput(pending_child) when finished */
                 if (S_ISDIR(pending_child->d_inode->i_mode)) {
-                        rc = ll_vfs_rmdir(pending_dir, pending_child, 
+                        rc = ll_vfs_rmdir(pending_dir, pending_child,
                                           mds->mds_vfsmnt);
                         if (rc)
                                 CERROR("error unlinking orphan dir %s: rc %d\n",