RFC: [PATCH 0/6] lustre: Improve locking in lu_object.s

RFC: [PATCH 0/6] lustre: Improve locking in lu_object.s

[NeilBrown]

This is a followup to the patches that James recently posted
which addressed some locking issues in lu_object.c.
I said at the time that I thought the locking could be
improved further.  Here is the result of that thought.

The lu_keys_guard lock is now gone, replaced by a spin lock, an rwsem,
and some lockless code.
The lu_context_exit() function, which appears to have been a
performance problem at one point, is now completely lockless.

This is an RFC at this stage: I haven't included Greg on the email and
don't expect this to go upstream without some review - and maybe even
some testing.

Thanks,
NeilBrown


---

NeilBrown (6):
      staging: lustre: make key_set_version an atomic_t
      staging: lustre: use an rwsem instead of lu_key_initing_cnt.
      staging: lustre: remove locking from lu_context_exit()
      staging: lustre: use wait_event_var() in lu_context_key_degister()
      staging: lustre: remove lock from key register/degister
      staging: lustre: rename lu_keys_guard to lu_context_remembered_guard


 drivers/staging/lustre/lustre/include/lu_object.h  |    1 
 drivers/staging/lustre/lustre/obdclass/lu_object.c |  187 ++++++++------------
 2 files changed, 78 insertions(+), 110 deletions(-)

--
Signature

[PATCH 1/6] staging: lustre: make key_set_version an atomic_t

[NeilBrown]

As a first step to simplifying the locking in lu_object.c,
change key_set_version to an atomic_t.  This will mean
we don't need to hold a spinlock when incrementing.

It is clear that keys_fill() (and so lu_context_refill())
cannot race with itself as it holds no locks for the main
part of the work.  So updates to lc_version, and testing of
that value cannot need locking either.
So remove the locking from keys_fill() and lu_context_refill()
as there is no longer anything to protect.  The locking around
lu_keys_initing_cnt is preserved for now.

Also, don't increment when deregistering or quiescing a key.
key_set_version is *only* use to avoid filling new key values
if there have been no changes to the set of key.
Deregistering a key does not mean that we need to try filling
any new value, so the increment is pointless.

Finally, remove the refill loop in keys_fill().  If a key
is registered or revived while keys_fill() is running it must be safe
to ignore it just as we would if it was registered immediately
after keys_fill() ran.  The important thing is that the
keys_set_version stored in ctx->lc_version must be sampled
*before* those unseen keys were added.
So sample keys_set_version early.

Signed-off-by: NeilBrown <neilb@suse.com>
---
 drivers/staging/lustre/lustre/obdclass/lu_object.c |   26 ++++----------------
 1 file changed, 5 insertions(+), 21 deletions(-)

diff --git a/drivers/staging/lustre/lustre/obdclass/lu_object.c b/drivers/staging/lustre/lustre/obdclass/lu_object.c
index f14e3509f059..b38adf75e8e4 100644
--- a/drivers/staging/lustre/lustre/obdclass/lu_object.c
+++ b/drivers/staging/lustre/lustre/obdclass/lu_object.c
@@ -1329,7 +1329,7 @@ static atomic_t lu_key_initing_cnt = ATOMIC_INIT(0);
  * lu_context_refill(). No locking is provided, as initialization and shutdown
  * are supposed to be externally serialized.
  */
