Whamcloud - gitweb
LU-5528 ptlrpc: fix race between connect vs resend
[fs/lustre-release.git] / lustre / ptlrpc / niobuf.c
index 4a4eeb4..c44fe83 100644 (file)
@@ -27,7 +27,7 @@
  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  *
- * Copyright (c) 2011, 2012, Intel Corporation.
+ * Copyright (c) 2012, 2013, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -35,9 +35,6 @@
  */
 
 #define DEBUG_SUBSYSTEM S_RPC
-#ifndef __KERNEL__
-#include <liblustre.h>
-#endif
 #include <obd_support.h>
 #include <lustre_net.h>
 #include <lustre_lib.h>
@@ -252,38 +249,38 @@ EXPORT_SYMBOL(ptlrpc_start_bulk_transfer);
  */
 void ptlrpc_abort_bulk(struct ptlrpc_bulk_desc *desc)
 {
-        struct l_wait_info       lwi;
-        int                      rc;
+       struct l_wait_info       lwi;
+       int                      rc;
 
-        LASSERT(!cfs_in_interrupt());           /* might sleep */
+       LASSERT(!in_interrupt());           /* might sleep */
 
-        if (!ptlrpc_server_bulk_active(desc))   /* completed or */
-                return;                         /* never started */
+       if (!ptlrpc_server_bulk_active(desc))   /* completed or */
+               return;                         /* never started */
 
-        /* We used to poison the pages with 0xab here because we did not want to
-         * send any meaningful data over the wire for evicted clients (bug 9297)
-         * However, this is no longer safe now that we use the page cache on the
-         * OSS (bug 20560) */
+       /* We used to poison the pages with 0xab here because we did not want to
+        * send any meaningful data over the wire for evicted clients (bug 9297)
+        * However, this is no longer safe now that we use the page cache on the
+        * OSS (bug 20560) */
 
-        /* The unlink ensures the callback happens ASAP and is the last
-         * one.  If it fails, it must be because completion just happened,
-         * but we must still l_wait_event() in this case, to give liblustre
-         * a chance to run server_bulk_callback()*/
+       /* The unlink ensures the callback happens ASAP and is the last
+        * one.  If it fails, it must be because completion just happened,
+        * but we must still l_wait_event() in this case, to give liblustre
+        * a chance to run server_bulk_callback()*/
        mdunlink_iterate_helper(desc->bd_mds, desc->bd_md_count);
 
-        for (;;) {
-                /* 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),
-                                           cfs_time_seconds(1), NULL, NULL);
-                rc = l_wait_event(desc->bd_waitq,
-                                  !ptlrpc_server_bulk_active(desc), &lwi);
-                if (rc == 0)
-                        return;
-
-                LASSERT(rc == -ETIMEDOUT);
-                CWARN("Unexpectedly long timeout: desc %p\n", desc);
-        }
+       for (;;) {
+               /* 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),
+                                          cfs_time_seconds(1), NULL, NULL);
+               rc = l_wait_event(desc->bd_waitq,
+                                 !ptlrpc_server_bulk_active(desc), &lwi);
+               if (rc == 0)
+                       return;
+
+               LASSERT(rc == -ETIMEDOUT);
+               CWARN("Unexpectedly long timeout: desc %p\n", desc);
+       }
 }
 EXPORT_SYMBOL(ptlrpc_abort_bulk);
 #endif /* HAVE_SERVER_SUPPORT */
@@ -362,7 +359,7 @@ int ptlrpc_register_bulk(struct ptlrpc_request *req)
                                  LNET_UNLINK, LNET_INS_AFTER, &me_h);
                if (rc != 0) {
                        CERROR("%s: LNetMEAttach failed x"LPU64"/%d: rc = %d\n",
-                              desc->bd_export->exp_obd->obd_name, xid,
+                              desc->bd_import->imp_obd->obd_name, xid,
                               posted_md, rc);
                        break;
                }
@@ -372,7 +369,7 @@ int ptlrpc_register_bulk(struct ptlrpc_request *req)
                                  &desc->bd_mds[posted_md]);
                if (rc != 0) {
                        CERROR("%s: LNetMDAttach failed x"LPU64"/%d: rc = %d\n",
-                              desc->bd_export->exp_obd->obd_name, xid,
+                              desc->bd_import->imp_obd->obd_name, xid,
                               posted_md, rc);
                        rc2 = LNetMEUnlink(me_h);
                        LASSERT(rc2 == 0);
@@ -402,7 +399,7 @@ int ptlrpc_register_bulk(struct ptlrpc_request *req)
        /* Holler if peer manages to touch buffers before he knows the xid */
        if (desc->bd_md_count != total_md)
                CWARN("%s: Peer %s touched %d buffers while I registered\n",
-                     desc->bd_export->exp_obd->obd_name, libcfs_id2str(peer),
+                     desc->bd_import->imp_obd->obd_name, libcfs_id2str(peer),
                      total_md - desc->bd_md_count);
        spin_unlock(&desc->bd_lock);
 
@@ -424,18 +421,17 @@ EXPORT_SYMBOL(ptlrpc_register_bulk);
  */
 int ptlrpc_unregister_bulk(struct ptlrpc_request *req, int async)
 {
-        struct ptlrpc_bulk_desc *desc = req->rq_bulk;
-        cfs_waitq_t             *wq;
-        struct l_wait_info       lwi;
-        int                      rc;
-        ENTRY;
+       struct ptlrpc_bulk_desc *desc = req->rq_bulk;
+       struct l_wait_info       lwi;
+       int                      rc;
+       ENTRY;
 
-        LASSERT(!cfs_in_interrupt());     /* might sleep */
+       LASSERT(!in_interrupt());     /* might sleep */
 
-        /* Let's setup deadline for reply unlink. */
-        if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_BULK_UNLINK) &&
-            async && req->rq_bulk_deadline == 0)
-                req->rq_bulk_deadline = cfs_time_current_sec() + LONG_UNLINK;
+       /* Let's setup deadline for reply unlink. */
+       if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_BULK_UNLINK) &&
+           async && req->rq_bulk_deadline == 0)
+               req->rq_bulk_deadline = cfs_time_current_sec() + LONG_UNLINK;
 
        if (ptlrpc_client_bulk_active(req) == 0)        /* completed or */
                RETURN(1);                              /* never registered */
