LU-911 ost: ost_handle() to refill environment on every requests

[fs/lustre-release.git] / lustre / ost / ost_handler.c

diff --git a/lustre/ost/ost_handler.c b/lustre/ost/ost_handler.c
index 52769e0..197381b 100644 (file)
--- a/lustre/ost/ost_handler.c
+++ b/lustre/ost/ost_handler.c
@@ -29,8 +29,7 @@
  * Copyright (c) 2001, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  *
- * Copyright (c) 2011 Whamcloud, Inc.
- *
+ * Copyright (c) 2011, 2012, Whamcloud, Inc.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -99,8 +98,8 @@ static int ost_validate_obdo(struct obd_export *exp, struct obdo *oa,
                 if (ioobj)
                         ioobj->ioo_seq = FID_SEQ_OST_MDT0;
         /* remove fid_seq_is_rsvd() after FID-on-OST allows SEQ > 9 */
-        } else if (oa == NULL ||
-                   !(fid_seq_is_rsvd(oa->o_seq) || fid_seq_is_idif(oa->o_seq))) {
+        } else if (oa == NULL || !(fid_seq_is_rsvd(oa->o_seq) ||
+                                   fid_seq_is_mdt0(oa->o_seq))) {
                 CERROR("%s: client %s sent invalid object "POSTID"\n",
                        exp->exp_obd->obd_name, obd_export_nid2str(exp),
                        oa ? oa->o_id : -1, oa ? oa->o_seq : -1);
@@ -784,7 +783,8 @@ static int ost_brw_read(struct ptlrpc_request *req, struct obd_trans_info *oti)
         /* Check if client was evicted while we were doing i/o before touching
            network */
         if (rc == 0) {
-                rc = target_bulk_io(exp, desc, &lwi);
+                if (likely(!CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CLIENT_BULK_CB2)))
+                        rc = target_bulk_io(exp, desc, &lwi);
                 no_reply = rc != 0;
         }
 
@@ -801,7 +801,7 @@ out_lock:
 out_tls:
         ost_tls_put(req);
 out_bulk:
-        if (desc)
+        if (desc && !CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CLIENT_BULK_CB2))
                 ptlrpc_free_bulk(desc);
 out:
         LASSERT(rc <= 0);
@@ -818,12 +818,25 @@ out:
         } else {
                 /* reply out callback would free */
                 ptlrpc_req_drop_rs(req);
-                CWARN("%s: ignoring bulk IO comm error with %s@%s id %s - "
-                      "client will retry\n",
-                      exp->exp_obd->obd_name,
-                      exp->exp_client_uuid.uuid,
-                      exp->exp_connection->c_remote_uuid.uuid,
-                      libcfs_id2str(req->rq_peer));
+                LCONSOLE_WARN("%s: Bulk IO read error with %s (at %s), "
+                              "client will retry: rc %d\n",
+                              exp->exp_obd->obd_name,
+                              obd_uuid2str(&exp->exp_client_uuid),
+                              obd_export_nid2str(exp), rc);
+        }
+        /* send a bulk after reply to simulate a network delay or reordering
+         * by a router */
+        if (unlikely(CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CLIENT_BULK_CB2))) {
+                cfs_waitq_t              waitq;
+                struct l_wait_info       lwi1;
+
+                CDEBUG(D_INFO, "reorder BULK\n");
+                cfs_waitq_init(&waitq);
+
+                lwi1 = LWI_TIMEOUT_INTR(cfs_time_seconds(3), NULL, NULL, NULL);
+                l_wait_event(waitq, 0, &lwi1);
+                rc = target_bulk_io(exp, desc, &lwi);
+                ptlrpc_free_bulk(desc);
         }
 
         RETURN(rc);
@@ -984,7 +997,7 @@ static int ost_brw_write(struct ptlrpc_request *req, struct obd_trans_info *oti)
         no_reply = rc != 0;
 
 skip_transfer:
-        if (unlikely(client_cksum != 0 && rc == 0)) {
+        if (client_cksum != 0 && rc == 0) {
                 static int cksum_counter;
                 repbody->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
                 repbody->oa.o_flags &= ~OBD_FL_CKSUM_ALL;
@@ -1025,7 +1038,7 @@ skip_transfer:
          */
         repbody->oa.o_valid &= ~(OBD_MD_FLMTIME | OBD_MD_FLATIME);
 
-        if (unlikely(client_cksum != server_cksum && rc == 0 &&  !mmap)) {
+        if (unlikely(client_cksum != server_cksum && rc == 0 && !mmap)) {
                 int  new_cksum = ost_checksum_bulk(desc, OST_WRITE, cksum_type);
                 char *msg;
                 char *via;
@@ -1110,12 +1123,11 @@ out:
         } else {
                 /* reply out callback would free */
                 ptlrpc_req_drop_rs(req);
-                CWARN("%s: ignoring bulk IO comm error with %s@%s id %s - "
-                      "client will retry\n",
-                      exp->exp_obd->obd_name,
-                      exp->exp_client_uuid.uuid,
-                      exp->exp_connection->c_remote_uuid.uuid,
-                      libcfs_id2str(req->rq_peer));
+                LCONSOLE_WARN("%s: Bulk IO write error with %s (at %s), "
+                              "client will retry: rc %d\n",
+                              exp->exp_obd->obd_name,
+                              obd_uuid2str(&exp->exp_client_uuid),
+                              obd_export_nid2str(exp), rc);
         }
         cfs_memory_pressure_clr();
         RETURN(rc);
@@ -1570,7 +1582,7 @@ static int ost_filter_recovery_request(struct ptlrpc_request *req,
                 RETURN(0);
 
         default:
-                DEBUG_REQ(D_ERROR, req, "not permitted during recovery");
+                DEBUG_REQ(D_WARNING, req, "not permitted during recovery");
                 *process = -EAGAIN;
                 RETURN(0);
         }
@@ -2047,6 +2059,12 @@ int ost_handle(struct ptlrpc_request *req)
         struct obd_device *obd = NULL;
         ENTRY;
 
+        /* OST module is kept between remounts, but the last reference
+         * to specific module (say, osd or ofd) kills all related keys
+         * from the environment. so we have to refill it until the root
+         * cause is fixed properly */
+        lu_env_refill(req->rq_svc_thread->t_env);
+
         LASSERT(current->journal_info == NULL);
 
         /* primordial rpcs don't affect server recovery */
@@ -2382,7 +2400,7 @@ static int ost_setup(struct obd_device *obd, struct lustre_cfg* lcfg)
         lprocfs_ost_init_vars(&lvars);
         lprocfs_obd_setup(obd, lvars.obd_vars);
 
-        cfs_sema_init(&ost->ost_health_sem, 1);
+        cfs_mutex_init(&ost->ost_health_mutex);
 
         if (oss_num_threads) {
                 /* If oss_num_threads is set, it is the min and the max. */
@@ -2499,13 +2517,13 @@ static int ost_cleanup(struct obd_device *obd)
         /* there is no recovery for OST OBD, all recovery is controlled by
          * obdfilter OBD */
         LASSERT(obd->obd_recovering == 0);
-        cfs_down(&ost->ost_health_sem);
+        cfs_mutex_lock(&ost->ost_health_mutex);
         ptlrpc_unregister_service(ost->ost_service);
         ptlrpc_unregister_service(ost->ost_create_service);
         ptlrpc_unregister_service(ost->ost_io_service);
         ost->ost_service = NULL;
         ost->ost_create_service = NULL;
-        cfs_up(&ost->ost_health_sem);
+        cfs_mutex_unlock(&ost->ost_health_mutex);
 
         lprocfs_obd_cleanup(obd);
 
@@ -2517,11 +2535,11 @@ static int ost_health_check(struct obd_device *obd)
         struct ost_obd *ost = &obd->u.ost;
         int rc = 0;
 
-        cfs_down(&ost->ost_health_sem);
+        cfs_mutex_lock(&ost->ost_health_mutex);
         rc |= ptlrpc_service_health_check(ost->ost_service);
         rc |= ptlrpc_service_health_check(ost->ost_create_service);
         rc |= ptlrpc_service_health_check(ost->ost_io_service);
-        cfs_up(&ost->ost_health_sem);
+        cfs_mutex_unlock(&ost->ost_health_mutex);
 
         /*
          * health_check to return 0 on healthy