X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fobdclass%2Fcl_io.c;h=93a26578ea940520096cb0da266b7f6f6b967130;hb=d1dded6e28473d889a9b24b47cbc804f90dd2956;hp=602b5ac9681fadc836d0dc115f87103347f71265;hpb=80a9ca6385fc8df64fa9f7fbd8fa14dfb40a609f;p=fs%2Flustre-release.git

diff --git a/lustre/obdclass/cl_io.c b/lustre/obdclass/cl_io.c
index 602b5ac..93a2657 100644
--- a/lustre/obdclass/cl_io.c
+++ b/lustre/obdclass/cl_io.c
@@ -15,11 +15,7 @@
  *
  * You should have received a copy of the GNU General Public License
  * version 2 along with this program; If not, see
- * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
- *
- * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
- * CA 95054 USA or visit www.sun.com if you need additional information or
- * have any questions.
+ * http://www.gnu.org/licenses/gpl-2.0.html
  *
  * GPL HEADER END
  */
@@ -27,7 +23,7 @@
  * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  *
- * Copyright (c) 2011, 2015, Intel Corporation.
+ * Copyright (c) 2011, 2017, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -42,10 +38,11 @@
 #define DEBUG_SUBSYSTEM S_CLASS
 
 #include <linux/sched.h>
+#include <linux/list.h>
+#include <linux/list_sort.h>
 #include <obd_class.h>
 #include <obd_support.h>
 #include <lustre_fid.h>
-#include <libcfs/list.h>
 #include <cl_object.h>
 #include "cl_internal.h"
 
@@ -55,11 +52,6 @@
  *
  */
 
-#define cl_io_for_each(slice, io) \
-	list_for_each_entry((slice), &io->ci_layers, cis_linkage)
-#define cl_io_for_each_reverse(slice, io)                 \
-	list_for_each_entry_reverse((slice), &io->ci_layers, cis_linkage)
-
 static inline int cl_io_type_is_valid(enum cl_io_type type)
 {
         return CIT_READ <= type && type < CIT_OP_NR;
@@ -71,14 +63,6 @@ static inline int cl_io_is_loopable(const struct cl_io *io)
 }
 
 /**
- * Returns true iff there is an IO ongoing in the given environment.
- */
-int cl_io_is_going(const struct lu_env *env)
-{
-        return cl_env_info(env)->clt_current_io != NULL;
-}
-
-/**
  * cl_io invariant that holds at all times when exported cl_io_*() functions
  * are entered and left.
  */
@@ -103,7 +87,6 @@ static int cl_io_invariant(const struct cl_io *io)
 void cl_io_fini(const struct lu_env *env, struct cl_io *io)
 {
 	struct cl_io_slice    *slice;
-	struct cl_thread_info *info;
 
         LINVRNT(cl_io_type_is_valid(io->ci_type));
         LINVRNT(cl_io_invariant(io));
@@ -123,9 +106,6 @@ void cl_io_fini(const struct lu_env *env, struct cl_io *io)
                 slice->cis_io = NULL;
         }
         io->ci_state = CIS_FINI;
