Move deferred-work orchestration from arena into pa/pac.

Move most arena-owned PAC deferral & background thread relevant codes
out to pa/pac for decoupling purposes.

include/jemalloc/internal/pa.h

@@ -152,8 +152,12 @@ void pa_shard_reset(tsdn_t *tsdn, pa_shard_t *shard);
  */
 void pa_shard_destroy(tsdn_t *tsdn, pa_shard_t *shard);
 
-/* Flush any caches used by shard */
+/*
-void pa_shard_flush(tsdn_t *tsdn, pa_shard_t *shard);
+ * Flush the shard's front caches (SEC + HPA) back to the ecaches.  If all is
+ * true, also fully decay-purge the PAC dirty (and muzzy, unless skipped) extents
+ * to the OS -- the "save as much memory as possible" path.
+ */
+void pa_shard_flush(tsdn_t *tsdn, pa_shard_t *shard, bool all);
 
 /* Gets an edata for the given allocation. */
 edata_t *pa_alloc(tsdn_t *tsdn, pa_shard_t *shard, size_t size,
@@ -178,21 +182,35 @@ bool pa_shrink(tsdn_t *tsdn, pa_shard_t *shard, edata_t *edata, size_t old_size,
 void    pa_dalloc(tsdn_t *tsdn, pa_shard_t *shard, edata_t *edata,
        bool *deferred_work_generated);
 bool    pa_decay_ms_set(tsdn_t *tsdn, pa_shard_t *shard, extent_state_t state,
-       ssize_t decay_ms, pac_purge_eagerness_t eagerness);
+       ssize_t decay_ms);
 ssize_t pa_decay_ms_get(pa_shard_t *shard, extent_state_t state);
 
-/*
- * Do deferred work on this PA shard: both PAC decay and the HPA deferred work
- * symetrically.  The only exception in the symetry is PAC decay-purge policy
- * requires the eagerness, which is passed in from the arena.
- */
 void pa_shard_set_deferral_allowed(
     tsdn_t *tsdn, pa_shard_t *shard, bool deferral_allowed);
+/*
+ * Do deferred work on this PA shard: dispatch to PAC (decay-purge) then HPA.
+ * Each allocator owns its own policy -- PAC decides eagerness from
+ * is_background_thread and, on the application path, notifies the background
+ * thread; the bg-thread driver passes is_background_thread=true and is never
+ * self-notified.
+ */
 void     pa_shard_do_deferred_work(
-    tsdn_t *tsdn, pa_shard_t *shard, pac_purge_eagerness_t eagerness);
+    tsdn_t *tsdn, pa_shard_t *shard, bool is_background_thread);
 void     pa_shard_try_deferred_work(tsdn_t *tsdn, pa_shard_t *shard);
 uint64_t pa_shard_time_until_deferred_work(tsdn_t *tsdn, pa_shard_t *shard);
 