-static unsigned int key_set_version;
+static atomic_t key_set_version = ATOMIC_INIT(0);
 
 /**
  * Register new key.
@@ -1352,7 +1352,7 @@ int lu_context_key_register(struct lu_context_key *key)
 			lu_keys[i] = key;
 			lu_ref_init(&key->lct_reference);
 			result = 0;
-			++key_set_version;
+			atomic_inc(&key_set_version);
 			break;
 		}
 	}
@@ -1390,7 +1390,6 @@ void lu_context_key_degister(struct lu_context_key *key)
 	lu_context_key_quiesce(key);
 
 	write_lock(&lu_keys_guard);
-	++key_set_version;
 	key_fini(&lu_shrink_env.le_ctx, key->lct_index);
 
 	/**
@@ -1550,7 +1549,6 @@ void lu_context_key_quiesce(struct lu_context_key *key)
 		list_for_each_entry(ctx, &lu_context_remembered, lc_remember)
 			key_fini(ctx, key->lct_index);
 
-		++key_set_version;
 		write_unlock(&lu_keys_guard);
 	}
 }
@@ -1559,7 +1557,7 @@ void lu_context_key_revive(struct lu_context_key *key)
 {
 	write_lock(&lu_keys_guard);
 	key->lct_tags &= ~LCT_QUIESCENT;
-	++key_set_version;
+	atomic_inc(&key_set_version);
 	write_unlock(&lu_keys_guard);
 }
 
@@ -1579,7 +1577,6 @@ static void keys_fini(struct lu_context *ctx)
 
 static int keys_fill(struct lu_context *ctx)
 {
-	unsigned int pre_version;
 	unsigned int i;
 
 	/*
@@ -1593,10 +1590,9 @@ static int keys_fill(struct lu_context *ctx)
 	 */
 	read_lock(&lu_keys_guard);
 	atomic_inc(&lu_key_initing_cnt);
-	pre_version = key_set_version;
 	read_unlock(&lu_keys_guard);
+	ctx->lc_version = atomic_read(&key_set_version);
 
-refill:
 	LINVRNT(ctx->lc_value);
 	for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
 		struct lu_context_key *key;
@@ -1639,15 +1635,7 @@ static int keys_fill(struct lu_context *ctx)
 		}
 	}
 
-	read_lock(&lu_keys_guard);
-	if (pre_version != key_set_version) {
-		pre_version = key_set_version;
-		read_unlock(&lu_keys_guard);
-		goto refill;
-	}
-	ctx->lc_version = key_set_version;
 	atomic_dec(&lu_key_initing_cnt);
-	read_unlock(&lu_keys_guard);
 	return 0;
 }
 
@@ -1756,13 +1744,9 @@ EXPORT_SYMBOL(lu_context_exit);
  */
 int lu_context_refill(struct lu_context *ctx)
 {
-	read_lock(&lu_keys_guard);
-	if (likely(ctx->lc_version == key_set_version)) {
-		read_unlock(&lu_keys_guard);
+	if (likely(ctx->lc_version == atomic_read(&key_set_version)))
 		return 0;
-	}
 
-	read_unlock(&lu_keys_guard);
 	return keys_fill(ctx);
 }
 

[PATCH 2/6] staging: lustre: use an rwsem instead of lu_key_initing_cnt.

[NeilBrown]

The main use of lu_key_initing_cnt is to wait for it to be zero, so
that lu_context_key_quiesce() can continue.  This is a lot
like the behavior of a semaphore.

So use an rwsem instead.

When keys_fill() calls down_read() it will opportunistically spin
while the writer is running.  As the writer is very short - just
setting a bit for keys_fill() to see, this is likey to always
be the case.
lu_context_key_quiesce() will now, if necessary, go to sleep until
woken, rather than spin repeatedly calling schedule.

Code is much more readable this way and lu_keys_guard is no longer
involved in this locking.

We can remove the write_lock from lu_context_key_revive() as there is
nothing to protect against.  This already mustn't race with
lu_context_key_quiesce(), and if keys_fill() runs concurrently and
doesn't see that LCT_QUIESCENT has been cleared, it hardly matters.
After it is cleared, lu_context_refill() will need to be run anyway.

Signed-off-by: NeilBrown <neilb@suse.com>
---
 drivers/staging/lustre/lustre/obdclass/lu_object.c |   53 +++++++-------------
 1 file changed, 18 insertions(+), 35 deletions(-)