-        info = cl_env_info(env);
-        if (info->clt_current_io == io)
-                info->clt_current_io = NULL;
 
 	/* sanity check for layout change */
 	switch(io->ci_type) {
@@ -142,6 +122,7 @@ void cl_io_fini(const struct lu_env *env, struct cl_io *io)
 		/* Check ignore layout change conf */
 		LASSERT(ergo(io->ci_ignore_layout || !io->ci_verify_layout,
 				!io->ci_need_restart));
+	case CIT_GLIMPSE:
 		break;
 	case CIT_LADVISE:
 		break;
@@ -189,11 +170,8 @@ static int cl_io_init0(const struct lu_env *env, struct cl_io *io,
 int cl_io_sub_init(const struct lu_env *env, struct cl_io *io,
                    enum cl_io_type iot, struct cl_object *obj)
 {
-        struct cl_thread_info *info = cl_env_info(env);
-
         LASSERT(obj != cl_object_top(obj));
-        if (info->clt_current_io == NULL)
-                info->clt_current_io = io;
+
         return cl_io_init0(env, io, iot, obj);
 }
 EXPORT_SYMBOL(cl_io_sub_init);
@@ -211,13 +189,12 @@ EXPORT_SYMBOL(cl_io_sub_init);
 int cl_io_init(const struct lu_env *env, struct cl_io *io,
                enum cl_io_type iot, struct cl_object *obj)
 {
-        struct cl_thread_info *info = cl_env_info(env);
+	LASSERT(obj == cl_object_top(obj));
 
-        LASSERT(obj == cl_object_top(obj));
-        LASSERT(info->clt_current_io == NULL);
+	/* clear I/O restart from previous instance */
+	io->ci_need_restart = 0;
 
-        info->clt_current_io = io;
-        return cl_io_init0(env, io, iot, obj);
+	return cl_io_init0(env, io, iot, obj);
 }
 EXPORT_SYMBOL(cl_io_init);
 
@@ -227,75 +204,36 @@ EXPORT_SYMBOL(cl_io_init);
  * \pre iot == CIT_READ || iot == CIT_WRITE
  */
 int cl_io_rw_init(const struct lu_env *env, struct cl_io *io,
-                  enum cl_io_type iot, loff_t pos, size_t count)
+		  enum cl_io_type iot, loff_t pos, size_t count)
 {
-        LINVRNT(iot == CIT_READ || iot == CIT_WRITE);
-        LINVRNT(io->ci_obj != NULL);
-        ENTRY;
+	LINVRNT(iot == CIT_READ || iot == CIT_WRITE);
+	LINVRNT(io->ci_obj != NULL);
+	ENTRY;
 
-        LU_OBJECT_HEADER(D_VFSTRACE, env, &io->ci_obj->co_lu,
-                         "io range: %u ["LPU64", "LPU64") %u %u\n",
-                         iot, (__u64)pos, (__u64)pos + count,
-                         io->u.ci_rw.crw_nonblock, io->u.ci_wr.wr_append);
-        io->u.ci_rw.crw_pos    = pos;
-        io->u.ci_rw.crw_count  = count;
-        RETURN(cl_io_init(env, io, iot, io->ci_obj));
+	LU_OBJECT_HEADER(D_VFSTRACE, env, &io->ci_obj->co_lu,
+			 "io range: %u [%llu, %llu) %u %u\n",
+			 iot, (__u64)pos, (__u64)pos + count,
+			 io->u.ci_rw.crw_nonblock, io->u.ci_wr.wr_append);
+	io->u.ci_rw.crw_pos    = pos;
+	io->u.ci_rw.crw_count  = count;
+	RETURN(cl_io_init(env, io, iot, io->ci_obj));
 }
 EXPORT_SYMBOL(cl_io_rw_init);
 
-static int cl_lock_descr_sort(const struct cl_lock_descr *d0,
-                              const struct cl_lock_descr *d1)
+static int cl_lock_descr_cmp(void *priv,
+			     struct list_head *a, struct list_head *b)
 {
+	const struct cl_io_lock_link *l0 = list_entry(a, struct cl_io_lock_link,
+						      cill_linkage);
+	const struct cl_io_lock_link *l1 = list_entry(b, struct cl_io_lock_link,
+						      cill_linkage);
+	const struct cl_lock_descr *d0 = &l0->cill_descr;
+	const struct cl_lock_descr *d1 = &l1->cill_descr;
+
 	return lu_fid_cmp(lu_object_fid(&d0->cld_obj->co_lu),
 			  lu_object_fid(&d1->cld_obj->co_lu));
 }
 
-/*
- * Sort locks in lexicographical order of their (fid, start-offset) pairs.
- */
-static void cl_io_locks_sort(struct cl_io *io)
-{
-        int done = 0;
-
-        ENTRY;
-        /* hidden treasure: bubble sort for now. */
-        do {
-                struct cl_io_lock_link *curr;
-                struct cl_io_lock_link *prev;
-                struct cl_io_lock_link *temp;
-
-                done = 1;
-                prev = NULL;
-
-		list_for_each_entry_safe(curr, temp, &io->ci_lockset.cls_todo,
-					 cill_linkage) {
-			if (prev != NULL) {
-				switch (cl_lock_descr_sort(&prev->cill_descr,
-							   &curr->cill_descr)) {
-				case 0:
-					/*
-					 * IMPOSSIBLE:	Identical locks are
-					 *		already removed at
-					 *		this point.
-					 */
-				default:
-					LBUG();
-				case +1:
-					list_move_tail(&curr->cill_linkage,
-						       &prev->cill_linkage);
-					done = 0;
-					continue; /* don't change prev: it's
-						   * still "previous" */
-				case -1: /* already in order */
-					break;
-				}
-			}
-			prev = curr;
-		}
-	} while (!done);
-	EXIT;
-}
-
 static void cl_lock_descr_merge(struct cl_lock_descr *d0,
 				const struct cl_lock_descr *d1)
 {
@@ -370,15 +308,19 @@ int cl_io_lock(const struct lu_env *env, struct cl_io *io)
         LINVRNT(cl_io_invariant(io));
 
         ENTRY;
-        cl_io_for_each(scan, io) {
-                if (scan->cis_iop->op[io->ci_type].cio_lock == NULL)
-                        continue;
-                result = scan->cis_iop->op[io->ci_type].cio_lock(env, scan);
-                if (result != 0)
-                        break;
-        }
+	list_for_each_entry(scan, &io->ci_layers, cis_linkage) {
+		if (scan->cis_iop->op[io->ci_type].cio_lock == NULL)
+			continue;
+		result = scan->cis_iop->op[io->ci_type].cio_lock(env, scan);
+		if (result != 0)
+			break;
+	}
         if (result == 0) {
-                cl_io_locks_sort(io);
+		/*
+		 * Sort locks in lexicographical order of their (fid,
+		 * start-offset) pairs to avoid deadlocks.
+		 */
+		list_sort(NULL, &io->ci_lockset.cls_todo, cl_lock_descr_cmp);
                 result = cl_lockset_lock(env, io, &io->ci_lockset);
         }
         if (result != 0)
@@ -419,7 +361,7 @@ void cl_io_unlock(const struct lu_env *env, struct cl_io *io)
 			link->cill_fini(env, link);
 	}
 
-	cl_io_for_each_reverse(scan, io) {
+	list_for_each_entry_reverse(scan, &io->ci_layers, cis_linkage) {
 		if (scan->cis_iop->op[io->ci_type].cio_unlock != NULL)
 			scan->cis_iop->op[io->ci_type].cio_unlock(env, scan);
 	}
@@ -446,14 +388,14 @@ int cl_io_iter_init(const struct lu_env *env, struct cl_io *io)
 
         ENTRY;
         result = 0;
-        cl_io_for_each(scan, io) {
-                if (scan->cis_iop->op[io->ci_type].cio_iter_init == NULL)
-                        continue;
-                result = scan->cis_iop->op[io->ci_type].cio_iter_init(env,
-                                                                      scan);
-                if (result != 0)
-                        break;
-        }
+	list_for_each_entry(scan, &io->ci_layers, cis_linkage) {
+		if (scan->cis_iop->op[io->ci_type].cio_iter_init == NULL)
+			continue;
+		result = scan->cis_iop->op[io->ci_type].cio_iter_init(env,
+								      scan);
+		if (result != 0)
+			break;
+	}
         if (result == 0)
                 io->ci_state = CIS_IT_STARTED;
         RETURN(result);
@@ -467,19 +409,20 @@ EXPORT_SYMBOL(cl_io_iter_init);
  */
 void cl_io_iter_fini(const struct lu_env *env, struct cl_io *io)
 {
-        const struct cl_io_slice *scan;
+	const struct cl_io_slice *scan;
 
-        LINVRNT(cl_io_is_loopable(io));
-        LINVRNT(io->ci_state == CIS_UNLOCKED);
-        LINVRNT(cl_io_invariant(io));
+	LINVRNT(cl_io_is_loopable(io));
+	LINVRNT(io->ci_state <= CIS_IT_STARTED ||
+		io->ci_state > CIS_IO_FINISHED);
+	LINVRNT(cl_io_invariant(io));
 
-        ENTRY;
-        cl_io_for_each_reverse(scan, io) {
-                if (scan->cis_iop->op[io->ci_type].cio_iter_fini != NULL)
-                        scan->cis_iop->op[io->ci_type].cio_iter_fini(env, scan);
-        }
-        io->ci_state = CIS_IT_ENDED;
-        EXIT;
+	ENTRY;
+	list_for_each_entry_reverse(scan, &io->ci_layers, cis_linkage) {
+		if (scan->cis_iop->op[io->ci_type].cio_iter_fini != NULL)
+			scan->cis_iop->op[io->ci_type].cio_iter_fini(env, scan);
+	}
+	io->ci_state = CIS_IT_ENDED;
+	EXIT;
 }
 EXPORT_SYMBOL(cl_io_iter_fini);
 
@@ -488,25 +431,25 @@ EXPORT_SYMBOL(cl_io_iter_fini);
  */
 void cl_io_rw_advance(const struct lu_env *env, struct cl_io *io, size_t nob)
 {
-        const struct cl_io_slice *scan;
+	const struct cl_io_slice *scan;
 
-        LINVRNT(io->ci_type == CIT_READ || io->ci_type == CIT_WRITE ||
-                nob == 0);
-        LINVRNT(cl_io_is_loopable(io));
-        LINVRNT(cl_io_invariant(io));
+	ENTRY;
 
-        ENTRY;
+	LINVRNT(io->ci_type == CIT_READ || io->ci_type == CIT_WRITE ||
+		nob == 0);
+	LINVRNT(cl_io_is_loopable(io));
+	LINVRNT(cl_io_invariant(io));
 
-        io->u.ci_rw.crw_pos   += nob;
-        io->u.ci_rw.crw_count -= nob;
+	io->u.ci_rw.crw_pos   += nob;
+	io->u.ci_rw.crw_count -= nob;
 
-        /* layers have to be notified. */
-        cl_io_for_each_reverse(scan, io) {
-                if (scan->cis_iop->op[io->ci_type].cio_advance != NULL)
-                        scan->cis_iop->op[io->ci_type].cio_advance(env, scan,
-                                                                   nob);
-        }
-        EXIT;
+	/* layers have to be notified. */
+	list_for_each_entry_reverse(scan, &io->ci_layers, cis_linkage) {
+		if (scan->cis_iop->op[io->ci_type].cio_advance != NULL)
+			scan->cis_iop->op[io->ci_type].cio_advance(env, scan,
+								   nob);
+	}
+	EXIT;
 }
 
 /**
@@ -572,13 +515,13 @@ int cl_io_start(const struct lu_env *env, struct cl_io *io)
         ENTRY;
 
         io->ci_state = CIS_IO_GOING;
-        cl_io_for_each(scan, io) {
-                if (scan->cis_iop->op[io->ci_type].cio_start == NULL)
-                        continue;
-                result = scan->cis_iop->op[io->ci_type].cio_start(env, scan);
-                if (result != 0)
-                        break;
-        }
+	list_for_each_entry(scan, &io->ci_layers, cis_linkage) {
+		if (scan->cis_iop->op[io->ci_type].cio_start == NULL)
+			continue;
+		result = scan->cis_iop->op[io->ci_type].cio_start(env, scan);
+		if (result != 0)
+			break;
+	}
         if (result >= 0)
                 result = 0;
         RETURN(result);
@@ -598,11 +541,11 @@ void cl_io_end(const struct lu_env *env, struct cl_io *io)
         LINVRNT(cl_io_invariant(io));
         ENTRY;
 
-        cl_io_for_each_reverse(scan, io) {
-                if (scan->cis_iop->op[io->ci_type].cio_end != NULL)
-                        scan->cis_iop->op[io->ci_type].cio_end(env, scan);
-                /* TODO: error handling. */
-        }
+	list_for_each_entry_reverse(scan, &io->ci_layers, cis_linkage) {
+		if (scan->cis_iop->op[io->ci_type].cio_end != NULL)
+			scan->cis_iop->op[io->ci_type].cio_end(env, scan);
+		/* TODO: error handling. */
+	}
         io->ci_state = CIS_IO_FINISHED;
         EXIT;
 }
