X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fptlrpc%2Fclient.c;h=1f9452d8a9545c97d74d4410fa6b260da2094aa3;hp=db3db5302ee8250b5e9c557d4c76a81ff0d194ba;hb=cd8dc58c76b43c5dfa9ec238cc3f4264e9f5f89f;hpb=3f92a017ae2d97506db89b34c374bbcf9670db54

diff --git a/lustre/ptlrpc/client.c b/lustre/ptlrpc/client.c
index db3db53..1f9452d 100644
--- a/lustre/ptlrpc/client.c
+++ b/lustre/ptlrpc/client.c
@@ -37,11 +37,6 @@
 /** Implementation of client-side PortalRPC interfaces */
 
 #define DEBUG_SUBSYSTEM S_RPC
-#ifndef __KERNEL__
-#include <errno.h>
-#include <signal.h>
-#include <liblustre.h>
-#endif
 
 #include <obd_support.h>
 #include <obd_class.h>
@@ -53,6 +48,7 @@
 #include "ptlrpc_internal.h"
 
 static int ptlrpc_send_new_req(struct ptlrpc_request *req);
+static int ptlrpcd_check_work(struct ptlrpc_request *req);
 
 /**
  * Initialize passed in client structure \a cl.
@@ -112,7 +108,7 @@ struct ptlrpc_bulk_desc *ptlrpc_new_bulk(unsigned npages, unsigned max_brw,
 		return NULL;
 
 	spin_lock_init(&desc->bd_lock);
-	cfs_waitq_init(&desc->bd_waitq);
+	init_waitqueue_head(&desc->bd_waitq);
 	desc->bd_max_iov = npages;
 	desc->bd_iov_count = 0;
 	desc->bd_portal = portal;
@@ -285,8 +281,8 @@ int ptlrpc_at_get_net_latency(struct ptlrpc_request *req)
 }
 
 /* Adjust expected network latency */