diff --git a/drivers/staging/lustre/lustre/obdclass/lu_object.c b/drivers/staging/lustre/lustre/obdclass/lu_object.c
index b38adf75e8e4..2f7a2589e508 100644
--- a/drivers/staging/lustre/lustre/obdclass/lu_object.c
+++ b/drivers/staging/lustre/lustre/obdclass/lu_object.c
@@ -1321,7 +1321,7 @@ enum {
 static struct lu_context_key *lu_keys[LU_CONTEXT_KEY_NR] = { NULL, };
 
 static DEFINE_RWLOCK(lu_keys_guard);
-static atomic_t lu_key_initing_cnt = ATOMIC_INIT(0);
+static DECLARE_RWSEM(lu_key_initing);
 
 /**
  * Global counter incremented whenever key is registered, unregistered,
@@ -1526,39 +1526,25 @@ void lu_context_key_quiesce(struct lu_context_key *key)
 
 	if (!(key->lct_tags & LCT_QUIESCENT)) {
 		/*
-		 * XXX memory barrier has to go here.
+		 * The write-lock on lu_key_initing will ensure that any
+		 * keys_fill() which didn't see LCT_QUIESCENT will have
+		 * finished before we call key_fini().
 		 */
-		write_lock(&lu_keys_guard);
+		down_write(&lu_key_initing);
 		key->lct_tags |= LCT_QUIESCENT;
+		up_write(&lu_key_initing);
 
-		/**
-		 * Wait until all lu_context_key::lct_init() methods
-		 * have completed.
-		 */
-		while (atomic_read(&lu_key_initing_cnt) > 0) {
-			write_unlock(&lu_keys_guard);
-			CDEBUG(D_INFO, "%s: \"%s\" %p, %d (%d)\n",
-			       __func__,
-			       module_name(key->lct_owner),
-			       key, atomic_read(&key->lct_used),
-			atomic_read(&lu_key_initing_cnt));
-			schedule();
-			write_lock(&lu_keys_guard);
-		}
-
+		write_lock(&lu_keys_guard);
 		list_for_each_entry(ctx, &lu_context_remembered, lc_remember)
 			key_fini(ctx, key->lct_index);
-
 		write_unlock(&lu_keys_guard);
 	}
 }
 
 void lu_context_key_revive(struct lu_context_key *key)
 {
-	write_lock(&lu_keys_guard);
 	key->lct_tags &= ~LCT_QUIESCENT;
 	atomic_inc(&key_set_version);
-	write_unlock(&lu_keys_guard);
 }
 
 static void keys_fini(struct lu_context *ctx)
@@ -1578,19 +1564,16 @@ static void keys_fini(struct lu_context *ctx)
 static int keys_fill(struct lu_context *ctx)
 {
 	unsigned int i;
+	int rc = 0;
 
 	/*
-	 * A serialisation with lu_context_key_quiesce() is needed, but some
-	 * "key->lct_init()" are calling kernel memory allocation routine and
-	 * can't be called while holding a spin_lock.
-	 * "lu_keys_guard" is held while incrementing "lu_key_initing_cnt"
-	 * to ensure the start of the serialisation.
-	 * An atomic_t variable is still used, in order not to reacquire the
-	 * lock when decrementing the counter.
+	 * A serialisation with lu_context_key_quiesce() is needed, to
+	 * ensure we see LCT_QUIESCENT and don't allocate a new value
+	 * after it freed one.  The rwsem provides this.  As down_read()
+	 * does optimistic spinning while the writer is active, this is
+	 * unlikely to ever sleep.
 	 */
-	read_lock(&lu_keys_guard);
-	atomic_inc(&lu_key_initing_cnt);
-	read_unlock(&lu_keys_guard);
+	down_read(&lu_key_initing);
 	ctx->lc_version = atomic_read(&key_set_version);
 
 	LINVRNT(ctx->lc_value);
@@ -1618,8 +1601,8 @@ static int keys_fill(struct lu_context *ctx)
 
 			value = key->lct_init(ctx, key);
 			if (unlikely(IS_ERR(value))) {
-				atomic_dec(&lu_key_initing_cnt);
-				return PTR_ERR(value);
+				rc = PTR_ERR(value);
+				break;
 			}
 
 			lu_ref_add_atomic(&key->lct_reference, "ctx", ctx);
@@ -1635,8 +1618,8 @@ static int keys_fill(struct lu_context *ctx)
 		}
 	}
 
