b=7178

 - let's try to catch the case when late MDS_CLOSE updates attrs. from
   out-of-date i/o epoch

diff --git a/lustre/llite/file.c b/lustre/llite/file.c
index 8b3e32c..3ca0446 100644 (file)
--- a/lustre/llite/file.c
+++ b/lustre/llite/file.c
@@ -74,7 +74,8 @@ finish:
 }
 
 int ll_md_och_close(struct obd_export *md_exp, struct inode *inode,
-                    struct obd_client_handle *och, int dirty)
+                    struct obd_client_handle *och, int dirty,
+                    __u64 epoch)
 {
         struct ptlrpc_request *req = NULL;
         struct mdc_op_data *op_data;
@@ -116,8 +117,9 @@ int ll_md_och_close(struct obd_export *md_exp, struct inode *inode,
 
         if (dirty) {
                 /* we modified data through this handle */
+                op_data->io_epoch = epoch;
                 op_data->flags |= MDS_BFLAG_DIRTY_EPOCH;
-                op_data->valid |= OBD_MD_FLFLAGS;
+                op_data->valid |= OBD_MD_FLFLAGS | OBD_MD_FLEPOCH;
                 if (ll_validate_size(inode, &op_data->size, &op_data->blocks))
                         op_data->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
         }
@@ -154,7 +156,7 @@ int ll_md_real_close(struct obd_export *md_exp,
         int freeing = inode->i_state & I_FREEING;
         struct obd_client_handle **och_p;
         struct obd_client_handle *och;
-        __u64 *och_usecount;
+        __u64 *och_usecount, epoch;
         int rc = 0, dirty = 0;
         ENTRY;
 
@@ -199,6 +201,8 @@ int ll_md_real_close(struct obd_export *md_exp,
 
         och = *och_p;
         *och_p = NULL;
+        epoch = lli->lli_io_epoch;
+        lli->lli_io_epoch = 0;
 
         up(&lli->lli_och_sem);
 
@@ -209,7 +213,7 @@ int ll_md_real_close(struct obd_export *md_exp,
          * and this will be called from block_ast callack.
         */
         if (och && och->och_fh.cookie != DEAD_HANDLE_MAGIC)
-                rc = ll_md_och_close(md_exp, inode, och, dirty);
+                rc = ll_md_och_close(md_exp, inode, och, dirty, epoch);
         
         RETURN(rc);
 }
@@ -352,6 +356,9 @@ int ll_och_fill(struct inode *inode, struct lookup_intent *it,
 
         memcpy(&och->och_fh, &body->handle, sizeof(body->handle));
         och->och_magic = OBD_CLIENT_HANDLE_MAGIC;
+        if (lli->lli_io_epoch != 0 && lli->lli_io_epoch != body->io_epoch)
+                CDEBUG(D_ERROR, "are we opening new epoch?! "LPD64
+                       " != "LPD64"\n", lli->lli_io_epoch, body->io_epoch);
         lli->lli_io_epoch = body->io_epoch;
         mdc_set_open_replay_data(ll_i2mdexp(inode), och, 
                                 LUSTRE_IT(it)->it_data);
@@ -494,7 +501,7 @@ int ll_file_open(struct inode *inode, struct file *file)
                         }
                         
                         /* ll_md_och_close() will free och */
-                        ll_md_och_close(ll_i2mdexp(inode), inode, och, 0);
+                        ll_md_och_close(ll_i2mdexp(inode), inode, och, 0, 0);
                 }
                 (*och_usecount)++;
                         
@@ -1348,7 +1355,7 @@ static int ll_lov_setstripe_ea_info(struct inode *inode, struct file *file,
         rc = ll_file_release(f->f_dentry->d_inode, f);
         
         /* Now also destroy our supplemental och */
-        ll_md_och_close(ll_i2mdexp(inode), f->f_dentry->d_inode, och, 0);
+        ll_md_och_close(ll_i2mdexp(inode), f->f_dentry->d_inode, och, 0, 0);
         EXIT;
 out:
         ll_intent_release(&oit);

diff --git a/lustre/llite/llite_internal.h b/lustre/llite/llite_internal.h
index dcf501b..9ed882d 100644 (file)
--- a/lustre/llite/llite_internal.h
+++ b/lustre/llite/llite_internal.h
@@ -340,7 +340,7 @@ int ll_local_open(struct file *file, struct lookup_intent *it,
 int ll_md_close(struct obd_export *md_exp, struct inode *inode,
                 struct file *file);
 int ll_md_och_close(struct obd_export *md_exp, struct inode *inode,
-                    struct obd_client_handle *och, int dirty);
+                    struct obd_client_handle *och, int dirty, __u64 epoch);
 int ll_och_fill(struct inode *inode, struct lookup_intent *it,
                 struct obd_client_handle *och);
 int ll_set_audit(struct inode *, __u64);
@@ -624,11 +624,6 @@ ll_inode2mdc_data(struct mdc_op_data *op_data,
                 op_data->ctime = LTIME_S(inode->i_ctime);
                 newvalid |= OBD_MD_FLCTIME;
         }
-        if (valid & OBD_MD_FLEPOCH) {
-                op_data->io_epoch = ll_i2info(inode)->lli_io_epoch;
-                newvalid |= OBD_MD_FLEPOCH;
-        }
-
         if (valid & OBD_MD_FLTYPE) {
                 op_data->mode = (op_data->mode & S_IALLUGO) |
                         (inode->i_mode & S_IFMT);

diff --git a/lustre/llite/llite_lib.c b/lustre/llite/llite_lib.c
index e2267ff..99903ff 100644 (file)
--- a/lustre/llite/llite_lib.c
+++ b/lustre/llite/llite_lib.c
@@ -583,6 +583,7 @@ void ll_lli_init(struct ll_inode_info *lli)
         lli->lli_audit_mask = AUDIT_OFF;
         lli->lli_key_info = NULL;
         init_waitqueue_head(&lli->lli_dirty_wait);
+        lli->lli_io_epoch = 0;
 }
 
 int ll_fill_super(struct super_block *sb, void *data, int silent)

diff --git a/lustre/llite/special.c b/lustre/llite/special.c
index dfdc2cf..46f2ab9 100644 (file)
--- a/lustre/llite/special.c
+++ b/lustre/llite/special.c
@@ -343,7 +343,7 @@ static int ll_special_open(struct inode *inode, struct file *filp)
                         }
                         ll_och_fill(inode, it, och);
                         /* ll_md_och_close() will free och */
-                        ll_md_och_close(ll_i2mdexp(inode), inode, och, 0);
+                        ll_md_och_close(ll_i2mdexp(inode), inode, och, 0, 0);
                 }       
                 (*och_usecount)++;        
 

diff --git a/lustre/mds/mds_open.c b/lustre/mds/mds_open.c
index a3ab2bb..31b7beb 100644 (file)
--- a/lustre/mds/mds_open.c
+++ b/lustre/mds/mds_open.c
@@ -1548,6 +1548,13 @@ int mds_mfd_close(struct ptlrpc_request *req, int offset,
                  * them until all write references are dropped.
                  * btw, we hold one reference */
                 LASSERT(mfd->mfd_mode & FMODE_WRITE);
+                LASSERT(request_body->valid & OBD_MD_FLEPOCH);
+                LASSERT(MDS_FILTERDATA(inode));
+                if (MDS_FILTERDATA(inode)->io_epoch != request_body->io_epoch)
+                        CDEBUG(D_ERROR, "try to update attr. for old epoch "
+                               LPD64" while current "LPD64"\n",
+                               MDS_FILTERDATA(inode)->io_epoch,
+                               request_body->io_epoch);
                 i_size_write(inode, request_body->size);
                 inode->i_blocks = request_body->blocks;
                 iattr.ia_size = inode->i_size;