Branch b1_6

problem:
when a quota request is delayed or dropped, the quota slave who has sent it
will wait for it forever.
solution:
When a quota request time out, dqacq_interpret will hanle it in case the
situation above happened.
b=14840
i=johann
i=andrew.perepechko

diff --git a/lustre/include/obd_support.h b/lustre/include/obd_support.h
index d0c7840..a273555 100644 (file)
--- a/lustre/include/obd_support.h
+++ b/lustre/include/obd_support.h
@@ -147,6 +147,8 @@ extern unsigned int obd_alloc_fail_rate;
 #define OBD_FAIL_MDS_OSC_PRECREATE       0x139
 #define OBD_FAIL_MDS_LLOG_SYNC_TIMEOUT   0x13a
 #define OBD_FAIL_MDS_CLOSE_NET_REP       0x13b
+#define OBD_FAIL_MDS_BLOCK_QUOTA_REQ     0x13c
+#define OBD_FAIL_MDS_DROP_QUOTA_REQ      0x13d
 
 #define OBD_FAIL_OST                     0x200
 #define OBD_FAIL_OST_CONNECT_NET         0x201

diff --git a/lustre/ldlm/ldlm_lib.c b/lustre/ldlm/ldlm_lib.c
index e052c27..f002733 100644 (file)
--- a/lustre/ldlm/ldlm_lib.c
+++ b/lustre/ldlm/ldlm_lib.c
@@ -1621,6 +1621,9 @@ int target_handle_dqacq_callback(struct ptlrpc_request *req)
         int repsize[2] = { sizeof(struct ptlrpc_body), 0 };
         ENTRY;
 
+        if (OBD_FAIL_CHECK_ONCE(OBD_FAIL_MDS_DROP_QUOTA_REQ))
+                RETURN(rc);
+
         repsize[1] = quota_get_qunit_data_size(req->rq_export->
                                                exp_connect_flags);
 
@@ -1662,6 +1665,9 @@ int target_handle_dqacq_callback(struct ptlrpc_request *req)
                 GOTO(out, rc = -EPROTO);
         }
 
+        /* Block the quota req. b=14840 */
+        OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_BLOCK_QUOTA_REQ,OBD_TIMEOUT_DEFAULT);
+
         rc = ptlrpc_reply(req);
 out:
         OBD_FREE(qdata, sizeof(struct qunit_data));

diff --git a/lustre/quota/quota_adjust_qunit.c b/lustre/quota/quota_adjust_qunit.c
index bcab209..3283a75 100644 (file)
--- a/lustre/quota/quota_adjust_qunit.c
+++ b/lustre/quota/quota_adjust_qunit.c
@@ -193,7 +193,7 @@ int quota_adjust_slave_lqs(struct quota_adjust_qunit *oqaq,
         struct lustre_qunit_size *lqs = NULL;
         unsigned long *lbunit, *liunit, *lbtune, *litune;
         signed long b_tmp = 0, i_tmp = 0;
-        static cfs_time_t time_limit = 0;
+        cfs_time_t time_limit = 0;
         int rc = 0;
         ENTRY;
 

diff --git a/lustre/quota/quota_context.c b/lustre/quota/quota_context.c
index 173a8d7..faba6c8 100644 (file)
--- a/lustre/quota/quota_context.c
+++ b/lustre/quota/quota_context.c
@@ -553,6 +553,8 @@ out:
         /* this qunit has been removed by qctxt_cleanup() */
         if (!qunit) {
                 spin_unlock(&qunit_hash_lock);
+                QDATA_DEBUG(qdata, "%s is discarded because qunit isn't found\n",
+                            opc == QUOTA_DQACQ ? "DQACQ" : "DQREL");
                 RETURN(err);
         }
 
@@ -570,6 +572,8 @@ out:
         }
 
         qunit_put(qunit);