-	atomic_dec(&lu_key_initing_cnt);
-	return 0;
+	up_read(&lu_key_initing);
+	return rc;
 }
 
 static int keys_init(struct lu_context *ctx)

[PATCH 3/6] staging: lustre: remove locking from lu_context_exit()

[NeilBrown]

Recent patches suggest that the locking in lu_context_exit() hurts
performance as the changes that make are to improve performance.
Let's go all the way and remove the locking completely.

The race of interest is between lu_context_exit() finalizing a
value with ->lct_exit, and lu_context_key_quiesce() freeing
the value with key_fini().

If lu_context_key_quiesce() has started, there is no need to
finalize the value - it can just be freed.  So lu_context_exit()
is changed to skip the call to ->lcu_exit if LCT_QUIESCENT it set.

If lc_context_exit() has started, lu_context_key_quiesce() must wait
for it to complete - it cannot just skip the freeing.  To allow
this we introduce a new lc_state, LCS_LEAVING.  This indicates that
->lcu_exit might be called.  Before calling key_fini() on a context,
lu_context_key_quiesce() waits (spinning) for lc_state to move on from
LCS_LEAVING.

Signed-off-by: NeilBrown <neilb@suse.com>
---
 drivers/staging/lustre/lustre/include/lu_object.h  |    1 
 drivers/staging/lustre/lustre/obdclass/lu_object.c |   41 ++++++++++++--------
 2 files changed, 26 insertions(+), 16 deletions(-)

diff --git a/drivers/staging/lustre/lustre/include/lu_object.h b/drivers/staging/lustre/lustre/include/lu_object.h
index f29bbca5af65..4153db762518 100644
--- a/drivers/staging/lustre/lustre/include/lu_object.h
+++ b/drivers/staging/lustre/lustre/include/lu_object.h
@@ -881,6 +881,7 @@ enum lu_xattr_flags {
 enum lu_context_state {
 	LCS_INITIALIZED = 1,
 	LCS_ENTERED,
+	LCS_LEAVING,
 	LCS_LEFT,
 	LCS_FINALIZED
 };
diff --git a/drivers/staging/lustre/lustre/obdclass/lu_object.c b/drivers/staging/lustre/lustre/obdclass/lu_object.c
index 2f7a2589e508..7bcf5ee47e04 100644
--- a/drivers/staging/lustre/lustre/obdclass/lu_object.c
+++ b/drivers/staging/lustre/lustre/obdclass/lu_object.c
@@ -44,6 +44,7 @@
 #include <linux/libcfs/libcfs.h>
 
 #include <linux/module.h>
+#include <linux/processor.h>
 
 /* hash_long() */
 #include <linux/libcfs/libcfs_hash.h>
@@ -1535,8 +1536,11 @@ void lu_context_key_quiesce(struct lu_context_key *key)
 		up_write(&lu_key_initing);
 
 		write_lock(&lu_keys_guard);
-		list_for_each_entry(ctx, &lu_context_remembered, lc_remember)
+		list_for_each_entry(ctx, &lu_context_remembered, lc_remember) {
+			spin_until_cond(READ_ONCE(ctx->lc_state) != LCS_LEAVING);
 			key_fini(ctx, key->lct_index);
+		}
+
 		write_unlock(&lu_keys_guard);
 	}
 }
@@ -1697,26 +1701,31 @@ void lu_context_exit(struct lu_context *ctx)
 	unsigned int i;
 
 	LINVRNT(ctx->lc_state == LCS_ENTERED);