@@ -458,12 +454,11 @@ int ptlrpc_unregister_bulk(struct ptlrpc_request *req, int async)
         if (async)
                 RETURN(0);
 
-        if (req->rq_set != NULL)
-                wq = &req->rq_set->set_waitq;
-        else
-                wq = &req->rq_reply_waitq;
-
         for (;;) {
+               /* The wq argument is ignored by user-space wait_event macros */
+               wait_queue_head_t *wq = (req->rq_set != NULL) ?
+                                       &req->rq_set->set_waitq :
+                                       &req->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),
@@ -581,7 +576,8 @@ int ptlrpc_send_reply(struct ptlrpc_request *req, int flags)
                 req->rq_type = PTL_RPC_MSG_REPLY;
 
         lustre_msg_set_type(req->rq_repmsg, req->rq_type);
-        lustre_msg_set_status(req->rq_repmsg, req->rq_status);
+       lustre_msg_set_status(req->rq_repmsg,
+                             ptlrpc_status_hton(req->rq_status));
         lustre_msg_set_opc(req->rq_repmsg,
                 req->rq_reqmsg ? lustre_msg_get_opc(req->rq_reqmsg) : 0);
 
@@ -677,7 +673,8 @@ int ptl_send_rpc(struct ptlrpc_request *request, int noreply)
         struct ptlrpc_connection *connection;
         lnet_handle_me_t  reply_me_h;
         lnet_md_t         reply_md;