-static void ptlrpc_at_adj_net_latency(struct ptlrpc_request *req,
-                                      unsigned int service_time)
+void ptlrpc_at_adj_net_latency(struct ptlrpc_request *req,
+			       unsigned int service_time)
 {
         unsigned int nl, oldnl;
         struct imp_at *at;
@@ -337,6 +333,7 @@ static int unpack_reply(struct ptlrpc_request *req)
  * If anything goes wrong just ignore it - same as if it never happened
  */
 static int ptlrpc_at_recv_early_reply(struct ptlrpc_request *req)
+__must_hold(&req->rq_lock)
 {
         struct ptlrpc_request *early_req;
         time_t                 olddl;
@@ -373,9 +370,11 @@ static int ptlrpc_at_recv_early_reply(struct ptlrpc_request *req)
 
 	spin_lock(&req->rq_lock);
 	olddl = req->rq_deadline;
-	/* server assumes it now has rq_timeout from when it sent the
-	 * early reply, so client should give it at least that long. */
-	req->rq_deadline = cfs_time_current_sec() + req->rq_timeout +
+	/* server assumes it now has rq_timeout from when the request
+	 * arrived, so the client should give it at least that long.
+	 * since we don't know the arrival time we'll use the original
+	 * sent time */
+	req->rq_deadline = req->rq_sent + req->rq_timeout +
 			   ptlrpc_at_get_net_latency(req);
 
 	DEBUG_REQ(D_ADAPTTO, req,
@@ -389,7 +388,7 @@ static int ptlrpc_at_recv_early_reply(struct ptlrpc_request *req)
 
 struct kmem_cache *request_cache;
 
-int ptlrpc_request_cache_init()
+int ptlrpc_request_cache_init(void)
 {
 	request_cache = kmem_cache_create("ptlrpc_cache",
 					  sizeof(struct ptlrpc_request),
@@ -397,12 +396,12 @@ int ptlrpc_request_cache_init()
 	return request_cache == NULL ? -ENOMEM : 0;
 }
 
-void ptlrpc_request_cache_fini()
+void ptlrpc_request_cache_fini(void)
 {
 	kmem_cache_destroy(request_cache);
 }
 
-struct ptlrpc_request *ptlrpc_request_cache_alloc(int flags)
+struct ptlrpc_request *ptlrpc_request_cache_alloc(gfp_t flags)
 {
 	struct ptlrpc_request *req;
 
@@ -421,15 +420,15 @@ void ptlrpc_request_cache_free(struct ptlrpc_request *req)
  */
 void ptlrpc_free_rq_pool(struct ptlrpc_request_pool *pool)
 {
-	cfs_list_t *l, *tmp;
+	struct list_head *l, *tmp;
 	struct ptlrpc_request *req;
 
 	LASSERT(pool != NULL);
 
 	spin_lock(&pool->prp_lock);
-	cfs_list_for_each_safe(l, tmp, &pool->prp_req_list) {
-		req = cfs_list_entry(l, struct ptlrpc_request, rq_list);
-		cfs_list_del(&req->rq_list);
+	list_for_each_safe(l, tmp, &pool->prp_req_list) {
+		req = list_entry(l, struct ptlrpc_request, rq_list);
+		list_del(&req->rq_list);
 		LASSERT(req->rq_reqbuf);
 		LASSERT(req->rq_reqbuf_len == pool->prp_rq_size);
 		OBD_FREE_LARGE(req->rq_reqbuf, pool->prp_rq_size);
@@ -451,7 +450,7 @@ void ptlrpc_add_rqs_to_pool(struct ptlrpc_request_pool *pool, int num_rq)
         while (size < pool->prp_rq_size)
                 size <<= 1;
 
-        LASSERTF(cfs_list_empty(&pool->prp_req_list) ||
+	LASSERTF(list_empty(&pool->prp_req_list) ||
                  size == pool->prp_rq_size,
                  "Trying to change pool size with nonempty pool "
                  "from %d to %d bytes\n", pool->prp_rq_size, size);
@@ -463,7 +462,7 @@ void ptlrpc_add_rqs_to_pool(struct ptlrpc_request_pool *pool, int num_rq)
 		struct lustre_msg *msg;
 
 		spin_unlock(&pool->prp_lock);
-		req = ptlrpc_request_cache_alloc(__GFP_IO);
+		req = ptlrpc_request_cache_alloc(GFP_NOFS);
 		if (!req)
 			return;
 		OBD_ALLOC_LARGE(msg, size);
@@ -475,7 +474,7 @@ void ptlrpc_add_rqs_to_pool(struct ptlrpc_request_pool *pool, int num_rq)
                 req->rq_reqbuf_len = size;
                 req->rq_pool = pool;
 		spin_lock(&pool->prp_lock);
-		cfs_list_add_tail(&req->rq_list, &pool->prp_req_list);
+		list_add_tail(&req->rq_list, &pool->prp_req_list);
 	}
 	spin_unlock(&pool->prp_lock);
 	return;
@@ -492,30 +491,30 @@ EXPORT_SYMBOL(ptlrpc_add_rqs_to_pool);
  */
 struct ptlrpc_request_pool *
 ptlrpc_init_rq_pool(int num_rq, int msgsize,
-                    void (*populate_pool)(struct ptlrpc_request_pool *, int))
+		    void (*populate_pool)(struct ptlrpc_request_pool *, int))
 {
-        struct ptlrpc_request_pool *pool;
+	struct ptlrpc_request_pool *pool;
 
-        OBD_ALLOC(pool, sizeof (struct ptlrpc_request_pool));
-        if (!pool)
-                return NULL;
+	OBD_ALLOC(pool, sizeof(struct ptlrpc_request_pool));
+	if (!pool)
+		return NULL;
 
-        /* Request next power of two for the allocation, because internally
-           kernel would do exactly this */
+	/* Request next power of two for the allocation, because internally
+	   kernel would do exactly this */
 
 	spin_lock_init(&pool->prp_lock);
-        CFS_INIT_LIST_HEAD(&pool->prp_req_list);
-        pool->prp_rq_size = msgsize + SPTLRPC_MAX_PAYLOAD;
-        pool->prp_populate = populate_pool;
+	INIT_LIST_HEAD(&pool->prp_req_list);
+	pool->prp_rq_size = msgsize + SPTLRPC_MAX_PAYLOAD;
+	pool->prp_populate = populate_pool;
 
-        populate_pool(pool, num_rq);
+	populate_pool(pool, num_rq);
 
-        if (cfs_list_empty(&pool->prp_req_list)) {
-                /* have not allocated a single request for the pool */
-                OBD_FREE(pool, sizeof (struct ptlrpc_request_pool));
-                pool = NULL;
-        }
-        return pool;
+	if (list_empty(&pool->prp_req_list)) {
+		/* have not allocated a single request for the pool */
+		OBD_FREE(pool, sizeof(struct ptlrpc_request_pool));
+		pool = NULL;
+	}
+	return pool;
 }
 EXPORT_SYMBOL(ptlrpc_init_rq_pool);
 
@@ -537,14 +536,14 @@ ptlrpc_prep_req_from_pool(struct ptlrpc_request_pool *pool)
 	 * in writeout path, where this matters, this is safe to do, because
 	 * nothing is lost in this case, and when some in-flight requests
 	 * complete, this code will be called again. */
-	if (unlikely(cfs_list_empty(&pool->prp_req_list))) {
+	if (unlikely(list_empty(&pool->prp_req_list))) {
 		spin_unlock(&pool->prp_lock);
 		return NULL;
 	}
 
-	request = cfs_list_entry(pool->prp_req_list.next, struct ptlrpc_request,
-				 rq_list);
-	cfs_list_del_init(&request->rq_list);
+	request = list_entry(pool->prp_req_list.next, struct ptlrpc_request,
+			     rq_list);
+	list_del_init(&request->rq_list);
 	spin_unlock(&pool->prp_lock);
 
         LASSERT(request->rq_reqbuf);
@@ -567,9 +566,9 @@ static void __ptlrpc_free_req_to_pool(struct ptlrpc_request *request)
 	struct ptlrpc_request_pool *pool = request->rq_pool;
 
 	spin_lock(&pool->prp_lock);
-	LASSERT(cfs_list_empty(&request->rq_list));
+	LASSERT(list_empty(&request->rq_list));
 	LASSERT(!request->rq_receiving_reply);
-	cfs_list_add_tail(&request->rq_list, &pool->prp_req_list);
+	list_add_tail(&request->rq_list, &pool->prp_req_list);
 	spin_unlock(&pool->prp_lock);
 }
 
@@ -620,26 +619,26 @@ static int __ptlrpc_request_bufs_pack(struct ptlrpc_request *request,
         ptlrpc_at_set_req_timeout(request);
 
 	spin_lock_init(&request->rq_lock);
-        CFS_INIT_LIST_HEAD(&request->rq_list);
-        CFS_INIT_LIST_HEAD(&request->rq_timed_list);
-        CFS_INIT_LIST_HEAD(&request->rq_replay_list);
-        CFS_INIT_LIST_HEAD(&request->rq_ctx_chain);
-        CFS_INIT_LIST_HEAD(&request->rq_set_chain);
-        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);
-
-        lustre_msg_set_opc(request->rq_reqmsg, opcode);
-
-        RETURN(0);
+	INIT_LIST_HEAD(&request->rq_list);
+	INIT_LIST_HEAD(&request->rq_timed_list);
+	INIT_LIST_HEAD(&request->rq_replay_list);
+	INIT_LIST_HEAD(&request->rq_ctx_chain);
+	INIT_LIST_HEAD(&request->rq_set_chain);
+	INIT_LIST_HEAD(&request->rq_history_list);
+	INIT_LIST_HEAD(&request->rq_exp_list);
+	init_waitqueue_head(&request->rq_reply_waitq);
+	init_waitqueue_head(&request->rq_set_waitq);
+	request->rq_xid = ptlrpc_next_xid();
+	atomic_set(&request->rq_refcount, 1);
+
+	lustre_msg_set_opc(request->rq_reqmsg, opcode);
+
+	RETURN(0);
 out_ctx:
-        sptlrpc_cli_ctx_put(request->rq_cli_ctx, 1);
+	sptlrpc_cli_ctx_put(request->rq_cli_ctx, 1);
 out_free:
-        class_import_put(imp);
-        return rc;
+	class_import_put(imp);
+	return rc;
 }
 
 int ptlrpc_request_bufs_pack(struct ptlrpc_request *request,
@@ -697,7 +696,7 @@ EXPORT_SYMBOL(ptlrpc_request_pack);
  */
 static inline
 struct ptlrpc_request *__ptlrpc_request_alloc(struct obd_import *imp,
-                                              struct ptlrpc_request_pool *pool)
+					      struct ptlrpc_request_pool *pool)
 {
 	struct ptlrpc_request *request = NULL;
 
@@ -705,21 +704,21 @@ struct ptlrpc_request *__ptlrpc_request_alloc(struct obd_import *imp,
 		request = ptlrpc_prep_req_from_pool(pool);
 
 	if (!request)
-		request = ptlrpc_request_cache_alloc(__GFP_IO);
+		request = ptlrpc_request_cache_alloc(GFP_NOFS);
 
-        if (request) {
-                LASSERTF((unsigned long)imp > 0x1000, "%p", imp);
-                LASSERT(imp != LP_POISON);
-                LASSERTF((unsigned long)imp->imp_client > 0x1000, "%p",
-                        imp->imp_client);
-                LASSERT(imp->imp_client != LP_POISON);
+	if (request) {
+		LASSERTF((unsigned long)imp > 0x1000, "%p\n", imp);
+		LASSERT(imp != LP_POISON);
+		LASSERTF((unsigned long)imp->imp_client > 0x1000, "%p\n",
+			imp->imp_client);
+		LASSERT(imp->imp_client != LP_POISON);
 
-                request->rq_import = class_import_get(imp);
-        } else {
-                CERROR("request allocation out of memory\n");
-        }
+		request->rq_import = class_import_get(imp);
+	} else {
+		CERROR("request allocation out of memory\n");
+	}
 
-        return request;
+	return request;
 }
 
 /**
@@ -859,14 +858,14 @@ struct ptlrpc_request_set *ptlrpc_prep_set(void)
 	OBD_ALLOC(set, sizeof *set);
 	if (!set)
 		RETURN(NULL);
-	cfs_atomic_set(&set->set_refcount, 1);
-	CFS_INIT_LIST_HEAD(&set->set_requests);
-	cfs_waitq_init(&set->set_waitq);
-	cfs_atomic_set(&set->set_new_count, 0);
-	cfs_atomic_set(&set->set_remaining, 0);
+	atomic_set(&set->set_refcount, 1);
+	INIT_LIST_HEAD(&set->set_requests);
+	init_waitqueue_head(&set->set_waitq);
+	atomic_set(&set->set_new_count, 0);
+	atomic_set(&set->set_remaining, 0);
 	spin_lock_init(&set->set_new_req_lock);
-	CFS_INIT_LIST_HEAD(&set->set_new_requests);
-	CFS_INIT_LIST_HEAD(&set->set_cblist);
+	INIT_LIST_HEAD(&set->set_new_requests);
+	INIT_LIST_HEAD(&set->set_cblist);
 	set->set_max_inflight = UINT_MAX;
 	set->set_producer     = NULL;
 	set->set_producer_arg = NULL;
@@ -912,40 +911,40 @@ EXPORT_SYMBOL(ptlrpc_prep_fcset);
  */
 void ptlrpc_set_destroy(struct ptlrpc_request_set *set)
 {
-        cfs_list_t       *tmp;
-        cfs_list_t       *next;
-        int               expected_phase;
-        int               n = 0;
-        ENTRY;
+	struct list_head	*tmp;
+	struct list_head	*next;
+	int			 expected_phase;
+	int			 n = 0;
+	ENTRY;
 
-        /* Requests on the set should either all be completed, or all be new */
-        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 =
-                        cfs_list_entry(tmp, struct ptlrpc_request,
-                                       rq_set_chain);
+	/* Requests on the set should either all be completed, or all be new */
+	expected_phase = (atomic_read(&set->set_remaining) == 0) ?
+			 RQ_PHASE_COMPLETE : RQ_PHASE_NEW;
+	list_for_each(tmp, &set->set_requests) {
+		struct ptlrpc_request *req =
+			list_entry(tmp, struct ptlrpc_request,
+				   rq_set_chain);
 
-                LASSERT(req->rq_phase == expected_phase);
-                n++;
-        }
+		LASSERT(req->rq_phase == expected_phase);
+		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);
+	LASSERTF(atomic_read(&set->set_remaining) == 0 ||
+		 atomic_read(&set->set_remaining) == n, "%d / %d\n",
+		 atomic_read(&set->set_remaining), n);
 
-        cfs_list_for_each_safe(tmp, next, &set->set_requests) {
-                struct ptlrpc_request *req =
-                        cfs_list_entry(tmp, struct ptlrpc_request,
-                                       rq_set_chain);
-                cfs_list_del_init(&req->rq_set_chain);
+	list_for_each_safe(tmp, next, &set->set_requests) {
+		struct ptlrpc_request *req =
+			list_entry(tmp, struct ptlrpc_request,
+				   rq_set_chain);
+		list_del_init(&req->rq_set_chain);
 
-                LASSERT(req->rq_phase == expected_phase);
+		LASSERT(req->rq_phase == expected_phase);
 
-                if (req->rq_phase == RQ_PHASE_NEW) {
-                        ptlrpc_req_interpret(NULL, req, -EBADR);
-                        cfs_atomic_dec(&set->set_remaining);
-                }
+		if (req->rq_phase == RQ_PHASE_NEW) {
+			ptlrpc_req_interpret(NULL, req, -EBADR);
+			atomic_dec(&set->set_remaining);
+		}
 
 		spin_lock(&req->rq_lock);
 		req->rq_set = NULL;
@@ -955,10 +954,10 @@ void ptlrpc_set_destroy(struct ptlrpc_request_set *set)
                 ptlrpc_req_finished (req);
         }
 
-        LASSERT(cfs_atomic_read(&set->set_remaining) == 0);
+	LASSERT(atomic_read(&set->set_remaining) == 0);
 
-        ptlrpc_reqset_put(set);
-        EXIT;
+	ptlrpc_reqset_put(set);
+	EXIT;
 }
 EXPORT_SYMBOL(ptlrpc_set_destroy);
 
@@ -970,17 +969,17 @@ EXPORT_SYMBOL(ptlrpc_set_destroy);
 int ptlrpc_set_add_cb(struct ptlrpc_request_set *set,
                       set_interpreter_func fn, void *data)
 {
-        struct ptlrpc_set_cbdata *cbdata;
+	struct ptlrpc_set_cbdata *cbdata;
 
-        OBD_ALLOC_PTR(cbdata);
-        if (cbdata == NULL)
-                RETURN(-ENOMEM);
+	OBD_ALLOC_PTR(cbdata);
+	if (cbdata == NULL)
+		RETURN(-ENOMEM);
 
-        cbdata->psc_interpret = fn;
-        cbdata->psc_data = data;
-        cfs_list_add_tail(&cbdata->psc_item, &set->set_cblist);
+	cbdata->psc_interpret = fn;
+	cbdata->psc_data = data;
+	list_add_tail(&cbdata->psc_item, &set->set_cblist);
 
-        RETURN(0);
+	RETURN(0);
 }
 EXPORT_SYMBOL(ptlrpc_set_add_cb);
 
@@ -991,12 +990,12 @@ EXPORT_SYMBOL(ptlrpc_set_add_cb);
 void ptlrpc_set_add_req(struct ptlrpc_request_set *set,
                         struct ptlrpc_request *req)
 {
-	LASSERT(cfs_list_empty(&req->rq_set_chain));
+	LASSERT(list_empty(&req->rq_set_chain));
 
 	/* The set takes over the caller's request reference */
-	cfs_list_add_tail(&req->rq_set_chain, &set->set_requests);
+	list_add_tail(&req->rq_set_chain, &set->set_requests);
 	req->rq_set = set;
-	cfs_atomic_inc(&set->set_remaining);
+	atomic_inc(&set->set_remaining);
 	req->rq_queued_time = cfs_time_current();
 
 	if (req->rq_reqmsg != NULL)
@@ -1029,20 +1028,20 @@ void ptlrpc_set_add_new_req(struct ptlrpcd_ctl *pc,
 	 */
 	req->rq_set = set;
 	req->rq_queued_time = cfs_time_current();
-	cfs_list_add_tail(&req->rq_set_chain, &set->set_new_requests);
-	count = cfs_atomic_inc_return(&set->set_new_count);
+	list_add_tail(&req->rq_set_chain, &set->set_new_requests);
+	count = atomic_inc_return(&set->set_new_count);
 	spin_unlock(&set->set_new_req_lock);
 
-        /* Only need to call wakeup once for the first entry. */
-        if (count == 1) {
-                cfs_waitq_signal(&set->set_waitq);
+	/* Only need to call wakeup once for the first entry. */
+	if (count == 1) {
+		wake_up(&set->set_waitq);
 
-                /* XXX: It maybe unnecessary to wakeup all the partners. But to
-                 *      guarantee the async RPC can be processed ASAP, we have
-                 *      no other better choice. It maybe fixed in future. */
-                for (i = 0; i < pc->pc_npartners; i++)
-                        cfs_waitq_signal(&pc->pc_partners[i]->pc_set->set_waitq);
-        }
+		/* XXX: It maybe unnecessary to wakeup all the partners. But to
+		 *      guarantee the async RPC can be processed ASAP, we have
+		 *      no other better choice. It maybe fixed in future. */
+		for (i = 0; i < pc->pc_npartners; i++)
+			wake_up(&pc->pc_partners[i]->pc_set->set_waitq);
+	}
 }
 EXPORT_SYMBOL(ptlrpc_set_add_new_req);
 
@@ -1076,16 +1075,16 @@ static int ptlrpc_import_delay_req(struct obd_import *imp,
 			  D_HA : D_ERROR, req, "IMP_CLOSED ");
 		*status = -EIO;
         } else if (ptlrpc_send_limit_expired(req)) {
-                /* probably doesn't need to be a D_ERROR after initial testing */
-                DEBUG_REQ(D_ERROR, req, "send limit expired ");
-                *status = -EIO;
-        } else if (req->rq_send_state == LUSTRE_IMP_CONNECTING &&
-                   imp->imp_state == LUSTRE_IMP_CONNECTING) {
-                /* allow CONNECT even if import is invalid */ ;
-                if (cfs_atomic_read(&imp->imp_inval_count) != 0) {
-                        DEBUG_REQ(D_ERROR, req, "invalidate in flight");
-                        *status = -EIO;
-                }
+		/* probably doesn't need to be a D_ERROR after initial testing*/
+		DEBUG_REQ(D_HA, req, "send limit expired ");
+		*status = -ETIMEDOUT;
+	} else if (req->rq_send_state == LUSTRE_IMP_CONNECTING &&
+		   imp->imp_state == LUSTRE_IMP_CONNECTING) {
+		/* allow CONNECT even if import is invalid */ ;
+		if (atomic_read(&imp->imp_inval_count) != 0) {
+			DEBUG_REQ(D_ERROR, req, "invalidate in flight");
+			*status = -EIO;
+		}
 	} else if (imp->imp_invalid || imp->imp_obd->obd_no_recov) {
 		if (!imp->imp_deactive)
 			DEBUG_REQ(D_NET, req, "IMP_INVALID");
@@ -1095,7 +1094,7 @@ static int ptlrpc_import_delay_req(struct obd_import *imp,
                 *status = -EIO;
         } else if (req->rq_send_state != imp->imp_state) {
                 /* invalidate in progress - any requests should be drop */
-                if (cfs_atomic_read(&imp->imp_inval_count) != 0) {
+		if (atomic_read(&imp->imp_inval_count) != 0) {
                         DEBUG_REQ(D_ERROR, req, "invalidate in flight");
                         *status = -EIO;
                 } else if (imp->imp_dlm_fake || req->rq_no_delay) {
@@ -1220,7 +1219,7 @@ static int after_reply(struct ptlrpc_request *req)
 
         LASSERT(obd != NULL);
         /* repbuf must be unlinked */
-        LASSERT(!req->rq_receiving_reply && !req->rq_must_unlink);
+	LASSERT(!req->rq_receiving_reply && !req->rq_reply_unlink);
 
         if (req->rq_reply_truncate) {
                 if (ptlrpc_no_resend(req)) {
@@ -1237,7 +1236,9 @@ static int after_reply(struct ptlrpc_request *req)
                  * will roundup it */
                 req->rq_replen       = req->rq_nob_received;
                 req->rq_nob_received = 0;
-                req->rq_resend       = 1;
+		spin_lock(&req->rq_lock);
+		req->rq_resend       = 1;
+		spin_unlock(&req->rq_lock);
                 RETURN(0);
         }
 
@@ -1267,7 +1268,9 @@ static int after_reply(struct ptlrpc_request *req)
 		time_t	now = cfs_time_current_sec();
 
 		DEBUG_REQ(D_RPCTRACE, req, "Resending request on EINPROGRESS");
+		spin_lock(&req->rq_lock);
 		req->rq_resend = 1;
+		spin_unlock(&req->rq_lock);
 		req->rq_nr_resend++;
 
 		/* allocate new xid to avoid reply reconstruction */
@@ -1360,7 +1363,11 @@ static int after_reply(struct ptlrpc_request *req)
                         /** version recovery */
                         ptlrpc_save_versions(req);
                         ptlrpc_retain_replayable_request(req, imp);
-                } else if (req->rq_commit_cb != NULL) {
+		} else if (req->rq_commit_cb != NULL &&
+			   list_empty(&req->rq_replay_list)) {
+			/* NB: don't call rq_commit_cb if it's already on
+			 * rq_replay_list, ptlrpc_free_committed() will call
+			 * it later, see LU-3618 for details */
 			spin_unlock(&imp->imp_lock);
 			req->rq_commit_cb(req);
 			spin_lock(&imp->imp_lock);
@@ -1376,12 +1383,12 @@ static int after_reply(struct ptlrpc_request *req)
 
 		ptlrpc_free_committed(imp);
 
-		if (!cfs_list_empty(&imp->imp_replay_list)) {
+		if (!list_empty(&imp->imp_replay_list)) {
 			struct ptlrpc_request *last;
 
-			last = cfs_list_entry(imp->imp_replay_list.prev,
-					      struct ptlrpc_request,
-					      rq_replay_list);
+			last = list_entry(imp->imp_replay_list.prev,
+					  struct ptlrpc_request,
+					  rq_replay_list);
 			/*
 			 * Requests with rq_replay stay on the list even if no
 			 * commit is expected.
@@ -1429,9 +1436,9 @@ static int ptlrpc_send_new_req(struct ptlrpc_request *req)
 			  "(%s != %s)", lustre_msg_get_status(req->rq_reqmsg),
 			  ptlrpc_import_state_name(req->rq_send_state),
 			  ptlrpc_import_state_name(imp->imp_state));
-		LASSERT(cfs_list_empty(&req->rq_list));
-		cfs_list_add_tail(&req->rq_list, &imp->imp_delayed_list);
-		cfs_atomic_inc(&req->rq_import->imp_inflight);
+		LASSERT(list_empty(&req->rq_list));
+		list_add_tail(&req->rq_list, &imp->imp_delayed_list);
+		atomic_inc(&req->rq_import->imp_inflight);
 		spin_unlock(&imp->imp_lock);
 		RETURN(0);
 	}
@@ -1443,9 +1450,9 @@ static int ptlrpc_send_new_req(struct ptlrpc_request *req)
 		RETURN(rc);
 	}
 
-	LASSERT(cfs_list_empty(&req->rq_list));
-	cfs_list_add_tail(&req->rq_list, &imp->imp_sending_list);
-	cfs_atomic_inc(&req->rq_import->imp_inflight);
+	LASSERT(list_empty(&req->rq_list));
+	list_add_tail(&req->rq_list, &imp->imp_sending_list);
+	atomic_inc(&req->rq_import->imp_inflight);
 	spin_unlock(&imp->imp_lock);
 
 	lustre_msg_set_status(req->rq_reqmsg, current_pid());
@@ -1456,7 +1463,9 @@ static int ptlrpc_send_new_req(struct ptlrpc_request *req)
                         req->rq_status = rc;
                         RETURN(1);
                 } else {
-                        req->rq_wait_ctx = 1;
+			spin_lock(&req->rq_lock);
+			req->rq_wait_ctx = 1;
+			spin_unlock(&req->rq_lock);
                         RETURN(0);
                 }
         }
@@ -1471,7 +1480,9 @@ static int ptlrpc_send_new_req(struct ptlrpc_request *req)
         rc = ptl_send_rpc(req, 0);
         if (rc) {
                 DEBUG_REQ(D_HA, req, "send failed (%d); expect timeout", rc);
-                req->rq_net_err = 1;
+		spin_lock(&req->rq_lock);
+		req->rq_net_err = 1;
+		spin_unlock(&req->rq_lock);
                 RETURN(rc);
         }
         RETURN(0);
@@ -1484,11 +1495,11 @@ static inline int ptlrpc_set_producer(struct ptlrpc_request_set *set)
 
 	LASSERT(set->set_producer != NULL);
 
-	remaining = cfs_atomic_read(&set->set_remaining);
+	remaining = atomic_read(&set->set_remaining);
 
 	/* populate the ->set_requests list with requests until we
 	 * reach the maximum number of RPCs in flight for this set */
-	while (cfs_atomic_read(&set->set_remaining) < set->set_max_inflight) {
+	while (atomic_read(&set->set_remaining) < set->set_max_inflight) {
 		rc = set->set_producer(set, set->set_producer_arg);
 		if (rc == -ENOENT) {
 			/* no more RPC to produce */
@@ -1498,7 +1509,7 @@ static inline int ptlrpc_set_producer(struct ptlrpc_request_set *set)
 		}
 	}
 
-	RETURN((cfs_atomic_read(&set->set_remaining) - remaining));
+	RETURN((atomic_read(&set->set_remaining) - remaining));
 }
 
 /**
@@ -1506,23 +1517,35 @@ static inline int ptlrpc_set_producer(struct ptlrpc_request_set *set)
  * and no more replies are expected.
  * (it is possible to get less replies than requests sent e.g. due to timed out
  * requests or requests that we had trouble to send out)
+ *
+ * NOTE: This function contains a potential schedule point (cond_resched()).
  */
 int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set)
 {
-        cfs_list_t *tmp, *next;
-        int force_timer_recalc = 0;
-        ENTRY;
+	struct list_head *tmp, *next;
+	struct list_head  comp_reqs;
+	int force_timer_recalc = 0;
+	ENTRY;
 
-        if (cfs_atomic_read(&set->set_remaining) == 0)
-                RETURN(1);
+	if (atomic_read(&set->set_remaining) == 0)
+		RETURN(1);
 
-        cfs_list_for_each_safe(tmp, next, &set->set_requests) {
-                struct ptlrpc_request *req =
-                        cfs_list_entry(tmp, struct ptlrpc_request,
-                                       rq_set_chain);
-                struct obd_import *imp = req->rq_import;
-                int unregistered = 0;
-                int rc = 0;
+	INIT_LIST_HEAD(&comp_reqs);
+	list_for_each_safe(tmp, next, &set->set_requests) {
+		struct ptlrpc_request *req =
+			list_entry(tmp, struct ptlrpc_request,
+				   rq_set_chain);
+		struct obd_import *imp = req->rq_import;
+		int unregistered = 0;
+		int rc = 0;
+
+		/* This schedule point is mainly for the ptlrpcd caller of this
+		 * function.  Most ptlrpc sets are not long-lived and unbounded
+		 * in length, but at the least the set used by the ptlrpcd is.
+		 * Since the processing time is unbounded, we need to insert an
+		 * explicit schedule point to make the thread well-behaved.
+		 */
+		cond_resched();
 
                 if (req->rq_phase == RQ_PHASE_NEW &&
                     ptlrpc_send_new_req(req)) {
@@ -1582,8 +1605,10 @@ int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set)
                         ptlrpc_rqphase_move(req, req->rq_next_phase);
                 }
 
-                if (req->rq_phase == RQ_PHASE_COMPLETE)
+                if (req->rq_phase == RQ_PHASE_COMPLETE) {
+			list_move_tail(&req->rq_set_chain, &comp_reqs);
                         continue;
+		}
 
                 if (req->rq_phase == RQ_PHASE_INTERPRET)
                         GOTO(interpret, req->rq_status);
@@ -1642,15 +1667,17 @@ int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set)
                             req->rq_waiting || req->rq_wait_ctx) {
                                 int status;
 
-                                if (!ptlrpc_unregister_reply(req, 1))
-                                        continue;
+				if (!ptlrpc_unregister_reply(req, 1)) {
+					ptlrpc_unregister_bulk(req, 1);
+					continue;
+				}
 
 				spin_lock(&imp->imp_lock);
 				if (ptlrpc_import_delay_req(imp, req, &status)){
 					/* put on delay list - only if we wait
 					 * recovery finished - before send */
-					cfs_list_del_init(&req->rq_list);
-					cfs_list_add_tail(&req->rq_list,
+					list_del_init(&req->rq_list);
+					list_add_tail(&req->rq_list,
 							  &imp->
 							  imp_delayed_list);
 					spin_unlock(&imp->imp_lock);
@@ -1673,8 +1700,8 @@ int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set)
 					GOTO(interpret, req->rq_status);
 				}
 
-				cfs_list_del_init(&req->rq_list);
-				cfs_list_add_tail(&req->rq_list,
+				list_del_init(&req->rq_list);
+				list_add_tail(&req->rq_list,
 						  &imp->imp_sending_list);
 
 				spin_unlock(&imp->imp_lock);
@@ -1737,6 +1764,7 @@ int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set)
 					spin_lock(&req->rq_lock);
 					req->rq_net_err = 1;
 					spin_unlock(&req->rq_lock);
+					continue;
 				}
 				/* need to reset the timeout */
 				force_timer_recalc = 1;
@@ -1822,7 +1850,11 @@ int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set)
 
                 ptlrpc_req_interpret(env, req, req->rq_status);
 
-                ptlrpc_rqphase_move(req, RQ_PHASE_COMPLETE);
+		if (ptlrpcd_check_work(req)) {
+			atomic_dec(&set->set_remaining);
+			continue;
+		}
+		ptlrpc_rqphase_move(req, RQ_PHASE_COMPLETE);
 
 		CDEBUG(req->rq_reqmsg != NULL ? D_RPCTRACE : 0,
 			"Completed RPC pname:cluuid:pid:xid:nid:"
@@ -1837,14 +1869,14 @@ int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set)
 		 * may happen in the case of marking it erroneous for the case
 		 * ptlrpc_import_delay_req(req, status) find it impossible to
 		 * allow sending this rpc and returns *status != 0. */
-		if (!cfs_list_empty(&req->rq_list)) {
-			cfs_list_del_init(&req->rq_list);
-			cfs_atomic_dec(&imp->imp_inflight);
+		if (!list_empty(&req->rq_list)) {
+			list_del_init(&req->rq_list);
+			atomic_dec(&imp->imp_inflight);
 		}
 		spin_unlock(&imp->imp_lock);
 
-                cfs_atomic_dec(&set->set_remaining);
-                cfs_waitq_broadcast(&imp->imp_recovery_waitq);
+		atomic_dec(&set->set_remaining);
+		wake_up_all(&imp->imp_recovery_waitq);
 
 		if (set->set_producer) {
 			/* produce a new request if possible */
@@ -1853,7 +1885,7 @@ int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set)
 
 			/* free the request that has just been completed
 			 * in order not to pollute set->set_requests */
-			cfs_list_del_init(&req->rq_set_chain);
+			list_del_init(&req->rq_set_chain);
 			spin_lock(&req->rq_lock);
 			req->rq_set = NULL;
 			req->rq_invalid_rqset = 0;
@@ -1863,11 +1895,17 @@ int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set)
 			if (req->rq_status != 0)
 				set->set_rc = req->rq_status;
 			ptlrpc_req_finished(req);
+		} else {
+			list_move_tail(&req->rq_set_chain, &comp_reqs);
 		}
-        }
+	}
+
+	/* move completed request at the head of list so it's easier for
+	 * caller to find them */
+	list_splice(&comp_reqs, &set->set_requests);
 
-        /* If we hit an error, we want to recover promptly. */
-        RETURN(cfs_atomic_read(&set->set_remaining) == 0 || force_timer_recalc);
+	/* If we hit an error, we want to recover promptly. */
+	RETURN(atomic_read(&set->set_remaining) == 0 || force_timer_recalc);
 }
 EXPORT_SYMBOL(ptlrpc_check_set);
 
@@ -1909,7 +1947,7 @@ int ptlrpc_expire_one_request(struct ptlrpc_request *req, int async_unlink)
                 RETURN(1);
         }
 
-        cfs_atomic_inc(&imp->imp_timeouts);
+	atomic_inc(&imp->imp_timeouts);
 
         /* The DLM server doesn't want recovery run on its imports. */
         if (imp->imp_dlm_fake)
@@ -1949,20 +1987,20 @@ int ptlrpc_expire_one_request(struct ptlrpc_request *req, int async_unlink)
  */
 int ptlrpc_expired_set(void *data)
 {
-        struct ptlrpc_request_set *set = data;
-        cfs_list_t                *tmp;
-        time_t                     now = cfs_time_current_sec();
-        ENTRY;
+	struct ptlrpc_request_set	*set = data;
+	struct list_head		*tmp;
+	time_t				now = cfs_time_current_sec();
+	ENTRY;
 
-        LASSERT(set != NULL);
+	LASSERT(set != NULL);
 
-        /*
-         * A timeout expired. See which reqs it applies to...
-         */
-        cfs_list_for_each (tmp, &set->set_requests) {
-                struct ptlrpc_request *req =
-                        cfs_list_entry(tmp, struct ptlrpc_request,
-                                       rq_set_chain);
+	/*
+	 * A timeout expired. See which reqs it applies to...
+	 */
+	list_for_each(tmp, &set->set_requests) {
+		struct ptlrpc_request *req =
+			list_entry(tmp, struct ptlrpc_request,
+				   rq_set_chain);
 
                 /* don't expire request waiting for context */
                 if (req->rq_wait_ctx)
@@ -2009,23 +2047,22 @@ EXPORT_SYMBOL(ptlrpc_mark_interrupted);
  */
 void ptlrpc_interrupted_set(void *data)
 {
-        struct ptlrpc_request_set *set = data;
-        cfs_list_t *tmp;
+	struct ptlrpc_request_set *set = data;
+	struct list_head *tmp;
 
-        LASSERT(set != NULL);
-        CDEBUG(D_RPCTRACE, "INTERRUPTED SET %p\n", set);
+	LASSERT(set != NULL);
+	CDEBUG(D_RPCTRACE, "INTERRUPTED SET %p\n", set);
 
-        cfs_list_for_each(tmp, &set->set_requests) {
-                struct ptlrpc_request *req =
-                        cfs_list_entry(tmp, struct ptlrpc_request,
-                                       rq_set_chain);
+	list_for_each(tmp, &set->set_requests) {
+		struct ptlrpc_request *req =
+			list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 
-                if (req->rq_phase != RQ_PHASE_RPC &&
-                    req->rq_phase != RQ_PHASE_UNREGISTERING)
-                        continue;
+		if (req->rq_phase != RQ_PHASE_RPC &&
+		    req->rq_phase != RQ_PHASE_UNREGISTERING)
+			continue;
 
-                ptlrpc_mark_interrupted(req);
-        }
+		ptlrpc_mark_interrupted(req);
+	}
 }
 EXPORT_SYMBOL(ptlrpc_interrupted_set);
 
@@ -2034,15 +2071,15 @@ EXPORT_SYMBOL(ptlrpc_interrupted_set);
  */
 int ptlrpc_set_next_timeout(struct ptlrpc_request_set *set)
 {
-        cfs_list_t            *tmp;
-        time_t                 now = cfs_time_current_sec();
-        int                    timeout = 0;
-        struct ptlrpc_request *req;
-        int                    deadline;
-        ENTRY;
+	struct list_head	*tmp;
+	time_t			 now = cfs_time_current_sec();
+	int			 timeout = 0;
+	struct ptlrpc_request	*req;
+	int			 deadline;
+	ENTRY;
 
-        cfs_list_for_each(tmp, &set->set_requests) {
-                req = cfs_list_entry(tmp, struct ptlrpc_request, rq_set_chain);
+	list_for_each(tmp, &set->set_requests) {
+		req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 
                 /*
                  * Request in-flight?
@@ -2088,7 +2125,7 @@ EXPORT_SYMBOL(ptlrpc_set_next_timeout);
  */
 int ptlrpc_set_wait(struct ptlrpc_request_set *set)
 {
-        cfs_list_t            *tmp;
+	struct list_head            *tmp;
         struct ptlrpc_request *req;
         struct l_wait_info     lwi;
         int                    rc, timeout;
@@ -2097,14 +2134,14 @@ int ptlrpc_set_wait(struct ptlrpc_request_set *set)
 	if (set->set_producer)
 		(void)ptlrpc_set_producer(set);
 	else
-		cfs_list_for_each(tmp, &set->set_requests) {
-			req = cfs_list_entry(tmp, struct ptlrpc_request,
-					     rq_set_chain);
+		list_for_each(tmp, &set->set_requests) {
+			req = list_entry(tmp, struct ptlrpc_request,
+					 rq_set_chain);
 			if (req->rq_phase == RQ_PHASE_NEW)
 				(void)ptlrpc_send_new_req(req);
 		}
 
-        if (cfs_list_empty(&set->set_requests))
+	if (list_empty(&set->set_requests))
                 RETURN(0);
 
         do {
@@ -2139,19 +2176,19 @@ int ptlrpc_set_wait(struct ptlrpc_request_set *set)
                 /* LU-769 - if we ignored the signal because it was already
                  * pending when we started, we need to handle it now or we risk
                  * it being ignored forever */
-                if (rc == -ETIMEDOUT && !lwi.lwi_allow_intr &&
-                    cfs_signal_pending()) {
-                        cfs_sigset_t blocked_sigs =
-                                           cfs_block_sigsinv(LUSTRE_FATAL_SIGS);
-
-                        /* In fact we only interrupt for the "fatal" signals
-                         * like SIGINT or SIGKILL. We still ignore less
-                         * important signals since ptlrpc set is not easily
-                         * reentrant from userspace again */
-                        if (cfs_signal_pending())
-                                ptlrpc_interrupted_set(set);
+		if (rc == -ETIMEDOUT && !lwi.lwi_allow_intr &&
+		    cfs_signal_pending()) {
+			sigset_t blocked_sigs =
+					   cfs_block_sigsinv(LUSTRE_FATAL_SIGS);
+
+			/* In fact we only interrupt for the "fatal" signals
+			 * like SIGINT or SIGKILL. We still ignore less
+			 * important signals since ptlrpc set is not easily
+			 * reentrant from userspace again */
+			if (cfs_signal_pending())
+				ptlrpc_interrupted_set(set);
 			cfs_restore_sigs(blocked_sigs);
-                }
+		}
 
                 LASSERT(rc == 0 || rc == -EINTR || rc == -ETIMEDOUT);
 
@@ -2162,22 +2199,22 @@ 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. */
-                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);
+		if (rc == 0 && atomic_read(&set->set_remaining) == 0) {
+			list_for_each(tmp, &set->set_requests) {
+				req = list_entry(tmp, struct ptlrpc_request,
+						 rq_set_chain);
 				spin_lock(&req->rq_lock);
 				req->rq_invalid_rqset = 1;
 				spin_unlock(&req->rq_lock);
-                        }
-                }
-        } while (rc != 0 || cfs_atomic_read(&set->set_remaining) != 0);
+			}
+		}
+	} while (rc != 0 || atomic_read(&set->set_remaining) != 0);
 
-        LASSERT(cfs_atomic_read(&set->set_remaining) == 0);
+	LASSERT(atomic_read(&set->set_remaining) == 0);
 
         rc = set->set_rc; /* rq_status of already freed requests if any */
-        cfs_list_for_each(tmp, &set->set_requests) {
-                req = cfs_list_entry(tmp, struct ptlrpc_request, rq_set_chain);
+	list_for_each(tmp, &set->set_requests) {
+		req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 
                 LASSERT(req->rq_phase == RQ_PHASE_COMPLETE);
                 if (req->rq_status != 0)
@@ -2192,9 +2229,9 @@ int ptlrpc_set_wait(struct ptlrpc_request_set *set)
                 struct ptlrpc_set_cbdata *cbdata, *n;
                 int err;
 
-                cfs_list_for_each_entry_safe(cbdata, n,
+		list_for_each_entry_safe(cbdata, n,
                                          &set->set_cblist, psc_item) {
-                        cfs_list_del_init(&cbdata->psc_item);
+			list_del_init(&cbdata->psc_item);
                         err = cbdata->psc_interpret(set, cbdata->psc_data, rc);
                         if (err && !rc)
                                 rc = err;
@@ -2224,9 +2261,9 @@ static void __ptlrpc_free_req(struct ptlrpc_request *request, int locked)
 
         LASSERTF(!request->rq_receiving_reply, "req %p\n", request);
         LASSERTF(request->rq_rqbd == NULL, "req %p\n",request);/* client-side */
-        LASSERTF(cfs_list_empty(&request->rq_list), "req %p\n", request);
-        LASSERTF(cfs_list_empty(&request->rq_set_chain), "req %p\n", request);
-        LASSERTF(cfs_list_empty(&request->rq_exp_list), "req %p\n", request);
+	LASSERTF(list_empty(&request->rq_list), "req %p\n", request);
+	LASSERTF(list_empty(&request->rq_set_chain), "req %p\n", request);
+	LASSERTF(list_empty(&request->rq_exp_list), "req %p\n", request);
         LASSERTF(!request->rq_replay, "req %p\n", request);
 
         req_capsule_fini(&request->rq_pill);
@@ -2236,17 +2273,17 @@ static void __ptlrpc_free_req(struct ptlrpc_request *request, int locked)
         if (request->rq_import != NULL) {
 		if (!locked)
 			spin_lock(&request->rq_import->imp_lock);
-		cfs_list_del_init(&request->rq_replay_list);
+		list_del_init(&request->rq_replay_list);
 		if (!locked)
 			spin_unlock(&request->rq_import->imp_lock);
         }
-        LASSERTF(cfs_list_empty(&request->rq_replay_list), "req %p\n", request);
+	LASSERTF(list_empty(&request->rq_replay_list), "req %p\n", request);
 
-        if (cfs_atomic_read(&request->rq_refcount) != 0) {
-                DEBUG_REQ(D_ERROR, request,
-                          "freeing request with nonzero refcount");
-                LBUG();
-        }
+	if (atomic_read(&request->rq_refcount) != 0) {
+		DEBUG_REQ(D_ERROR, request,
+			  "freeing request with nonzero refcount");
+		LBUG();
+	}
 
         if (request->rq_repbuf != NULL)
                 sptlrpc_cli_free_repbuf(request);
@@ -2281,7 +2318,7 @@ static int __ptlrpc_req_finished(struct ptlrpc_request *request, int locked);
  */
 void ptlrpc_req_finished_with_imp_lock(struct ptlrpc_request *request)
 {
-	LASSERT(spin_is_locked(&request->rq_import->imp_lock));
+	assert_spin_locked(&request->rq_import->imp_lock);
 	(void)__ptlrpc_req_finished(request, 1);
 }
 EXPORT_SYMBOL(ptlrpc_req_finished_with_imp_lock);
@@ -2306,9 +2343,9 @@ static int __ptlrpc_req_finished(struct ptlrpc_request *request, int locked)
         }
 
         DEBUG_REQ(D_INFO, request, "refcount now %u",
-                  cfs_atomic_read(&request->rq_refcount) - 1);
+		  atomic_read(&request->rq_refcount) - 1);
 
-        if (cfs_atomic_dec_and_test(&request->rq_refcount)) {
+	if (atomic_dec_and_test(&request->rq_refcount)) {
                 __ptlrpc_free_req(request, locked);
                 RETURN(1);
         }
@@ -2343,17 +2380,17 @@ EXPORT_SYMBOL(ptlrpc_req_xid);
  */
 int ptlrpc_unregister_reply(struct ptlrpc_request *request, int async)
 {
-        int                rc;
-        struct l_wait_info lwi;
+	int                rc;
+	struct l_wait_info lwi;
 
-        /*
-         * Might sleep.
-         */
-        LASSERT(!cfs_in_interrupt());
+	/*
+	 * Might sleep.
+	 */
+	LASSERT(!in_interrupt());
 
-        /*
-         * Let's setup deadline for reply unlink.
-         */
+	/*
+	 * Let's setup deadline for reply unlink.
+	 */
         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_REPL_UNLINK) &&
             async && request->rq_reply_deadline == 0)
                 request->rq_reply_deadline = cfs_time_current_sec()+LONG_UNLINK;
@@ -2389,12 +2426,10 @@ int ptlrpc_unregister_reply(struct ptlrpc_request *request, int async)
          * unlinked before returning a req to the pool.
          */
         for (;;) {
-#ifdef __KERNEL__
 		/* The wq argument is ignored by user-space wait_event macros */
-		cfs_waitq_t *wq = (request->rq_set != NULL) ?
-				  &request->rq_set->set_waitq :
-				  &request->rq_reply_waitq;
-#endif
+		wait_queue_head_t *wq = (request->rq_set != NULL) ?
+					&request->rq_set->set_waitq :
+					&request->rq_reply_waitq;
                 /* Network access will complete in finite time but the HUGE
                  * timeout lets us CWARN for visibility of sluggish NALs */
                 lwi = LWI_TIMEOUT_INTERVAL(cfs_time_seconds(LONG_UNLINK),
@@ -2408,13 +2443,46 @@ int ptlrpc_unregister_reply(struct ptlrpc_request *request, int async)
 
                 LASSERT(rc == -ETIMEDOUT);
                 DEBUG_REQ(D_WARNING, request, "Unexpectedly long timeout "
-                          "rvcng=%d unlnk=%d", request->rq_receiving_reply,
-                          request->rq_must_unlink);
+			  "rvcng=%d unlnk=%d/%d", request->rq_receiving_reply,
+			  request->rq_req_unlink, request->rq_reply_unlink);
         }
         RETURN(0);
 }
 EXPORT_SYMBOL(ptlrpc_unregister_reply);
 
+static void ptlrpc_free_request(struct ptlrpc_request *req)
+{
+	spin_lock(&req->rq_lock);
+	req->rq_replay = 0;
+	spin_unlock(&req->rq_lock);
+
+	if (req->rq_commit_cb != NULL)
+		req->rq_commit_cb(req);
+	list_del_init(&req->rq_replay_list);
+
+	__ptlrpc_req_finished(req, 1);
+}
+
+/**
+ * the request is committed and dropped from the replay list of its import
+ */
+void ptlrpc_request_committed(struct ptlrpc_request *req, int force)
+{
+	struct obd_import 	*imp = req->rq_import;
+
+	spin_lock(&imp->imp_lock);
+	if (list_empty(&req->rq_replay_list)) {
+		spin_unlock(&imp->imp_lock);
+		return;
+	}
+
+	if (force || req->rq_transno <= imp->imp_peer_committed_transno)
+		ptlrpc_free_request(req);
+
+	spin_unlock(&imp->imp_lock);
+}
+EXPORT_SYMBOL(ptlrpc_request_committed);
+
 /**
  * Iterates through replay_list on import and prunes
  * all requests have transno smaller than last_committed for the
@@ -2425,32 +2493,32 @@ EXPORT_SYMBOL(ptlrpc_unregister_reply);
  */
 void ptlrpc_free_committed(struct obd_import *imp)
 {
-        cfs_list_t *tmp, *saved;
-        struct ptlrpc_request *req;
-        struct ptlrpc_request *last_req = NULL; /* temporary fire escape */
-        ENTRY;
+	struct ptlrpc_request	*req, *saved;
+	struct ptlrpc_request	*last_req = NULL; /* temporary fire escape */
+	bool			 skip_committed_list = true;
+	ENTRY;
 
 	LASSERT(imp != NULL);
-	LASSERT(spin_is_locked(&imp->imp_lock));
-
+	assert_spin_locked(&imp->imp_lock);
 
         if (imp->imp_peer_committed_transno == imp->imp_last_transno_checked &&
             imp->imp_generation == imp->imp_last_generation_checked) {
                 CDEBUG(D_INFO, "%s: skip recheck: last_committed "LPU64"\n",
                        imp->imp_obd->obd_name, imp->imp_peer_committed_transno);
-                EXIT;
-                return;
+		RETURN_EXIT;
         }
         CDEBUG(D_RPCTRACE, "%s: committing for last_committed "LPU64" gen %d\n",
                imp->imp_obd->obd_name, imp->imp_peer_committed_transno,
                imp->imp_generation);
+
+	if (imp->imp_generation != imp->imp_last_generation_checked)
+		skip_committed_list = false;
+
         imp->imp_last_transno_checked = imp->imp_peer_committed_transno;
         imp->imp_last_generation_checked = imp->imp_generation;
 
-        cfs_list_for_each_safe(tmp, saved, &imp->imp_replay_list) {
-                req = cfs_list_entry(tmp, struct ptlrpc_request,
-                                     rq_replay_list);
-
+	list_for_each_entry_safe(req, saved, &imp->imp_replay_list,
+				     rq_replay_list) {
                 /* XXX ok to remove when 1357 resolved - rread 05/29/03  */
                 LASSERT(req != last_req);
                 last_req = req;
@@ -2464,38 +2532,44 @@ void ptlrpc_free_committed(struct obd_import *imp)
                         GOTO(free_req, 0);
                 }
 
-                if (req->rq_replay) {
-                        DEBUG_REQ(D_RPCTRACE, req, "keeping (FL_REPLAY)");
-                        continue;
-                }
-
                 /* not yet committed */
                 if (req->rq_transno > imp->imp_peer_committed_transno) {
                         DEBUG_REQ(D_RPCTRACE, req, "stopping search");
                         break;
                 }
 
+		if (req->rq_replay) {
+			DEBUG_REQ(D_RPCTRACE, req, "keeping (FL_REPLAY)");
+			list_move_tail(&req->rq_replay_list,
+					   &imp->imp_committed_list);
+			continue;
+		}
+
                 DEBUG_REQ(D_INFO, req, "commit (last_committed "LPU64")",
                           imp->imp_peer_committed_transno);
 free_req:
-		spin_lock(&req->rq_lock);
-		req->rq_replay = 0;
-		spin_unlock(&req->rq_lock);
-                if (req->rq_commit_cb != NULL)
-                        req->rq_commit_cb(req);
-                cfs_list_del_init(&req->rq_replay_list);
-                __ptlrpc_req_finished(req, 1);
+		ptlrpc_free_request(req);
         }
 
+	if (skip_committed_list)
+		GOTO(out, 0);
+
+	list_for_each_entry_safe(req, saved, &imp->imp_committed_list,
+				     rq_replay_list) {
+		LASSERT(req->rq_transno != 0);
+		if (req->rq_import_generation < imp->imp_generation) {
+			DEBUG_REQ(D_RPCTRACE, req, "free stale open request");
+			ptlrpc_free_request(req);
+		}
+	}
+out:
         EXIT;
-        return;
 }
 
 void ptlrpc_cleanup_client(struct obd_import *imp)
 {
         ENTRY;
         EXIT;
-        return;
 }
 EXPORT_SYMBOL(ptlrpc_cleanup_client);
 
@@ -2508,10 +2582,19 @@ EXPORT_SYMBOL(ptlrpc_cleanup_client);
 void ptlrpc_resend_req(struct ptlrpc_request *req)
 {
         DEBUG_REQ(D_HA, req, "going to resend");
+	spin_lock(&req->rq_lock);
+
+	/* Request got reply but linked to the import list still.
+	   Let ptlrpc_check_set() to process it. */
+	if (ptlrpc_client_replied(req)) {
+		spin_unlock(&req->rq_lock);
+		DEBUG_REQ(D_HA, req, "it has reply, so skip it");
+		return;
+	}
+
         lustre_msg_set_handle(req->rq_reqmsg, &(struct lustre_handle){ 0 });
         req->rq_status = -EAGAIN;
 
-	spin_lock(&req->rq_lock);
         req->rq_resend = 1;
         req->rq_net_err = 0;
         req->rq_timedout = 0;
@@ -2547,9 +2630,9 @@ EXPORT_SYMBOL(ptlrpc_restart_req);
  */
 struct ptlrpc_request *ptlrpc_request_addref(struct ptlrpc_request *req)
 {
-        ENTRY;
-        cfs_atomic_inc(&req->rq_refcount);
-        RETURN(req);
+	ENTRY;
+	atomic_inc(&req->rq_refcount);
+	RETURN(req);
 }
 EXPORT_SYMBOL(ptlrpc_request_addref);
 
@@ -2560,9 +2643,9 @@ EXPORT_SYMBOL(ptlrpc_request_addref);
 void ptlrpc_retain_replayable_request(struct ptlrpc_request *req,
                                       struct obd_import *imp)
 {
-	cfs_list_t *tmp;
+	struct list_head *tmp;
 
-	LASSERT(spin_is_locked(&imp->imp_lock));
+	assert_spin_locked(&imp->imp_lock);
 
         if (req->rq_transno == 0) {
                 DEBUG_REQ(D_EMERG, req, "saving request with zero transno");
@@ -2573,19 +2656,19 @@ void ptlrpc_retain_replayable_request(struct ptlrpc_request *req,
            as resent replayed requests. */
         lustre_msg_clear_flags(req->rq_reqmsg, MSG_RESENT);
 
-        /* don't re-add requests that have been replayed */
-        if (!cfs_list_empty(&req->rq_replay_list))
-                return;
+	/* don't re-add requests that have been replayed */
+	if (!list_empty(&req->rq_replay_list))
+		return;
 
-        lustre_msg_add_flags(req->rq_reqmsg, MSG_REPLAY);
+	lustre_msg_add_flags(req->rq_reqmsg, MSG_REPLAY);
 
-        LASSERT(imp->imp_replayable);
-        /* Balanced in ptlrpc_free_committed, usually. */
-        ptlrpc_request_addref(req);
-        cfs_list_for_each_prev(tmp, &imp->imp_replay_list) {
-                struct ptlrpc_request *iter =
-                        cfs_list_entry(tmp, struct ptlrpc_request,
-                                       rq_replay_list);
+	LASSERT(imp->imp_replayable);
+	/* Balanced in ptlrpc_free_committed, usually. */
+	ptlrpc_request_addref(req);
+	list_for_each_prev(tmp, &imp->imp_replay_list) {
+		struct ptlrpc_request *iter = list_entry(tmp,
+							 struct ptlrpc_request,
+							 rq_replay_list);
 
                 /* We may have duplicate transnos if we create and then
                  * open a file, or for closes retained if to match creating
@@ -2602,11 +2685,11 @@ void ptlrpc_retain_replayable_request(struct ptlrpc_request *req,
                                 continue;
                 }
 
-                cfs_list_add(&req->rq_replay_list, &iter->rq_replay_list);
-                return;
-        }
+		list_add(&req->rq_replay_list, &iter->rq_replay_list);
+		return;
+	}
 
-        cfs_list_add(&req->rq_replay_list, &imp->imp_replay_list);
+	list_add(&req->rq_replay_list, &imp->imp_replay_list);
 }
 EXPORT_SYMBOL(ptlrpc_retain_replayable_request);
 
@@ -2623,11 +2706,11 @@ int ptlrpc_queue_wait(struct ptlrpc_request *req)
         LASSERT(req->rq_set == NULL);
         LASSERT(!req->rq_receiving_reply);
 
-        set = ptlrpc_prep_set();
-        if (set == NULL) {
-                CERROR("Unable to allocate ptlrpc set.");
-                RETURN(-ENOMEM);
-        }
+	set = ptlrpc_prep_set();
+	if (set == NULL) {
+		CERROR("cannot allocate ptlrpc set: rc = %d\n", -ENOMEM);
+		RETURN(-ENOMEM);
+	}
 
 	/* for distributed debugging */
 	lustre_msg_set_status(req->rq_reqmsg, current_pid());
@@ -2653,14 +2736,14 @@ struct ptlrpc_replay_async_args {
  * In case of error restart replay process.
  */
 static int ptlrpc_replay_interpret(const struct lu_env *env,
-                                   struct ptlrpc_request *req,
-                                   void * data, int rc)
+				   struct ptlrpc_request *req,
+				   void * data, int rc)
 {
-        struct ptlrpc_replay_async_args *aa = data;
-        struct obd_import *imp = req->rq_import;
+	struct ptlrpc_replay_async_args *aa = data;
+	struct obd_import *imp = req->rq_import;
 
-        ENTRY;
-        cfs_atomic_dec(&imp->imp_replay_inflight);
+	ENTRY;
+	atomic_dec(&imp->imp_replay_inflight);
 
         if (!ptlrpc_client_replied(req)) {
                 CERROR("request replay timed out, restarting recovery\n");
@@ -2777,11 +2860,11 @@ int ptlrpc_replay_req(struct ptlrpc_request *req)
                                     ptlrpc_at_get_net_latency(req));
         DEBUG_REQ(D_HA, req, "REPLAY");
 
-        cfs_atomic_inc(&req->rq_import->imp_replay_inflight);
-        ptlrpc_request_addref(req); /* ptlrpcd needs a ref */
+	atomic_inc(&req->rq_import->imp_replay_inflight);
+	ptlrpc_request_addref(req);	/* ptlrpcd needs a ref */
 
-        ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
-        RETURN(0);
+	ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
+	RETURN(0);
 }
 EXPORT_SYMBOL(ptlrpc_replay_req);
 
@@ -2790,22 +2873,23 @@ EXPORT_SYMBOL(ptlrpc_replay_req);
  */
 void ptlrpc_abort_inflight(struct obd_import *imp)
 {
-        cfs_list_t *tmp, *n;
-        ENTRY;
+	struct list_head *tmp, *n;
+	ENTRY;
 
-        /* Make sure that no new requests get processed for this import.
-         * ptlrpc_{queue,set}_wait must (and does) hold imp_lock while testing
-         * this flag and then putting requests on sending_list or delayed_list.
-         */
+	/* Make sure that no new requests get processed for this import.
+	 * ptlrpc_{queue,set}_wait must (and does) hold imp_lock while testing
+	 * this flag and then putting requests on sending_list or delayed_list.
+	 */
 	spin_lock(&imp->imp_lock);
 
-        /* XXX locking?  Maybe we should remove each request with the list
-         * locked?  Also, how do we know if the requests on the list are
-         * being freed at this time?
-         */
-        cfs_list_for_each_safe(tmp, n, &imp->imp_sending_list) {
-                struct ptlrpc_request *req =
-                        cfs_list_entry(tmp, struct ptlrpc_request, rq_list);
+	/* XXX locking?  Maybe we should remove each request with the list
+	 * locked?  Also, how do we know if the requests on the list are
+	 * being freed at this time?
+	 */
+	list_for_each_safe(tmp, n, &imp->imp_sending_list) {
+		struct ptlrpc_request *req = list_entry(tmp,
+							struct ptlrpc_request,
+							rq_list);
 
                 DEBUG_REQ(D_RPCTRACE, req, "inflight");
 
@@ -2818,9 +2902,9 @@ void ptlrpc_abort_inflight(struct obd_import *imp)
 		spin_unlock(&req->rq_lock);
 	}
 
-	cfs_list_for_each_safe(tmp, n, &imp->imp_delayed_list) {
+	list_for_each_safe(tmp, n, &imp->imp_delayed_list) {
 		struct ptlrpc_request *req =
-			cfs_list_entry(tmp, struct ptlrpc_request, rq_list);
+			list_entry(tmp, struct ptlrpc_request, rq_list);
 
 		DEBUG_REQ(D_RPCTRACE, req, "aborting waiting req");
 
@@ -2849,14 +2933,14 @@ EXPORT_SYMBOL(ptlrpc_abort_inflight);
  */
 void ptlrpc_abort_set(struct ptlrpc_request_set *set)
 {
-        cfs_list_t *tmp, *pos;
+	struct list_head *tmp, *pos;
 
-        LASSERT(set != NULL);
+	LASSERT(set != NULL);
 
-        cfs_list_for_each_safe(pos, tmp, &set->set_requests) {
-                struct ptlrpc_request *req =
-                        cfs_list_entry(pos, struct ptlrpc_request,
-                                       rq_set_chain);
+	list_for_each_safe(pos, tmp, &set->set_requests) {
+		struct ptlrpc_request *req =
+			list_entry(pos, struct ptlrpc_request,
+				   rq_set_chain);
 
 		spin_lock(&req->rq_lock);
 		if (req->rq_phase != RQ_PHASE_RPC) {
@@ -2973,73 +3057,101 @@ EXPORT_SYMBOL(ptlrpc_sample_next_xid);
  *    have delay before it really runs by ptlrpcd thread.
  */
 struct ptlrpc_work_async_args {
-        __u64   magic;
-        int   (*cb)(const struct lu_env *, void *);
-        void   *cbdata;
+	int   (*cb)(const struct lu_env *, void *);
+	void   *cbdata;
 };
 
-#define PTLRPC_WORK_MAGIC 0x6655436b676f4f44ULL /* magic code */
+static void ptlrpcd_add_work_req(struct ptlrpc_request *req)
+{
+	/* re-initialize the req */
+	req->rq_timeout		= obd_timeout;
+	req->rq_sent		= cfs_time_current_sec();
+	req->rq_deadline	= req->rq_sent + req->rq_timeout;
+	req->rq_reply_deadline	= req->rq_deadline;
+	req->rq_phase		= RQ_PHASE_INTERPRET;
+	req->rq_next_phase	= RQ_PHASE_COMPLETE;
+	req->rq_xid		= ptlrpc_next_xid();
+	req->rq_import_generation = req->rq_import->imp_generation;
+
+	ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
+}
 
 static int work_interpreter(const struct lu_env *env,
-                            struct ptlrpc_request *req, void *data, int rc)
+			    struct ptlrpc_request *req, void *data, int rc)
 {
-        struct ptlrpc_work_async_args *arg = data;
+	struct ptlrpc_work_async_args *arg = data;
+
+	LASSERT(ptlrpcd_check_work(req));
+	LASSERT(arg->cb != NULL);
+
+	rc = arg->cb(env, arg->cbdata);
+
+	list_del_init(&req->rq_set_chain);
+	req->rq_set = NULL;
+
+	if (atomic_dec_return(&req->rq_refcount) > 1) {
+		atomic_set(&req->rq_refcount, 2);
+		ptlrpcd_add_work_req(req);
+	}
+	return rc;
+}
 
-        LASSERT(arg->magic == PTLRPC_WORK_MAGIC);
-        LASSERT(arg->cb != NULL);
+static int worker_format;
 
-        return arg->cb(env, arg->cbdata);
+static int ptlrpcd_check_work(struct ptlrpc_request *req)
+{
+	return req->rq_pill.rc_fmt == (void *)&worker_format;
 }
 
 /**
  * Create a work for ptlrpc.
  */
 void *ptlrpcd_alloc_work(struct obd_import *imp,
-                         int (*cb)(const struct lu_env *, void *), void *cbdata)
+			 int (*cb)(const struct lu_env *, void *), void *cbdata)
 {
-        struct ptlrpc_request         *req = NULL;
-        struct ptlrpc_work_async_args *args;
-        ENTRY;
+	struct ptlrpc_request         *req = NULL;
+	struct ptlrpc_work_async_args *args;
+	ENTRY;
 
-        cfs_might_sleep();
+	might_sleep();
 
-        if (cb == NULL)
-                RETURN(ERR_PTR(-EINVAL));
+	if (cb == NULL)
+		RETURN(ERR_PTR(-EINVAL));
 
-        /* copy some code from deprecated fakereq. */
-        req = ptlrpc_request_cache_alloc(__GFP_IO);
-        if (req == NULL) {
-                CERROR("ptlrpc: run out of memory!\n");
-                RETURN(ERR_PTR(-ENOMEM));
-        }
+	/* copy some code from deprecated fakereq. */
+	req = ptlrpc_request_cache_alloc(GFP_NOFS);
+	if (req == NULL) {
+		CERROR("ptlrpc: run out of memory!\n");
+		RETURN(ERR_PTR(-ENOMEM));
+	}
 
-        req->rq_send_state = LUSTRE_IMP_FULL;
-        req->rq_type = PTL_RPC_MSG_REQUEST;
-        req->rq_import = class_import_get(imp);
-        req->rq_export = NULL;
-        req->rq_interpret_reply = work_interpreter;
-        /* don't want reply */
-        req->rq_receiving_reply = 0;
-        req->rq_must_unlink = 0;
-        req->rq_no_delay = req->rq_no_resend = 1;
+	req->rq_send_state = LUSTRE_IMP_FULL;
+	req->rq_type = PTL_RPC_MSG_REQUEST;
+	req->rq_import = class_import_get(imp);
+	req->rq_export = NULL;
+	req->rq_interpret_reply = work_interpreter;
+	/* don't want reply */
+	req->rq_receiving_reply = 0;
+	req->rq_req_unlink = req->rq_reply_unlink = 0;
+	req->rq_no_delay = req->rq_no_resend = 1;
+	req->rq_pill.rc_fmt = (void *)&worker_format;
 
 	spin_lock_init(&req->rq_lock);
-        CFS_INIT_LIST_HEAD(&req->rq_list);
-        CFS_INIT_LIST_HEAD(&req->rq_replay_list);
-        CFS_INIT_LIST_HEAD(&req->rq_set_chain);
-        CFS_INIT_LIST_HEAD(&req->rq_history_list);
-        CFS_INIT_LIST_HEAD(&req->rq_exp_list);
-        cfs_waitq_init(&req->rq_reply_waitq);
-        cfs_waitq_init(&req->rq_set_waitq);
-        cfs_atomic_set(&req->rq_refcount, 1);
-
-        CLASSERT (sizeof(*args) <= sizeof(req->rq_async_args));
-        args = ptlrpc_req_async_args(req);
-        args->magic  = PTLRPC_WORK_MAGIC;
-        args->cb     = cb;
-        args->cbdata = cbdata;
-
-        RETURN(req);
+	INIT_LIST_HEAD(&req->rq_list);
+	INIT_LIST_HEAD(&req->rq_replay_list);
+	INIT_LIST_HEAD(&req->rq_set_chain);
+	INIT_LIST_HEAD(&req->rq_history_list);
+	INIT_LIST_HEAD(&req->rq_exp_list);
+	init_waitqueue_head(&req->rq_reply_waitq);
+	init_waitqueue_head(&req->rq_set_waitq);
+	atomic_set(&req->rq_refcount, 1);
+
+	CLASSERT (sizeof(*args) <= sizeof(req->rq_async_args));
+	args = ptlrpc_req_async_args(req);
+	args->cb     = cb;
+	args->cbdata = cbdata;
+
+	RETURN(req);
 }
 EXPORT_SYMBOL(ptlrpcd_alloc_work);
 
@@ -3054,7 +3166,7 @@ EXPORT_SYMBOL(ptlrpcd_destroy_work);
 
 int ptlrpcd_queue_work(void *handler)
 {
-        struct ptlrpc_request *req = handler;
+	struct ptlrpc_request *req = handler;
 
         /*
          * Check if the req is already being queued.
@@ -3064,26 +3176,9 @@ int ptlrpcd_queue_work(void *handler)
          * for this purpose. This is okay because the caller should use this
          * req as opaque data. - Jinshan
          */
-        LASSERT(cfs_atomic_read(&req->rq_refcount) > 0);
-        if (cfs_atomic_read(&req->rq_refcount) > 1)
-                return -EBUSY;
-
-        if (cfs_atomic_inc_return(&req->rq_refcount) > 2) { /* race */
-                cfs_atomic_dec(&req->rq_refcount);
-                return -EBUSY;
-        }
-
-        /* re-initialize the req */
-        req->rq_timeout        = obd_timeout;
-        req->rq_sent           = cfs_time_current_sec();
-        req->rq_deadline       = req->rq_sent + req->rq_timeout;
-        req->rq_reply_deadline = req->rq_deadline;
-        req->rq_phase          = RQ_PHASE_INTERPRET;
-        req->rq_next_phase     = RQ_PHASE_COMPLETE;
-        req->rq_xid            = ptlrpc_next_xid();
-        req->rq_import_generation = req->rq_import->imp_generation;
-
-        ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
-        return 0;
+	LASSERT(atomic_read(&req->rq_refcount) > 0);
+	if (atomic_inc_return(&req->rq_refcount) == 2)
+		ptlrpcd_add_work_req(req);
+	return 0;
 }
 EXPORT_SYMBOL(ptlrpcd_queue_work);