-	ctx->lc_state = LCS_LEFT;
+	/*
+	 * Ensure lu_context_key_quiesce() sees LCS_LEAVING
+	 * or we see LCT_QUIESCENT
+	 */
+	smp_store_mb(ctx->lc_state, LCS_LEAVING);
+	/*
+	 * Disable preempt to ensure we get a warning if
+	 * any lct_exit ever tries to sleep.  That would hurt
+	 * lu_context_key_quiesce() which spins waiting for us.
+	 */
+	preempt_disable();
 	if (ctx->lc_tags & LCT_HAS_EXIT && ctx->lc_value) {
-		/* could race with key quiescency */
-		if (ctx->lc_tags & LCT_REMEMBER)
-			read_lock(&lu_keys_guard);
-
 		for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
-			if (ctx->lc_value[i]) {
-				struct lu_context_key *key;
+			struct lu_context_key *key;
 
-				key = lu_keys[i];
-				if (key->lct_exit)
-					key->lct_exit(ctx,
-						      key, ctx->lc_value[i]);
-			}
+			key = lu_keys[i];
+			if (ctx->lc_value[i] &&
+			    !(key->lct_tags & LCT_QUIESCENT) &&
+			    key->lct_exit)
+				key->lct_exit(ctx, key, ctx->lc_value[i]);
 		}
-
-		if (ctx->lc_tags & LCT_REMEMBER)
-			read_unlock(&lu_keys_guard);
 	}
+
+	preempt_enable();
+	smp_store_release(&ctx->lc_state, LCS_LEFT);
 }
 EXPORT_SYMBOL(lu_context_exit);
 

[PATCH 4/6] staging: lustre: use wait_event_var() in lu_context_key_degister()

[NeilBrown]

lu_context_key_degister() has an open coded loop which calls
schedule() without setting a new task state.  This is generally
a bad idea - it could easily just spin.

Instead, use wait_event_var() to wait for ->lct_used to be zero,
and arrange to get a wakeup when that happens.
Previously ->lct_used would only fall down to 1.  Now we decrement
it an extra time so that wake_up, which only happens when the
count reaches zero, will only happen when lu_context_key_degister()
is actually waiting for it.

Note that this patch removes key_fini() from protection by
lu_keys_guard.  key_fini() calls are not always protected
by a lock, and there seems to be no need here.  Nothing else
can be acting on the given key in that context at this point,
so no race is possible.

Signed-off-by: NeilBrown <neilb@suse.com>
---
 drivers/staging/lustre/lustre/obdclass/lu_object.c |   18 +++++++-----------
 1 file changed, 7 insertions(+), 11 deletions(-)

diff --git a/drivers/staging/lustre/lustre/obdclass/lu_object.c b/drivers/staging/lustre/lustre/obdclass/lu_object.c
index 7bcf5ee47e04..f1c35a71c413 100644
--- a/drivers/staging/lustre/lustre/obdclass/lu_object.c
+++ b/drivers/staging/lustre/lustre/obdclass/lu_object.c
@@ -1372,7 +1372,8 @@ static void key_fini(struct lu_context *ctx, int index)
 
 		key->lct_fini(ctx, key, ctx->lc_value[index]);
 		lu_ref_del(&key->lct_reference, "ctx", ctx);
-		atomic_dec(&key->lct_used);
+		if (atomic_dec_and_test(&key->lct_used))
+			wake_up_var(&key->lct_used);
 
 		if ((ctx->lc_tags & LCT_NOREF) == 0)
 			module_put(key->lct_owner);
@@ -1390,28 +1391,23 @@ void lu_context_key_degister(struct lu_context_key *key)
 
 	lu_context_key_quiesce(key);
 
-	write_lock(&lu_keys_guard);
 	key_fini(&lu_shrink_env.le_ctx, key->lct_index);
 
 	/**
 	 * Wait until all transient contexts referencing this key have
 	 * run lu_context_key::lct_fini() method.
 	 */