+/*
+ * Called on the application path after a pa operation (alloc/expand/shrink/
+ * dalloc) reports deferred_work_generated -- i.e. it left PAC decay due or
+ * HPA work (hugify/dehugify/purge) pending.  If PAC dirty decay is immediate
+ * (dirty_decay_ms == 0), purge dirty synchronously; then, if a background
+ * thread is enabled but idle (sleeping indefinitely), wake it early so the
+ * pending work runs promptly rather than at the next scheduled wakeup.
+ *
+ * Application thread only; must never be called on the background thread.
+ */
+void pa_shard_handle_deferred_work(tsdn_t *tsdn, pa_shard_t *shard);
+
 /******************************************************************************/
 /*
  * Various bits of "boring" functionality that are still part of this module,

include/jemalloc/internal/pac.h

@@ -240,10 +240,10 @@ bool pac_maybe_decay_purge(tsdn_t *tsdn, pac_t *pac, decay_t *decay,
     pac_purge_eagerness_t eagerness);
 
 /*
- * Result of a pac_do_deferred_work() pass.  Reported per decay state so the
+ * Result of a pac_decay_deferred() pass.  Reported per decay state so
- * caller (on the user-inline path) can decide, per state, whether to notify the
+ * pac_do_deferred_work can decide, per state, whether to notify the background
- * background thread.  npages_new is the per-state epoch backlog delta, only
+ * thread.  npages_new is the per-state epoch backlog delta, only meaningful when
- * meaningful when the corresponding *_epoch_advanced is true.
+ * the corresponding *_epoch_advanced is true.
  */
 typedef struct pac_deferred_work_result_s pac_deferred_work_result_t;
 struct pac_deferred_work_result_s {
@@ -254,11 +254,20 @@ struct pac_deferred_work_result_s {
 };
 
 /*
- * Non-forced deferred decay work for both the dirty and muzzy states.
+ * All deferred decay-purge work for a PAC shard: decide the eagerness from the
- * Corresponding decay->mtx are acquired internally.
+ * caller context, run pac_decay_deferred, and (application path only) notify the
+ * background thread for any decay epoch that advanced.  The bg-thread driver
+ * passes is_background_thread=true and is never self-notified.
  */
-void pac_do_deferred_work(tsdn_t *tsdn, pac_t *pac,
+void pac_do_deferred_work(
-    pac_purge_eagerness_t eagerness, pac_deferred_work_result_t *result);
+    tsdn_t *tsdn, pac_t *pac, bool is_background_thread);
+
+/*
+ * Application-path hook (after deferred_work_generated): wake the background
+ * thread if it is sleeping idle.  Runs the same early-wake logic as the notify
+ * path (pac_maybe_wake_bg), gated on the thread being idle.
+ */
+void pac_wake_bg_on_deferred(tsdn_t *tsdn, pac_t *pac);
 
 /*
  * Fully decay the extents of the given state, acquiring decay->mtx internally.
@@ -278,7 +287,7 @@ bool pac_retain_grow_limit_get_set(
     tsdn_t *tsdn, pac_t *pac, size_t *old_limit, size_t *new_limit);
 
 bool    pac_decay_ms_set(tsdn_t *tsdn, pac_t *pac, extent_state_t state,
-       ssize_t decay_ms, pac_purge_eagerness_t eagerness);
+       ssize_t decay_ms);
 ssize_t pac_decay_ms_get(pac_t *pac, extent_state_t state);
 
 /* Whether the muzzy decay path is relevant (muzzy pages exist, or decay is on). */
@@ -288,6 +297,12 @@ pac_should_decay_muzzy(pac_t *pac) {
 	    || pac_decay_ms_get(pac, extent_state_muzzy) > 0;
 }
 
+/* Whether dirty decay is immediate (dirty_decay_ms == 0). */
+static inline bool
+pac_decay_immediately(pac_t *pac) {
+	return decay_immediately(&pac->decay_dirty);
+}
+
 void pac_destroy(tsdn_t *tsdn, pac_t *pac);
 
 void pac_sec_flush(tsdn_t *tsdn, pac_t *pac);

src/arena.c

@@ -83,9 +83,6 @@ const arena_config_t arena_config_default = {
  * definition.
  */
 
-static void arena_maybe_do_deferred_work(
-    tsdn_t *tsdn, arena_t *arena, decay_t *decay, size_t npages_new);
-
 /******************************************************************************/
 
 void
@@ -314,32 +311,13 @@ arena_cache_bins_stats_merge(tsdn_t *tsdn, arena_t *arena) {
 	malloc_mutex_unlock(tsdn, &arena->cache_bin_array_descriptor_ql_mtx);
 }
 
-static void
-arena_background_thread_inactivity_check(
-    tsdn_t *tsdn, arena_t *arena, bool is_background_thread) {
-	if (!background_thread_enabled() || is_background_thread) {
-		return;
-	}
-	background_thread_info_t *info = arena_background_thread_info_get(
-	    arena);
-	if (background_thread_indefinite_sleep(info)) {
-		arena_maybe_do_deferred_work(
-		    tsdn, arena, &arena->pa_shard.pac.decay_dirty, 0);
-	}
-}
-
 /*
- * React to deferred work generated by a PAI function.
+ * React to deferred work generated by a PAI function.  Delegates to pa, which
+ * owns the PAC/HPA decay + background-thread interaction.
  */
 void
 arena_handle_deferred_work(tsdn_t *tsdn, arena_t *arena) {
-	witness_assert_depth_to_rank(
+	pa_shard_handle_deferred_work(tsdn, &arena->pa_shard);
-	    tsdn_witness_tsdp_get(tsdn), WITNESS_RANK_CORE, 0);
-
-	if (decay_immediately(&arena->pa_shard.pac.decay_dirty)) {
-		pac_decay_all_now(tsdn, &arena->pa_shard.pac, extent_state_dirty);
-	}
-	arena_background_thread_inactivity_check(tsdn, arena, false);
 }
 
 static void
@@ -474,29 +452,10 @@ arena_extent_ralloc_large_expand(
 	}
 }
 
-/*
- * In situations where we're not forcing a decay (i.e. because the user
- * specifically requested it), should we purge ourselves, or wait for the
- * background thread to get to it.
- */
-static pac_purge_eagerness_t
-arena_decide_unforced_purge_eagerness(bool is_background_thread) {
-	if (is_background_thread) {
-		return PAC_PURGE_ALWAYS;
-	} else if (!is_background_thread && background_thread_enabled()) {
-		return PAC_PURGE_NEVER;
-	} else {
-		return PAC_PURGE_ON_EPOCH_ADVANCE;
-	}
-}
-
 bool
 arena_decay_ms_set(
     tsdn_t *tsdn, arena_t *arena, extent_state_t state, ssize_t decay_ms) {
-	pac_purge_eagerness_t eagerness = arena_decide_unforced_purge_eagerness(
+	return pa_decay_ms_set(tsdn, &arena->pa_shard, state, decay_ms);
-	    /* is_background_thread */ false);
-	return pa_decay_ms_set(
-	    tsdn, &arena->pa_shard, state, decay_ms, eagerness);
 }
 
 ssize_t
@@ -508,144 +467,31 @@ void
 arena_decay(tsdn_t *tsdn, arena_t *arena, bool is_background_thread, bool all) {
 	if (all) {
 		/*
-		 * We should take a purge of "all" to mean "save as much memory
+		 * A purge of "all" means "save as much memory as possible"
-		 * as possible", including flushing any caches (for situations
+		 * (thread death, manual purge): flush caches and fully decay.
-		 * like thread death, or manual purge calls).  This blocking
-		 * flush-and-fully-decay path is kept separate from the deferred
-		 * (all=false) path below.
 		 */
-		pa_shard_flush(tsdn, &arena->pa_shard);
+		pa_shard_flush(tsdn, &arena->pa_shard, /* all */ true);
-		pac_decay_all_now(
-		    tsdn, &arena->pa_shard.pac, extent_state_dirty);
-		if (pac_should_decay_muzzy(&arena->pa_shard.pac)) {
-			pac_decay_all_now(
-			    tsdn, &arena->pa_shard.pac, extent_state_muzzy);
-		}
 		return;
 	}
 
 	/*
-	 * Deferred (non-forced) decay-purge.  The PAC layer owns the decay
+	 * Deferred (non-forced) decay-purge: pa owns the whole bridge (eagerness
-	 * orchestration (lock acquisition, dirty-before-muzzy ordering, the
+	 * decision, PAC decay orchestration, and background-thread notify).
-	 * dirty-contended-skip-muzzy and muzzy short-circuit rules).   We call
-	 * it with the eagerness decided here and then notify the background
-	 * thread per decay state for any epoch that advanced.
-	 *
-	 * A concurrent background_thread enable/disable (mallctl) can race
-	 * this path: the enable state is read lock-free twice below (for the
-	 * eagerness decision, then the notify guard), so the two reads may
-	 * disagree.  Worst case is benign and self-healing:
-	 *   disabled->enabled: purged immediately, plus a possibly-redundant
-	 *     wake;
-	 *   enabled->disabled: deferred but not notified this pass -- the
-	 *     pages stay in the decay backlog and are reclaimed on the next
-	 *     decay tick or by the bg thread before it stops.
-	 * It stays safe regardless: info is allocated once and never freed;
-	 * the wake is gated by info->mtx + background_thread_is_started() (so
-	 * we never signal a not-yet-running thread); the bg-thread locks here
-	 * are trylocks; and the purge runs under decay->mtx, which the toggle
-	 * never touches.
 	 */
-	pac_purge_eagerness_t eagerness = arena_decide_unforced_purge_eagerness(
+	pa_shard_do_deferred_work(
-	    is_background_thread);
+	    tsdn, &arena->pa_shard, is_background_thread);
-	pac_deferred_work_result_t result;
-	pac_do_deferred_work(
-	    tsdn, &arena->pa_shard.pac, eagerness, &result);
-
-	if (have_background_thread && background_thread_enabled()
-	    && !is_background_thread) {
-		if (result.dirty_epoch_advanced) {
-			arena_maybe_do_deferred_work(tsdn, arena,
-			    &arena->pa_shard.pac.decay_dirty,
-			    result.dirty_npages_new);
-		}
-		if (result.muzzy_epoch_advanced) {
-			arena_maybe_do_deferred_work(tsdn, arena,
-			    &arena->pa_shard.pac.decay_muzzy,
-			    result.muzzy_npages_new);
-		}
-	}
-}
-
-static bool
-arena_should_decay_early(tsdn_t *tsdn, arena_t *arena, decay_t *decay,
-    background_thread_info_t *info, nstime_t *remaining_sleep,
-    size_t npages_new) {
-	malloc_mutex_assert_owner(tsdn, &info->mtx);
-
-	if (malloc_mutex_trylock(tsdn, &decay->mtx)) {
-		return false;
-	}
-
-	if (!decay_gradually(decay)) {
-		malloc_mutex_unlock(tsdn, &decay->mtx);
-		return false;
-	}
-
-	nstime_init(remaining_sleep, background_thread_wakeup_time_get(info));
-	if (nstime_compare(remaining_sleep, &decay->epoch) <= 0) {
-		malloc_mutex_unlock(tsdn, &decay->mtx);
-		return false;
-	}
-	nstime_subtract(remaining_sleep, &decay->epoch);
-	if (npages_new > 0) {
-		uint64_t npurge_new = decay_npages_purge_in(
-		    decay, remaining_sleep, npages_new);
-		info->npages_to_purge_new += npurge_new;
-	}
-	malloc_mutex_unlock(tsdn, &decay->mtx);
-	return info->npages_to_purge_new
-	    > PAC_DECAY_PURGE_NPAGES_THRESHOLD;
-}
-
-/*
- * Check if deferred work needs to be done sooner than planned.
- * For decay we might want to wake up earlier because of an influx of dirty
- * pages. Rather than waiting for previously estimated time, we proactively
- * purge those pages.
- * If background thread sleeps indefinitely, always wake up because some
- * deferred work has been generated.
- */
-static void
-arena_maybe_do_deferred_work(
-    tsdn_t *tsdn, arena_t *arena, decay_t *decay, size_t npages_new) {
-	background_thread_info_t *info = arena_background_thread_info_get(
-	    arena);
-	if (malloc_mutex_trylock(tsdn, &info->mtx)) {
-		/*
-		 * Background thread may hold the mutex for a long period of
-		 * time.  We'd like to avoid the variance on application
-		 * threads.  So keep this non-blocking, and leave the work to a
-		 * future epoch.
-		 */
-		return;
-	}
-	if (!background_thread_is_started(info)) {
-		goto label_done;
-	}
-
-	nstime_t remaining_sleep;
-	if (background_thread_indefinite_sleep(info)) {
-		background_thread_wakeup_early(info, NULL);
-	} else if (arena_should_decay_early(tsdn, arena, decay, info,
-	               &remaining_sleep, npages_new)) {
-		info->npages_to_purge_new = 0;
-		background_thread_wakeup_early(info, &remaining_sleep);
-	}
-label_done:
-	malloc_mutex_unlock(tsdn, &info->mtx);
 }
 
 /* Called from background threads. */
 void
 arena_do_deferred_work(tsdn_t *tsdn, arena_t *arena) {
 	/*
-	 * The background thread forces decay (PAC_PURGE_ALWAYS) and drives both
+	 * Runs on the background thread.  is_background_thread=true makes pa
-	 * PAC and HPA deferred work through the symmetric pa_shard facade.  No
+	 * force the PAC purge (eagerness ALWAYS) and skip the early-wake --
-	 * PAC result is needed here to notify background thread for an early
+	 * the background thread must not notify itself.
-	 * wake because this function should be called in background thread.
 	 */
-	pa_shard_do_deferred_work(tsdn, &arena->pa_shard, PAC_PURGE_ALWAYS);
+	pa_shard_do_deferred_work(
+	    tsdn, &arena->pa_shard, /* is_background_thread */ true);
 }
 
 static void

src/pa.c

@@ -1,6 +1,7 @@
 #include "jemalloc/internal/jemalloc_preamble.h"
 
 #include "jemalloc/internal/background_thread.h"
+#include "jemalloc/internal/background_thread_inlines.h"
 #include "jemalloc/internal/deferral.h"
 #include "jemalloc/internal/hpa.h"
 #include "jemalloc/internal/pa.h"
@@ -90,15 +91,21 @@ pa_shard_disable_hpa(tsdn_t *tsdn, pa_shard_t *shard) {
 void
 pa_shard_reset(tsdn_t *tsdn, pa_shard_t *shard) {
 	atomic_store_zu(&shard->nactive, 0, ATOMIC_RELAXED);
-	pa_shard_flush(tsdn, shard);
+	pa_shard_flush(tsdn, shard, /* all */ false);
 }
 
 void
-pa_shard_flush(tsdn_t *tsdn, pa_shard_t *shard) {
+pa_shard_flush(tsdn_t *tsdn, pa_shard_t *shard, bool all) {
 	pac_sec_flush(tsdn, &shard->pac);
 	if (shard->ever_used_hpa) {
 		hpa_shard_flush(tsdn, &shard->hpa);
 	}
+	if (all) {
+		pac_decay_all_now(tsdn, &shard->pac, extent_state_dirty);
+		if (pac_should_decay_muzzy(&shard->pac)) {
+			pac_decay_all_now(tsdn, &shard->pac, extent_state_muzzy);
+		}
+	}
 }
 
 static bool
@@ -232,8 +239,8 @@ pa_dalloc(tsdn_t *tsdn, pa_shard_t *shard, edata_t *edata,
 
 bool
 pa_decay_ms_set(tsdn_t *tsdn, pa_shard_t *shard, extent_state_t state,
-    ssize_t decay_ms, pac_purge_eagerness_t eagerness) {
+    ssize_t decay_ms) {
-	return pac_decay_ms_set(tsdn, &shard->pac, state, decay_ms, eagerness);
+	return pac_decay_ms_set(tsdn, &shard->pac, state, decay_ms);
 }
 
 ssize_t
@@ -250,17 +257,29 @@ pa_shard_set_deferral_allowed(
 	}
 }
 
+void
+pa_shard_handle_deferred_work(tsdn_t *tsdn, pa_shard_t *shard) {
+	witness_assert_depth_to_rank(
+	    tsdn_witness_tsdp_get(tsdn), WITNESS_RANK_CORE, 0);
+
+	if (pac_decay_immediately(&shard->pac)) {
+		pac_decay_all_now(tsdn, &shard->pac, extent_state_dirty);
+	}
+	if (background_thread_enabled()) {
+		pac_wake_bg_on_deferred(tsdn, &shard->pac);
+	}
+}
+
 void
 pa_shard_do_deferred_work(
-    tsdn_t *tsdn, pa_shard_t *shard, pac_purge_eagerness_t eagerness) {
+    tsdn_t *tsdn, pa_shard_t *shard, bool is_background_thread) {
+	pac_do_deferred_work(tsdn, &shard->pac, is_background_thread);
 	/*
-	 * The PAC result is only consumed on the application notification path,
+	 * Application threads self-throttle HPA deferred work inline from their
-	 * which calls pac_do_deferred_work directly.  This facade just drives the
+	 * own alloc/dalloc path (hpa_shard_maybe_do_deferred_work, capped), so
-	 * work, so the result stays local.
+	 * only drive it (forced, uncapped) from here on the background thread.
 	 */
-	pac_deferred_work_result_t result;
+	if (is_background_thread && pa_shard_uses_hpa(shard)) {
-	pac_do_deferred_work(tsdn, &shard->pac, eagerness, &result);
-	if (pa_shard_uses_hpa(shard)) {
 		hpa_shard_do_deferred_work(tsdn, &shard->hpa);
 	}
 }

src/pac.c

@@ -2,6 +2,7 @@
 
 #include "jemalloc/internal/arena.h"
 #include "jemalloc/internal/background_thread.h"
+#include "jemalloc/internal/background_thread_inlines.h"
 #include "jemalloc/internal/deferral.h"
 #include "jemalloc/internal/extent.h"
 #include "jemalloc/internal/pac.h"
@@ -807,8 +808,14 @@ pac_decay_deferred_one(tsdn_t *tsdn, pac_t *pac, decay_t *decay,
 	return epoch_advanced;
 }
 
-void
+/*
-pac_do_deferred_work(tsdn_t *tsdn, pac_t *pac,
+ * Non-forced deferred decay-purge for both the dirty and muzzy states, at the
+ * given eagerness; corresponding decay->mtx are acquired internally.  Reports
+ * per-state epoch-advance in *result.  JET_EXTERN: exposed for deterministic
+ * unit testing only; production callers use pac_do_deferred_work.
+ */
+JET_EXTERN void
+pac_decay_deferred(tsdn_t *tsdn, pac_t *pac,
     pac_purge_eagerness_t eagerness, pac_deferred_work_result_t *result) {
 	memset(result, 0, sizeof(*result));
 
@@ -829,6 +836,151 @@ pac_do_deferred_work(tsdn_t *tsdn, pac_t *pac,
 	    eagerness, &contended, &result->muzzy_npages_new);
 }
 
+/*
+ * This is the single, deliberate place where PAC reaches jointly into both a
+ * decay_t and a background_thread_info_t.  It is intentionally NOT decoupled
+ * further, for two reasons:
+ *   (a) The early-wake decision is intrinsically a JOINT info+decay
+ *       computation: holding info->mtx (outer) we trylock decay->mtx (inner)
+ *       to read remaining_sleep = wakeup_time - decay->epoch and the
+ *       decay_npages_purge_in() estimate atomically against epoch advance.
+ *       Splitting it would require copying these values across a new API while
+ *       still holding both locks, adding interface surface for no behavioral
+ *       gain.
+ *   (b) It runs on the application FREE path (pac_do_deferred_work ->
+ *       pac_maybe_wake_bg), where perf parity matters most.
+ *
+ * The info->mtx-outer / decay->mtx-inner trylock nesting is load-bearing for
+ * lock ordering; keep it.
+ *
+ * Parity trap: when npages_new == 0 (e.g. an epoch advanced with no new
+ * backlog) the accumulation below is skipped, but the THRESHOLD compare on the
+ * EXISTING info->npages_to_purge_new backlog still runs and can still trigger
+ * an early wakeup.  Do not "simplify" by early-returning when npages_new == 0.
+ */
+static bool
+pac_decay_should_wake_early(tsdn_t *tsdn, decay_t *decay,
+    background_thread_info_t *info, nstime_t *remaining_sleep,
+    size_t npages_new) {
+	malloc_mutex_assert_owner(tsdn, &info->mtx);
+
+	if (malloc_mutex_trylock(tsdn, &decay->mtx)) {
+		return false;
+	}
+	if (!decay_gradually(decay)) {
+		malloc_mutex_unlock(tsdn, &decay->mtx);
+		return false;
+	}
+	nstime_init(remaining_sleep, background_thread_wakeup_time_get(info));
+	if (nstime_compare(remaining_sleep, &decay->epoch) <= 0) {
+		malloc_mutex_unlock(tsdn, &decay->mtx);
+		return false;
+	}
+	nstime_subtract(remaining_sleep, &decay->epoch);
+	if (npages_new > 0) {
+		uint64_t npurge_new = decay_npages_purge_in(
+		    decay, remaining_sleep, npages_new);
+		info->npages_to_purge_new += npurge_new;
+	}
+	malloc_mutex_unlock(tsdn, &decay->mtx);
+	return info->npages_to_purge_new
+	    > PAC_DECAY_PURGE_NPAGES_THRESHOLD;
+}
+
+/*
+ * The PAC's base index is, under the current PA/PAC construction contract, the
+ * owning arena index (pa_shard_init asserts base_ind_get(base) == ind).  This
+ * is a current construction invariant, not a permanent PAC guarantee.
+ */
+static unsigned
+pac_ind_get(const pac_t *pac) {
+	return base_ind_get(pac->base);
+}
+
+/*
+ * Notify the background thread that a decay epoch advanced: if it sleeps
+ * indefinitely, wake it now; otherwise wake it early when the projected backlog
+ * crosses the purge threshold before its next scheduled wakeup.  Non-blocking
+ * (trylocks info->mtx) to keep the application free path cheap.
+ */
+static void
+pac_maybe_wake_bg(tsdn_t *tsdn, pac_t *pac, decay_t *decay, size_t npages_new) {
+	background_thread_info_t *info =
+	    background_thread_info_get(pac_ind_get(pac));
+	if (malloc_mutex_trylock(tsdn, &info->mtx)) {
+		/*
+		 * The background thread may hold the mutex for a while; keep this
+		 * non-blocking and leave the work to a future epoch.
+		 */
+		return;
+	}
+	if (!background_thread_is_started(info)) {
+		goto label_done;
+	}
+	nstime_t remaining_sleep;
+	if (background_thread_indefinite_sleep(info)) {
+		background_thread_wakeup_early(info, NULL);
+	} else if (pac_decay_should_wake_early(tsdn, decay, info,
+	               &remaining_sleep, npages_new)) {
+		info->npages_to_purge_new = 0;
+		background_thread_wakeup_early(info, &remaining_sleep);
+	}
+label_done:
+	malloc_mutex_unlock(tsdn, &info->mtx);
+}
+
+/*
+ * A hook prepared for calling in pa: after a deferred work generated, wake the
+ * background thread only when it looks idle (lock-free acquire read of
+ * indefinite_sleep), then run the full early-wake decision under info->mtx.
+ */
+void
+pac_wake_bg_on_deferred(tsdn_t *tsdn, pac_t *pac) {
+	background_thread_info_t *info =
+	    background_thread_info_get(pac_ind_get(pac));
+	if (background_thread_indefinite_sleep(info)) {
+		pac_maybe_wake_bg(
+		    tsdn, pac, &pac->decay_dirty, /* npages_new */ 0);
+	}
+}
+
+static pac_purge_eagerness_t pac_decide_purge_eagerness(
+    bool is_background_thread);
+
+void
+pac_do_deferred_work(tsdn_t *tsdn, pac_t *pac, bool is_background_thread) {
+	/*
+	 * A concurrent background_thread enable/disable (mallctl) can race this
+	 * path: the enable state is read lock-free twice below (for the eagerness
+	 * decision, then the notify guard), so the two reads may disagree.  Worst
+	 * case is benign and self-healing:
+	 *   disabled->enabled: purged immediately, plus a possibly-redundant wake;
+	 *   enabled->disabled: deferred but not notified this pass -- the pages
+	 *     stay in the decay backlog and are reclaimed on the next decay tick or
+	 *     by the bg thread before it stops.
+	 * It stays safe regardless: info is allocated once and never freed; the
+	 * wake is gated by info->mtx + background_thread_is_started(); the bg-thread
+	 * locks here are trylocks; and the purge runs under decay->mtx, which the
+	 * toggle never touches.
+	 */
+	pac_purge_eagerness_t eagerness =
+	    pac_decide_purge_eagerness(is_background_thread);
+	pac_deferred_work_result_t result;
+	pac_decay_deferred(tsdn, pac, eagerness, &result);
+
+	if (have_background_thread && background_thread_enabled()
+	    && !is_background_thread) {
+		if (result.dirty_epoch_advanced) {
+			pac_maybe_wake_bg(tsdn, pac, &pac->decay_dirty,
+			    result.dirty_npages_new);
+		}
+		if (result.muzzy_epoch_advanced) {
+			pac_maybe_wake_bg(tsdn, pac, &pac->decay_muzzy,
+			    result.muzzy_npages_new);
+		}
+	}
+}
+
 void
 pac_decay_all_now(tsdn_t *tsdn, pac_t *pac, extent_state_t state) {
 	decay_t           *decay;
@@ -842,9 +994,25 @@ pac_decay_all_now(tsdn_t *tsdn, pac_t *pac, extent_state_t state) {
 	malloc_mutex_unlock(tsdn, &decay->mtx);
 }
 
+/*
+ * Decide the unforced decay-purge eagerness.  On the background thread, force
+ * the purge; on an application thread, defer to the background thread when it is
+ * enabled, otherwise purge on the current thread on epoch advance.
+ */
+static pac_purge_eagerness_t
+pac_decide_purge_eagerness(bool is_background_thread) {
+	if (is_background_thread) {
+		return PAC_PURGE_ALWAYS;
+	} else if (background_thread_enabled()) {
+		return PAC_PURGE_NEVER;
+	} else {
+		return PAC_PURGE_ON_EPOCH_ADVANCE;
+	}
+}
+
 bool
 pac_decay_ms_set(tsdn_t *tsdn, pac_t *pac, extent_state_t state,
-    ssize_t decay_ms, pac_purge_eagerness_t eagerness) {
+    ssize_t decay_ms) {
 	decay_t           *decay;
 	pac_decay_stats_t *decay_stats;
 	ecache_t          *ecache;
@@ -873,6 +1041,12 @@ pac_decay_ms_set(tsdn_t *tsdn, pac_t *pac, extent_state_t state,
 	nstime_t cur_time;
 	nstime_init_update(&cur_time);
 	decay_reinit(decay, &cur_time, decay_ms);
+	/*
+	 * decay_ms is only ever set from a non-background thread (mallctl or
+	 * arena init), so decide the eagerness here rather than threading it in.
+	 */
+	pac_purge_eagerness_t eagerness =
+	    pac_decide_purge_eagerness(/* is_background_thread */ false);
 	pac_maybe_decay_purge(tsdn, pac, decay, decay_stats, ecache, eagerness);
 	malloc_mutex_unlock(tsdn, &decay->mtx);