LU-13814 osc: add osc_queue_dio_pages

For dio, we need to replace osc_queue_sync_pages with a
specialized function.  This adds that function and starts
specializing it to DIO only.

Test-Parameters: fortestonly
Test-Parameters: forjanitoronly
Signed-off-by: Patrick Farrell <pfarrell@whamcloud.com>
Change-Id: I1eb996578cf2fc4d758d959ffc7f9b48225374ce

diff --git a/lustre/include/lustre_osc.h b/lustre/include/lustre_osc.h
index 18f2e3e..f5745a0 100644 (file)
--- a/lustre/include/lustre_osc.h
+++ b/lustre/include/lustre_osc.h
@@ -569,6 +569,9 @@ int osc_teardown_async_page(const struct lu_env *env, struct osc_object *obj,
                            struct osc_page *ops);
 int osc_flush_async_page(const struct lu_env *env, struct cl_io *io,
                         struct osc_page *ops);
+int osc_queue_dio_pages(const struct lu_env *env, struct cl_io *io,
+                       struct osc_object *obj, struct list_head *list,
+                       int brw_flags);
 int osc_queue_sync_pages(const struct lu_env *env, struct cl_io *io,
                         struct osc_object *obj, struct list_head *list,
                         int brw_flags);

diff --git a/lustre/osc/osc_cache.c b/lustre/osc/osc_cache.c
index 3440eb1..7c251e4 100644 (file)
--- a/lustre/osc/osc_cache.c
+++ b/lustre/osc/osc_cache.c
@@ -2539,6 +2539,119 @@ out:
        return rc;
 }
 