-	while (atomic_read(&key->lct_used) > 1) {
-		write_unlock(&lu_keys_guard);
-		CDEBUG(D_INFO, "%s: \"%s\" %p, %d\n",
-		       __func__, module_name(key->lct_owner),
-		       key, atomic_read(&key->lct_used));
-		schedule();
-		write_lock(&lu_keys_guard);
-	}
+	atomic_dec(&key->lct_used);
+	wait_var_event(&key->lct_used, atomic_read(&key->lct_used) == 0);
+
+	write_lock(&lu_keys_guard);
 	if (lu_keys[key->lct_index]) {
 		lu_keys[key->lct_index] = NULL;
 		lu_ref_fini(&key->lct_reference);
 	}
 	write_unlock(&lu_keys_guard);
 
-	LASSERTF(atomic_read(&key->lct_used) == 1,
+	LASSERTF(atomic_read(&key->lct_used) == 0,
 		 "key has instances: %d\n",
 		 atomic_read(&key->lct_used));
 }

[PATCH 5/6] staging: lustre: remove lock from key register/degister

[NeilBrown]

lu_context_key_register() doesn't really need locking.
It can use cmpxchg() to atomically install a key, and
check the result to see if it succeeded.
This requires the key to be completely ready before we
try to install it, so lct_used and lct_reference are
set up first, then reverted on failure.

With this done, lu_context_key_degister() no longer
needs locking.  It just need to set the slot to NULL.
This is done with suitable memory barriers so that the
slot cannot be reused until we are completely finished
with it.

Note that I added a warning if the slot holds NULL.
The code currently tested that code, but I don't think it
can really happen.

Signed-off-by: NeilBrown <neilb@suse.com>
---
 drivers/staging/lustre/lustre/obdclass/lu_object.c |   37 +++++++++-----------
 1 file changed, 17 insertions(+), 20 deletions(-)

diff --git a/drivers/staging/lustre/lustre/obdclass/lu_object.c b/drivers/staging/lustre/lustre/obdclass/lu_object.c
index f1c35a71c413..1c874781066d 100644
--- a/drivers/staging/lustre/lustre/obdclass/lu_object.c
+++ b/drivers/staging/lustre/lustre/obdclass/lu_object.c
@@ -1345,19 +1345,23 @@ int lu_context_key_register(struct lu_context_key *key)
 	LASSERT(key->lct_tags != 0);
 
 	result = -ENFILE;
-	write_lock(&lu_keys_guard);
+	atomic_set(&key->lct_used, 1);
+	lu_ref_init(&key->lct_reference);
 	for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
-		if (!lu_keys[i]) {
-			key->lct_index = i;
-			atomic_set(&key->lct_used, 1);
-			lu_keys[i] = key;
-			lu_ref_init(&key->lct_reference);
-			result = 0;
-			atomic_inc(&key_set_version);
-			break;
-		}
+		if (lu_keys[i])
+			continue;
+		key->lct_index = i;
+		if (cmpxchg(&lu_keys[i], NULL, key) != NULL)
+			continue;
+
+		result = 0;
+		atomic_inc(&key_set_version);
+		break;
+	}
+	if (result) {
+		lu_ref_fini(&key->lct_reference);
+		atomic_set(&key->lct_used, 0);
 	}
-	write_unlock(&lu_keys_guard);
 	return result;
 }
 EXPORT_SYMBOL(lu_context_key_register);
@@ -1400,16 +1404,9 @@ void lu_context_key_degister(struct lu_context_key *key)
 	atomic_dec(&key->lct_used);
 	wait_var_event(&key->lct_used, atomic_read(&key->lct_used) == 0);
 
-	write_lock(&lu_keys_guard);
-	if (lu_keys[key->lct_index]) {
-		lu_keys[key->lct_index] = NULL;
+	if (!WARN_ON(lu_keys[key->lct_index] == NULL))
 		lu_ref_fini(&key->lct_reference);
-	}
-	write_unlock(&lu_keys_guard);
-
-	LASSERTF(atomic_read(&key->lct_used) == 0,
-		 "key has instances: %d\n",
-		 atomic_read(&key->lct_used));
+	smp_store_release(&lu_keys[key->lct_index], NULL);
 }
 EXPORT_SYMBOL(lu_context_key_degister);
 