-        struct obd_device *obd = request->rq_import->imp_obd;
+       struct obd_import *imp = request->rq_import;
+       struct obd_device *obd = imp->imp_obd;
         ENTRY;
 
         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_DROP_RPC))
@@ -689,30 +686,42 @@ int ptl_send_rpc(struct ptlrpc_request *request, int noreply)
         /* If this is a re-transmit, we're required to have disengaged
          * cleanly from the previous attempt */
         LASSERT(!request->rq_receiving_reply);
-
-        if (request->rq_import->imp_obd &&
-            request->rq_import->imp_obd->obd_fail) {
-                CDEBUG(D_HA, "muting rpc for failed imp obd %s\n",
-                       request->rq_import->imp_obd->obd_name);
-                /* this prevents us from waiting in ptlrpc_queue_wait */
-                request->rq_err = 1;
+       LASSERT(!((lustre_msg_get_flags(request->rq_reqmsg) & MSG_REPLAY) &&
+               (imp->imp_state == LUSTRE_IMP_FULL)));
+
+       if (unlikely(obd != NULL && obd->obd_fail)) {
+               CDEBUG(D_HA, "muting rpc for failed imp obd %s\n",
+                       obd->obd_name);
+               /* this prevents us from waiting in ptlrpc_queue_wait */
+               spin_lock(&request->rq_lock);
+               request->rq_err = 1;
+               spin_unlock(&request->rq_lock);
                 request->rq_status = -ENODEV;
                 RETURN(-ENODEV);
         }
 
-        connection = request->rq_import->imp_connection;
-
-        lustre_msg_set_handle(request->rq_reqmsg,
-                              &request->rq_import->imp_remote_handle);
-        lustre_msg_set_type(request->rq_reqmsg, PTL_RPC_MSG_REQUEST);
-        lustre_msg_set_conn_cnt(request->rq_reqmsg,
-                                request->rq_import->imp_conn_cnt);
-        lustre_msghdr_set_flags(request->rq_reqmsg,
-                                request->rq_import->imp_msghdr_flags);
-
-        if (request->rq_resend)
-                lustre_msg_add_flags(request->rq_reqmsg, MSG_RESENT);
-
+       connection = imp->imp_connection;
+
+       lustre_msg_set_handle(request->rq_reqmsg,
+                             &imp->imp_remote_handle);
+       lustre_msg_set_type(request->rq_reqmsg, PTL_RPC_MSG_REQUEST);
+       lustre_msg_set_conn_cnt(request->rq_reqmsg,
+                               imp->imp_conn_cnt);
+       lustre_msghdr_set_flags(request->rq_reqmsg,
+                               imp->imp_msghdr_flags);
+
+       /** For enabled AT all request should have AT_SUPPORT in the
+        * FULL import state when OBD_CONNECT_AT is set */
+       LASSERT(AT_OFF || imp->imp_state != LUSTRE_IMP_FULL ||
+               (imp->imp_msghdr_flags & MSGHDR_AT_SUPPORT) ||
+               !(imp->imp_connect_data.ocd_connect_flags &
+               OBD_CONNECT_AT));
+
+       if (request->rq_resend) {
+               lustre_msg_add_flags(request->rq_reqmsg, MSG_RESENT);
+               if (request->rq_resend_cb != NULL)
+                       request->rq_resend_cb(request, &request->rq_async_args);
+       }
         if (request->rq_memalloc)
                 mpflag = cfs_memory_pressure_get_and_set();
 
@@ -737,7 +746,9 @@ int ptl_send_rpc(struct ptlrpc_request *request, int noreply)
                         if (rc) {
                                 /* this prevents us from looping in
                                  * ptlrpc_queue_wait */
-                                request->rq_err = 1;
+                               spin_lock(&request->rq_lock);
+                               request->rq_err = 1;
+                               spin_unlock(&request->rq_lock);
                                 request->rq_status = rc;
                                 GOTO(cleanup_bulk, rc);
                         }
@@ -757,18 +768,18 @@ int ptl_send_rpc(struct ptlrpc_request *request, int noreply)
         }
 
        spin_lock(&request->rq_lock);