+        if (rc < 0 && rc != -EDQUOT)
+                 RETURN(err);
 
         /* don't reschedule in such cases:
          *   - acq/rel failure and qunit isn't changed,
@@ -628,7 +632,15 @@ static int dqacq_interpret(struct ptlrpc_request *req, void *data, int rc)
         OBD_ALLOC(qdata, sizeof(struct qunit_data));
         if (!qdata)
                 RETURN(-ENOMEM);
-        rc1 = quota_get_qdata(req, qdata, QUOTA_REPLY, QUOTA_IMPORT);
+
+        if (rc == -EIO || rc == -EINTR || rc == -ENOTCONN )
+                /* if a quota req timeouts or is dropped, we should update quota
+                 * statistics which will be handled in dqacq_completion. And in
+                 * this situation we should get qdata from request instead of
+                 * reply */
+                rc1 = quota_get_qdata(req, qdata, QUOTA_REQUEST, QUOTA_IMPORT);
+        else
+                rc1 = quota_get_qdata(req, qdata, QUOTA_REPLY, QUOTA_IMPORT);
         if (rc1 < 0) {
                 DEBUG_REQ(D_ERROR, req, "error unpacking qunit_data\n");
                 GOTO(exit, rc = -EPROTO);
@@ -717,8 +729,6 @@ schedule_dqacq(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
         }
         qunit = empty;
         insert_qunit_nolock(qctxt, qunit);
-        if (wait)
-                list_add_tail(&qw.qw_entry, &qunit->lq_waiters);
         spin_unlock(&qunit_hash_lock);
 
         LASSERT(qunit);
@@ -753,8 +763,6 @@ schedule_dqacq(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
                 spin_unlock(&qctxt->lqc_lock);
                 QDATA_DEBUG(qdata, "lqc_import is invalid.\n");
                 spin_lock(&qunit_hash_lock);
-                if (wait)
-                        list_del_init(&qw.qw_entry);
                 remove_qunit_nolock(qunit);
                 qunit = NULL;
                 spin_unlock(&qunit_hash_lock);
@@ -806,6 +814,11 @@ schedule_dqacq(struct obd_device *obd, struct lustre_quota_ctxt *qctxt,
         req->rq_interpret_reply = dqacq_interpret;
         ptlrpcd_add_req(req);
 
+        spin_lock(&qunit_hash_lock);
+        if (wait && qunit)
+                list_add_tail(&qw.qw_entry, &qunit->lq_waiters);
+        spin_unlock(&qunit_hash_lock);
+
         QDATA_DEBUG(qdata, "%s scheduled.\n",
                     opc == QUOTA_DQACQ ? "DQACQ" : "DQREL");
 wait_completion:

diff --git a/lustre/tests/sanity-quota.sh b/lustre/tests/sanity-quota.sh
index d410638..7049eaa 100644 (file)
--- a/lustre/tests/sanity-quota.sh
+++ b/lustre/tests/sanity-quota.sh
@@ -931,6 +931,11 @@ test_12() {
        [ "$(grep $DIR2 /proc/mounts)" ] || mount_client $DIR2 || \
                { skip "Need lustre mounted on $MOUNT2 " && retutn 0; }
 
+       if [ $OSTCOUNT -lt 2 ]; then
+               skip "$OSTCOUNT < 2, too few osts"
+               return 0;
+       fi
+
        LIMIT=$(( $BUNIT_SZ * $(($OSTCOUNT + 1)) * 10)) # 10 bunits each sever
        TESTFILE="$DIR/$tdir/$tfile-0"
        TESTFILE2="$DIR2/$tdir/$tfile-1"
@@ -942,7 +947,7 @@ test_12() {
 
        $LFS setstripe $TESTFILE -i 0 -c 1
        chown $TSTUSR.$TSTUSR $TESTFILE
-       $LFS setstripe $TESTFILE2 -i 0 -c 1
+       $LFS setstripe $TESTFILE2 -i 1 -c 1
         chown $TSTUSR2.$TSTUSR2 $TESTFILE2
 
        #define OBD_FAIL_OST_HOLD_WRITE_RPC      0x21f
@@ -1140,7 +1145,7 @@ test_14a(){
         quota_set_version 1
         $LFS quotacheck -ug $DIR
 
-        for i in `seq 1 30`; do 
+        for i in `seq 1 30`; do
                 $LFS setquota -u quota15_$i $i $i $i $i $DIR || error "lfs setquota failed"
         done
 
@@ -1149,7 +1154,7 @@ test_14a(){
         $LFS quotainv -ug $DIR
         $LFS quotacheck -ug $DIR
 
-        for i in `seq 1 30`; do 
+        for i in `seq 1 30`; do
                 # the format is "mntpnt   curspace[*]   bsoftlimit   bhardlimit   [time]   curinodes[*]    isoftlimit  ihardlimit"
                 ($LFS quota -u quota15_$i $DIR | grep -E '^ *'$DIR' *[0-9]+\** *'$i' *'$i' *[0-9]+\** *'$i' *'$i) \
                  || error "lfs quota output is unexpected"
@@ -1302,6 +1307,116 @@ test_17() {
 }
 run_test 17 "run for fixing bug14526 ==========="
 
+# test when mds takes a long time to handle a quota req so that
+# the ost has dropped it, the ost still could work well b=14840
+test_18() {
+       LIMIT=$((100 * 1024 * 1024)) # 100G
+       TESTFILE="$DIR/$tdir/$tfile"
+
+       wait_delete_completed
+
+       set_blk_tunesz 512
+       set_blk_unitsz 1024
+
+       log "   User quota (limit: $LIMIT kbytes)"
+       $LFS setquota -u $TSTUSR 0 $LIMIT 0 0 $MOUNT
+       $SHOW_QUOTA_USER
+
+       $LFS setstripe $TESTFILE -i 0 -c 1
+       chown $TSTUSR.$TSTUSR $TESTFILE
+
+       #define OBD_FAIL_MDS_BLOCK_QUOTA_REQ      0x13c
+       lustre_fail mds 0x13c
+
+       log "   step1: write 100M block ..."
+       $RUNAS dd if=/dev/zero of=$TESTFILE bs=$BLK_SZ count=$((1024 * 100)) &
+       DDPID=$!
+
+        sleep 5
+        lustre_fail mds 0
+
+       echo  "   step2: testing ......"
+       count=0
+       while [ true ]; do
+           if [ -z `ps -ef | awk '$2 == '${DDPID}' { print $8 }'` ]; then break; fi
+           count=$[count+1]
+           if [ $count -gt 200 ]; then
+               error "dd should be finished!"
+           fi
+           sleep 1
+       done
+        log -n "(dd_pid=$DDPID, time=$count)"
+        if [ $count -lt 90 ]; then
+            error " should take longer!"
+        else
+            echo " successful"
+        fi
+
+       rm -f $TESTFILE
+       sync; sleep 3; sync;
+
+       $LFS setquota -u $TSTUSR 0 0 0 0 $MOUNT         # clear user limit
+
+       set_blk_unitsz $((128 * 1024))
+       set_blk_tunesz $((128 * 1024 / 2))
+}
+run_test 18 "run for fixing bug14840 ==========="
+
+# test when mds drops a quota req, the ost still could work well b=14840
+test_18a() {
+        LIMIT=$((100 * 1024 * 1024)) # 100G
+       TESTFILE="$DIR/$tdir/$tfile-a"
+
+       wait_delete_completed
+
+       set_blk_tunesz 512
+       set_blk_unitsz 1024
+
+       log "   User quota (limit: $LIMIT kbytes)"
+       $LFS setquota -u $TSTUSR 0 $LIMIT 0 0 $MOUNT
+       $SHOW_QUOTA_USER
+
+       $LFS setstripe $TESTFILE -i 0 -c 1
+       chown $TSTUSR.$TSTUSR $TESTFILE
+
+       #define OBD_FAIL_MDS_DROP_QUOTA_REQ | OBD_FAIL_ONCE   0x8000013d
+       lustre_fail mds 0x8000013d
+
+       log "   step1: write 100M block ..."
+       $RUNAS dd if=/dev/zero of=$TESTFILE bs=$BLK_SZ count=$((1024 * 100)) &
+       DDPID=$!
+
+       echo  "   step2: testing ......"
+       count=0
+       while [ true ]; do
+           if [ -z `ps -ef | awk '$2 == '${DDPID}' { print $8 }'` ]; then break; fi
+           count=$[count+1]
+           if [ $count -gt 200 ]; then
+               lustre_fail mds 0
+               error "dd should be finished!"
+           fi
+           sleep 1
+       done
+        log -n "(dd_pid=$DDPID, time=$count)"
+        if [ $count -lt 90 ]; then
+           lustre_fail mds 0
+            error " should take longer!"
+        else
+            echo " successful"
+        fi
+
+        lustre_fail mds 0
+
+       rm -f $TESTFILE
+       sync; sleep 3; sync;
+
+       $LFS setquota -u $TSTUSR 0 0 0 0 $MOUNT         # clear user limit
+
+       set_blk_unitsz $((128 * 1024))
+       set_blk_tunesz $((128 * 1024 / 2))
+}
+run_test 18a "run for fixing bug14840 ==========="
+
 # turn off quota
 test_99()
 {