[lustre-devel] [PATCH 04/20] lustre: osc: take ldlm lock when queue sync pages

[James Simmons]

From: Bobi Jam <bobijam at whamcloud.com>

osc_queue_sync_pages() add osc_extent to osc_object's IO extent
list without taking ldlm locks, and then it calls
osc_io_unplug_async() to queue the IO work for the client.

This patch make sync page queuing take ldlm lock in the
osc_extent.

WC-bug-id: https://jira.whamcloud.com/browse/LU-16160
Lustre-commit: 67aca1fcc6bed2079 ("LU-16160 osc: take ldlm lock when queue sync pages")
Signed-off-by: Bobi Jam <bobijam at whamcloud.com>
Reviewed-on: https://review.whamcloud.com/48557
Reviewed-by: Mikhail Pershin <mpershin at whamcloud.com>
Reviewed-by: Alex Zhuravlev <bzzz at whamcloud.com>
Reviewed-by: Oleg Drokin <green at whamcloud.com>
Signed-off-by: James Simmons <jsimmons at infradead.org>
---
 fs/lustre/include/lustre_osc.h |  3 +++
 fs/lustre/mdc/mdc_dev.c        |  3 +++
 fs/lustre/osc/osc_cache.c      |  7 +++++++
 fs/lustre/osc/osc_io.c         |  1 +
 fs/lustre/osc/osc_lock.c       | 19 +++++++++++++++++++
 5 files changed, 33 insertions(+)

diff --git a/fs/lustre/include/lustre_osc.h b/fs/lustre/include/lustre_osc.h
index 89f02c51dbb3..323eeba3da42 100644
--- a/fs/lustre/include/lustre_osc.h
+++ b/fs/lustre/include/lustre_osc.h
@@ -157,6 +157,7 @@ struct osc_io {
 
 	/* write osc_lock for this IO, used by osc_extent_find(). */
 	struct osc_lock		*oi_write_osclock;
+	struct osc_lock		*oi_read_osclock;
 	struct obdo		oi_oa;
 	struct osc_async_cbargs {
 		bool			opc_rpc_sent;
@@ -724,6 +725,8 @@ int osc_lock_enqueue_wait(const struct lu_env *env, struct osc_object *obj,
 			  struct osc_lock *oscl);
 void osc_lock_set_writer(const struct lu_env *env, const struct cl_io *io,
 			 struct cl_object *obj, struct osc_lock *oscl);
+void osc_lock_set_reader(const struct lu_env *env, const struct cl_io *io,
+			 struct cl_object *obj, struct osc_lock *oscl);
 int osc_lock_print(const struct lu_env *env, void *cookie,
 		   lu_printer_t p, const struct cl_lock_slice *slice);
 void osc_lock_cancel(const struct lu_env *env,
diff --git a/fs/lustre/mdc/mdc_dev.c b/fs/lustre/mdc/mdc_dev.c
index fd0e36225c13..2fd137d2fa4b 100644
--- a/fs/lustre/mdc/mdc_dev.c
+++ b/fs/lustre/mdc/mdc_dev.c
@@ -966,6 +966,9 @@ int mdc_lock_init(const struct lu_env *env, struct cl_object *obj,
 
 	if (io->ci_type == CIT_WRITE || cl_io_is_mkwrite(io))
 		osc_lock_set_writer(env, io, obj, ols);
+	else if (io->ci_type == CIT_READ ||
+		 (io->ci_type == CIT_FAULT && !io->u.ci_fault.ft_mkwrite))
+		osc_lock_set_reader(env, io, obj, ols);
 
 	LDLM_DEBUG_NOLOCK("lock %p, mdc lock %p, flags %llx\n",
 			  lock, ols, ols->ols_flags);
diff --git a/fs/lustre/osc/osc_cache.c b/fs/lustre/osc/osc_cache.c
index b6f0cdb92bdc..36fec837d93e 100644
--- a/fs/lustre/osc/osc_cache.c
+++ b/fs/lustre/osc/osc_cache.c
@@ -2655,6 +2655,7 @@ 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)
 {
+	struct osc_io *oio = osc_env_io(env);
 	struct client_obd *cli = osc_cli(obj);
 	struct osc_extent *ext;
 	struct osc_async_page *oap;
@@ -2663,6 +2664,7 @@ int osc_queue_sync_pages(const struct lu_env *env, struct cl_io *io,
 	bool can_merge = true;
 	pgoff_t start = CL_PAGE_EOF;
 	pgoff_t end = 0;
+	struct osc_lock *oscl;
 
 	list_for_each_entry(oap, list, oap_pending_item) {
 		struct osc_page *opg = oap2osc_page(oap);
@@ -2703,6 +2705,11 @@ int osc_queue_sync_pages(const struct lu_env *env, struct cl_io *io,
 	ext->oe_srvlock = !!(brw_flags & OBD_BRW_SRVLOCK);
 	ext->oe_ndelay = !!(brw_flags & OBD_BRW_NDELAY);
 	ext->oe_dio = !!(brw_flags & OBD_BRW_NOCACHE);
+	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_dio && !ext->oe_rw) { /* direct io write */
 		int grants;
 		int ppc;
diff --git a/fs/lustre/osc/osc_io.c b/fs/lustre/osc/osc_io.c
index 4c9b3d2bb481..aa8f61d710b9 100644
--- a/fs/lustre/osc/osc_io.c
+++ b/fs/lustre/osc/osc_io.c
@@ -461,6 +461,7 @@ void osc_io_rw_iter_fini(const struct lu_env *env,
 		oio->oi_lru_reserved = 0;
 	}
 	oio->oi_write_osclock = NULL;
+	oio->oi_read_osclock = NULL;
 
 	osc_io_iter_fini(env, ios);
 }
diff --git a/fs/lustre/osc/osc_lock.c b/fs/lustre/osc/osc_lock.c
index c8f85020c8fe..dd109496d260 100644
--- a/fs/lustre/osc/osc_lock.c
+++ b/fs/lustre/osc/osc_lock.c
@@ -1178,6 +1178,22 @@ void osc_lock_set_writer(const struct lu_env *env, const struct cl_io *io,
 }
 EXPORT_SYMBOL(osc_lock_set_writer);
 
+void osc_lock_set_reader(const struct lu_env *env, const struct cl_io *io,
+			 struct cl_object *obj, struct osc_lock *oscl)
+{
+	struct osc_io *oio = osc_env_io(env);
+
+	if (!cl_object_same(io->ci_obj, obj))
+		return;
+
+	if (oscl->ols_glimpse || osc_lock_is_lockless(oscl))
+		return;
+
+	if (oio->oi_read_osclock == NULL)
+		oio->oi_read_osclock = oscl;
+}
+EXPORT_SYMBOL(osc_lock_set_reader);
+
 int osc_lock_init(const struct lu_env *env,
 		  struct cl_object *obj, struct cl_lock *lock,
 		  const struct cl_io *io)
@@ -1224,6 +1240,9 @@ int osc_lock_init(const struct lu_env *env,
 
 	if (io->ci_type == CIT_WRITE || cl_io_is_mkwrite(io))
 		osc_lock_set_writer(env, io, obj, oscl);
+	else if (io->ci_type == CIT_READ ||
+		 (io->ci_type == CIT_FAULT && !io->u.ci_fault.ft_mkwrite))
+		osc_lock_set_reader(env, io, obj, oscl);
 
 
 	LDLM_DEBUG_NOLOCK("lock %p, osc lock %p, flags %llx",
-- 
2.27.0