+int osc_queue_dio_pages(const struct lu_env *env, struct cl_io *io,
+                       struct osc_object *obj, struct list_head *list,
+                       int brw_flags)
+{
+       struct client_obd *cli = osc_cli(obj);
+       struct osc_io *oio = osc_env_io(env);
+       struct osc_async_page *oap;
+       struct osc_extent *ext;
+       struct osc_lock *oscl;
+       int mppr = cli->cl_max_pages_per_rpc;
+       pgoff_t start = CL_PAGE_EOF;
+       bool can_merge = true;
+       int page_count = 0;
+       pgoff_t end = 0;
+
+       ENTRY;
+
+       list_for_each_entry(oap, list, oap_pending_item) {
+               struct osc_page *opg = oap2osc_page(oap);
+               pgoff_t index = osc_index(opg);
+
+               if (index > end)
+                       end = index;
+               if (index < start)
+                       start = index;
+               ++page_count;
+               mppr <<= (page_count > mppr);
+
+               if (unlikely(oap->oap_count < PAGE_SIZE))
+                       can_merge = false;
+       }
+
+       ext = osc_extent_alloc(obj);
+       if (ext == NULL) {
+               struct osc_async_page *tmp;
+
+               list_for_each_entry_safe(oap, tmp, list, oap_pending_item) {
+                       list_del_init(&oap->oap_pending_item);
+                       osc_ap_completion(env, cli, obj, oap, 0, -ENOMEM);
+               }
+               RETURN(-ENOMEM);
+       }
+
+       ext->oe_rw = !!(brw_flags & OBD_BRW_READ);
+       ext->oe_sync = 1;
+       ext->oe_no_merge = !can_merge;
+       ext->oe_urgent = 1;
+       ext->oe_start = start;
+       ext->oe_end = ext->oe_max_end = end;
+       ext->oe_obj = obj;
+       ext->oe_srvlock = !!(brw_flags & OBD_BRW_SRVLOCK);
+       ext->oe_ndelay = !!(brw_flags & OBD_BRW_NDELAY);
+       ext->oe_dio = true;
+       oscl = oio->oi_write_osclock ? : oio->oi_read_osclock;
+       if (oscl && oscl->ols_dlmlock != NULL) {
+               ext->oe_dlmlock = LDLM_LOCK_GET(oscl->ols_dlmlock);
+               lu_ref_add(&ext->oe_dlmlock->l_reference, "osc_extent", ext);
+       }
+       if (!ext->oe_rw) { /* direct io write */
+               int grants;
+               int ppc;
+
+               ppc = 1 << (cli->cl_chunkbits - PAGE_SHIFT);
+               grants = cli->cl_grant_extent_tax;
+               grants += (1 << cli->cl_chunkbits) *
+                       ((page_count + ppc - 1) / ppc);
+
+               CDEBUG(D_CACHE, "requesting %d bytes grant\n", grants);
+               spin_lock(&cli->cl_loi_list_lock);
+               if (osc_reserve_grant(cli, grants) == 0) {
+                       list_for_each_entry(oap, list, oap_pending_item) {
+                               osc_consume_write_grant(cli,
+                                                       &oap->oap_brw_page);
+                       }
+                       atomic_long_add(page_count, &obd_dirty_pages);
+                       osc_unreserve_grant_nolock(cli, grants, 0);
+                       ext->oe_grants = grants;
+               } else {
+                       /* We cannot report ENOSPC correctly if we do parallel
+                        * DIO (async RPC submission), so turn off parallel dio
+                        * if there is not sufficient grant available.  This
+                        * makes individual RPCs synchronous.
+                        */
+                       io->ci_parallel_dio = false;
+                       CDEBUG(D_CACHE,
+                       "not enough grant available, switching to sync for this i/o\n");
+               }
+               spin_unlock(&cli->cl_loi_list_lock);
+               osc_update_next_shrink(cli);
+       }
+
+       ext->oe_is_rdma_only = !!(brw_flags & OBD_BRW_RDMA_ONLY);
+       ext->oe_nr_pages = page_count;
+       ext->oe_mppr = mppr;
+       list_splice_init(list, &ext->oe_pages);
+       ext->oe_layout_version = io->ci_layout_version;
+
+       osc_object_lock(obj);
+       /* Reuse the initial refcount for RPC, don't drop it */
+       osc_extent_state_set(ext, OES_LOCK_DONE);
+       if (!ext->oe_rw) { /* write */
+               list_add_tail(&ext->oe_link, &obj->oo_urgent_exts);
+               osc_update_pending(obj, OBD_BRW_WRITE, page_count);
+       } else {
+               list_add_tail(&ext->oe_link, &obj->oo_reading_exts);
+               osc_update_pending(obj, OBD_BRW_READ, page_count);
+       }
+       osc_object_unlock(obj);
+
+       osc_io_unplug_async(env, cli, obj);
+       RETURN(0);
+}
+
 int osc_queue_sync_pages(const struct lu_env *env, struct cl_io *io,
                         struct osc_object *obj, struct list_head *list,
                         int brw_flags)

diff --git a/lustre/osc/osc_io.c b/lustre/osc/osc_io.c
index 434e81c..b515187 100644 (file)
--- a/lustre/osc/osc_io.c
+++ b/lustre/osc/osc_io.c
@@ -331,8 +331,8 @@ int __osc_dio_submit(const struct lu_env *env, struct cl_io *io,
                }
 
                if (sync_queue) {
-                       result = osc_queue_sync_pages(env, top_io, osc, &list,
-                                                     brw_flags);
+                       result = osc_queue_dio_pages(env, top_io, osc, &list,
+                                                    brw_flags);
                        if (result < 0)
                                break;
                        queued = 0;
@@ -341,8 +341,8 @@ int __osc_dio_submit(const struct lu_env *env, struct cl_io *io,
        }
 
        if (queued > 0)
-               result = osc_queue_sync_pages(env, top_io, osc, &list,
-                                             brw_flags);
+               result = osc_queue_dio_pages(env, top_io, osc, &list,
+                                            brw_flags);
 
        /* Update c/mtime for sync write. LU-7310 */
        if (crt == CRT_WRITE && qout->pl_nr > 0 && result == 0) {