b=21938 use the same set during replay

[fs/lustre-release.git] / lustre / ptlrpc / client.c

diff --git a/lustre/ptlrpc/client.c b/lustre/ptlrpc/client.c
index eb69133..b04fccb 100644 (file)
--- a/lustre/ptlrpc/client.c
+++ b/lustre/ptlrpc/client.c
@@ -533,6 +533,7 @@ static int __ptlrpc_request_bufs_pack(struct ptlrpc_request *request,
         CFS_INIT_LIST_HEAD(&request->rq_history_list);
         CFS_INIT_LIST_HEAD(&request->rq_exp_list);
         cfs_waitq_init(&request->rq_reply_waitq);
+        cfs_waitq_init(&request->rq_set_waitq);
         request->rq_xid = ptlrpc_next_xid();
         cfs_atomic_set(&request->rq_refcount, 1);
 
@@ -715,6 +716,7 @@ struct ptlrpc_request *ptlrpc_prep_fakereq(struct obd_import *imp,
         CFS_INIT_LIST_HEAD(&request->rq_history_list);
         CFS_INIT_LIST_HEAD(&request->rq_exp_list);
         cfs_waitq_init(&request->rq_reply_waitq);
+        cfs_waitq_init(&request->rq_set_waitq);
 
         request->rq_xid = ptlrpc_next_xid();
         cfs_atomic_set(&request->rq_refcount, 1);
@@ -729,7 +731,7 @@ void ptlrpc_fakereq_finished(struct ptlrpc_request *req)
                 struct ptlrpc_request_set *set = req->rq_set;
 
                 if (set)
-                        set->set_remaining --;
+                        cfs_atomic_dec(&set->set_remaining);
         }
 
         ptlrpc_rqphase_move(req, RQ_PHASE_COMPLETE);
@@ -747,7 +749,7 @@ struct ptlrpc_request_set *ptlrpc_prep_set(void)
                 RETURN(NULL);
         CFS_INIT_LIST_HEAD(&set->set_requests);
         cfs_waitq_init(&set->set_waitq);
-        set->set_remaining = 0;
+        cfs_atomic_set(&set->set_remaining, 0);
         cfs_spin_lock_init(&set->set_new_req_lock);
         CFS_INIT_LIST_HEAD(&set->set_new_requests);
         CFS_INIT_LIST_HEAD(&set->set_cblist);
@@ -765,7 +767,7 @@ void ptlrpc_set_destroy(struct ptlrpc_request_set *set)
         ENTRY;
 
         /* Requests on the set should either all be completed, or all be new */
-        expected_phase = (set->set_remaining == 0) ?
+        expected_phase = (cfs_atomic_read(&set->set_remaining) == 0) ?
                          RQ_PHASE_COMPLETE : RQ_PHASE_NEW;
         cfs_list_for_each (tmp, &set->set_requests) {
                 struct ptlrpc_request *req =
@@ -776,8 +778,9 @@ void ptlrpc_set_destroy(struct ptlrpc_request_set *set)
                 n++;
         }
 
-        LASSERTF(set->set_remaining == 0 || set->set_remaining == n, "%d / %d\n",
-                 set->set_remaining, n);
+        LASSERTF(cfs_atomic_read(&set->set_remaining) == 0 || 
+                 cfs_atomic_read(&set->set_remaining) == n, "%d / %d\n",
+                 cfs_atomic_read(&set->set_remaining), n);
 
         cfs_list_for_each_safe(tmp, next, &set->set_requests) {
                 struct ptlrpc_request *req =
@@ -789,14 +792,14 @@ void ptlrpc_set_destroy(struct ptlrpc_request_set *set)
 
                 if (req->rq_phase == RQ_PHASE_NEW) {
                         ptlrpc_req_interpret(NULL, req, -EBADR);
-                        set->set_remaining--;
+                        cfs_atomic_dec(&set->set_remaining);
                 }
 
                 req->rq_set = NULL;
                 ptlrpc_req_finished (req);
         }
 
-        LASSERT(set->set_remaining == 0);
+        LASSERT(cfs_atomic_read(&set->set_remaining) == 0);
 
         OBD_FREE(set, sizeof(*set));
         EXIT;
@@ -824,7 +827,7 @@ void ptlrpc_set_add_req(struct ptlrpc_request_set *set,
         /* The set takes over the caller's request reference */
         cfs_list_add_tail(&req->rq_set_chain, &set->set_requests);
         req->rq_set = set;
-        set->set_remaining++;
+        cfs_atomic_inc(&set->set_remaining);
 }
 
 /**
@@ -1227,7 +1230,7 @@ int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set)
         int force_timer_recalc = 0;
         ENTRY;
 
-        if (set->set_remaining == 0)
+        if (cfs_atomic_read(&set->set_remaining) == 0)
                 RETURN(1);
 
         cfs_list_for_each(tmp, &set->set_requests) {
@@ -1539,12 +1542,12 @@ int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set)
                 }
                 cfs_spin_unlock(&imp->imp_lock);
 
-                set->set_remaining--;
+                cfs_atomic_dec(&set->set_remaining);
                 cfs_waitq_broadcast(&imp->imp_recovery_waitq);
         }
 
         /* If we hit an error, we want to recover promptly. */
-        RETURN(set->set_remaining == 0 || force_timer_recalc);
+        RETURN(cfs_atomic_read(&set->set_remaining) == 0 || force_timer_recalc);
 }
 
 /* Return 1 if we should give up, else 0 */
@@ -1794,9 +1797,18 @@ int ptlrpc_set_wait(struct ptlrpc_request_set *set)
                  * EINTR.
                  * I don't really care if we go once more round the loop in
                  * the error cases -eeb. */
-        } while (rc != 0 || set->set_remaining != 0);
+                if (rc == 0 && cfs_atomic_read(&set->set_remaining) == 0) {
+                        cfs_list_for_each(tmp, &set->set_requests) {
+                                req = cfs_list_entry(tmp, struct ptlrpc_request,
+                                                     rq_set_chain);
+                                cfs_spin_lock(&req->rq_lock);
+                                req->rq_invalid_rqset = 1;
+                                cfs_spin_unlock(&req->rq_lock);
+                        }
+                }
+        } while (rc != 0 || cfs_atomic_read(&set->set_remaining) != 0);
 
-        LASSERT(set->set_remaining == 0);
+        LASSERT(cfs_atomic_read(&set->set_remaining) == 0);
 
         rc = 0;
         cfs_list_for_each(tmp, &set->set_requests) {