[PATCH v4] KEYS: key_garbage_collector(): process in chunks

[PATCH v4] KEYS: key_garbage_collector(): process in chunks

[Jarkko Sakkinen]

key_garbage_collector() has a design that is not resistant enough for
workqueue lockups. The duration scales up relative to the number of items
in the serial tree, which causes a reoccuring theme of being guilty of
causing workqueue lockups.

Resolve the problem by processing in chunks:

1. Declare a static variable key_gc_work, which owns the garbage collection
   in-flight state, and key_schedule_gc_work() for re-scheduling the work.
2. Define KEY_GC_CHUNK_SIZE, set to 1024 keys, as the bound for the number
   of items scanned and destroyed.
3. Define KEY_GC_DESTROYING to mark resume to destruction without repeating
   the RCU grace period. Drain the graveyard before advancing key-type
   reaping.
4. Define KEY_GC_REAP_REQUESTED for preserving full-pass requests across
   the continuations.

keygcbench results without the patch:

unused:   min=3016.754 avg=11859.097 max=28667.645 var=141391574.712
expired:  min=0.015 avg=12758.165 max=21516.237 var=85158284.646
combined: min=0.015 avg=12308.631 max=28667.645 var=113477010.596

keygcbench results with the patch:

unused:   min=0.225 avg=13.991 max=853.100 var=858.700
expired:  min=0.016 avg=30.442 max=2791.695 var=14187.539
combined: min=0.016 avg=19.249 max=2791.695 var=5177.436

Tests were with QEMU/KVM hosted by AMD 4650U (Zen 2) CPU.

Fixes: 0c061b5707ab ("KEYS: Correctly destroy key payloads when their keytype is removed")
Reported-by: syzbot+c15de43f1e7fffbb7b9b@syzkaller.appspotmail.com
Closes: https://syzkaller.appspot.com/bug?extid=c15de43f1e7fffbb7b9b
Link: https://git.kernel.org/pub/scm/linux/kernel/git/jarkko/linux-tpmdd-test.git/tree/package/keygcbench
Signed-off-by: Jarkko Sakkinen <jarkko@kernel.org>
---
v4:
- Added syzkaller link, as the resolution can be reasoned from the benchmark
  results.
v3:
- Added new tests results made by using keygcbench.
- Added link to keygcbench.
v2:
- Fixed typo in the commit message.
- Drop the local variable gc_state.
---
 security/keys/gc.c       | 151 ++++++++++++++++++++++++++-------------
 security/keys/internal.h |   1 -
 security/keys/key.c      |   2 +-
 3 files changed, 102 insertions(+), 52 deletions(-)

diff --git a/security/keys/gc.c b/security/keys/gc.c
index eda445f815d4..ddf51879971c 100644
--- a/security/keys/gc.c
+++ b/security/keys/gc.c
@@ -10,16 +10,27 @@
 #include <keys/keyring-type.h>
 #include "internal.h"
 
-/*
- * Delay between key revocation/expiry in seconds
- */
-unsigned key_gc_delay = 5 * 60;
+#define KEY_GC_CHUNK_SIZE 1024
 
 /*
  * Reaper for unused keys.
  */
 static void key_garbage_collector(struct work_struct *work);
-DECLARE_WORK(key_gc_work, key_garbage_collector);
+
+static struct {
+	time64_t start;			/* expiry range start */
+	time64_t end;			/* expiry range end */
+	struct rb_node *cursor;		/* serial tree position */
+	u8 state;			/* state flags */
+	struct work_struct work;	/* work item */
+} key_gc_work = {
+	.work = __WORK_INITIALIZER(key_gc_work.work, key_garbage_collector),
+};
+
+/*
+ * Delay between key revocation/expiry in seconds
+ */
+unsigned int key_gc_delay = 5 * 60;
 
 /*
  * Reaper for links from keyrings to dead keys.
@@ -34,6 +45,7 @@ static unsigned long key_gc_flags;
 #define KEY_GC_KEY_EXPIRED	0	/* A key expired and needs unlinking */
 #define KEY_GC_REAP_KEYTYPE	1	/* A keytype is being unregistered */
 #define KEY_GC_REAPING_KEYTYPE	2	/* Cleared when keytype reaped */