[PATCH 6/6] staging: lustre: rename lu_keys_guard to lu_context_remembered_guard

[NeilBrown]

The only remaining use of lu_keys_guard is to protect the
lu_context_remembers linked list, and always write_lock()
is used.
So rename it to reflect this, and change to a spinlock.
We move keys_fini() out of the locked region in
lu_context_fini() - once we have removed the context from
the lc_remembers list, there can no longer be a race.

Signed-off-by: NeilBrown <neilb@suse.com>
---
 drivers/staging/lustre/lustre/obdclass/lu_object.c |   22 +++++++++-----------
 1 file changed, 10 insertions(+), 12 deletions(-)

diff --git a/drivers/staging/lustre/lustre/obdclass/lu_object.c b/drivers/staging/lustre/lustre/obdclass/lu_object.c
index 1c874781066d..688a0428262d 100644
--- a/drivers/staging/lustre/lustre/obdclass/lu_object.c
+++ b/drivers/staging/lustre/lustre/obdclass/lu_object.c
@@ -1321,7 +1321,6 @@ enum {
 
 static struct lu_context_key *lu_keys[LU_CONTEXT_KEY_NR] = { NULL, };
 
-static DEFINE_RWLOCK(lu_keys_guard);
 static DECLARE_RWSEM(lu_key_initing);
 
 /**
@@ -1508,6 +1507,7 @@ EXPORT_SYMBOL(lu_context_key_get);
  * List of remembered contexts. XXX document me.
  */
 static LIST_HEAD(lu_context_remembered);
+static DEFINE_SPINLOCK(lu_context_remembered_guard);
 
 /**
  * Destroy \a key in all remembered contexts. This is used to destroy key
@@ -1528,13 +1528,12 @@ void lu_context_key_quiesce(struct lu_context_key *key)
 		key->lct_tags |= LCT_QUIESCENT;
 		up_write(&lu_key_initing);
 
-		write_lock(&lu_keys_guard);
+		spin_lock(&lu_context_remembered_guard);
 		list_for_each_entry(ctx, &lu_context_remembered, lc_remember) {
 			spin_until_cond(READ_ONCE(ctx->lc_state) != LCS_LEAVING);
 			key_fini(ctx, key->lct_index);
 		}
-
-		write_unlock(&lu_keys_guard);
+		spin_unlock(&lu_context_remembered_guard);
 	}
 }
 
@@ -1640,9 +1639,9 @@ int lu_context_init(struct lu_context *ctx, __u32 tags)
 	ctx->lc_state = LCS_INITIALIZED;
 	ctx->lc_tags = tags;
 	if (tags & LCT_REMEMBER) {
-		write_lock(&lu_keys_guard);
+		spin_lock(&lu_context_remembered_guard);
 		list_add(&ctx->lc_remember, &lu_context_remembered);
-		write_unlock(&lu_keys_guard);
+		spin_unlock(&lu_context_remembered_guard);
 	} else {
 		INIT_LIST_HEAD(&ctx->lc_remember);
 	}
@@ -1665,14 +1664,13 @@ void lu_context_fini(struct lu_context *ctx)
 
 	if ((ctx->lc_tags & LCT_REMEMBER) == 0) {
 		LASSERT(list_empty(&ctx->lc_remember));
-		keys_fini(ctx);
-
-	} else { /* could race with key degister */
-		write_lock(&lu_keys_guard);
-		keys_fini(ctx);
+	} else {
+		/* could race with key degister */
+		spin_lock(&lu_context_remembered_guard);
 		list_del_init(&ctx->lc_remember);
-		write_unlock(&lu_keys_guard);
+		spin_unlock(&lu_context_remembered_guard);
 	}
+	keys_fini(ctx);
 }
 EXPORT_SYMBOL(lu_context_fini);