LU-4198 clio: turn on lockless for some kind of IO

[fs/lustre-release.git] / lustre / osc / osc_io.c

diff --git a/lustre/osc/osc_io.c b/lustre/osc/osc_io.c
index b4a196f..03b4cf8 100644 (file)
--- a/lustre/osc/osc_io.c
+++ b/lustre/osc/osc_io.c
@@ -87,12 +87,12 @@ static int osc_io_read_ahead(const struct lu_env *env,
                        ldlm_lock_decref(&lockh, dlmlock->l_req_mode);
                }
 
-               ra->cra_rpc_size = osc_cli(osc)->cl_max_pages_per_rpc;
-               ra->cra_end = cl_index(osc2cl(osc),
-                                      dlmlock->l_policy_data.l_extent.end);
+               ra->cra_rpc_pages = osc_cli(osc)->cl_max_pages_per_rpc;
+               ra->cra_end_idx = cl_index(osc2cl(osc),
+                                          dlmlock->l_policy_data.l_extent.end);
                ra->cra_release = osc_read_ahead_release;
                ra->cra_cbdata = dlmlock;
-               if (ra->cra_end != CL_PAGE_EOF)
+               if (ra->cra_end_idx != CL_PAGE_EOF)
                        ra->cra_contention = true;
                result = 0;
        }
@@ -116,7 +116,7 @@ int osc_io_submit(const struct lu_env *env, const struct cl_io_slice *ios,
        struct osc_object *osc  = NULL; /* to keep gcc happy */
        struct osc_page   *opg;
        struct cl_io      *io;
-       struct list_head  list = LIST_HEAD_INIT(list);
+       LIST_HEAD(list);
 
        struct cl_page_list *qin      = &queue->c2_qin;
        struct cl_page_list *qout     = &queue->c2_qout;
@@ -143,6 +143,10 @@ int osc_io_submit(const struct lu_env *env, const struct cl_io_slice *ios,
        if (crt == CRT_READ && ios->cis_io->ci_ndelay)
                brw_flags |= OBD_BRW_NDELAY;
 
+       page = cl_page_list_first(qin);
+       if (page->cp_type == CPT_TRANSIENT)
+               brw_flags |= OBD_BRW_NOCACHE;
+
         /*
          * NOTE: here @page is a top-level page. This is done to avoid
          *       creation of sub-page-list.
@@ -368,15 +372,30 @@ int osc_io_commit_async(const struct lu_env *env,
 }
 EXPORT_SYMBOL(osc_io_commit_async);
 
+static bool osc_import_not_healthy(struct obd_import *imp)
+{
+       return imp->imp_invalid || imp->imp_deactive ||
+              !(imp->imp_state == LUSTRE_IMP_FULL ||
+                imp->imp_state == LUSTRE_IMP_IDLE);
+}
+
 int osc_io_iter_init(const struct lu_env *env, const struct cl_io_slice *ios)
 {
        struct osc_object *osc = cl2osc(ios->cis_obj);
        struct obd_import *imp = osc_cli(osc)->cl_import;
        struct osc_io *oio = osc_env_io(env);
        int rc = -EIO;
+       ENTRY;
 
        spin_lock(&imp->imp_lock);
-       if (likely(!imp->imp_invalid)) {
+       /**
+        * check whether this OSC device is available for non-delay read,
+        * fast switching mirror if we haven't tried all mirrors.
+        */
+       if (ios->cis_io->ci_type == CIT_READ && ios->cis_io->ci_ndelay &&
+           !ios->cis_io->ci_tried_all_mirrors && osc_import_not_healthy(imp)) {
+               rc = -EWOULDBLOCK;
+       } else if (likely(!imp->imp_invalid)) {
                atomic_inc(&osc->oo_nr_ios);
                oio->oi_is_active = 1;
                rc = 0;
@@ -386,7 +405,7 @@ int osc_io_iter_init(const struct lu_env *env, const struct cl_io_slice *ios)
        if (cfs_capable(CFS_CAP_SYS_RESOURCE))
                oio->oi_cap_sys_resource = 1;
 
-       return rc;
+       RETURN(rc);
 }
 EXPORT_SYMBOL(osc_io_iter_init);