+#define KEY_GC_REAP_REQUESTED	3	/* Another complete pass was requested */
 
 
 /*
@@ -44,6 +56,20 @@ struct key_type key_type_dead = {
 	.name = ".dead",
 };
 
+/*
+ * Schedule a new GC pass or re-schedule an ongoing GC pass.
+ */
+static void key_schedule_gc_work(void)
+{
+	/*
+	 * Order caller updates before setting KEY_GC_REAP_REQUESTED, given that
+	 * set_bit() is unordered.
+	 */
+	smp_mb__before_atomic();
+	set_bit(KEY_GC_REAP_REQUESTED, &key_gc_flags);
+	schedule_work(&key_gc_work.work);
+}
+
 /*
  * Schedule a garbage collection run.
  * - time precision isn't particularly important
@@ -57,7 +83,7 @@ void key_schedule_gc(time64_t gc_at)
 
 	if (gc_at <= now || test_bit(KEY_GC_REAP_KEYTYPE, &key_gc_flags)) {
 		kdebug("IMMEDIATE");
-		schedule_work(&key_gc_work);
+		key_schedule_gc_work();
 	} else if (gc_at < key_gc_next_run) {
 		kdebug("DEFERRED");
 		key_gc_next_run = gc_at;
@@ -85,7 +111,7 @@ void key_set_expiry(struct key *key, time64_t expiry)
 void key_schedule_gc_links(void)
 {
 	set_bit(KEY_GC_KEY_EXPIRED, &key_gc_flags);
-	schedule_work(&key_gc_work);
+	key_schedule_gc_work();
 }
 
 /*
@@ -119,7 +145,7 @@ void key_gc_keytype(struct key_type *ktype)
 	set_bit(KEY_GC_REAP_KEYTYPE, &key_gc_flags);
 
 	kdebug("schedule");
-	schedule_work(&key_gc_work);
+	key_schedule_gc_work();
 
 	kdebug("sleep");
 	wait_on_bit(&key_gc_flags, KEY_GC_REAPING_KEYTYPE,
@@ -134,7 +160,9 @@ void key_gc_keytype(struct key_type *ktype)
  */
 static noinline void key_gc_unused_keys(struct list_head *keys)
 {
-	while (!list_empty(keys)) {
+	unsigned int chunk = KEY_GC_CHUNK_SIZE;
+
+	while (!list_empty(keys) && chunk--) {
 		struct key *key =
 			list_entry(keys->next, struct key, graveyard_link);
 		short state = key->state;
@@ -178,54 +206,64 @@ static noinline void key_gc_unused_keys(struct list_head *keys)
 static void key_garbage_collector(struct work_struct *work)
 {
 	static LIST_HEAD(graveyard);
-	static u8 gc_state;		/* Internal persistent state */
+	typeof(key_gc_work) *gc = &key_gc_work;
 #define KEY_GC_REAP_AGAIN	0x01	/* - Need another cycle */
 #define KEY_GC_REAPING_LINKS	0x02	/* - We need to reap links */
+#define KEY_GC_DESTROYING	0x04	/* - We need to destroy keys */
 #define KEY_GC_REAPING_DEAD_1	0x10	/* - We need to mark dead keys */
 #define KEY_GC_REAPING_DEAD_2	0x20	/* - We need to reap dead key links */
 #define KEY_GC_REAPING_DEAD_3	0x40	/* - We need to reap dead keys */
 #define KEY_GC_FOUND_DEAD_KEY	0x80	/* - We found at least one dead key */
 
-	struct rb_node *cursor;
+	unsigned int chunk = KEY_GC_CHUNK_SIZE;
 	struct key *key;
-	time64_t new_timer, limit, expiry;
+	time64_t expiry;
 
-	kenter("[%lx,%x]", key_gc_flags, gc_state);
+	kenter("[%lx,%x]", key_gc_flags, gc->state);
 
-	limit = ktime_get_real_seconds();
+	if (gc->state & KEY_GC_DESTROYING)
+		goto destroy_keys;
+	if (gc->cursor) {
+		spin_lock(&key_serial_lock);
+		goto continue_scanning;
+	}
+
+	test_and_clear_bit(KEY_GC_REAP_REQUESTED, &key_gc_flags);
+	gc->start = ktime_get_real_seconds();
 
 	/* Work out what we're going to be doing in this pass */
-	gc_state &= KEY_GC_REAPING_DEAD_1 | KEY_GC_REAPING_DEAD_2;
-	gc_state <<= 1;
+	gc->state &= KEY_GC_REAPING_DEAD_1 | KEY_GC_REAPING_DEAD_2;
+	gc->state <<= 1;
 	if (test_and_clear_bit(KEY_GC_KEY_EXPIRED, &key_gc_flags))
-		gc_state |= KEY_GC_REAPING_LINKS;
+		gc->state |= KEY_GC_REAPING_LINKS;
 
 	if (test_and_clear_bit(KEY_GC_REAP_KEYTYPE, &key_gc_flags))
-		gc_state |= KEY_GC_REAPING_DEAD_1;
-	kdebug("new pass %x", gc_state);
+		gc->state |= KEY_GC_REAPING_DEAD_1;
+	kdebug("new pass %x", gc->state);
 
-	new_timer = TIME64_MAX;
+	gc->end = TIME64_MAX;
 
 	/* As only this function is permitted to remove things from the key
 	 * serial tree, if cursor is non-NULL then it will always point to a
 	 * valid node in the tree - even if lock got dropped.
 	 */
 	spin_lock(&key_serial_lock);
-	cursor = rb_first(&key_serial_tree);
+	gc->cursor = rb_first(&key_serial_tree);
 
 continue_scanning:
-	while (cursor) {
-		key = rb_entry(cursor, struct key, serial_node);
-		cursor = rb_next(cursor);
+	while (gc->cursor && chunk) {
+		key = rb_entry(gc->cursor, struct key, serial_node);
+		gc->cursor = rb_next(gc->cursor);
+		chunk--;
 
 		if (!test_bit_acquire(KEY_FLAG_USER_ALIVE, &key->flags)) {
 			/* Clobber key->user after final put seen. */
 			goto found_unreferenced_key;
 		}
 
-		if (unlikely(gc_state & KEY_GC_REAPING_DEAD_1)) {
+		if (unlikely(gc->state & KEY_GC_REAPING_DEAD_1)) {
 			if (key->type == key_gc_dead_keytype) {
-				gc_state |= KEY_GC_FOUND_DEAD_KEY;
+				gc->state |= KEY_GC_FOUND_DEAD_KEY;
 				set_bit(KEY_FLAG_DEAD, &key->flags);
 				key->perm = 0;
 				goto skip_dead_key;
@@ -239,24 +277,24 @@ static void key_garbage_collector(struct work_struct *work)
 		if (expiry != TIME64_MAX) {
 			if (!(key->type->flags & KEY_TYPE_INSTANT_REAP))
 				expiry += key_gc_delay;
-			if (expiry > limit && expiry < new_timer) {
+			if (expiry > gc->start && expiry < gc->end) {
 				kdebug("will expire %x in %lld",
-				       key_serial(key), key->expiry - limit);
-				new_timer = key->expiry;
+				       key_serial(key), key->expiry - gc->start);
+				gc->end = key->expiry;
 			}
 		}
 
-		if (unlikely(gc_state & KEY_GC_REAPING_DEAD_2))
+		if (unlikely(gc->state & KEY_GC_REAPING_DEAD_2))
 			if (key->type == key_gc_dead_keytype)
-				gc_state |= KEY_GC_FOUND_DEAD_KEY;
+				gc->state |= KEY_GC_FOUND_DEAD_KEY;
 
-		if ((gc_state & KEY_GC_REAPING_LINKS) ||
-		    unlikely(gc_state & KEY_GC_REAPING_DEAD_2)) {
+		if ((gc->state & KEY_GC_REAPING_LINKS) ||
+		    unlikely(gc->state & KEY_GC_REAPING_DEAD_2)) {
 			if (key->type == &key_type_keyring)
 				goto found_keyring;
 		}
 
-		if (unlikely(gc_state & KEY_GC_REAPING_DEAD_3))
+		if (unlikely(gc->state & KEY_GC_REAPING_DEAD_3))
 			if (key->type == key_gc_dead_keytype)
 				goto destroy_dead_key;
 
@@ -269,7 +307,11 @@ static void key_garbage_collector(struct work_struct *work)
 	spin_unlock(&key_serial_lock);
 
 maybe_resched:
-	if (cursor) {
+	if (gc->cursor) {
+		if (!chunk) {
+			schedule_work(&key_gc_work.work);
+			return;
+		}
 		cond_resched();
 		spin_lock(&key_serial_lock);
 		goto continue_scanning;
@@ -281,12 +323,12 @@ static void key_garbage_collector(struct work_struct *work)
 	 */
 	kdebug("pass complete");
 
-	if (new_timer != TIME64_MAX) {
-		new_timer += key_gc_delay;
-		key_schedule_gc(new_timer);
+	if (gc->end != TIME64_MAX) {
+		gc->end += key_gc_delay;
+		key_schedule_gc(gc->end);
 	}
 
-	if (unlikely(gc_state & KEY_GC_REAPING_DEAD_2) ||
+	if (unlikely(gc->state & KEY_GC_REAPING_DEAD_2) ||
 	    !list_empty(&graveyard)) {
 		/* Make sure that all pending keyring payload destructions are
 		 * fulfilled and that people aren't now looking at dead or
@@ -297,33 +339,42 @@ static void key_garbage_collector(struct work_struct *work)
 		synchronize_rcu();
 	}
 
+	gc->state |= KEY_GC_DESTROYING;
+
+destroy_keys:
 	if (!list_empty(&graveyard)) {
 		kdebug("gc keys");
 		key_gc_unused_keys(&graveyard);
+		if (!list_empty(&graveyard)) {
+			schedule_work(&key_gc_work.work);
+			return;
+		}
 	}
+	gc->state &= ~KEY_GC_DESTROYING;
 
-	if (unlikely(gc_state & (KEY_GC_REAPING_DEAD_1 |
+	if (unlikely(gc->state & (KEY_GC_REAPING_DEAD_1 |
 				 KEY_GC_REAPING_DEAD_2))) {
-		if (!(gc_state & KEY_GC_FOUND_DEAD_KEY)) {
+		if (!(gc->state & KEY_GC_FOUND_DEAD_KEY)) {
 			/* No remaining dead keys: short circuit the remaining
 			 * keytype reap cycles.
 			 */
 			kdebug("dead short");
-			gc_state &= ~(KEY_GC_REAPING_DEAD_1 | KEY_GC_REAPING_DEAD_2);
-			gc_state |= KEY_GC_REAPING_DEAD_3;
+			gc->state &= ~(KEY_GC_REAPING_DEAD_1 | KEY_GC_REAPING_DEAD_2);
+			gc->state |= KEY_GC_REAPING_DEAD_3;
 		} else {
-			gc_state |= KEY_GC_REAP_AGAIN;
+			gc->state |= KEY_GC_REAP_AGAIN;
 		}
 	}
 
-	if (unlikely(gc_state & KEY_GC_REAPING_DEAD_3)) {
+	if (unlikely(gc->state & KEY_GC_REAPING_DEAD_3)) {
 		kdebug("dead wake");
 		clear_and_wake_up_bit(KEY_GC_REAPING_KEYTYPE, &key_gc_flags);
 	}
 
-	if (gc_state & KEY_GC_REAP_AGAIN)
-		schedule_work(&key_gc_work);
-	kleave(" [end %x]", gc_state);
+	if ((gc->state & KEY_GC_REAP_AGAIN) ||
+	    test_bit(KEY_GC_REAP_REQUESTED, &key_gc_flags))
+		schedule_work(&key_gc_work.work);
+	kleave(" [end %x]", gc->state);
 	return;
 
 	/* We found an unreferenced key - once we've removed it from the tree,
@@ -335,7 +386,7 @@ static void key_garbage_collector(struct work_struct *work)
 	spin_unlock(&key_serial_lock);
 
 	list_add_tail(&key->graveyard_link, &graveyard);
-	gc_state |= KEY_GC_REAP_AGAIN;
+	gc->state |= KEY_GC_REAP_AGAIN;
 	goto maybe_resched;
 
 	/* We found a restricted keyring and need to update the restriction if
@@ -353,7 +404,7 @@ static void key_garbage_collector(struct work_struct *work)
 	 */
 found_keyring:
 	spin_unlock(&key_serial_lock);
-	keyring_gc(key, limit);
+	keyring_gc(key, gc->start);
 	goto maybe_resched;
 
 	/* We found a dead key that is still referenced.  Reset its type and
diff --git a/security/keys/internal.h b/security/keys/internal.h
index b7b622bc36a1..cca936f5015d 100644
--- a/security/keys/internal.h
+++ b/security/keys/internal.h
@@ -162,7 +162,6 @@ extern bool lookup_user_key_possessed(const struct key *key,
 extern long join_session_keyring(const char *name);
 extern void key_change_session_keyring(struct callback_head *twork);
 
-extern struct work_struct key_gc_work;
 extern unsigned key_gc_delay;
 extern void keyring_gc(struct key *keyring, time64_t limit);
 extern void keyring_restriction_gc(struct key *keyring,
diff --git a/security/keys/key.c b/security/keys/key.c
index e88548f1a704..5c2d0e3b49e6 100644
--- a/security/keys/key.c
+++ b/security/keys/key.c
@@ -670,7 +670,7 @@ void key_put(struct key *key)
 			}
 			/* Mark key as safe for GC after key->user done. */
 			clear_bit_unlock(KEY_FLAG_USER_ALIVE, &key->flags);
-			schedule_work(&key_gc_work);
+			key_schedule_gc(0);
 		}
 	}
 }
-- 
2.47.3

Re: [PATCH v4] KEYS: key_garbage_collector(): process in chunks

[Jarkko Sakkinen]

On Thu, Oct 29, 2026 at 04:07:05PM +0200, Jarkko Sakkinen wrote:
> keygcbench results without the patch:
> 
> unused:   min=3016.754 avg=11859.097 max=28667.645 var=141391574.712
> expired:  min=0.015 avg=12758.165 max=21516.237 var=85158284.646
> combined: min=0.015 avg=12308.631 max=28667.645 var=113477010.596
> 
> keygcbench results with the patch:
> 
> unused:   min=0.225 avg=13.991 max=853.100 var=858.700
> expired:  min=0.016 avg=30.442 max=2791.695 var=14187.539
> combined: min=0.016 avg=19.249 max=2791.695 var=5177.436

I personally would optimize all-in for variance because:

1. It is insignificant drawback for any practical keyring workload
   if the total time increases slightly.
2. Latency peaks are a signifiant drawback for any possible latency
   sensitive workload running in the system.

What I'm doing now is to find sweet spot chunk size below 1024 where
variances reaches its lowest value.

The latency part makes this change more valuable than focusing on
only on the regression itself.

Br, Jarkko

Re: [PATCH v4] KEYS: key_garbage_collector(): process in chunks

[Jarkko Sakkinen]

On Thu, Oct 29, 2026 at 04:33:10PM +0200, Jarkko Sakkinen wrote:
> On Thu, Oct 29, 2026 at 04:07:05PM +0200, Jarkko Sakkinen wrote:
> > keygcbench results without the patch:
> > 
> > unused:   min=3016.754 avg=11859.097 max=28667.645 var=141391574.712
> > expired:  min=0.015 avg=12758.165 max=21516.237 var=85158284.646
> > combined: min=0.015 avg=12308.631 max=28667.645 var=113477010.596
> > 
> > keygcbench results with the patch:
> > 
> > unused:   min=0.225 avg=13.991 max=853.100 var=858.700
> > expired:  min=0.016 avg=30.442 max=2791.695 var=14187.539
> > combined: min=0.016 avg=19.249 max=2791.695 var=5177.436
> 
> I personally would optimize all-in for variance because:
> 
> 1. It is insignificant drawback for any practical keyring workload
>    if the total time increases slightly.
> 2. Latency peaks are a signifiant drawback for any possible latency
>    sensitive workload running in the system.
> 
> What I'm doing now is to find sweet spot chunk size below 1024 where
> variances reaches its lowest value.
> 
> The latency part makes this change more valuable than focusing on
> only on the regression itself.

The best block size 1 and this was actually my guess. Just in case I ran
the benchmark three times for each 11 block sizes.

It is simple refactor to do. I simply substitute KEY_GC_BLOCK_SIZE with
1 and carefully refactor the constant out.

I think that converging to the correct solution through this route was
worth of the additional trouble because at least now properly sanity
checked.

Br, Jarkko

Re: [PATCH v4] KEYS: key_garbage_collector(): process in chunks

[Jarkko Sakkinen]

On Thu, Oct 29, 2026 at 06:45:19PM +0200, Jarkko Sakkinen wrote:
> On Thu, Oct 29, 2026 at 04:33:10PM +0200, Jarkko Sakkinen wrote:
> > On Thu, Oct 29, 2026 at 04:07:05PM +0200, Jarkko Sakkinen wrote:
> > > keygcbench results without the patch:
> > > 
> > > unused:   min=3016.754 avg=11859.097 max=28667.645 var=141391574.712
> > > expired:  min=0.015 avg=12758.165 max=21516.237 var=85158284.646
> > > combined: min=0.015 avg=12308.631 max=28667.645 var=113477010.596
> > > 
> > > keygcbench results with the patch:
> > > 
> > > unused:   min=0.225 avg=13.991 max=853.100 var=858.700
> > > expired:  min=0.016 avg=30.442 max=2791.695 var=14187.539
> > > combined: min=0.016 avg=19.249 max=2791.695 var=5177.436
> > 
> > I personally would optimize all-in for variance because:
> > 
> > 1. It is insignificant drawback for any practical keyring workload
> >    if the total time increases slightly.
> > 2. Latency peaks are a signifiant drawback for any possible latency
> >    sensitive workload running in the system.
> > 
> > What I'm doing now is to find sweet spot chunk size below 1024 where
> > variances reaches its lowest value.
> > 
> > The latency part makes this change more valuable than focusing on
> > only on the regression itself.
> 
> The best block size 1 and this was actually my guess. Just in case I ran
> the benchmark three times for each 11 block sizes.
> 
> It is simple refactor to do. I simply substitute KEY_GC_BLOCK_SIZE with
> 1 and carefully refactor the constant out.
> 
> I think that converging to the correct solution through this route was
> worth of the additional trouble because at least now properly sanity
> checked.

Taking this back! It has 133% total time increase :-) So yeah constant
is required. I have all eleven (x3 each) available so I pick something
more appropriate (more like 13% at most).

Br, Jarkko