LU-13814 clio: Add cl_dio_submit_rw

This pushes the queue handling to the cl_io_submit_rw
level.

Signed-off-by: Patrick Farrell <patrick.farrell@oracle.com>
Change-Id: I38ca5ff2f497147e8246ebc43ff109eecbd127bb
Reviewed-on: https://review.whamcloud.com/c/fs/lustre-release/+/52104
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>
Reviewed-by: Qian Yingjin <qian@ddn.com>
Reviewed-by: Marc Vef <mvef@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/include/cl_object.h b/lustre/include/cl_object.h
index b8987bd..b075c58 100644 (file)
--- a/lustre/include/cl_object.h
+++ b/lustre/include/cl_object.h
@@ -2324,6 +2324,8 @@ int cl_lock_enqueue(const struct lu_env *env, struct cl_io *io,
                    struct cl_lock *lock, struct cl_sync_io *anchor);
 void cl_lock_cancel(const struct lu_env *env, struct cl_lock *lock);
 
+struct cl_dio_pages;
+
 /* cl_io */
 int   cl_io_init(const struct lu_env *env, struct cl_io *io,
                 enum cl_io_type iot, struct cl_object *obj);
@@ -2344,6 +2346,8 @@ int   cl_io_lock_add(const struct lu_env *env, struct cl_io *io,
                     struct cl_io_lock_link *link);
 int   cl_io_lock_alloc_add(const struct lu_env *env, struct cl_io *io,
                           struct cl_lock_descr *descr);
+int   cl_dio_submit_rw    (const struct lu_env *env, struct cl_io *io,
+                          enum cl_req_type iot, struct cl_dio_pages *cdp);
 int   cl_io_submit_rw(const struct lu_env *env, struct cl_io *io,
                      enum cl_req_type iot, struct cl_2queue *queue);
 int   cl_io_submit_sync(const struct lu_env *env, struct cl_io *io,
@@ -2462,7 +2466,6 @@ void cl_req_attr_set(const struct lu_env *env, struct cl_object *obj,
 struct cl_sync_io;
 struct cl_dio_aio;
 struct cl_sub_dio;
-struct cl_dio_pages;
 
 typedef void (cl_sync_io_end_t)(const struct lu_env *, struct cl_sync_io *);
 

diff --git a/lustre/llite/rw26.c b/lustre/llite/rw26.c
index bfcaa24..2525d04 100644 (file)
--- a/lustre/llite/rw26.c
+++ b/lustre/llite/rw26.c
@@ -364,11 +364,9 @@ ll_direct_rw_pages(const struct lu_env *env, struct cl_io *io, size_t size,
        struct cl_dio_pages *cdp = &sdio->csd_dio_pages;
        struct cl_sync_io *anchor = &sdio->csd_sync;
        struct cl_object *obj = io->ci_obj;
-       struct cl_2queue *queue = NULL;
        struct cl_page *page;
        int iot = rw == READ ? CRT_READ : CRT_WRITE;
        loff_t offset = cdp->cdp_file_offset;
-       int io_pages = 0;
        ssize_t rc = 0;
        int i = 0;
 
@@ -416,7 +414,6 @@ ll_direct_rw_pages(const struct lu_env *env, struct cl_io *io, size_t size,
                 */
                if (from != 0 || to != PAGE_SIZE)
                        cl_page_clip(env, page, from, to);
-               ++io_pages;
                i++;
 
                offset += to - from;
@@ -428,38 +425,21 @@ ll_direct_rw_pages(const struct lu_env *env, struct cl_io *io, size_t size,
        LASSERT(i == cdp->cdp_count);
        LASSERT(size == 0);
 
-       cl_dio_pages_2queue(cdp);
-       queue = &cdp->cdp_queue;
-       atomic_add(io_pages, &anchor->csi_sync_nr);
+       atomic_add(cdp->cdp_count, &anchor->csi_sync_nr);
        /*
         * Avoid out-of-order execution of adding inflight
         * modifications count and io submit.
         */
        smp_mb();
-       rc = cl_io_submit_rw(env, io, iot, queue);
-       /* pages must be off the queue when they're freed */
-       if (rc == 0) {
-               while (queue->c2_qout.pl_nr > 0) {
-                       page = cl_page_list_first(&queue->c2_qout);
-                       cl_page_list_del(env, &queue->c2_qout, page,
-                                                false);
-               }
-       } else {
-               atomic_add(-queue->c2_qin.pl_nr,
+       rc = cl_dio_submit_rw(env, io, iot, cdp);
+       if (rc != 0) {
+               atomic_add(-cdp->cdp_count,
                           &anchor->csi_sync_nr);
                for (i = 0; i < cdp->cdp_count; i++) {
                        page = cdp->cdp_cl_pages[i];
                        page->cp_sync_io = NULL;
                }
        }
-       /* handle partially submitted reqs */
-       if (queue->c2_qin.pl_nr > 0) {
-               CERROR(DFID " failed to submit %d dio pages: %zd\n",
-                      PFID(lu_object_fid(&obj->co_lu)),
-                      queue->c2_qin.pl_nr, rc);
-               if (rc == 0)
-                       rc = -EIO;
-       }
 
 out:
        /* cleanup of the page array is handled by cl_sub_dio_end, so there's

diff --git a/lustre/obdclass/cl_io.c b/lustre/obdclass/cl_io.c
index 826d4a0..ea3f456 100644 (file)
--- a/lustre/obdclass/cl_io.c
+++ b/lustre/obdclass/cl_io.c
@@ -669,6 +669,39 @@ void cl_io_extent_release(const struct lu_env *env, struct cl_io *io)
 }
 EXPORT_SYMBOL(cl_io_extent_release);
 
+int cl_dio_submit_rw(const struct lu_env *env, struct cl_io *io,
+                    enum cl_req_type crt, struct cl_dio_pages *cdp)
+{
+       const struct cl_io_slice *scan;
+       struct cl_2queue *queue;
+       int result = 0;
+
+       ENTRY;
+
+       cl_dio_pages_2queue(cdp);
+       queue = &cdp->cdp_queue;
+
+       list_for_each_entry(scan, &io->ci_layers, cis_linkage) {
+               if (scan->cis_iop->cio_submit == NULL)
+                       continue;
+               result = scan->cis_iop->cio_submit(env, io, scan, crt, queue);
+               if (result != 0)
+                       break;
+       }
+       /*
+        * If ->cio_submit() failed, no pages were sent.
+        */
+       LASSERT(ergo(result != 0, list_empty(&queue->c2_qout.pl_pages)));
+       while (queue->c2_qout.pl_nr > 0) {
+               struct cl_page *page;
+
+               page = cl_page_list_first(&queue->c2_qout);
+               cl_page_list_del(env, &queue->c2_qout, page, false);
+       }
+       RETURN(result);
+}
+EXPORT_SYMBOL(cl_dio_submit_rw);
+
 /**
  * cl_io_submit_rw() - Submits a list of pages for immediate IO.
  * @env: execution environment