@@ -619,12 +562,14 @@ int cl_io_read_ahead(const struct lu_env *env, struct cl_io *io,
 	const struct cl_io_slice *scan;
 	int                       result = 0;
 
-	LINVRNT(io->ci_type == CIT_READ || io->ci_type == CIT_FAULT);
+	LINVRNT(io->ci_type == CIT_READ ||
+		io->ci_type == CIT_FAULT ||
+		io->ci_type == CIT_WRITE);
 	LINVRNT(io->ci_state == CIS_IO_GOING || io->ci_state == CIS_LOCKED);
 	LINVRNT(cl_io_invariant(io));
 	ENTRY;
 
-	cl_io_for_each(scan, io) {
+	list_for_each_entry(scan, &io->ci_layers, cis_linkage) {
 		if (scan->cis_iop->cio_read_ahead == NULL)
 			continue;
 
@@ -643,14 +588,14 @@ EXPORT_SYMBOL(cl_io_read_ahead);
  * \see cl_io_operations::cio_commit_async()
  */
 int cl_io_commit_async(const struct lu_env *env, struct cl_io *io,
-			struct cl_page_list *queue, int from, int to,
-			cl_commit_cbt cb)
+		       struct cl_page_list *queue, int from, int to,
+		       cl_commit_cbt cb)
 {
 	const struct cl_io_slice *scan;
 	int result = 0;
 	ENTRY;
 
-	cl_io_for_each(scan, io) {
+	list_for_each_entry(scan, &io->ci_layers, cis_linkage) {
 		if (scan->cis_iop->cio_commit_async == NULL)
 			continue;
 		result = scan->cis_iop->cio_commit_async(env, scan, queue,
@@ -679,7 +624,7 @@ int cl_io_submit_rw(const struct lu_env *env, struct cl_io *io,
 	int result = 0;
 	ENTRY;
 
-	cl_io_for_each(scan, io) {
+	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, scan, crt, queue);
@@ -711,7 +656,7 @@ int cl_io_submit_sync(const struct lu_env *env, struct cl_io *io,
 		pg->cp_sync_io = anchor;
 	}
 
-	cl_sync_io_init(anchor, queue->c2_qin.pl_nr, &cl_sync_io_end);
+	cl_sync_io_init(anchor, queue->c2_qin.pl_nr);
 	rc = cl_io_submit_rw(env, io, iot, queue);
 	if (rc == 0) {
 		/*
@@ -777,41 +722,47 @@ int cl_io_cancel(const struct lu_env *env, struct cl_io *io,
  */
 int cl_io_loop(const struct lu_env *env, struct cl_io *io)
 {
-        int result   = 0;
+	int result   = 0;
 
-        LINVRNT(cl_io_is_loopable(io));
-        ENTRY;
+	LINVRNT(cl_io_is_loopable(io));
+	ENTRY;
+
+	do {
+		size_t nob;
+
+		io->ci_continue = 0;
+		result = cl_io_iter_init(env, io);
+		if (result == 0) {
+			nob    = io->ci_nob;
+			result = cl_io_lock(env, io);
+			if (result == 0) {
+				/*
+				 * Notify layers that locks has been taken,
+				 * and do actual i/o.
+				 *
+				 *   - llite: kms, short read;
+				 *   - llite: generic_file_read();
+				 */
+				result = cl_io_start(env, io);
+				/*
+				 * Send any remaining pending
+				 * io, etc.
+				 *
+				 **   - llite: ll_rw_stats_tally.
+				 */
+				cl_io_end(env, io);
+				cl_io_unlock(env, io);
+				cl_io_rw_advance(env, io, io->ci_nob - nob);
+			}
+		}
+		cl_io_iter_fini(env, io);
+	} while (result == 0 && io->ci_continue);
+
+	if (result == -EWOULDBLOCK && io->ci_ndelay) {
+		io->ci_need_restart = 1;
+		result = 0;
+	}
 
-        do {
-                size_t nob;
-
-                io->ci_continue = 0;
-                result = cl_io_iter_init(env, io);
-                if (result == 0) {
-                        nob    = io->ci_nob;
-                        result = cl_io_lock(env, io);
-                        if (result == 0) {
-                                /*
-                                 * Notify layers that locks has been taken,
-                                 * and do actual i/o.
-                                 *
-                                 *   - llite: kms, short read;
-                                 *   - llite: generic_file_read();
-                                 */
-                                result = cl_io_start(env, io);
-                                /*
-                                 * Send any remaining pending
-                                 * io, etc.
-                                 *
-                                 *   - llite: ll_rw_stats_tally.
-                                 */
-                                cl_io_end(env, io);
-                                cl_io_unlock(env, io);
-                                cl_io_rw_advance(env, io, io->ci_nob - nob);
-                        }
-                }
-                cl_io_iter_fini(env, io);
-        } while (result == 0 && io->ci_continue);
 	if (result == 0)
 		result = io->ci_result;
 	RETURN(result < 0 ? result : 0);
@@ -828,20 +779,20 @@ EXPORT_SYMBOL(cl_io_loop);
  * \see cl_lock_slice_add(), cl_req_slice_add(), cl_page_slice_add()
  */
 void cl_io_slice_add(struct cl_io *io, struct cl_io_slice *slice,
-                     struct cl_object *obj,
-                     const struct cl_io_operations *ops)
+		     struct cl_object *obj,
+		     const struct cl_io_operations *ops)
 {
 	struct list_head *linkage = &slice->cis_linkage;
 
-        LASSERT((linkage->prev == NULL && linkage->next == NULL) ||
+	LASSERT((linkage->prev == NULL && linkage->next == NULL) ||
 		list_empty(linkage));
-        ENTRY;
+	ENTRY;
 
 	list_add_tail(linkage, &io->ci_layers);
-        slice->cis_io  = io;
-        slice->cis_obj = obj;
-        slice->cis_iop = ops;
-        EXIT;
+	slice->cis_io  = io;
+	slice->cis_obj = obj;
+	slice->cis_iop = ops;
+	EXIT;
 }
 EXPORT_SYMBOL(cl_io_slice_add);
 
@@ -1036,6 +987,7 @@ void cl_page_list_discard(const struct lu_env *env, struct cl_io *io,
 		cl_page_discard(env, io, page);
 	EXIT;
 }
+EXPORT_SYMBOL(cl_page_list_discard);
 
 /**
  * Initialize dual page queue.
@@ -1161,34 +1113,29 @@ void cl_req_attr_set(const struct lu_env *env, struct cl_object *obj,
 }
 EXPORT_SYMBOL(cl_req_attr_set);
 
-/* cl_sync_io_callback assumes the caller must call cl_sync_io_wait() to
- * wait for the IO to finish. */
-void cl_sync_io_end(const struct lu_env *env, struct cl_sync_io *anchor)
-{
-	wake_up_all(&anchor->csi_waitq);
-
-	/* it's safe to nuke or reuse anchor now */
-	atomic_set(&anchor->csi_barrier, 0);
-}
-EXPORT_SYMBOL(cl_sync_io_end);
-
 /**
- * Initialize synchronous io wait anchor
+ * Initialize synchronous io wait \a anchor for \a nr pages with optional
+ * \a end handler.
+ * \param anchor owned by caller, initialzied here.
+ * \param nr number of pages initally pending in sync.
+ * \param end optional callback sync_io completion, can be used to
+ *  trigger erasure coding, integrity, dedupe, or similar operation.
+ * \q end is called with a spinlock on anchor->csi_waitq.lock
  */
-void cl_sync_io_init(struct cl_sync_io *anchor, int nr,
-		     void (*end)(const struct lu_env *, struct cl_sync_io *))
+
+void cl_sync_io_init_notify(struct cl_sync_io *anchor, int nr,
+			    struct cl_dio_aio *aio, cl_sync_io_end_t *end)
 {
 	ENTRY;
 	memset(anchor, 0, sizeof(*anchor));
 	init_waitqueue_head(&anchor->csi_waitq);
 	atomic_set(&anchor->csi_sync_nr, nr);
-	atomic_set(&anchor->csi_barrier, nr > 0);
 	anchor->csi_sync_rc = 0;
 	anchor->csi_end_io = end;
-	LASSERT(end != NULL);
+	anchor->csi_aio = aio;
 	EXIT;
 }
-EXPORT_SYMBOL(cl_sync_io_init);
+EXPORT_SYMBOL(cl_sync_io_init_notify);
 
 /**
  * Wait until all IO completes. Transfer completion routine has to call
@@ -1197,33 +1144,30 @@ EXPORT_SYMBOL(cl_sync_io_init);
 int cl_sync_io_wait(const struct lu_env *env, struct cl_sync_io *anchor,
 		    long timeout)
 {
-	struct l_wait_info lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(timeout),
-						  NULL, NULL, NULL);
-	int rc;
+	int rc = 0;
 	ENTRY;
 
 	LASSERT(timeout >= 0);
 
-	rc = l_wait_event(anchor->csi_waitq,
-			  atomic_read(&anchor->csi_sync_nr) == 0,
-			  &lwi);
-	if (rc < 0) {
+	if (timeout > 0 &&
+	    wait_event_idle_timeout(anchor->csi_waitq,
+				    atomic_read(&anchor->csi_sync_nr) == 0,
+				    cfs_time_seconds(timeout)) == 0) {
+		rc = -ETIMEDOUT;
 		CERROR("IO failed: %d, still wait for %d remaining entries\n",
 		       rc, atomic_read(&anchor->csi_sync_nr));
+	}
 
-		lwi = (struct l_wait_info) { 0 };
-		(void)l_wait_event(anchor->csi_waitq,
-				   atomic_read(&anchor->csi_sync_nr) == 0,
-				   &lwi);
-	} else {
+	wait_event_idle(anchor->csi_waitq,
+			atomic_read(&anchor->csi_sync_nr) == 0);
+	if (!rc)
 		rc = anchor->csi_sync_rc;
-	}
+
+	/* We take the lock to ensure that cl_sync_io_note() has finished */
+	spin_lock(&anchor->csi_waitq.lock);
 	LASSERT(atomic_read(&anchor->csi_sync_nr) == 0);
+	spin_unlock(&anchor->csi_waitq.lock);
 
-	/* wait until cl_sync_io_note() has done wakeup */
-	while (unlikely(atomic_read(&anchor->csi_barrier) != 0)) {
-		cpu_relax();
-	}
 	RETURN(rc);
 }
 EXPORT_SYMBOL(cl_sync_io_wait);
@@ -1243,10 +1187,34 @@ void cl_sync_io_note(const struct lu_env *env, struct cl_sync_io *anchor,
 	 * IO.
 	 */
 	LASSERT(atomic_read(&anchor->csi_sync_nr) > 0);
-	if (atomic_dec_and_test(&anchor->csi_sync_nr)) {
-		LASSERT(anchor->csi_end_io != NULL);
-		anchor->csi_end_io(env, anchor);
-		/* Can't access anchor any more */
+	if (atomic_dec_and_lock(&anchor->csi_sync_nr,
+				&anchor->csi_waitq.lock)) {
+		struct cl_dio_aio *aio = NULL;
+
+		cl_sync_io_end_t *end_io = anchor->csi_end_io;
+
+		/*
+		 * Holding the lock across both the decrement and
+		 * the wakeup ensures cl_sync_io_wait() doesn't complete
+		 * before the wakeup completes and the contents of
+		 * of anchor become unsafe to access as the owner is free
+		 * to immediately reclaim anchor when cl_sync_io_wait()
+		 * completes.
+		 */
+		wake_up_all_locked(&anchor->csi_waitq);
+		if (end_io)
+			end_io(env, anchor);
+		if (anchor->csi_aio)
+			aio = anchor->csi_aio;
+
+		spin_unlock(&anchor->csi_waitq.lock);
+
+		/**
+		 * If anchor->csi_aio is set, we are responsible for freeing
+		 * memory here rather than when cl_sync_io_wait() completes.
+		 */
+		if (aio)
+			OBD_FREE_PTR(aio);
 	}
 	EXIT;
 }