LU-904 ptlrpc: redo io on -EINPROGRESS

[fs/lustre-release.git] / lustre / obdfilter / filter_io_26.c

diff --git a/lustre/obdfilter/filter_io_26.c b/lustre/obdfilter/filter_io_26.c
index 806734b..cafb080 100644 (file)
--- a/lustre/obdfilter/filter_io_26.c
+++ b/lustre/obdfilter/filter_io_26.c
@@ -28,6 +28,8 @@
 /*
  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
+ *
+ * Copyright (c) 2011, 2012, Whamcloud, Inc.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -157,7 +159,7 @@ static int dio_complete_routine(struct bio *bio, unsigned int done, int error)
                 CERROR("***** bio->bi_private is NULL!  This should never "
                        "happen.  Normally, I would crash here, but instead I "
                        "will dump the bio contents to the console.  Please "
-                       "report this to <http://bugzilla.lustre.org/> , along "
+                       "report this to <http://bugs.whamcloud.com/>, along "
                        "with any interesting messages leading up to this point "
                        "(like SCSI errors, perhaps).  Because bi_private is "
                        "NULL, I can't wake up the thread that initiated this "
@@ -481,7 +483,7 @@ int filter_direct_io(int rw, struct dentry *dchild, struct filter_iobuf *iobuf,
         struct inode *inode = dchild->d_inode;
         int blocks_per_page = CFS_PAGE_SIZE >> inode->i_blkbits;
         int rc, rc2, create;
-        cfs_semaphore_t *sem;
+        cfs_mutex_t *mutex;
         ENTRY;
 
         LASSERTF(iobuf->dr_npages <= iobuf->dr_max_pages, "%d,%d\n",
@@ -492,20 +494,25 @@ int filter_direct_io(int rw, struct dentry *dchild, struct filter_iobuf *iobuf,
                 if (iobuf->dr_npages == 0)
                         RETURN(0);
                 create = 0;
-                sem = NULL;
+                mutex = NULL;
         } else {
                 LASSERTF(rw == OBD_BRW_WRITE, "%x\n", rw);
                 LASSERT(iobuf->dr_npages > 0);
                 create = 1;
-                sem = &obd->u.filter.fo_alloc_lock;
+                mutex = &obd->u.filter.fo_alloc_lock;
 
                 lquota_enforce(filter_quota_interface_ref, obd,
                                iobuf->dr_ignore_quota);
         }
 
-        rc = fsfilt_map_inode_pages(obd, inode, iobuf->dr_pages,
+        if (rw == OBD_BRW_WRITE &&
+            OBD_FAIL_CHECK(OBD_FAIL_OST_MAPBLK_ENOSPC)) {
+                rc = -ENOSPC;
+        } else {
+                rc = fsfilt_map_inode_pages(obd, inode, iobuf->dr_pages,
                                     iobuf->dr_npages, iobuf->dr_blocks,
-                                    obdfilter_created_scratchpad, create, sem);
+                                    obdfilter_created_scratchpad, create, mutex);
+        }
 
         if (rw == OBD_BRW_WRITE) {
                 if (rc == 0) {
@@ -520,6 +527,13 @@ int filter_direct_io(int rw, struct dentry *dchild, struct filter_iobuf *iobuf,
 
                 UNLOCK_INODE_MUTEX(inode);
 
+                /* Force commit to make the just-deleted blocks
+                 * reusable. LU-456 */
+                if (rc == -ENOSPC) {
+                        fsfilt_commit(obd, inode, oti->oti_handle, 1);
+                        RETURN(rc);
+                }
+
                 rc2 = filter_finish_transno(exp, inode, oti, 0, 0);
                 if (rc2 != 0) {
                         CERROR("can't close transaction: %d\n", rc2);
@@ -597,6 +611,7 @@ int filter_commitrw_write(struct obd_export *exp, struct obdo *oa,
         unsigned int qcids[MAXQUOTAS] = { oa->o_uid, oa->o_gid };
         int rec_pending[MAXQUOTAS] = { 0, 0 }, quota_pages = 0;
         int sync_journal_commit = obd->u.filter.fo_syncjournal;
+        int retries = 0;
         ENTRY;
 
         LASSERT(oti != NULL);
@@ -678,12 +693,15 @@ int filter_commitrw_write(struct obd_export *exp, struct obdo *oa,
         if (rc == -ENOTCONN)
                 GOTO(cleanup, rc);
 
+        if (OBD_FAIL_CHECK(OBD_FAIL_OST_DQACQ_NET))
+                GOTO(cleanup, rc = -EINPROGRESS);
+
         push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
         cleanup_phase = 2;
 
-        ll_vfs_dq_init(inode);
         fsfilt_check_slow(obd, now, "quota init");
 
+retry:
         LOCK_INODE_MUTEX(inode);
         fsfilt_check_slow(obd, now, "i_mutex");
         oti->oti_handle = fsfilt_brw_start(obd, objcount, &fso, niocount, res,
@@ -700,6 +718,9 @@ int filter_commitrw_write(struct obd_export *exp, struct obdo *oa,
 
         fsfilt_check_slow(obd, now, "brw_start");
 
+        /* Locking order: i_mutex -> journal_lock -> dqptr_sem. LU-952 */
+        ll_vfs_dq_init(inode);
+
         i = OBD_MD_FLATIME | OBD_MD_FLMTIME | OBD_MD_FLCTIME;
 
         /* If the inode still has SUID+SGID bits set (see filter_precreate())
@@ -742,9 +763,16 @@ int filter_commitrw_write(struct obd_export *exp, struct obdo *oa,
         /* filter_direct_io drops i_mutex */
         rc = filter_direct_io(OBD_BRW_WRITE, res->dentry, iobuf, exp, &iattr,
                               oti, sync_journal_commit ? &wait_handle : NULL);
+        if (rc == -ENOSPC && retries++ < 3) {
+                CDEBUG(D_INODE, "retry after force commit, retries:%d\n",
+                       retries);
+                oti->oti_handle = NULL;
+                fsfilt_check_slow(obd, now, "direct_io");
+                goto retry;
+        }
 
-        obdo_from_inode(oa, inode, NULL, rc == 0 ? FILTER_VALID_FLAGS : 0 |
-                                                   OBD_MD_FLUID |OBD_MD_FLGID);
+        obdo_from_inode(oa, inode, (rc == 0 ? FILTER_VALID_FLAGS : 0) |
+                                   OBD_MD_FLUID | OBD_MD_FLGID);
 
         lquota_getflag(filter_quota_interface_ref, obd, oa);
 
@@ -761,10 +789,28 @@ int filter_commitrw_write(struct obd_export *exp, struct obdo *oa,
                         rc = err;
         }
 
-        if (obd->obd_replayable && !rc && wait_handle)
-                LASSERTF(oti->oti_transno <= obd->obd_last_committed,
-                         "oti_transno "LPU64" last_committed "LPU64"\n",
-                         oti->oti_transno, obd->obd_last_committed);
+        /* In rare cases fsfilt_commit_wait() will wake up and return after
+         * the transaction has finished its work and updated j_commit_sequence
+         * but the commit callbacks have not been run yet.  Wait here until
+         * that is finished so that clients requesting sync IO don't see the
+         * reply transno < last_committed.  LU-753 */
+        if (unlikely(obd->obd_replayable && !rc && wait_handle &&
+                     oti->oti_transno > obd->obd_last_committed)) {
+                cfs_waitq_t wq;
+                struct l_wait_info lwi =
+                        LWI_TIMEOUT_INTERVAL(cfs_time_seconds(5),
+                                             (cfs_duration_t)((HZ + 4)/5),
+                                             NULL, NULL);
+                cfs_waitq_init(&wq);
+                l_wait_event(wq,
+                             oti->oti_transno <= obd->obd_last_committed,
+                             &lwi);
+
+                /* commit callback isn't done after waiting for 5 secs ? */
+                if (unlikely(oti->oti_transno > obd->obd_last_committed))
+                        CERROR("transno:"LPU64" > last_committed:"LPU64"\n",
+                               oti->oti_transno, obd->obd_last_committed);
+        }
 
         fsfilt_check_slow(obd, now, "commitrw commit");