-        /* If the MD attach succeeds, there _will_ be a reply_in callback */
-        request->rq_receiving_reply = !noreply;
-        /* We are responsible for unlinking the reply buffer */
-        request->rq_must_unlink = !noreply;
-        /* Clear any flags that may be present from previous sends. */
+       /* We are responsible for unlinking the reply buffer */
+       request->rq_reply_unlinked = noreply;
+       request->rq_receiving_reply = !noreply;
+       /* Clear any flags that may be present from previous sends. */
+       request->rq_req_unlinked = 0;
         request->rq_replied = 0;
         request->rq_err = 0;
         request->rq_timedout = 0;
         request->rq_net_err = 0;
         request->rq_resend = 0;
         request->rq_restart = 0;
-        request->rq_reply_truncate = 0;
+       request->rq_reply_truncated = 0;
        spin_unlock(&request->rq_lock);
 
         if (!noreply) {
@@ -783,8 +794,8 @@ int ptl_send_rpc(struct ptlrpc_request *request, int noreply)
                 reply_md.user_ptr  = &request->rq_reply_cbid;
                 reply_md.eq_handle = ptlrpc_eq_h;
 
-                /* We must see the unlink callback to unset rq_must_unlink,
-                   so we can't auto-unlink */
+               /* We must see the unlink callback to set rq_reply_unlinked,
+                * so we can't auto-unlink */
                 rc = LNetMDAttach(reply_me_h, reply_md, LNET_RETAIN,
                                   &request->rq_reply_md_h);
                 if (rc != 0) {
@@ -805,20 +816,20 @@ int ptl_send_rpc(struct ptlrpc_request *request, int noreply)
 
         /* add references on request for request_out_callback */
         ptlrpc_request_addref(request);
-        if (obd->obd_svc_stats != NULL)
-                lprocfs_counter_add(obd->obd_svc_stats, PTLRPC_REQACTIVE_CNTR,
-                        cfs_atomic_read(&request->rq_import->imp_inflight));
+       if (obd != NULL && obd->obd_svc_stats != NULL)
+               lprocfs_counter_add(obd->obd_svc_stats, PTLRPC_REQACTIVE_CNTR,
+                       atomic_read(&imp->imp_inflight));
 
-        OBD_FAIL_TIMEOUT(OBD_FAIL_PTLRPC_DELAY_SEND, request->rq_timeout + 5);
+       OBD_FAIL_TIMEOUT(OBD_FAIL_PTLRPC_DELAY_SEND, request->rq_timeout + 5);
 
-        cfs_gettimeofday(&request->rq_arrival_time);
-        request->rq_sent = cfs_time_current_sec();
-        /* We give the server rq_timeout secs to process the req, and
-           add the network latency for our local timeout. */
+       do_gettimeofday(&request->rq_sent_tv);
+       request->rq_sent = cfs_time_current_sec();
+       /* We give the server rq_timeout secs to process the req, and
+          add the network latency for our local timeout. */
         request->rq_deadline = request->rq_sent + request->rq_timeout +
                 ptlrpc_at_get_net_latency(request);
 
-        ptlrpc_pinger_sending_on_import(request->rq_import);
+       ptlrpc_pinger_sending_on_import(imp);
 
         DEBUG_REQ(D_INFO, request, "send flg=%x",
                   lustre_msg_get_flags(request->rq_reqmsg));
@@ -828,9 +839,10 @@ int ptl_send_rpc(struct ptlrpc_request *request, int noreply)
                           connection,
                           request->rq_request_portal,
                           request->rq_xid, 0);
-        if (rc == 0)
-                GOTO(out, rc);
+       if (likely(rc == 0))
+               GOTO(out, rc);
 
+       request->rq_req_unlinked = 1;
         ptlrpc_req_finished(request);
         if (noreply)
                 GOTO(out, rc);