LU-904 ptlrpc: redo io on -EINPROGRESS

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

diff --git a/lustre/ptlrpc/client.c b/lustre/ptlrpc/client.c
index 0a2ab28..6b50207 100644 (file)
--- a/lustre/ptlrpc/client.c
+++ b/lustre/ptlrpc/client.c
@@ -96,7 +96,7 @@ struct ptlrpc_connection *ptlrpc_uuid_to_connection(struct obd_uuid *uuid)
  * Allocate and initialize new bulk descriptor
  * Returns pointer to the descriptor or NULL on error.
  */
-static inline struct ptlrpc_bulk_desc *new_bulk(int npages, int type, int portal)
+struct ptlrpc_bulk_desc *new_bulk(int npages, int type, int portal)
 {
         struct ptlrpc_bulk_desc *desc;
 
@@ -148,39 +148,6 @@ struct ptlrpc_bulk_desc *ptlrpc_prep_bulk_imp(struct ptlrpc_request *req,
 }
 
 /**
- * Prepare bulk descriptor for specified incoming request \a req that
- * can fit \a npages * pages. \a type is bulk type. \a portal is where
- * the bulk to be sent. Used on server-side after request was already
- * received.
- * Returns pointer to newly allocatrd initialized bulk descriptor or NULL on
- * error.
- */
-struct ptlrpc_bulk_desc *ptlrpc_prep_bulk_exp(struct ptlrpc_request *req,
-                                              int npages, int type, int portal)
-{
-        struct obd_export *exp = req->rq_export;
-        struct ptlrpc_bulk_desc *desc;
-
-        ENTRY;
-        LASSERT(type == BULK_PUT_SOURCE || type == BULK_GET_SINK);
-
-        desc = new_bulk(npages, type, portal);
-        if (desc == NULL)
-                RETURN(NULL);
-
-        desc->bd_export = class_export_get(exp);
-        desc->bd_req = req;
-
-        desc->bd_cbid.cbid_fn  = server_bulk_callback;
-        desc->bd_cbid.cbid_arg = desc;
-
-        /* NB we don't assign rq_bulk here; server-side requests are
-         * re-used, and the handler frees the bulk desc explicitly. */
-
-        return desc;
-}
-
-/**
  * Add a page \a page to the bulk descriptor \a desc.
  * Data to transfer in the page starts at offset \a pageoffset and
  * amount of data to transfer from the page is \a len
@@ -1100,9 +1067,16 @@ static int ptlrpc_import_delay_req(struct obd_import *imp,
  */
 static int ptlrpc_console_allow(struct ptlrpc_request *req)
 {
-        __u32 opc = lustre_msg_get_opc(req->rq_reqmsg);
+        __u32 opc;
         int err;
 
+        /* Fake requests include no rq_reqmsg */
+        if (req->rq_fake)
+                return 0;
+
+        LASSERT(req->rq_reqmsg != NULL);
+        opc = lustre_msg_get_opc(req->rq_reqmsg);
+
         /* Suppress particular reconnect errors which are to be expected.  No
          * errors are suppressed for the initial connection on an import */
         if ((lustre_handle_is_used(&req->rq_import->imp_remote_handle)) &&
@@ -1343,23 +1317,25 @@ static int after_reply(struct ptlrpc_request *req)
  * Helper function to send request \a req over the network for the first time
  * Also adjusts request phase.
  * Returns 0 on success or error code.
- */ 
+ */
 static int ptlrpc_send_new_req(struct ptlrpc_request *req)
 {
-        struct obd_import     *imp;
+        struct obd_import     *imp = req->rq_import;
         int rc;
         ENTRY;
 
         LASSERT(req->rq_phase == RQ_PHASE_NEW);
-        if (req->rq_sent && (req->rq_sent > cfs_time_current_sec()))
+        if (req->rq_sent && (req->rq_sent > cfs_time_current_sec()) &&
+            (!req->rq_generation_set ||
+             req->rq_import_generation == imp->imp_generation))
                 RETURN (0);
 
         ptlrpc_rqphase_move(req, RQ_PHASE_RPC);
 
-        imp = req->rq_import;
         cfs_spin_lock(&imp->imp_lock);
 
-        req->rq_import_generation = imp->imp_generation;
+        if (!req->rq_generation_set)
+                req->rq_import_generation = imp->imp_generation;
 
         if (ptlrpc_import_delay_req(imp, req, &rc)) {
                 cfs_spin_lock(&req->rq_lock);
@@ -1690,7 +1666,7 @@ int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set)
                          * process the reply. Similarly if the RPC returned
                          * an error, and therefore the bulk will never arrive.
                          */
-                        if (req->rq_bulk == NULL || req->rq_status != 0) {
+                        if (req->rq_bulk == NULL || req->rq_status < 0) {
                                 ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);
                                 GOTO(interpret, req->rq_status);
                         }
@@ -1708,7 +1684,7 @@ int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set)
                          * was good after getting the REPLY for her GET or
                          * the ACK for her PUT. */
                         DEBUG_REQ(D_ERROR, req, "bulk transfer failed");
-                        LBUG();
+                        req->rq_status = -EIO;
                 }
 
                 ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);