not send lmm in replay to client this avoid problems with reply size in case

ost count is different between original processing and replay.
not count ost twice in case ost connected before recovery finished.

Branch b_1_6
b=16736
i=tappro
i=green

lustre/mds/handler.c

@@ -422,6 +422,7 @@ static int mds_destroy_export(struct obd_export *export)
         struct mds_obd *mds = &obd->u.mds;
         struct lvfs_run_ctxt saved;
         struct lov_mds_md *lmm;
+        __u32 lmm_sz, cookie_sz;
         struct llog_cookie *logcookies;
         int rc = 0;
         ENTRY;
@@ -435,18 +436,20 @@ static int mds_destroy_export(struct obd_export *export)
         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
         /* Close any open files (which may also cause orphan unlinking). */
 
-        OBD_ALLOC(lmm, mds->mds_max_mdsize);
+        lmm_sz = mds->mds_max_mdsize;
+        OBD_ALLOC(lmm, lmm_sz);
         if (lmm == NULL) {
                 CWARN("%s: allocation failure during cleanup; can not force "
                       "close file handles on this service.\n", obd->obd_name);
                 GOTO(out, rc = -ENOMEM);
         }
 
-        OBD_ALLOC(logcookies, mds->mds_max_cookiesize);
+        cookie_sz = mds->mds_max_cookiesize;
+        OBD_ALLOC(logcookies, cookie_sz);
         if (logcookies == NULL) {
                 CWARN("%s: allocation failure during cleanup; can not force "
                       "close file handles on this service.\n", obd->obd_name);
-                OBD_FREE(lmm, mds->mds_max_mdsize);
+                OBD_FREE(lmm, lmm_sz);
                 GOTO(out, rc = -ENOMEM);
         }
 
@@ -455,7 +458,7 @@ static int mds_destroy_export(struct obd_export *export)
                 struct list_head *tmp = med->med_open_head.next;
                 struct mds_file_data *mfd =
                         list_entry(tmp, struct mds_file_data, mfd_list);
-                int lmm_size = mds->mds_max_mdsize;
+                int lmm_size = lmm_sz;
                 umode_t mode = mfd->mfd_dentry->d_inode->i_mode;
                 __u64 valid = 0;
 
@@ -505,8 +508,8 @@ static int mds_destroy_export(struct obd_export *export)
                 spin_lock(&med->med_open_lock);
         }
 
-        OBD_FREE(logcookies, mds->mds_max_cookiesize);
-        OBD_FREE(lmm, mds->mds_max_mdsize);
+        OBD_FREE(logcookies, cookie_sz);
+        OBD_FREE(lmm, lmm_sz);
 
         spin_unlock(&med->med_open_lock);
 
@@ -632,7 +635,10 @@ int mds_pack_md(struct obd_device *obd, struct lustre_msg *msg, int offset,
                        inode->i_ino);
                 RETURN(0);
         }
+        /* if this replay request we should be silencely exist without fill md*/
         lmm_size = lustre_msg_buflen(msg, offset);
+        if (lmm_size == 0)
+                RETURN(0);
 
         /* I don't really like this, but it is a sanity check on the client
          * MD request.  However, if the client doesn't know how much space
@@ -2443,6 +2449,13 @@ static int mds_intent_policy(struct ldlm_namespace *ns,
         else if (it->opc & IT_UNLINK)
                 repsize[repbufcnt++] = mds->mds_max_cookiesize;
 
+        /* if we do recovery we isn't send reply mds state is restored */
+        if(lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) {
+                repsize[DLM_REPLY_REC_OFF+1] = 0;
+                if (it->opc & IT_UNLINK)
+                        repsize[DLM_REPLY_REC_OFF+2] = 0;
+        }
+
         rc = lustre_pack_reply(req, repbufcnt, repsize, NULL);
         if (rc)
                 RETURN(req->rq_status = rc);
