LU-2099 osd: set dr_elapsed before dr_numreqs

[fs/lustre-release.git] / lustre / osd-ldiskfs / osd_io.c

diff --git a/lustre/osd-ldiskfs/osd_io.c b/lustre/osd-ldiskfs/osd_io.c
index d641331..5626f5f 100644 (file)
--- a/lustre/osd-ldiskfs/osd_io.c
+++ b/lustre/osd-ldiskfs/osd_io.c
@@ -93,8 +93,14 @@ int generic_error_remove_page(struct address_space *mapping, struct page *page)
 }
 #endif
 
-static void osd_init_iobuf(struct osd_device *d, struct osd_iobuf *iobuf,int rw)
+static void __osd_init_iobuf(struct osd_device *d, struct osd_iobuf *iobuf,
+                            int rw, int line)
 {
+       LASSERTF(iobuf->dr_elapsed_valid == 0,
+                "iobuf %p, reqs %d, rw %d, line %d\n", iobuf,
+                cfs_atomic_read(&iobuf->dr_numreqs), iobuf->dr_rw,
+                iobuf->dr_init_at);
+
         cfs_waitq_init(&iobuf->dr_wait);
         cfs_atomic_set(&iobuf->dr_numreqs, 0);
         iobuf->dr_max_pages = PTLRPC_MAX_BRW_PAGES;
@@ -104,9 +110,10 @@ static void osd_init_iobuf(struct osd_device *d, struct osd_iobuf *iobuf,int rw)
         iobuf->dr_frags = 0;
         iobuf->dr_elapsed = 0;
         /* must be counted before, so assert */
-        LASSERT(iobuf->dr_elapsed_valid == 0);
         iobuf->dr_rw = rw;
+       iobuf->dr_init_at = line;
 }
+#define osd_init_iobuf(dev,iobuf,rw) __osd_init_iobuf(dev, iobuf, rw, __LINE__)
 
 static void osd_iobuf_add_page(struct osd_iobuf *iobuf, struct page *page)
 {
@@ -188,11 +195,19 @@ static int dio_complete_routine(struct bio *bio, unsigned int done, int error)
         if (error != 0 && iobuf->dr_error == 0)
                 iobuf->dr_error = error;
 
-        if (cfs_atomic_dec_and_test(&iobuf->dr_numreqs)) {
-                iobuf->dr_elapsed = jiffies - iobuf->dr_start_time;
-                iobuf->dr_elapsed_valid = 1;
-                cfs_waitq_signal(&iobuf->dr_wait);
-        }
+       /*
+        * set dr_elapsed before dr_numreqs turns to 0, otherwise
+        * it's possible that service thread will see dr_numreqs
+        * is zero, but dr_elapsed is not set yet, leading to lost
+        * data in this processing and an assertion in a subsequent
+        * call to OSD.
+        */
+       if (cfs_atomic_read(&iobuf->dr_numreqs) == 1) {
+               iobuf->dr_elapsed = jiffies - iobuf->dr_start_time;
+               iobuf->dr_elapsed_valid = 1;
+       }
+       if (cfs_atomic_dec_and_test(&iobuf->dr_numreqs))
+               cfs_waitq_signal(&iobuf->dr_wait);
 
         /* Completed bios used to be chained off iobuf->dr_bios and freed in
          * filter_clear_dreq().  It was then possible to exhaust the biovec-256
@@ -1115,9 +1130,6 @@ static ssize_t osd_write(const struct lu_env *env, struct dt_object *dt,
        struct inode            *inode = osd_dt_obj(dt)->oo_inode;
        struct osd_thandle      *oh;
        ssize_t                 result;
-#ifdef HAVE_QUOTA_SUPPORT
-       cfs_cap_t               save = cfs_curproc_cap_pack();
-#endif
        int                     is_link;
 
         LASSERT(dt_object_exists(dt));
@@ -1126,18 +1138,14 @@ static ssize_t osd_write(const struct lu_env *env, struct dt_object *dt,
                 return -EACCES;
 
         LASSERT(handle != NULL);
+       LASSERT(inode != NULL);
+       ll_vfs_dq_init(inode);
 
         /* XXX: don't check: one declared chunk can be used many times */
         /* OSD_EXEC_OP(handle, write); */
 
         oh = container_of(handle, struct osd_thandle, ot_super);
         LASSERT(oh->ot_handle->h_transaction != NULL);
-#ifdef HAVE_QUOTA_SUPPORT
-        if (ignore_quota)
-                cfs_cap_raise(CFS_CAP_SYS_RESOURCE);
-        else
-                cfs_cap_lower(CFS_CAP_SYS_RESOURCE);
-#endif
        /* Write small symlink to inode body as we need to maintain correct
         * on-disk symlinks for ldiskfs.
         * Note: the buf->lb_buf contains a NUL terminator while buf->lb_len
@@ -1150,9 +1158,6 @@ static ssize_t osd_write(const struct lu_env *env, struct dt_object *dt,
                result = osd_ldiskfs_write_record(inode, buf->lb_buf,
                                                  buf->lb_len, is_link, pos,
                                                  oh->ot_handle);
-#ifdef HAVE_QUOTA_SUPPORT
-        cfs_curproc_cap_unpack(save);
-#endif
         if (result == 0)
                 result = buf->lb_len;
         return result;
@@ -1205,6 +1210,8 @@ static int osd_punch(const struct lu_env *env, struct dt_object *dt,
         LASSERT(end == OBD_OBJECT_EOF);
         LASSERT(dt_object_exists(dt));
         LASSERT(osd_invariant(obj));
+       LASSERT(inode != NULL);
+       ll_vfs_dq_init(inode);
 
         LASSERT(th);
         oh = container_of(th, struct osd_thandle, ot_super);