LU-13799 llite: Adjust dio refcounting

[fs/lustre-release.git] / lustre / llite / rw26.c

diff --git a/lustre/llite/rw26.c b/lustre/llite/rw26.c
index 36d9c56..6fb6d53 100644 (file)
--- a/lustre/llite/rw26.c
+++ b/lustre/llite/rw26.c
@@ -367,7 +367,10 @@ ll_direct_rw_pages(const struct lu_env *env, struct cl_io *io, size_t size,
                         */
                        page->cp_inode = inode;
                }
-               cl_2queue_add(queue, page);
+               /* We keep the refcount from cl_page_find, so we don't need
+                * another one here
+                */
+               cl_2queue_add(queue, page, false);
                /*
                 * Set page clip to tell transfer formation engine
                 * that page has to be sent even if it is beyond KMS.
@@ -376,8 +379,6 @@ ll_direct_rw_pages(const struct lu_env *env, struct cl_io *io, size_t size,
                        cl_page_clip(env, page, 0, size);
                ++io_pages;
 
-               /* drop the reference count for cl_page_find */
-               cl_page_put(env, page);
                offset += page_size;
                size -= page_size;
        }
@@ -518,10 +519,19 @@ out:
        else
                vio->u.readwrite.vui_read += tot_bytes;
 
-       /* If async dio submission is not allowed, we must wait here. */
-       if (is_sync_kiocb(iocb) && !io->ci_parallel_dio) {
+       /* We cannot do async submission - for AIO or regular DIO - unless
+        * lockless because it causes us to release the lock early.
+        *
+        * There are also several circumstances in which we must disable
+        * parallel DIO, so we check if it is enabled.
+        *
+        * The check for "is_sync_kiocb" excludes AIO, which does not need to
+        * be disabled in these situations.
+        */
+       if (io->ci_dio_lock || (is_sync_kiocb(iocb) && !io->ci_parallel_dio)) {
                ssize_t rc2;
 
+               /* Wait here rather than doing async submission */
                rc2 = cl_sync_io_wait_recycle(env, &aio->cda_sync, 0, 0);
                if (result == 0 && rc2)
                        result = rc2;
@@ -864,7 +874,7 @@ static int ll_write_end(struct file *file, struct address_space *mapping,
                lcc->lcc_page = NULL; /* page will be queued */
 
                /* Add it into write queue */
-               cl_page_list_add(plist, page);
+               cl_page_list_add(plist, page, true);
                if (plist->pl_nr == 1) /* first page */
                        vio->u.readwrite.vui_from = from;
                else