@@ -2450,7 +2463,6 @@ static int mds_intent_policy(struct ldlm_namespace *ns,
         rep = lustre_msg_buf(req->rq_repmsg, DLM_LOCKREPLY_OFF, sizeof(*rep));
         intent_set_disposition(rep, DISP_IT_EXECD);
 
-
         /* execute policy */
         switch ((long)it->opc) {
         case IT_OPEN:

lustre/mds/mds_lov.c

@@ -202,6 +202,7 @@ static int mds_lov_update_max_ost(struct mds_obd *mds, obd_id index)
         if (data[off] == 0) {
                 __u32 stripes;
 
+                data[off] = 1;
                 mds->mds_lov_objid_count++;
                 stripes = min_t(__u32, LOV_MAX_STRIPE_COUNT,
                                 mds->mds_lov_objid_count);
@@ -503,7 +504,10 @@ static int mds_lov_update_desc(struct obd_device *obd, struct obd_export *lov,
         CDEBUG(D_CONFIG, "updated lov_desc, tgt_count: %d\n",
                mds->mds_lov_desc.ld_tgt_count);
 
-        if (mds_lov_update_max_ost(mds, index))
+        mutex_down(&obd->obd_dev_sem);
+        rc = mds_lov_update_max_ost(mds, index);
+        mutex_up(&obd->obd_dev_sem);
+        if (rc)
                 GOTO(out, rc = -ENOMEM);
 
         /* If we added a target we have to reconnect the llogs */
@@ -532,11 +536,8 @@ static int mds_lov_update_mds(struct obd_device *obd,
         ENTRY;
 
         /* Don't let anyone else mess with mds_lov_objids now */
-        mutex_down(&obd->obd_dev_sem);
-
         old_count = mds->mds_lov_desc.ld_tgt_count;
         rc = mds_lov_update_desc(obd, mds->mds_osc_exp, idx);
-        mutex_up(&obd->obd_dev_sem);
         if (rc)
                 GOTO(out, rc);
 
@@ -1084,10 +1085,7 @@ int mds_notify(struct obd_device *obd, struct obd_device *watched,
                    mds_lov_connect. */
                 idx = mds_lov_get_idx(obd->u.mds.mds_osc_exp,
                                       &watched->u.cli.cl_target_uuid);
-                mutex_down(&obd->obd_dev_sem);
                 rc = mds_lov_update_desc(obd, obd->u.mds.mds_osc_exp, idx);
-                mutex_up(&obd->obd_dev_sem);
-
                 mds_allow_cli(obd, CONFIG_SYNC);
                 RETURN(rc);
         }

lustre/mds/mds_open.c

@@ -371,11 +371,11 @@ static int mds_create_objects(struct ptlrpc_request *req, int offset,
                 rc = fsfilt_set_md(obd, inode, *handle, lmm, lmm_size, "lov");
                 if (rc)
                         CERROR("open replay failed to set md:%d\n", rc);
-                lmm_buf = lustre_msg_buf(req->rq_repmsg, offset, lmm_size);
-                LASSERT(lmm_buf);
-                memcpy(lmm_buf, lmm, lmm_size);
 
-                *objid = lmm_buf;
+                /* for replay we not need send lmm to client, this not used now */
+                lustre_shrink_reply(req, offset, 0, 1);
+                *objid = lmm;
+
                 RETURN(rc);
         }
 
@@ -395,7 +395,6 @@ static int mds_create_objects(struct ptlrpc_request *req, int offset,
 
         obdo_from_inode(oinfo.oi_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 */
                 if (rec->ur_flags & MDS_OPEN_HAS_EA) {
@@ -405,18 +404,19 @@ static int mds_create_objects(struct ptlrpc_request *req, int offset,
                         if (rc)
                                 GOTO(out_oa, rc);
                 } else {
-                        OBD_ALLOC(lmm, mds->mds_max_mdsize);
+                        __u32 lmm_sz = mds->mds_max_mdsize;
+                        OBD_ALLOC(lmm, lmm_sz);
                         if (lmm == NULL)
                                 GOTO(out_oa, rc = -ENOMEM);
 
-                        lmm_size = mds->mds_max_mdsize;
+                        lmm_size = lmm_sz;
                         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_osc_exp,
                                                    0, &oinfo.oi_md, lmm);
-                        OBD_FREE(lmm, mds->mds_max_mdsize);
+                        OBD_FREE(lmm, lmm_sz);
                         if (rc)
                                 GOTO(out_oa, rc);
                 }